Add hdc (HarmonyOS Device Connector)

src/linktools/android/adb.py

@@ -31,7 +31,7 @@
 import os
 import re
 import time
-from typing import Any, Generator, List, Callable, TYPE_CHECKING, TypeVar, Optional
+from typing import Any, Generator, List, Callable, TYPE_CHECKING, TypeVar
 
 from .struct import App, UnixSocket, InetSocket, Process, File, SystemService
 from .. import utils, environ
@@ -71,9 +71,8 @@ def list_devices(self, alive: bool = None) -> Generator["Device", None, None]:
         :return: 设备号数组
         """
         result = self.exec("devices")
-        lines = result.splitlines()
-        for i in range(1, len(lines)):
-            splits = lines[i].split(maxsplit=1)
+        for line in result.splitlines():
+            splits = line.split(maxsplit=1)
             if len(splits) >= 2:
                 device, status = splits
                 if alive is None:
@@ -658,11 +657,11 @@ def get_uid(self, package_name: str = None, timeout: utils.Timeout = None) -> in
         else:
             default = -1
             out = self.shell("id", "-u", timeout=timeout)
-            uid = utils.int(out, default=default)
+            uid = utils.int(out.strip(), default=default)
             if uid != default:
                 return uid
             out = self.shell("echo", "-n", "${USER_ID}", timeout=timeout)
-            uid = utils.int(out, default=default)
+            uid = utils.int(out.strip(), default=default)
             if uid != default:
                 return uid
             raise AdbError("unknown adb uid: %s" % out)

src/linktools/device.py

@@ -24,10 +24,13 @@ def __init__(self, tool: Tool = None, options: [str] = None, error_type: Type[Br
     def list_devices(self, alive: bool = None) -> Generator["BaseDevice", None, None]:
         from .android import Adb
         from .ios import Sib
+        from .harmony import Hdc
         for device in Adb().list_devices(alive=alive):
             yield device
         for device in Sib().list_devices(alive=alive):
             yield device
+        for device in Hdc().list_devices(alive=alive):
+            yield device
 
     def popen(self, *args: [Any], **kwargs) -> utils.Process:
         if self._tool is None:

src/linktools/harmony/__init__.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+@author  : Hu Ji
+@file    : __init__.py.py
+@time    : 2024/7/15 11:38 
+@site    : https://github.com/ice-black-tea
+@software: PyCharm
+
+              ,----------------,              ,---------,
+         ,-----------------------,          ,"        ,"|
+       ,"                      ,"|        ,"        ,"  |
+      +-----------------------+  |      ,"        ,"    |
+      |  .-----------------.  |  |     +---------+      |
+      |  |                 |  |  |     | -==----'|      |
+      |  | $ sudo rm -rf / |  |  |     |         |      |
+      |  |                 |  |  |/----|`---=    |      |
+      |  |                 |  |  |   ,/|==== ooo |      ;
+      |  |                 |  |  |  // |(((( [33]|    ,"
+      |  `-----------------'  |," .;'| |((((     |  ,"
+      +-----------------------+  ;;  | |         |,"
+         /_)______________(_/  //'   | +---------+
+    ___________________________/___  `,
+   /  oooooooooooooooo  .o.  oooo /,   `,"-----------
+  / ==ooooooooooooooo==.o.  ooo= //   ,``--{)B     ,"
+ /_==__==========__==_ooo__ooo=_/'   /___________,"
+"""
+
+from .hdc import Hdc, Device

src/linktools/harmony/hdc.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+# -*- coding:utf-8 -*-
+
+from typing import Any, Generator, List, Callable, TYPE_CHECKING, TypeVar
+
+from .. import utils
+from .._environ import environ
+from ..decorator import cached_property
+from ..device import BridgeError, Bridge, BaseDevice
+
+if TYPE_CHECKING:
+    DEVICE_TYPE = TypeVar("DEVICE_TYPE", bound="Device")
+
+_logger = environ.get_logger("harmony.hdc")
+
+
+class HdcError(BridgeError):
+    pass
+
+
+class Hdc(Bridge):
+
+    def __init__(self, options: List[str] = None):
+        super().__init__(
+            tool=environ.get_tool("hdc"),
+            options=options,
+            error_type=HdcError
+        )
+
+    def list_devices(self, alive: bool = None) -> Generator["Device", None, None]:
+        """
+        获取所有设备列表
+        :param alive: 只显示在线的设备
+        :return: 设备号数组
+        """
+        result = self.exec("list", "targets", "-v")
+        for line in result.splitlines():
+            splits = line.split()
+            if len(splits) == 4:
+                id, mode, status, address = splits[0], splits[1], splits[2], splits[3]
+                if alive is None:
+                    yield Device(id)
+                elif alive == (status in ("Connected",)):
+                    yield Device(id)
+
+
+class Device(BaseDevice):
+
+    def __init__(self, id: str = None, hdc: Hdc = None):
+        """
+        :param id: 设备号
+        """
+        self._hdc = hdc or Hdc()
+        if id is None:
+            devices = list(self._hdc.list_devices(alive=True))
+            if len(devices) == 0:
+                raise HdcError("no devices/emulators found")
+            elif len(devices) > 1:
+                raise HdcError("more than one device/emulator")
+            self._id = devices[0]._id
+        else:
+            self._id = id
+
+    @property
+    def id(self) -> str:
+        """
+        获取设备号
+        :return: 设备号
+        """
+        return self._id
+
+    @property
+    def name(self) -> str:
+        return self.get_prop("const.product.model", timeout=1)
+
+    @cached_property
+    def abi(self) -> str:
+        """
+        获取设备abi类型
+        :return: abi类型
+        """
+        result = self.get_prop("const.product.cpu.abilist")
+        if result.find("arm64") >= 0:
+            return "arm64"
+        elif result.find("armeabi") >= 0:
+            return "arm"
+        elif result.find("x86_64") >= 0:
+            return "x86_64"
+        elif result.find("x86") >= 0:
+            return "x86"
+        raise HdcError("unknown abi: %s" % result)
+
+    @cached_property
+    def uid(self) -> int:
+        """
+        获取shell的uid
+        :return: uid
+        """
+        return self.get_uid()
+
+    def copy(self, type: "Callable[[str, Hdc], DEVICE_TYPE]" = None) -> "DEVICE_TYPE":
+        return (type or Device)(self._id, self._hdc)
+
+    @utils.timeoutable
+    def exec(self, *args: Any, **kwargs) -> str:
+        """
+        执行命令
+        :param args: 命令行参数
+        :return: hdc输出结果
+        """
+        args = ["-t", self.id, *args]
+        return self._hdc.exec(*args, **kwargs)
+
+    def make_shell_args(self, *args: Any):
+        cmd = utils.list2cmdline([str(arg) for arg in args])
+        return ["shell", cmd]
+
+    @utils.timeoutable
+    def shell(self, *args: Any, **kwargs) -> str:
+        """
+        执行shell
+        :param args: shell命令
+        :return: hdc输出结果
+        """
+        args = self.make_shell_args(*args)
+        return self.exec(*args, **kwargs)
+
+    @utils.timeoutable
+    def get_prop(self, prop: str, **kwargs) -> str:
+        """
+        获取属性值
+        :param prop: 属性名
+        :return: 属性值
+        """
+        return self.shell("param", "get", prop, **kwargs).rstrip()
+
+    @utils.timeoutable
+    def get_uid(self, timeout: utils.Timeout = None) -> int:
+        default = -1
+        out = self.shell("id", "-u", timeout=timeout)
+        uid = utils.int(out.strip(), default=default)
+        if uid != default:
+            return uid
+        out = self.shell("echo", "-n", "${USER_ID}", timeout=timeout)
+        uid = utils.int(out.strip(), default=default)
+        if uid != default:
+            return uid
+        raise HdcError("unknown hdc uid: %s" % out)
+
+    def __repr__(self):
+        return f"HdcDevice<{self.id}>"