# -*- coding: utf-8 -*- import os import sys import time import json import wda import threading import subprocess from pathlib import Path from typing import List, Dict, Optional from tidevice import Usbmux from Entity.DeviceModel import DeviceModel from Entity.Variables import WdaAppBundleId from Module.FlaskSubprocessManager import FlaskSubprocessManager from Utils.LogManager import LogManager class Deviceinfo(object): def __init__(self): self.deviceIndex = 0 # 投屏端口（本地映射端口起始值，会递增） self.screenProxy = 9110 # 记录 iproxy Popen 进程：[{ "id": udid, "target": Popen }, ...] self.pidList: List[Dict] = [] # 当前已连接的设备（tidevice 的 Device 对象列表） self.deviceArray: List = [] # 子进程通信（向前端发送设备信息） self.manager = FlaskSubprocessManager.get_instance() # 已发给前端的设备模型列表（用于拔出时发 type=2） self.deviceModelList: List[DeviceModel] = [] # ---------------------------- # 监听设备连接（死循环，内部捕获异常） # ---------------------------- def startDeviceListener(self): while True: try: lists = Usbmux().device_list() except Exception as e: # 另一台电脑常见：usbmuxd 连接失败（未安装 iTunes/Apple Mobile Device Support） LogManager.warning(f"usbmuxd 连接失败: {e}。请确认已安装 iTunes/Apple Mobile Device Support，并在手机上“信任此电脑”", "listener") time.sleep(2) continue # 新接入设备 for device in lists: if device not in self.deviceArray: self.screenProxy += 1 try: self.connectDevice(device.udid) self.deviceArray.append(device) except Exception as e: LogManager.error(f"连接设备失败 {device.udid}: {e}", device.udid) # 拔出设备处理 self._removeDisconnected(lists) time.sleep(1) # ---------------------------- # 连接单台设备：启动 WDA、读取屏参、通知前端、映射投屏端口 # ---------------------------- def connectDevice(self, identifier: str): # 1) 连接 WDA（USBClient -> 设备 8100） try: d = wda.USBClient(identifier, 8100) LogManager.info("启动 WDA 成功", identifier) except Exception as e: LogManager.error(f"启动 WDA 失败，请检查手机是否已信任、WDA 是否正常。错误: {e}", identifier) return # 不抛出到外层，保持监听循环健壮 # 2) 读取屏幕信息（失败不影响主流程） width, height, scale = 0, 0, 1.0 try: size = d.window_size() width, height = size.width, size.height scale = d.scale except Exception as e: LogManager.warning(f"读取屏幕信息失败：{e}", identifier) # 3) 组装模型并发送给前端 model = DeviceModel(identifier, self.screenProxy, width, height, scale, type=1) self.deviceModelList.append(model) try: self.manager.send(model.toDict()) except Exception as e: LogManager.warning(f"向前端发送设备模型失败：{e}", identifier) # 4) 可选：启动你的 app 并回到桌面 try: d.app_start(WdaAppBundleId) d.home() except Exception as e: LogManager.warning(f"启动/切回桌面失败：{e}", identifier) time.sleep(2) # 5) 本地端口 -> 设备端口 的映射（投屏：本地 self.screenProxy -> 设备 9100） target = self.relayDeviceScreenPort(identifier) self.pidList.append({"target": target, "id": identifier}) # ---------------------------- # 处理拔出设备：发通知、关掉 iproxy、移出状态 # ---------------------------- def _removeDisconnected(self, current_list): set1 = set(self.deviceArray) set2 = set(current_list) difference = list(set1 - set2) # 在旧集合中但不在新集合中 -> 已拔出 for i in difference: udid = i.udid # 1) 通知前端：type = 2 for a in list(self.deviceModelList): if udid == a.deviceId: a.type = 2 try: self.manager.send(a.toDict()) except Exception as e: LogManager.warning(f"发送下线事件失败：{e}", udid) self.deviceModelList.remove(a) # 2) 关掉对应的 iproxy for k in list(self.pidList): if udid == k["id"]: target = k.get("target") try: if target and target.poll() is None: target.kill() except Exception: pass self.pidList.remove(k) # 3) 从已连接集合中移除 try: self.deviceArray.remove(i) except Exception: pass # ---------------------------- # 根目录与 iproxy 可执行文件定位 # ---------------------------- def _base_dir(self) -> Path: """ 打包后：返回 exe 所在目录； 源码运行：返回项目根目录（Module 的上一级） """ if getattr(sys, "frozen", False): return Path(sys.executable).resolve().parent return Path(__file__).resolve().parents[1] # iOSAI/ 作为根 def _iproxy_path(self) -> Path: """返回 iproxy 可执行文件的完整路径""" exe = "iproxy.exe" if os.name == "nt" else "iproxy" base = self._base_dir() # 常见放置位置（按优先级） candidates = [ base / "resources" / "iproxy" / exe # 推荐：打包资源目录 ] for p in candidates: if p.exists(): return p tried = [str(c) for c in candidates] raise FileNotFoundError(f"iproxy not found, tried: {tried}") # ---------------------------- # 端口映射：启动 iproxy # ---------------------------- def relayDeviceScreenPort(self, udid: str) -> Optional[subprocess.Popen]: try: iproxy = self._iproxy_path() # 例如 .../resources/iproxy/iproxy.exe iproxy_dir = iproxy.parent # 继承环境并把 iproxy 目录加入 PATH（放最前） env = os.environ.copy() env["PATH"] = str(iproxy_dir) + os.pathsep + env.get("PATH", "") # 可选：帮助本进程解析该目录下 DLL（py3.8+） try: os.add_dll_directory(str(iproxy_dir)) except Exception: pass # Windows 隐藏子进程窗口 creationflags = 0x08000000 if os.name == "nt" else 0 # 绝对路径 + shell=False，避免 PATH/别名干扰 args = [str(iproxy), "-u", udid, str(self.screenProxy), "9100"] # （可选）把子进程输出写到你的日志里，排查更方便 def _pipe_to_log(name: str, stream): try: for line in iter(stream.readline, ''): line = line.rstrip('\r\n') if line: LogManager.info(f"[iproxy {name}] {line}", udid) except Exception: print("遇到错误了") pass p = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(iproxy_dir), env=env, shell=False, text=True, encoding="utf-8", bufsize=1, creationflags=creationflags, ) # 启动异步日志转发（不要阻塞主线程；需要时可删除） try: import threading threading.Thread(target=_pipe_to_log, args=("STDOUT", p.stdout), daemon=True).start() threading.Thread(target=_pipe_to_log, args=("STDERR", p.stderr), daemon=True).start() except Exception as e: print("这里有错误") print(e) pass LogManager.info(f"启动 iproxy 成功，本地 {self.screenProxy} -> 设备 9100", udid) return p except Exception as e: LogManager.error(f"启动 iproxy 失败：{e}", udid) return None