tunneld: seperate server from api to improve import performance

pymobiledevice3/cli/cli_common.py

@@ -20,7 +20,7 @@
 from pymobiledevice3.lockdown import LockdownClient, create_using_usbmux
 from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
-from pymobiledevice3.tunneld import TUNNELD_DEFAULT_ADDRESS, async_get_tunneld_devices
+from pymobiledevice3.tunneld.api import TUNNELD_DEFAULT_ADDRESS, async_get_tunneld_devices
 from pymobiledevice3.usbmux import select_devices_by_connection_type
 
 COLORED_OUTPUT = True

pymobiledevice3/cli/remote.py

@@ -20,7 +20,8 @@
 from pymobiledevice3.remote.tunnel_service import RemotePairingManualPairingService, get_core_device_tunnel_services, \
     get_remote_pairing_tunnel_services
 from pymobiledevice3.remote.utils import get_rsds
-from pymobiledevice3.tunneld import TUNNELD_DEFAULT_ADDRESS, TunneldRunner
+from pymobiledevice3.tunneld.api import TUNNELD_DEFAULT_ADDRESS
+from pymobiledevice3.tunneld.server import TunneldRunner
 
 logger = logging.getLogger(__name__)
 

pymobiledevice3/tunneld/api.py

@@ -0,0 +1,60 @@
+
+
+from typing import Optional
+
+import requests
+
+from pymobiledevice3.exceptions import TunneldConnectionError
+from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
+from pymobiledevice3.utils import get_asyncio_loop
+
+TUNNELD_DEFAULT_ADDRESS = ('127.0.0.1', 49151)
+
+
+async def async_get_tunneld_devices(tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
+        -> list[RemoteServiceDiscoveryService]:
+    tunnels = _list_tunnels(tunneld_address)
+    return await _create_rsds_from_tunnels(tunnels)
+
+
+def get_tunneld_devices(tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
+        -> list[RemoteServiceDiscoveryService]:
+    return get_asyncio_loop().run_until_complete(async_get_tunneld_devices(tunneld_address))
+
+
+async def async_get_tunneld_device_by_udid(udid: str, tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
+        -> Optional[RemoteServiceDiscoveryService]:
+    tunnels = _list_tunnels(tunneld_address)
+    if udid not in tunnels:
+        return None
+    rsds = await _create_rsds_from_tunnels({udid: tunnels[udid]})
+    return rsds[0]
+
+
+def get_tunneld_device_by_udid(udid: str, tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
+        -> Optional[RemoteServiceDiscoveryService]:
+    return get_asyncio_loop().run_until_complete(async_get_tunneld_device_by_udid(udid, tunneld_address))
+
+
+def _list_tunnels(tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) -> dict[str, list[dict]]:
+    try:
+        # Get the list of tunnels from the specified address
+        resp = requests.get(f'http://{tunneld_address[0]}:{tunneld_address[1]}')
+        tunnels = resp.json()
+    except requests.exceptions.ConnectionError:
+        raise TunneldConnectionError()
+    return tunnels
+
+
+async def _create_rsds_from_tunnels(tunnels: dict[str, list[dict]]) -> list[RemoteServiceDiscoveryService]:
+    rsds = []
+    for udid, details in tunnels.items():
+        for tunnel_details in details:
+            rsd = RemoteServiceDiscoveryService((tunnel_details['tunnel-address'], tunnel_details['tunnel-port']),
+                                                name=tunnel_details['interface'])
+            try:
+                await rsd.connect()
+                rsds.append(rsd)
+            except (TimeoutError, ConnectionError):
+                continue
+    return rsds

pymobiledevice3/tunneld.py → pymobiledevice3/tunneld/server.py

@@ -10,7 +10,6 @@
 
 import construct
 import fastapi
-import requests
 import uvicorn
 from construct import StreamError
 from fastapi import FastAPI
@@ -19,7 +18,7 @@
 from pymobiledevice3 import usbmux
 from pymobiledevice3.bonjour import REMOTED_SERVICE_NAMES, browse
 from pymobiledevice3.exceptions import ConnectionFailedError, ConnectionFailedToUsbmuxdError, DeviceNotFoundError, \
-    GetProhibitedError, InvalidServiceError, LockdownError, MuxException, PairingError, TunneldConnectionError
+    GetProhibitedError, InvalidServiceError, LockdownError, MuxException, PairingError
 from pymobiledevice3.lockdown import create_using_usbmux, get_mobdev2_lockdowns
 from pymobiledevice3.osu.os_utils import get_os_utils
 from pymobiledevice3.remote.common import TunnelProtocol
@@ -28,12 +27,10 @@
 from pymobiledevice3.remote.tunnel_service import CoreDeviceTunnelProxy, RemotePairingProtocol, TunnelResult, \
     create_core_device_tunnel_service_using_rsd, get_remote_pairing_tunnel_services
 from pymobiledevice3.remote.utils import get_rsds, stop_remoted
-from pymobiledevice3.utils import asyncio_print_traceback, get_asyncio_loop
+from pymobiledevice3.utils import asyncio_print_traceback
 
 logger = logging.getLogger(__name__)
 
-TUNNELD_DEFAULT_ADDRESS = ('127.0.0.1', 49151)
-
 # bugfix: after the device reboots, it might take some time for remoted to start answering the bonjour queries
 REATTEMPT_INTERVAL = 5
 REATTEMPT_COUNT = 5
@@ -475,52 +472,3 @@ async def start_tunnel(
 
     def _run_app(self) -> None:
         uvicorn.run(self._app, host=self.host, port=self.port, loop='asyncio')
-
-
-async def async_get_tunneld_devices(tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
-        -> list[RemoteServiceDiscoveryService]:
-    tunnels = _list_tunnels(tunneld_address)
-    return await _create_rsds_from_tunnels(tunnels)
-
-
-def get_tunneld_devices(tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
-        -> list[RemoteServiceDiscoveryService]:
-    return get_asyncio_loop().run_until_complete(async_get_tunneld_devices(tunneld_address))
-
-
-async def async_get_tunneld_device_by_udid(udid: str, tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
-        -> Optional[RemoteServiceDiscoveryService]:
-    tunnels = _list_tunnels(tunneld_address)
-    if udid not in tunnels:
-        return None
-    rsds = await _create_rsds_from_tunnels({udid: tunnels[udid]})
-    return rsds[0]
-
-
-def get_tunneld_device_by_udid(udid: str, tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) \
-        -> Optional[RemoteServiceDiscoveryService]:
-    return get_asyncio_loop().run_until_complete(async_get_tunneld_device_by_udid(udid, tunneld_address))
-
-
-def _list_tunnels(tunneld_address: tuple[str, int] = TUNNELD_DEFAULT_ADDRESS) -> dict[str, list[dict]]:
-    try:
-        # Get the list of tunnels from the specified address
-        resp = requests.get(f'http://{tunneld_address[0]}:{tunneld_address[1]}')
-        tunnels = resp.json()
-    except requests.exceptions.ConnectionError:
-        raise TunneldConnectionError()
-    return tunnels
-
-
-async def _create_rsds_from_tunnels(tunnels: dict[str, list[dict]]) -> list[RemoteServiceDiscoveryService]:
-    rsds = []
-    for udid, details in tunnels.items():
-        for tunnel_details in details:
-            rsd = RemoteServiceDiscoveryService((tunnel_details['tunnel-address'], tunnel_details['tunnel-port']),
-                                                name=tunnel_details['interface'])
-            try:
-                await rsd.connect()
-                rsds.append(rsd)
-            except (TimeoutError, ConnectionError):
-                continue
-    return rsds

tests/conftest.py

@@ -8,7 +8,7 @@
 from pymobiledevice3.lockdown_service_provider import LockdownServiceProvider
 from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
 from pymobiledevice3.services.dvt.dvt_secure_socket_proxy import DvtSecureSocketProxyService
-from pymobiledevice3.tunneld import async_get_tunneld_devices
+from pymobiledevice3.tunneld.server import async_get_tunneld_devices
 
 logging.getLogger('quic').disabled = True
 logging.getLogger('asyncio').disabled = True