# -*- 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): LogManager.info("Device Listener started", "listener") 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 目录和 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_dir(self) -> Path: """返回打包后的 iproxy 目录（你现在放在 Module/iproxy/）""" return self._base_dir() / "resources" / "iproxy" def _iproxy_path(self) -> Path: base = self._base_dir() candidates = [ base / "resources" / "iproxy", # 新推荐：resources/iproxy/ ] for p in candidates: if p.exists(): print("iproxy 路径为 ", p) return p # 打一条清晰的错误便于排查 tried = [str(c) for c in candidates] raise FileNotFoundError(f"iproxy not found, tried: {tried}") # ---------------------------- # 端口映射：启动 iproxy（设置 cwd 和 PATH，隐藏窗口） # ---------------------------- def relayDeviceScreenPort(self, udid: str) -> Optional[subprocess.Popen]: try: iproxy = self._iproxy_path() iproxy_dir = self._iproxy_dir() if not iproxy.exists(): raise FileNotFoundError(f"iproxy not found: {iproxy}") # 继承环境并把 iproxy 目录加入 PATH，方便 DLL 解析 env = os.environ.copy() env["PATH"] = str(iproxy_dir) + os.pathsep + env.get("PATH", "") # Windows 隐藏子进程窗口 CREATE_NO_WINDOW = 0x08000000 if os.name == "nt" else 0 p = subprocess.Popen( ["iproxy", "-u", udid, str(self.screenProxy), "9100"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, creationflags=CREATE_NO_WINDOW, cwd=str(iproxy_dir), # 关键：工作目录设为 iproxy 所在目录 env=env, # 关键：把 iproxy_dir 注入 PATH text=True, # 你后面如果要读 stdout/stderr 的话更方便 encoding="utf-8", bufsize=1 ) LogManager.info(f"启动 iproxy 成功，本地 {self.screenProxy} -> 设备 9100", udid) return p except Exception as e: LogManager.error(f"启动 iproxy 失败：{e}", udid) return None