代码优化一下

tidevice/__main__.py

@@ -28,9 +28,8 @@
 from ._imagemounter import cache_developer_image
 from ._ipautil import IPAReader
 from ._perf import DataType
-from ._proto import LOG, MODELS, PROGRAM_NAME
+from ._proto import LOG, MODELS, PROGRAM_NAME, ConnectionType
 from ._relay import relay
-from ._types import ConnectionType
 from ._usbmux import Usbmux
 from ._utils import get_app_dir, get_binary_by_name, is_atty
 from ._version import __version__

tidevice/_device.py

@@ -73,16 +73,6 @@ class BaseDevice():
     def __init__(self,
                  udid: Optional[str] = None,
                  usbmux: Union[Usbmux, str, None] = None):
-        if udid is None:
-            udid = os.environ.get("TMQ_DEVICE_UDID")
-            if udid:
-                logger.info("use udid from env: %s=%s", "TMQ_DEVICE_UDID",
-                            udid)
-        if usbmux is None:
-            usbmux = os.environ.get("TMQ_USBMUX")
-            if usbmux:
-                logger.info("use usbmux from env: %s=%s", "TMQ_USBMUX", usbmux)
-
         if not usbmux:
             self._usbmux = Usbmux()
         elif isinstance(usbmux, str):

tidevice/_instruments.py

@@ -338,7 +338,6 @@ def make_channel(self, identifier: str) -> int:
             return self._channels[identifier]
 
         channel_id = self._next_channel_id()
-        args = [channel_id, identifier]
         aux = AUXMessageBuffer()
         aux.append_u32(channel_id)
         aux.append_obj(identifier)
@@ -683,10 +682,21 @@ def prepare(self):
 
     def app_launch(self,
                    bundle_id: str,
-                   app_env: dict = {},
-                   args: list = [],
-                   kill_running: bool = False) -> int:
+                   app_env: typing.Dict[str, str] = {},
+                   args: typing.List[str] = [],
+                   kill_running: bool = True) -> int:
         """
+        Launch an app with bundle id
+
+        Args:
+            bundle_id: bundle id of the app
+            app_env: environment variables
+            args: arguments
+            kill_running: deprecated argument, useless now
+
+        Returns:
+            pid of the app
+
         Raises:
             ServiceError
         """
@@ -698,8 +708,8 @@ def app_launch(self,
         options = {
             # don't suspend the process after starting it
             "StartSuspendedKey": 0,
-            # kill the application if it is already running
-            "KillExisting": kill_running,
+            # Note: set KillExisting to False and call 60+ times, instruments service will crash
+            "KillExisting": True,
             # I donot know much about it, when set to True, app will have a pid, but not show up
             # "ActivateSuspended": False,
         }
@@ -738,16 +748,18 @@ def app_process_list(self, app_infos: List[dict]) -> Iterator[dict]:
         """
         Args:
             app_infos: value from self.instrumentation.app_list()
-        Returns yield of
-        {
-            'isApplication': True,
-            'name': 'timed',
-            'pid': 58,
-            'realAppName': '/usr/libexec/timed',
-            'startDate': datetime.datetime(2020, 5, 25, 2, 22, 29, 603427)},
-            'bundle_id': 'com.libexec.xxx',
-            'display_name': "xxxxx",
-        }
+        
+        Returns:
+            yield of
+            {
+                'isApplication': True,
+                'name': 'timed',
+                'pid': 58,
+                'realAppName': '/usr/libexec/timed',
+                'startDate': datetime.datetime(2020, 5, 25, 2, 22, 29, 603427)},
+                'bundle_id': 'com.libexec.xxx',
+                'display_name': "xxxxx",
+            }
         """
         def exefile2appinfo(exe_abspath: str, app_infos: List[dict]):
             for info in app_infos:

tidevice/_proto.py

@@ -5,7 +5,7 @@
 __all__ = [
     'Color', 'AFC_MAGIC', 'AFCMode', 'AFC', 'AFCStatus', 'AFCPacket', 'LOCKDOWN_PORT', 'PROGRAM_NAME',
     'SYSMON_PROC_ATTRS', 'SYSMON_SYS_ATTRS', 'MODELS', 'LockdownService',
-    "UsbmuxReplyCode", "InstrumentsService", "LOG", "StatResult"
+    "UsbmuxReplyCode", "InstrumentsService", "LOG", "StatResult", "UsbmuxMessageType", "ConnectionType"
 ]
 
 from dataclasses import dataclass
@@ -152,7 +152,7 @@ class AFCPacket(typing.NamedTuple):
     status: AFCStatus
     data: bytes # contains fd when open file, other time always empty
     payload: bytes
-    
+
 
 # See also: https://www.theiphonewiki.com/wiki/Models
 MODELS = {
@@ -265,16 +265,11 @@ class AFCPacket(typing.NamedTuple):
 }
 
 
-class UsbmuxMessageType(str, enum.Enum):
-    Attached = "Attached"
-    Detached = "Detached"
-
-
 class LockdownService(str, enum.Enum):
     # hdiutil mount /Applications/Xcode.app/Contents/Developer/Platforms/iPhoneOS.platform/DeviceSupport/14.0/DeveloperDiskImage.dmg
     # tree /Volumes/DeveloperDiskImage/Library/Lockdown
     MobileLockdown = 'com.apple.mobile.lockdown'
-    
+
     CRASH_REPORT_MOVER_SERVICE = "com.apple.crashreportmover"
     CRASH_REPORT_COPY_MOBILE_SERVICE = "com.apple.crashreportcopymobile"
 
@@ -292,8 +287,6 @@ class LockdownService(str, enum.Enum):
     TestmanagerdLockdownSecure = "com.apple.testmanagerd.lockdown.secure"  # for iOS 14.0
     AmfiLockdown = "com.apple.amfi.lockdown" # iOS >= 15.7
 
-
-
 
 class InstrumentsService(str, enum.Enum):
     DeviceInfo = 'com.apple.instruments.server.services.deviceinfo'
@@ -331,13 +324,22 @@ class StatResult:
 
     def is_dir(self) -> bool:
         return self.st_ifmt == "S_IFDIR"
-    
+
     def is_link(self) -> bool:
         return self.st_ifmt == "S_IFLNK"
 
 
 
+class UsbmuxMessageType(enum.IntEnum):
+    RESULT = 1
+    CONNECT = 2
+    LISTEN = 3
+    ADD = 4
+    REMOVE = 5
+    PAIRED = 6
+    PLIST = 8
+
 
-if __name__ == "__main__":
-    print(Color.wrap_text("Hello", Color.RED))
-    print(AFC.GET_DEVINFO)
+class ConnectionType(str, enum.Enum):
+    USB = "usb"
+    NETWORK = "network"

tidevice/_safe_socket.py

@@ -15,7 +15,7 @@
 import weakref
 from typing import Any, Union
 
-from ._proto import PROGRAM_NAME
+from ._proto import PROGRAM_NAME, UsbmuxMessageType
 from ._utils import set_socket_timeout
 from .exceptions import *
 
@@ -46,11 +46,11 @@ def release_uid(id: int):
 
 
 class SafeStreamSocket:
-    def __init__(self, addr: Union[str, tuple, socket.socket,
+    def __init__(self, addr: Union[str, typing.Tuple[str, int], socket.socket,
                                    Any]):
         """
         Args:
-            addr: can be /var/run/usbmuxd or (localhost, 27015)
+            addr: can be /var/run/usbmuxd or localhost:27015 or (localhost, 27015)
         """
         self._id = acquire_uid()
         self._sock = None
@@ -131,7 +131,10 @@ def recvall(self, size: int) -> bytearray:
         return buf
 
     def sendall(self, data: Union[bytes, bytearray]) -> int:
-        return self._sock.sendall(data)
+        try:
+            return self._sock.sendall(data)
+        except Exception as e:
+            raise SocketError("sendall error") from e
 
     def ssl_unwrap(self):
         assert isinstance(self._sock, ssl.SSLSocket)
@@ -180,10 +183,10 @@ def __init__(self, addr: str, tag: int = 0):
     def prepare(self):
         pass
 
-    def is_secure(self):
+    def is_secure(self) -> bool:
         return isinstance(self._sock, ssl.SSLSocket)
 
-    def send_packet(self, payload: dict, message_type: int = 8):
+    def send_packet(self, payload: dict, message_type: int = UsbmuxMessageType.PLIST):
         """
         Args:
             payload: required
@@ -192,9 +195,6 @@ def send_packet(self, payload: dict, message_type: int = 8):
             message_type: 8 (Plist)
             tag: int
         """
-        #if self.is_secure():
-        #    logger.debug(secure_text + " send: {}", payload)
-        #else:
         logger.debug("SEND({}): {}", self.id, payload)
 
         body_data = plistlib.dumps(payload)
@@ -222,10 +222,6 @@ def recv_packet(self, header_size=None) -> dict:
         if 'PairRecordData' in payload:
             logger.debug("Recv pair record data ...")
         else:
-            # if self.is_secure():
-            #    logger.debug(secure_text + " recv" + Color.END + ": {}",
-            #                 payload)
-            # else:
             logger.debug("RECV({}): {}", self.id, payload)
         return payload
 

tidevice/_types.py

@@ -1,30 +1,51 @@
 # -*- coding: utf-8 -*-
-
 """Created on Fri Sep 24 2021 14:13:00 by codeskyblue
 """
 
+import dataclasses
 import typing
-import enum
+from dataclasses import dataclass
+
+from ._proto import ConnectionType
+
+_T = typing.TypeVar("_T")
 
 
-class ConnectionType(str, enum.Enum):
-    USB = "usb"
-    NETWORK = "network"
+def alias_field(name: str) -> dataclasses.Field:
+    return dataclasses.field(metadata={"alias": name})
 
 
 class _BaseInfo:
+
     def _asdict(self) -> dict:
         """ for simplejson """
         return self.__dict__.copy()
-
+
+    @classmethod
+    def from_json(cls: _T, data: dict) -> _T:
+        kwargs = {}
+        for field in dataclasses.fields(cls):
+            possible_names = [field.name]
+            if "alias" in field.metadata:
+                possible_names.append(field.metadata["alias"])
+            for name in possible_names:
+                if name in data:
+                    value = data[name]
+                    if field.type != type(value):
+                        value = field.type(value)
+                    kwargs[field.name] = value
+                    break
+        return cls(**kwargs)
+
     def __repr__(self) -> str:
         attrs = []
         for k, v in self.__dict__.items():
             attrs.append(f"{k}={v!r}")
         return f"<{self.__class__.__name__} " + ", ".join(attrs) + ">"
 
 
+@dataclass(frozen=True)
 class DeviceInfo(_BaseInfo):
-    udid: str
-    device_id: int
-    conn_type: ConnectionType
+    udid: str = alias_field("SerialNumber")
+    device_id: int = alias_field("DeviceID")
+    conn_type: ConnectionType = alias_field("ConnectionType")

tidevice/_usbmux.py

@@ -12,8 +12,8 @@
 import uuid
 from typing import Optional, Union
 
-from ._types import DeviceInfo, ConnectionType
-from ._proto import PROGRAM_NAME, UsbmuxReplyCode, LOG
+from ._types import DeviceInfo
+from ._proto import PROGRAM_NAME, ConnectionType, UsbmuxReplyCode, LOG
 from ._safe_socket import PlistSocket, PlistSocketProxy
 from .exceptions import * # pragma warning disables S2208
 
@@ -81,10 +81,10 @@ def device_list(self) -> typing.List[DeviceInfo]:
         result = {}
         for item in data['DeviceList']:
             prop = item['Properties']
-            info = DeviceInfo()
-            info.udid = prop['SerialNumber']
-            info.device_id = prop['DeviceID']
-            info.conn_type = prop['ConnectionType'].lower()
+            prop['ConnectionType'] = prop['ConnectionType'].lower() # 兼容旧代码
+            info = DeviceInfo.from_json(prop)
+
+            # always skip network device
             if info.udid in result and info.conn_type == ConnectionType.NETWORK:
                 continue
             result[info.udid] = info