From e5a0aecfa556c04f8aaf8c8258af2be1b550b8e0 Mon Sep 17 00:00:00 2001 From: vadvolo Date: Sat, 22 Nov 2025 15:13:18 +0000 Subject: [PATCH] feat: Add Telnet support (dev_port and streamer_type) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add dev_port and streamer_type to AppSettings - Add get_streamer_type method to convert string to proto enum - Update GnetcliFetcher and GnetcliDeployer with new parameters - Pass port and streamer_type to all HostParams calls 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/gnetcli_adapter/gnetcli_adapter.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/src/gnetcli_adapter/gnetcli_adapter.py b/src/gnetcli_adapter/gnetcli_adapter.py index 26cea52..50b8af0 100644 --- a/src/gnetcli_adapter/gnetcli_adapter.py +++ b/src/gnetcli_adapter/gnetcli_adapter.py @@ -69,8 +69,16 @@ class AppSettings(BaseSettings): password: Optional[str] = None dev_login: Optional[str] = None dev_password: Optional[str] = None + dev_port: Optional[int] = None + streamer_type: Optional[str] = None # "ssh" or "telnet" ssh_agent_enabled: bool = True + def get_streamer_type(self): + """Convert string streamer_type to proto enum value""" + if self.streamer_type == "telnet": + return pb.StreamerType.StreamerType_telnet + return pb.StreamerType.StreamerType_ssh + def make_dev_credentials(self) -> Optional[Credentials]: if not self.dev_login and not self.dev_password: return None @@ -151,6 +159,8 @@ def __init__( password: Optional[str] = None, dev_login: Optional[str] = None, dev_password: Optional[str] = None, + dev_port: Optional[int] = None, + streamer_type: Optional[str] = None, ssh_agent_enabled: bool = True, server_path: Optional[str] = None, server_conf: Config = DEFAULT_GNETCLI_SERVER_CONF, @@ -160,6 +170,8 @@ def __init__( "password": password, "dev_login": dev_login, "dev_password": dev_password, + "dev_port": dev_port, + "streamer_type": streamer_type, "server_path": server_path, "url": url, "server_conf": server_conf, @@ -247,6 +259,8 @@ async def afetch_dev(self, api: Gnetcli, device: Device) -> str: credentials=self.conf.make_dev_credentials(), device=gnetcli_device, ip=ip, + port=self.conf.dev_port, + streamer_type=self.conf.get_streamer_type(), ), ) if res.status != 0: @@ -264,6 +278,8 @@ async def adownload_dev(self, api: Gnetcli, device: Device, files: List[str]) -> credentials=self.conf.make_dev_credentials(), device=gnetcli_device, ip=ip, + port=self.conf.dev_port, + streamer_type=self.conf.get_streamer_type(), ), ) res: Dict[str, str] = {} @@ -301,6 +317,8 @@ def __init__( password: Optional[str] = None, dev_login: Optional[str] = None, dev_password: Optional[str] = None, + dev_port: Optional[int] = None, + streamer_type: Optional[str] = None, ssh_agent_enabled: bool = True, server_path: Optional[str] = None, server_conf: Optional[Config] = DEFAULT_GNETCLI_SERVER_CONF, @@ -311,6 +329,8 @@ def __init__( "password": password, "dev_login": dev_login, "dev_password": dev_password, + "dev_port": dev_port, + "streamer_type": streamer_type, "server_path": server_path, "url": url, "server_conf": server_conf, @@ -447,6 +467,8 @@ async def deploy( credentials=self.conf.make_dev_credentials(), device=gnetcli_device, ip=ip, + port=self.conf.dev_port, + streamer_type=self.conf.get_streamer_type(), ) command_groups: list[tuple[str, CommandList]]= [] if isinstance(cmds, dict): # PC