feat: add dump_hierarchy and ToastWatcher

Author: codematrixer

.flake8

@@ -5,4 +5,5 @@ ignore =
     F401
     E402
     W292
-    F403
+    F403
+    F821

.github/workflows/release.yml

@@ -1,8 +1,12 @@
 name: Release
 
+# on:
+#   release:
+#     types: [created]
 on:
-  release:
-    types: [created]
+  push:
+    tags:
+      - '*.*.*'
 
 jobs:
   build-n-publish:

README.md

@@ -63,7 +63,7 @@ d.screenshot(lpath)
 
 # Device touch
 d.click(500, 1000)
-d.click(0.5, 0.4)
+d.click(0.5, 0.4)  # "If of type float, it represents percentage coordinates."
 d.double_click(500, 1000)
 d.double_click(0.5, 0.4)
 d.long_click(500, 1000)
@@ -100,21 +100,28 @@ d(text="showToast").info
 # }
 
 d(id="swiper").exists()
-d(type="Button", text="showToast").exists()
-d(text="showToast", isAfter=True).exists()
-d(text="showToast").click_if_exists()
+d(type="Button", text="tab_recrod").exists()
+d(text="tab_recrod", isAfter=True).exists()
+d(text="tab_recrod").click_if_exists()
 d(type="Button", index=3).click()
-d(text="showToast").double_click()
-d(text="showToast").long_click()
+d(text="tab_recrod").double_click()
+d(text="tab_recrod").long_click()
 
 component: ComponentData = d(type="ListItem", index=1).find_component()
 d(type="ListItem").drag_to(component)
 
-d(text="showToast").input_text("abc")
-d(text="showToast").clear_text()
-d(text="showToast").pinch_in()
-d(text="showToast").pinch_out()
+d(text="tab_recrod").input_text("abc")
+d(text="tab_recrod").clear_text()
+d(text="tab_recrod").pinch_in()
+d(text="tab_recrod").pinch_out()
 
+# Dump hierarchy
+d.dump_hierarchy()
+
+# Toast Watcher
+d.toast_watcher.start()
+d(type="Button", text="tab_recrod").click()  # 触发toast的操作
+toast = d.toast_watcher.get()
 
 ```
 

docs/DEVELOP.md

@@ -168,6 +168,27 @@
 {"result":"UiWindow#10"}
 ```
 
+### uiEventObserverOnce
+**send**
+```
+{"module":"com.ohos.devicetest.hypiumApiHelper","method":"callHypiumApi","params":{"api":"Driver.uiEventObserverOnce","this":"Driver#0","args":["toastShow"],"message_type":"hypium"},"request_id":"20240905144543056211","client":"127.0.0.1"}
+```
+**recv**
+```
+{"result":true}
+```
+
+
+### getRecentUiEvent
+**send**
+```
+{"module":"com.ohos.devicetest.hypiumApiHelper","method":"callHypiumApi","params":{"api":"Driver.getRecentUiEvent","this":"Driver#0","args":[3000],"message_type":"hypium"},"request_id":"20240905143857794307","client":"127.0.0.1"}
+```
+**recv**
+```
+{"result":{"bundleName":"com.samples.test.uitest","text":"testMessage","type":"Toast"}}
+```
+
 
 ## Component
 ### Component.getId

hmdriver2/_client.py

@@ -64,15 +64,15 @@ def _send_msg(self, msg: typing.Dict):
             }
         """
         msg = json.dumps(msg, ensure_ascii=False, separators=(',', ':'))
-        logger.debug(f"send msg: {msg}")
+        logger.debug(f"sendMsg: {msg}")
         self.sock.sendall(msg.encode('utf-8') + b'\n')
 
     def _recv_msg(self, buff_size: int = 1024, decode=False) -> typing.Union[bytearray, str]:
         try:
             relay = self.sock.recv(buff_size)
             if decode:
                 relay = relay.decode()
-            logger.debug(f"recv msg: {relay}")
+            logger.debug(f"recvMsg: {relay}")
             return relay
         except (socket.timeout, UnicodeDecodeError) as e:
             logger.warning(e)
@@ -100,26 +100,13 @@ def invoke(self, api: str, this: str = "Driver#0", args: typing.List = []) -> Hy
         return HypiumResponse(**(json.loads(result)))
 
     def start(self):
-        """
-
-        Args:
-            file_path (str): Path where the recorded video will be saved.
-
-        Raises:
-            RuntimeError: If the screen capture fails to start.
-        """
         logger.info("Start client connection")
         self._init_so_resource()
         self._restart_uitest_service()
 
         self._connect_sock()
 
     def release(self):
-        """
-
-        Raises:
-            RuntimeError: If the screen capture fails to start.
-        """
         logger.info("Release client connection")
         try:
             if self.sock:
@@ -176,6 +163,7 @@ def __get_local_md5sum(f: str) -> str:
     def _restart_uitest_service(self):
         """
         Restart the UITest daemon.
+
         Note: 'hdc shell aa test' will also start a uitest daemon.
         $ hdc shell ps -ef |grep uitest
         shell        44306     1 25 11:03:37 ?    00:00:16 uitest start-daemon singleness

hmdriver2/_toast.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+
+from .proto import HypiumResponse
+
+
+class ToastWatcher:
+    def __init__(self, session: "Driver"):  # type: ignore
+        self.session = session
+
+    def start(self) -> bool:
+        """
+        Initiates the observer to listen for a UI toast event
+
+        Returns:
+            bool: True if the observer starts successfully, else False.
+        """
+        api = "Driver.uiEventObserverOnce"
+        resp: HypiumResponse = self.session._invoke(api, args=["toastShow"])
+        return resp.result
+
+    def get(self, timeout: int = 5) -> str:
+        """
+        Read the latest toast message content from the recent period.
+
+        Args:
+            timeout (int): The maximum time to wait for a toast to appear if there are no matching toasts within the given time.
+
+        Returns:
+            str: The content of the latest toast message.
+        """
+        api = "Driver.getRecentUiEvent"
+        resp: HypiumResponse = self.session._invoke(api, args=[timeout])
+        if resp.result:
+            return resp.result.get("text")
+        return None

hmdriver2/driver.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 
-import atexit
 import time
 import uuid
 from typing import Type, Any, Tuple, Dict, Union, List
@@ -11,11 +10,11 @@
 except ImportError:
     from cached_property import cached_property
 
-from . import logger
 from .exception import DeviceNotFoundError
 from ._client import HMClient
 from ._uiobject import UiObject
 from .hdc import list_devices
+from ._toast import ToastWatcher
 from .proto import HypiumResponse, KeyCode, Point, DisplayRotation, DeviceInfo
 
 
@@ -24,25 +23,30 @@ class Driver:
 
     def __init__(self, serial: str):
         self.serial = serial
-        if not self._check_serial():
+        if not self._is_device_online():
             raise DeviceNotFoundError(f"Device [{self.serial}] not found")
 
         self._client = HMClient(self.serial)
         self._this_driver = self._client._hdriver.value  # "Driver#0"
         self.hdc = self._client.hdc
 
     def __new__(cls: Type[Any], serial: str) -> Any:
+        """
+        Ensure that only one instance of Driver exists per device serial number.
+        """
         if serial not in cls._instance:
             cls._instance[serial] = super().__new__(cls)
         return cls._instance[serial]
 
     def __call__(self, **kwargs) -> UiObject:
+
         return UiObject(self._client, **kwargs)
 
     def __del__(self):
-        self._client.release()
+        if hasattr(self, '_client') and self._client:
+            self._client.release()
 
-    def _check_serial(self):
+    def _is_device_online(self):
         _serials = list_devices()
         return True if self.serial in _serials else False
 
@@ -59,6 +63,9 @@ def stop_app(self, package_name: str):
         self.hdc.stop_app(package_name)
 
     def clear_app(self, package_name: str):
+        """
+        Clear the application's cache and data.
+        """
         self.hdc.shell(f"bm clean -n {package_name} -c")  # clear cache
         self.hdc.shell(f"bm clean -n {package_name} -d")  # clear data
 
@@ -96,21 +103,30 @@ def unlock(self):
         self.hdc.swipe(0.5 * w, 0.8 * h, 0.5 * w, 0.2 * h)
         time.sleep(.5)
 
+    def _invoke(self, api: str, args: List = []) -> HypiumResponse:
+        return self._client.invoke(api, this=self._this_driver, args=args)
+
     @cached_property
     def display_size(self) -> Tuple[int, int]:
         api = "Driver.getDisplaySize"
-        resp: HypiumResponse = self._client.invoke(api, self._this_driver)
+        resp: HypiumResponse = self._invoke(api)
         w, h = resp.result.get("x"), resp.result.get("y")
         return w, h
 
     @cached_property
     def display_rotation(self) -> DisplayRotation:
         api = "Driver.getDisplayRotation"
-        value = self._client.invoke(api, self._this_driver).result
+        value = self._invoke(api).result
         return DisplayRotation.from_value(value)
 
     @cached_property
     def device_info(self) -> DeviceInfo:
+        """
+        Get detailed information about the device.
+
+        Returns:
+            DeviceInfo: An object containing various properties of the device.
+        """
         hdc = self.hdc
         return DeviceInfo(
             productName=hdc.product_name(),
@@ -123,16 +139,43 @@ def device_info(self) -> DeviceInfo:
             displayRotation=self.display_rotation
         )
 
+    @cached_property
+    def toast_watcher(self):
+        return ToastWatcher(self)
+
     def open_url(self, url: str):
         self.hdc.shell(f"aa start -U {url}")
 
     def pull_file(self, rpath: str, lpath: str):
+        """
+        Pull a file from the device to the local machine.
+
+        Args:
+            rpath (str): The remote path of the file on the device.
+            lpath (str): The local path where the file should be saved.
+        """
         self.hdc.recv_file(rpath, lpath)
 
     def push_file(self, lpath: str, rpath: str):
+        """
+        Push a file from the local machine to the device.
+
+        Args:
+            lpath (str): The local path of the file.
+            rpath (str): The remote path where the file should be saved on the device.
+        """
         self.hdc.send_file(lpath, rpath)
 
     def screenshot(self, path: str) -> str:
+        """
+        Take a screenshot of the device display.
+
+        Args:
+            path (str): The local path to save the screenshot.
+
+        Returns:
+            str: The path where the screenshot is saved.
+        """
         _uuid = uuid.uuid4().hex
         _tmp_path = f"/data/local/tmp/_tmp_{_uuid}.jpeg"
         self.shell(f"snapshot_display -f {_tmp_path}")
@@ -145,7 +188,14 @@ def shell(self, cmd):
 
     def _to_abs_pos(self, x: Union[int, float], y: Union[int, float]) -> Point:
         """
-        returns a function which can convert percent size to abs size
+        Convert percentages to absolute screen coordinates.
+
+        Args:
+            x (Union[int, float]): X coordinate as a percentage or absolute value.
+            y (Union[int, float]): Y coordinate as a percentage or absolute value.
+
+        Returns:
+            Point: A Point object with absolute screen coordinates.
         """
         assert x >= 0
         assert y >= 0
@@ -163,21 +213,28 @@ def click(self, x: Union[int, float], y: Union[int, float]):
         # self.hdc.tap(point.x, point.y)
         point = self._to_abs_pos(x, y)
         api = "Driver.click"
-        self._client.invoke(api, self._this_driver, args=[point.x, point.y])
+        self._invoke(api, args=[point.x, point.y])
 
     def double_click(self, x: Union[int, float], y: Union[int, float]):
         point = self._to_abs_pos(x, y)
         api = "Driver.doubleClick"
-        self._client.invoke(api, self._this_driver, args=[point.x, point.y])
+        self._invoke(api, args=[point.x, point.y])
 
     def long_click(self, x: Union[int, float], y: Union[int, float]):
         point = self._to_abs_pos(x, y)
         api = "Driver.longClick"
-        self._client.invoke(api, self._this_driver, args=[point.x, point.y])
+        self._invoke(api, args=[point.x, point.y])
 
     def swipe(self, x1, y1, x2, y2, speed=1000):
         """
-        speed为滑动速率, 范围:200-40000, 不在范围内设为默认值为600, 单位: 像素点/秒
+        Perform a swipe action on the device screen.
+
+        Args:
+            x1 (float): The start X coordinate as a percentage or absolute value.
+            y1 (float): The start Y coordinate as a percentage or absolute value.
+            x2 (float): The end X coordinate as a percentage or absolute value.
+            y2 (float): The end Y coordinate as a percentage or absolute value.
+            speed (int, optional): The swipe speed in pixels per second. Default is 1000. Range: 200-40000. If not within the range, set to default value of 600.
         """
         point1 = self._to_abs_pos(x1, y1)
         point2 = self._to_abs_pos(x2, y2)
@@ -187,3 +244,12 @@ def swipe(self, x1, y1, x2, y2, speed=1000):
     def input_text(self, x, y, text: str):
         point = self._to_abs_pos(x, y)
         self.hdc.input_text(point.x, point.y, text)
+
+    def dump_hierarchy(self) -> Dict:
+        """
+        Dump the UI hierarchy of the device screen.
+
+        Returns:
+            Dict: The dumped UI hierarchy as a dictionary.
+        """
+        return self.hdc.dump_hierarchy()