# -*- coding: utf-8 -*- import os import signal 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] = [] # 最大可连接设备限制 self.maxDeviceCount = 6 # 操作锁 self._lock = threading.Lock() # ===== iproxy：一次性完成 路径定位 + 环境变量配置 + 启动器准备 ===== try: self.iproxy_path = self._iproxy_path() # 绝对路径 self.iproxy_dir = self.iproxy_path.parent # 1) 配置环境（PATH/DLL），放到初始化里一次性处理 os.environ["PATH"] = str(self.iproxy_dir) + os.pathsep + os.environ.get("PATH", "") try: # 仅 Windows 有效；其他平台忽略 os.add_dll_directory(str(self.iproxy_dir)) except Exception: pass # 2) 预构建通用 Popen 参数（隐藏窗口、工作目录、文本模式等） self._creationflags = 0x08000000 if os.name == "nt" else 0 self._popen_kwargs = dict( stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(self.iproxy_dir), shell=False, text=True, creationflags=self._creationflags, encoding="utf-8", bufsize=1, ) # 3) 准备一个“启动器”（闭包）：仅接受 (udid, local_port, remote_port) 参数 def _spawn_iproxy(udid: str, local_port: int, remote_port: int = 9100) -> subprocess.Popen: args = [str(self.iproxy_path), "-u", udid, str(local_port), str(remote_port)] p = subprocess.Popen(args, **self._popen_kwargs) # 异步日志转发（可选） def _pipe_to_log(name: str, stream): try: for line in iter(stream.readline, ''): s = line.strip() if s: LogManager.info(f"[iproxy {name}] {s}", udid) except Exception: pass 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: pass return p self._spawn_iproxy = _spawn_iproxy # 保存启动器 LogManager.info(f"iproxy 启动器已就绪，目录: {self.iproxy_dir}") except Exception as e: # 没找到 iproxy 也允许实例化成功，但后续启动会失败并给出明确日志 self.iproxy_path = None self.iproxy_dir = None self._spawn_iproxy = None LogManager.error(f"初始化 iproxy 失败：{e}") # ---------------------------- # 监听设备连接（死循环，内部捕获异常） # ---------------------------- 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，并在手机上“信任此电脑”") time.sleep(2) continue # 新接入设备 for device in lists: if (device not in self.deviceArray) and (len(self.deviceArray) < self.maxDeviceCount): 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) # 加个非空判断 if target is not None: with self._lock: self.pidList.append({"target": target, "id": identifier}) # 安全杀死iproxy进程 def _terminate_proc(self, p: subprocess.Popen): if not p: return if p.poll() is not None: return try: p.terminate() # 先温柔 p.wait(timeout=3) except Exception: try: if os.name == "posix": try: # 如果 iproxy 启动时用了 setsid，这里可杀整个进程组 os.killpg(os.getpgid(p.pid), signal.SIGKILL) except Exception: p.kill() else: p.kill() # Windows 直接 kill p.wait(timeout=2) # 一定要 wait，避免僵尸 except Exception: pass # ---------------------------- # 处理拔出设备：发通知、关掉 iproxy、移出状态 # ---------------------------- def _removeDisconnected(self, current_list): # 1) 计算“被拔出”的 UDID 集合 —— 用 UDID，而不是对象做集合运算 try: prev_udids = {getattr(d, "udid", None) for d in self.deviceArray if getattr(d, "udid", None)} now_udids = {getattr(d, "udid", None) for d in current_list if getattr(d, "udid", None)} except Exception as e: LogManager.error(f"收集 UDID 失败：{e}", "") return removed_udids = prev_udids - now_udids if not removed_udids: return # 2) 加锁，避免多线程同时改三个列表 if not hasattr(self, "_lock"): self._lock = threading.RLock() with self._lock: # 2.1 通知前端并清理 deviceModelList for udid in list(removed_udids): for a in list(self.deviceModelList): if udid == getattr(a, "deviceId", None): a.type = 2 try: self.manager.send(a.toDict()) except Exception as e: LogManager.warning(f"发送下线事件失败：{e}", udid) try: self.deviceModelList.remove(a) except ValueError: pass # 2.2 关闭该 UDID 的所有 iproxy survivors = [] for k in list(self.pidList): kid = k.get("id") if kid in removed_udids: p = k.get("target") try: self._terminate_proc(p) except Exception as e: LogManager.warning(f"关闭 iproxy 异常：{e}", kid) # 不再把该项放回 survivors，相当于移除 else: survivors.append(k) self.pidList = survivors # 2.3 从已连接集合中移除（按 UDID 过滤，避免对象引用不一致导致 remove 失败） self.deviceArray = [d for d in self.deviceArray if getattr(d, "udid", None) not in removed_udids] # 3) 打点 for udid in removed_udids: LogManager.info("设备已拔出，清理完成（下线通知 + 端口映射关闭 + 状态移除）", udid) # ---------------------------- # 根目录与 iproxy 可执行文件定位 # ---------------------------- def _base_dir(self) -> Path: if getattr(sys, "frozen", False): return Path(sys.executable).resolve().parent return Path(__file__).resolve().parents[1] # iOSAI/ 作为根 def _iproxy_path(self) -> Path: 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 raise FileNotFoundError(f"iproxy not found, tried: {[str(c) for c in candidates]}") # ---------------------------- # 端口映射：仅做“转发端口”这件事（调用已准备好的启动器） # ---------------------------- def relayDeviceScreenPort(self, udid: str) -> Optional[subprocess.Popen]: if not self._spawn_iproxy: LogManager.error("iproxy 启动器未就绪，无法建立端口映射（初始化时未找到 iproxy）。", udid) return None try: p = self._spawn_iproxy(udid, self.screenProxy, 9100) LogManager.info(f"启动 iproxy 成功，本地 {self.screenProxy} -> 设备 9100", udid) return p except Exception as e: LogManager.error(f"启动 iproxy 失败：{e}", udid) return None