diff --git a/tools/otci/otci/__init__.py b/tools/otci/otci/__init__.py index 7cb0aa45b83b..eed56c615ada 100644 --- a/tools/otci/otci/__init__.py +++ b/tools/otci/otci/__init__.py @@ -37,7 +37,8 @@ connect_ncp_sim, \ connect_cmd_handler, \ connect_otbr_ssh, \ - connect_otbr_adb + connect_otbr_adb_tcp, \ + connect_otbr_adb_usb from .types import Rloc16, ChildId, NetifIdentifier @@ -46,7 +47,8 @@ 'connect_cli_serial', 'connect_ncp_sim', 'connect_otbr_ssh', - 'connect_otbr_adb', + 'connect_otbr_adb_tcp', + 'connect_otbr_adb_usb', 'connect_cmd_handler', ] diff --git a/tools/otci/otci/command_handlers.py b/tools/otci/otci/command_handlers.py index 7c790e6d839d..ac24e4a641b0 100644 --- a/tools/otci/otci/command_handlers.py +++ b/tools/otci/otci/command_handlers.py @@ -285,19 +285,13 @@ def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]): class OtbrAdbCommandRunner(OTCommandHandler): - def __init__(self, host, port): - from adb_shell.adb_device import AdbDeviceTcp - - self.__host = host - self.__port = port - self.__adb = AdbDeviceTcp(host, port, default_transport_timeout_s=9.0) + from adb_shell.adb_device import AdbDevice + def __init__(self, adb: AdbDevice): + self.__adb = adb self.__line_read_callback = None self.__adb.connect(rsa_keys=None, auth_timeout_s=0.1) - def __repr__(self): - return f'{self.__host}:{self.__port}' - def execute_command(self, cmd: str, timeout: float) -> List[str]: sh_cmd = f'ot-ctl {cmd}' @@ -324,3 +318,32 @@ def wait(self, duration: float) -> List[str]: def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]): self.__line_read_callback = callback + + +class OtbrAdbTcpCommandRunner(OTCommandHandler): + + def __init__(self, host: str, port: int): + from adb_shell.adb_device import AdbDeviceTcp + + self.__host = host + self.__port = port + + adb = AdbDeviceTcp(host, port, default_transport_timeout_s=9.0) + super(OtbrAdbTcpCommandRunner, self).__init__(adb) + + def __repr__(self): + return f'{self.__host}:{self.__port}' + + +class OtbrAdbUsbCommandRunner(OtbrAdbCommandRunner): + + def __init__(self, serial: str): + from adb_shell.adb_device import AdbDeviceUsb + + self.__serial = serial + + adb = AdbDeviceUsb(serial, port_path=None, default_transport_timeout_s=9.0) + super(OtbrAdbUsbCommandRunner, self).__init__(adb) + + def __repr__(self): + return f'USB:{self.__serial}' diff --git a/tools/otci/otci/otci.py b/tools/otci/otci/otci.py index 46a73734db26..93bb56a67bec 100644 --- a/tools/otci/otci/otci.py +++ b/tools/otci/otci/otci.py @@ -33,7 +33,7 @@ from typing import Callable, List, Collection, Union, Tuple, Optional, Dict, Pattern, Any from . import connectors -from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner, OtbrAdbCommandRunner +from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner, OtbrAdbTcpCommandRunner, OtbrAdbUsbCommandRunner from .connectors import Simulator from .errors import UnexpectedCommandOutput, ExpectLineTimeoutError, CommandError, InvalidArgumentsError from .types import ChildId, Rloc16, Ip6Addr, ThreadState, PartitionId, DeviceMode, RouterId, SecurityPolicy, Ip6Prefix, \ @@ -2503,8 +2503,13 @@ def connect_otbr_ssh(host: str, port: int = 22, username='pi', password='raspber return OTCI(cmd_handler) -def connect_otbr_adb(host: str, port: int = 5555): - cmd_handler = OtbrAdbCommandRunner(host, port) +def connect_otbr_adb_tcp(host: str, port: int = 5555): + cmd_handler = OtbrAdbTcpCommandRunner(host, port) + return OTCI(cmd_handler) + + +def connect_otbr_adb_usb(serial: str): + cmd_handler = OtbrAdbUsbCommandRunner(serial) return OTCI(cmd_handler)