hdc: reformat files & small fixes

benchkit/hdc/__init__.py

@@ -3,22 +3,25 @@
 Module to handle hdc (OpenHarmony Device Connector) interactions between host and remote phone.
 HDC documentation: https://docs.openharmony.cn/pages/v5.0/en/application-dev/dfx/hdc.md 
 """
-from typing import Iterable, Optional, Callable, List
 from enum import Enum
+from platform import system as os_system
+from typing import Callable, Iterable, List, Optional
 
+from benchkit.dependencies.executables import ExecutableDependency
+from benchkit.dependencies.packages import Dependency
 from benchkit.shell.shell import get_args, shell_out
 from benchkit.utils.types import Command, PathType
-from benchkit.dependencies.packages import Dependency
-from benchkit.dependencies.executables import ExecutableDependency
 
 
 class HDCError(Exception):
     """Handle for errors from hdc."""
 
+
 class DeviceIdentifierKind(Enum):
-    IP_AND_PORT = 0,
-    SERIAL = 1,
-    DONT_CARE = 2,
+    IP_AND_PORT = 0
+    SERIAL = 1
+    DONT_CARE = 2
+
 
 class HDCDevice:
     """Representation of a device connected through hdc."""
@@ -33,7 +36,7 @@ def __init__(
 
     def __str__(self) -> str:
         return f"{self.identifier}"
-    
+
 
 class OpenHarmonyDeviceConnector:
     """Operations with the phone for high-level hdc operations."""
@@ -51,71 +54,66 @@ def __init__(
         self._keep_connected = keep_connected
         self._wait_connected = wait_connected
         self._expected_os = expected_os
+        self._bin = "hdc.exe" if "Windows" == os_system() else "hdc"
 
-
-    def __init__(
-        self, 
+    @staticmethod
+    def from_device(
         device: HDCDevice,
         keep_connected: bool = False,
         wait_connected: bool = False,
         expected_os: Optional[str] = None,
-    ) -> None:
-        self.identifier = device.identifier
-        self.kind = device.kind
-        self._keep_connected = keep_connected
-        self._wait_connected = wait_connected
-        self._expected_os = expected_os
-
-
-    @staticmethod
-    def dependencies() -> List[Dependency]:
-        return [
-            ExecutableDependency("hdc.exe")
-        ]
+    ) -> "OpenHarmonyDeviceConnector":
+        return OpenHarmonyDeviceConnector(
+            identifier=device.identifier,
+            kind=device.kind,
+            keep_connected=keep_connected,
+            wait_connected=wait_connected,
+            expected_os=expected_os,
+        )
 
+    def dependencies(self) -> List[Dependency]:
+        return [ExecutableDependency(self._bin)]
 
     def _find_device(self) -> Optional[HDCDevice]:
         devices = [dev for dev in self._devices() if dev.identifier == self.identifier]
         match len(devices):
             case 0:
                 return None
-            case 1: 
+            case 1:
                 return devices[0]
             case _:
                 raise ValueError("Wrong device list.")
-
 
-    @staticmethod
-    def _devices() -> Iterable[str]:
+    def _devices(self) -> List[HDCDevice]:
         """Get list of devices recognized by hdc.
 
         Returns:
             Iterable[HDCDevice]: list of devices recognized by hdc.
         """
-        output = OpenHarmonyDeviceConnector._host_shell_out(command="hdc.exe list targets")
+        output = OpenHarmonyDeviceConnector._host_shell_out(command=f"{self._bin} list targets")
         device_ids = output.strip().splitlines()
         devices = []
         for dev in device_ids:
             if ":" in dev:
                 devices.append(HDCDevice(dev, DeviceIdentifierKind.IP_AND_PORT))
             else:
                 devices.append(HDCDevice(dev, DeviceIdentifierKind.SERIAL))
-        
+
         return devices
-
-
-    @staticmethod
-    def query_devices(filter: Callable[[HDCDevice], bool] = lambda _: True) -> Iterable[HDCDevice]:
+
+    def query_devices(
+        self,
+        filter_callback: Callable[[HDCDevice], bool] = lambda _: True,
+    ) -> Iterable[HDCDevice]:
         """Get filtered list of devices recognized by hdc.
 
         Returns:
             Iterable[HDCDevice]: filtered list of devices recognized by hdc.
         """
-        devices = OpenHarmonyDeviceConnector._devices()
-        filtered = [dev for dev in devices if filter(dev)]
+        devices = self._devices()
+        filtered = [dev for dev in devices if filter_callback(dev)]
         return filtered
 
-
     @staticmethod
     def _host_shell_out(
         command: Command,
@@ -130,8 +128,7 @@ def _host_shell_out(
             print_output=print_output,
         )
         return output
-
-
+
     def _target_shell_out(
         self,
         command: Command,
@@ -141,12 +138,7 @@ def _target_shell_out(
         dir_args = ["cd", f"{current_dir}", "&&"] if current_dir is not None else []
         command_args = dir_args + get_args(command)
 
-        hdc_command = [
-            "hdc.exe",
-            "-t",
-            f"{self.identifier}",
-            "shell"
-        ] + command_args
+        hdc_command = [f"{self._bin}", "-t", f"{self.identifier}", "shell"] + command_args
 
         output = shell_out(
             command=hdc_command,
@@ -156,7 +148,6 @@ def _target_shell_out(
 
         return output
 
-
     def shell_out(
         self,
         command: Command,
@@ -180,7 +171,6 @@ def shell_out(
             current_dir=current_dir,
             output_is_log=output_is_log,
         )
-
 
     def push(
         self,
@@ -201,7 +191,7 @@ def push(
             "file",
             "send",
             f"{local_path}",
-            f"{remote_path}"
+            f"{remote_path}",
         ]
         self._host_shell_out(command=command)
 
@@ -217,13 +207,13 @@ def pull(
             remote_path (PathType): path on the device where the file is.
             local_path (PathType): path where to pull the file on the host.
         """
-        commmand = [
-            "hdc.exe",
+        command = [
+            f"{self._bin}",
             "-t",
             f"{self.identifier}",
             "file",
             "recv",
             f"{remote_path}",
-            f"{local_path}"
+            f"{local_path}",
         ]
-        self._host_shell_out(command=commmand)
+        self._host_shell_out(command=command)

tests/test_hdc_api.py

@@ -1,34 +1,36 @@
-#!/usr/bin/env python3
-# SPDX-License-Identifier: MIT
-"""
-Module to test the HDC api basics
-"""
-
-import os
-from benchkit.hdc import *
-
-HOST_FILE = ".gitignore"
-TMP_FILE = "cloned-gitignore"
-TARGET_FILE = "/storage/media/100/local/testing/.gitignore"
-
-def main() -> None:
-    device = OpenHarmonyDeviceConnector.query_devices(lambda _: True)[0]
-    print(device)
-
-    bridge = OpenHarmonyDeviceConnector(device)
-    output = bridge._target_shell_out("cd data && ls")
-    print(output)
-
-    bridge.push(HOST_FILE, TARGET_FILE)
-    bridge.pull(TARGET_FILE, TMP_FILE)
-
-    f = open(TMP_FILE, 'r')
-    content=f.read()
-    print(content)
-    f.close()
-
-    os.remove(f.name)
-
-
-if __name__ == '__main__':
-    main()
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+"""
+Module to test the HDC api basics
+"""
+
+import os
+
+from benchkit.hdc import *
+
+HOST_FILE = ".gitignore"
+TMP_FILE = "cloned-gitignore"
+TARGET_FILE = "/storage/media/100/local/testing/.gitignore"
+
+
+def main() -> None:
+    device = OpenHarmonyDeviceConnector.query_devices(lambda _: True)[0]
+    print(device)
+
+    bridge = OpenHarmonyDeviceConnector(device)
+    output = bridge._target_shell_out("cd data && ls")
+    print(output)
+
+    bridge.push(HOST_FILE, TARGET_FILE)
+    bridge.pull(TARGET_FILE, TMP_FILE)
+
+    f = open(TMP_FILE, "r")
+    content = f.read()
+    print(content)
+    f.close()
+
+    os.remove(f.name)
+
+
+if __name__ == "__main__":
+    main()