add cmd runwda

runtest.sh

@@ -0,0 +1,5 @@
+#!/bin/bash -ex
+#
+
+poetry run isort . -m HANGING_INDENT -l 120 --check-only
+poetry run pytest -v

tidevice3/__main__.py

@@ -18,7 +18,7 @@
 def main():
     try:
         cli(auto_envvar_prefix='T3')
-    except FatalError as e:
+    except (FatalError, ValueError) as e:
         click.echo(f"Error: {e}")
         sys.exit(1)
     except NoDeviceConnectedError:

tidevice3/cli/cli_common.py

@@ -56,6 +56,6 @@ def new_func(ctx, *args, **kwargs):
     return update_wrapper(new_func, func)
 
 
-CLI_GROUPS = ["list", "developer", "screenshot", "screenrecord", "fsync", "app", "reboot", "tunneld", "exec"]
+CLI_GROUPS = ["list", "developer", "screenshot", "screenrecord", "fsync", "app", "reboot", "tunneld", "runwda", "exec"]
 for group in CLI_GROUPS:
     __import__(f"tidevice3.cli.{group}")

tidevice3/cli/runwda.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""Created on Mon Jan 29 2024 14:15:52 by codeskyblue
+"""
+
+
+import logging
+import threading
+import time
+import typing
+
+import click
+from pymobiledevice3.lockdown import LockdownClient
+from pymobiledevice3.services.dvt.testmanaged.xcuitest import XCUITestService
+from pymobiledevice3.services.installation_proxy import InstallationProxyService
+from pymobiledevice3.tcp_forwarder import UsbmuxTcpForwarder
+
+from tidevice3.cli.cli_common import cli, pass_rsd
+
+logger = logging.getLogger(__name__)
+
+
+def guess_wda_bundle_id(service_provider: LockdownClient) -> typing.Optional[str]:
+    app_infos = InstallationProxyService(lockdown=service_provider).get_apps('User')
+    wda_bundle_ids = []
+    for bundle_id in app_infos.keys():
+        if bundle_id.endswith(".xctrunner"):
+            wda_bundle_ids.append(bundle_id)
+    wda_bundle_ids.sort(key=lambda x: x.find('WebDriverAgentRunner'), reverse=True)
+    if not wda_bundle_ids:
+        return None
+    return wda_bundle_ids[0]
+
+
+@cli.command("runwda")
+@click.option('--bundle-id', default=None, help="WebDriverAgent bundle id")
+@click.option("--src-port", default=8100, help="WebDriverAgent listen port")
+@click.option("--dst-port", default=8100, help="local listen port")
+@pass_rsd
+def cli_runwda(service_provider: LockdownClient, bundle_id: str, src_port: int, dst_port: int):
+    """run WebDriverAgent"""
+    if not bundle_id:
+        bundle_id = guess_wda_bundle_id(service_provider)
+        if not bundle_id:
+            raise ValueError("WebDriverAgent not found")
+
+    def tcp_forwarder():
+        logger.info("forwarder started, listen on %s", dst_port)
+        forwarder = UsbmuxTcpForwarder(service_provider.udid, dst_port, src_port)
+        forwarder.start()
+
+    def xcuitest():
+        XCUITestService(service_provider).run(bundle_id)
+
+    thread1 = threading.Thread(target=tcp_forwarder, daemon=True)
+    thread2 = threading.Thread(target=xcuitest, daemon=True)
+    thread1.start()
+    thread2.start()
+
+    while thread1.is_alive() and thread2.is_alive():
+        time.sleep(0.1)
+    logger.info("Program exited")
+

tidevice3/cli/screenrecord.py

@@ -94,7 +94,7 @@ def resize_for_ffmpeg(img: Image.Image) -> Image.Image:
 @pass_rsd
 def cli_screenrecord(service_provider: LockdownClient, out: str, fps: int, show_time: bool):
     """ screenrecord to mp4 """
-    writer = imageio.get_writer(out, fps=fps)#, macro_block_size=1)
+    writer = imageio.get_writer(out, fps=fps)
     frame_index = 0
     try:
         for png_data in limit_fps(iter_screenshot(service_provider), fps, debug=True):
@@ -108,7 +108,6 @@ def cli_screenrecord(service_provider: LockdownClient, out: str, fps: int, show_
             frame_index += 1
     except KeyboardInterrupt:
         print("")
-        pass
     finally:
         writer.close()
         logger.info("screenrecord saved to %s", out)