upgrade pmd3 version from 2 to 4

pyproject.toml

@@ -9,14 +9,15 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = "^3.8"
-pymobiledevice3 = "^2.46.2"
+pymobiledevice3 = "^4.2.3"
 click = "*"
 pydantic = "^2.5.3"
 fastapi = "*"
 requests = "*"
 numpy = "*"
 imageio = {extras = ["ffmpeg"], version = "^2.33.1"}
 pillow = "^10.0"
+zeroconf = "^0.132.2"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.4"

tests/test_api.py

@@ -11,5 +11,6 @@ def test_api(tmp_path: Path):
     for d in list_devices(usb=True):
         print("UDID:", d.Identifier)
         service_provider = connect_service_provider(d.Identifier)
-        pil_im = screenshot(service_provider)
-        pil_im.save(tmp_path / "screenshot.png")
+        with service_provider:
+            pil_im = screenshot(service_provider)
+            pil_im.save(tmp_path / "screenshot.png")

tidevice3/api.py

@@ -3,24 +3,33 @@
 
 import datetime
 import io
+import logging
 import os
-from typing import Iterator, Optional
+import socket
+from typing import Any, Dict, Iterator, Optional
 
 import requests
+from packaging.version import Version
 from PIL import Image
 from pydantic import BaseModel
+from pymobiledevice3.common import get_home_folder
+from pymobiledevice3.exceptions import AlreadyMountedError
 from pymobiledevice3.lockdown import LockdownClient, create_using_usbmux, usbmux
 from pymobiledevice3.lockdown_service_provider import LockdownServiceProvider
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
+from pymobiledevice3.services.amfi import AmfiService
 from pymobiledevice3.services.dvt.dvt_secure_socket_proxy import DvtSecureSocketProxyService
 from pymobiledevice3.services.dvt.instruments.device_info import DeviceInfo
 from pymobiledevice3.services.dvt.instruments.screenshot import Screenshot
 from pymobiledevice3.services.installation_proxy import InstallationProxyService
+from pymobiledevice3.services.mobile_image_mounter import auto_mount
 from pymobiledevice3.services.screenshot import ScreenshotService
+from pymobiledevice3.utils import get_asyncio_loop
 
 from tidevice3.exceptions import FatalError
 from tidevice3.utils.download import download_file, is_hyperlink
 
+logger = logging.getLogger(__name__)
 
 class DeviceShortInfo(BaseModel):
     BuildVersion: str
@@ -78,15 +87,34 @@ def connect_service_provider(udid: Optional[str], force_usbmux: bool = False, us
     return lockdown
 
 
-def connect_remote_service_discovery_service(udid: str, tunneld_url: str = 'http://localhost:5555') -> RemoteServiceDiscoveryService:
+class EnterableRemoteServiceDiscoveryService(RemoteServiceDiscoveryService):
+    def __enter__(self) -> EnterableRemoteServiceDiscoveryService:
+        get_asyncio_loop().run_until_complete(self.connect())
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        get_asyncio_loop().run_until_complete(self.close())
+
+
+def is_port_open(ip: str, port: int) -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        return s.connect_ex((ip, port)) == 0
+
+
+def connect_remote_service_discovery_service(udid: str, tunneld_url: str = None) -> EnterableRemoteServiceDiscoveryService:
+    if tunneld_url is None:
+        if is_port_open("localhost", 49151):
+            tunneld_url = "http://localhost:49151"
+        else:
+            tunneld_url = "http://localhost:5555" # for backward compatibility
+
     try:
         resp = requests.get(tunneld_url, timeout=DEFAULT_TIMEOUT)
-        tunnels = resp.json()
+        tunnels: Dict[str, Any] = resp.json()
         ipv6_address = tunnels.get(udid)
         if ipv6_address is None:
             raise FatalError("tunneld not ready for device", udid)
-        rsd = RemoteServiceDiscoveryService(ipv6_address)
-        rsd.connect()
+        rsd = EnterableRemoteServiceDiscoveryService(ipv6_address)
         return rsd
     except requests.RequestException:
         raise FatalError("Please run `sudo t3 tunneld` first")
@@ -136,4 +164,23 @@ def app_install(service_provider: LockdownClient, path_or_url: str):
         ipa_path = path_or_url
     else:
         raise ValueError("local file not found", path_or_url)
-    InstallationProxyService(lockdown=service_provider).install_from_local(ipa_path)
+    InstallationProxyService(lockdown=service_provider).install_from_local(ipa_path)
+
+
+def enable_developer_mode(service_provider: LockdownClient):
+    """ enable developer mode """
+    if Version(service_provider.product_version) >= Version("16"):
+        if not service_provider.developer_mode_status:
+            logger.info('enable developer mode')
+            AmfiService(service_provider).enable_developer_mode()
+        else:
+            logger.info('developer mode already enabled')
+
+    try:
+        xcode = get_home_folder() / 'Xcode.app'
+        xcode.mkdir(parents=True, exist_ok=True)
+        auto_mount(service_provider, xcode=xcode)
+        logger.info('mount developer image')
+    except AlreadyMountedError:
+        logger.info('developer image already mounted')
+

tidevice3/cli/app.py

@@ -40,7 +40,7 @@ def cli_app_install(service_provider: LockdownClient, path_or_url: str):
 
 
 @app.command("list")
-@click.option('app_type', '-t', '--type', type=click.Choice(['System', 'User', 'Hidden', 'Any']), default='Any',
+@click.option('app_type', '-t', '--type', type=click.Choice(['System', 'User', 'Hidden', 'Any']), default='User',
               help='include only applications of given type')
 @click.option("--calculate-sizes/--no-calculate-size", default=False)
 @pass_service_provider

tidevice3/cli/relay.py

@@ -6,10 +6,11 @@
 Ref: https://github.com/doronz88/pymobiledevice3/blob/master/pymobiledevice3/cli/usbmux.py#L32
 """
 
-from functools import partial
 import logging
 import tempfile
 import threading
+from functools import partial
+
 import click
 from pymobiledevice3.lockdown import LockdownClient
 from pymobiledevice3.tcp_forwarder import UsbmuxTcpForwarder

tidevice3/cli/tunneld.py

@@ -19,14 +19,14 @@
 from fastapi import FastAPI
 from packaging.version import Version
 from pymobiledevice3.exceptions import MuxException
-from pymobiledevice3.cli.cli_common import is_admin_user
+from pymobiledevice3.osu.os_utils import OsUtils
 
 from tidevice3.cli.cli_common import cli
 from tidevice3.cli.list import list_devices
 from tidevice3.utils.common import threadsafe_function
 
 logger = logging.getLogger(__name__)
-
+os_utils = OsUtils.create()
 
 class Address(NamedTuple):
     ip: str
@@ -157,7 +157,7 @@ def run_forever(self):
 @click.option("--port", "port", help="listen port", default=5555)
 def tunneld(pmd3_path: str, port: int):
     """start server for iOS >= 17 auto start-tunnel, function like pymobiledevice3 remote tunneld"""
-    if not is_admin_user():
+    if not os_utils.is_admin:
         logger.error("Please run as root(Mac) or administrator(Windows)")
         sys.exit(1)