diff --git a/padsleep.py b/padsleep.py index 217b989..bf25556 100644 --- a/padsleep.py +++ b/padsleep.py @@ -35,7 +35,9 @@ DEFAULT_CONFIG = { "check_interval_seconds": 30, "auto_sleep_delay_minutes": 5, "adb_device_serial": "", - "adb_path": "adb", + "adb_wireless_host": "", + "adb_wireless_port": 5555, + "adb_wireless_auto_connect": False, } # ── 日志 ──────────────────────────────────────────────────────── @@ -182,6 +184,8 @@ class AdbManager: self._screen_on_since: float | None = None self._last_known_screen_on = False self._last_adb_check = 0 + self._wireless_addr: str = "" # 当前无线连接地址 (IP:PORT) + self._wireless_retries: int = 0 # 无线重连尝试次数 # ── 连接管理 ── @@ -209,6 +213,107 @@ class AdbManager: self._last_error = str(e) return None + # ── 无线连接管理 ── + + @staticmethod + def is_wireless_serial(serial: str) -> bool: + """判断 serial 是否为无线格式（IP:PORT）。""" + parts = serial.split(":") + if len(parts) != 2: + return False + try: + port = int(parts[1]) + ip_parts = parts[0].split(".") + if len(ip_parts) == 4 and all(0 <= int(p) <= 255 for p in ip_parts): + return True + except (ValueError, IndexError): + pass + return False + + def connect_wireless(self, host: str, port: int = 5555, timeout: float = 5.0) -> tuple[bool, str]: + """连接无线 ADB 设备。返回 (success, message)。""" + addr = f"{host}:{port}" + try: + client = self._ensure_client() + result = client.connect(addr, timeout=timeout) + if "connected" in result.lower(): + self._wireless_addr = addr + self._wireless_retries = 0 + log.info("无线 ADB 已连接: %s", addr) + return True, result + return False, result + except Exception as e: + self._last_error = str(e) + log.error("无线连接失败: %s", e) + return False, str(e) + + def disconnect_wireless(self, addr: str = "") -> bool: + """断开无线 ADB 连接。""" + target = addr or self._wireless_addr + if not target: + return False + try: + client = self._ensure_client() + client.disconnect(target) + log.info("无线 ADB 已断开: %s", target) + if target == self._wireless_addr: + self._wireless_addr = "" + return True + except Exception as e: + self._last_error = str(e) + log.error("断开无线连接失败: %s", e) + return False + + def get_connection_type(self) -> str: + """返回当前连接类型：'usb' / 'wireless' / 'none'。""" + if self._device is None: + return "none" + if self.is_wireless_serial(self._device.serial): + return "wireless" + return "usb" + + def ensure_wireless_connection(self) -> bool: + """确保无线连接存活，失败时尝试重连。返回是否成功。""" + config = self._config_getter() + wireless_host = config.get("adb_wireless_host", "") + wireless_port = config.get("adb_wireless_port", 5555) + + if not wireless_host: + return False + + addr = f"{wireless_host}:{wireless_port}" + + # 检查是否已有连接 + if self._device and self._device.serial == addr: + try: + self._device.getprop("ro.build.version.sdk") + return True + except Exception: + # 连接已断开，尝试重连 + pass + + # 尝试重连（最多 3 次） + max_retries = 3 + for i in range(max_retries): + self._wireless_retries += 1 + log.info("尝试重连无线设备 (%d/%d): %s", i + 1, max_retries, addr) + success, msg = self.connect_wireless(wireless_host, wireless_port, timeout=3.0) + if success: + # 重新获取设备 + try: + client = self._ensure_client() + self._device = client.device(addr) + self._device.getprop("ro.build.version.sdk") + self._ready = True + self._last_error = "" + return True + except Exception: + pass + time.sleep(2) + + self._last_error = f"无线重连失败: {addr}" + return False + # ── 设备管理 ── def check_available(self) -> bool: @@ -222,11 +327,19 @@ class AdbManager: return False def list_devices(self) -> list[dict]: - """返回 [{"serial": "...", "status": "device"}]。""" + """返回 [{"serial": "...", "status": "device", "connection_type": "usb"/"wireless"}]。""" try: client = self._ensure_client() devices = client.device_list() - return [{"serial": d.serial, "status": "device"} for d in devices] + result = [] + for d in devices: + conn_type = "wireless" if self.is_wireless_serial(d.serial) else "usb" + result.append({ + "serial": d.serial, + "status": "device", + "connection_type": conn_type, + }) + return result except Exception as e: self._last_error = str(e) return [] @@ -246,22 +359,45 @@ class AdbManager: config = self._config_getter() target_serial = config.get("adb_device_serial", "") + wireless_host = config.get("adb_wireless_host", "") try: client = self._ensure_client() devices = client.device_list() + # USB 连接优先 for d in devices: - if not target_serial or d.serial == target_serial: - # 尝试获取设备属性以验证连接 - d.getprop("ro.build.version.sdk") - self._device = d - self._ready = True - self._last_error = "" + if not self.is_wireless_serial(d.serial): + if not target_serial or d.serial == target_serial: + d.getprop("ro.build.version.sdk") + self._device = d + self._ready = True + self._last_error = "" + return True + + # 尝试无线连接 + if wireless_host: + wireless_port = config.get("adb_wireless_port", 5555) + wireless_addr = f"{wireless_host}:{wireless_port}" + + # 检查是否已在设备列表中 + for d in devices: + if d.serial == wireless_addr: + d.getprop("ro.build.version.sdk") + self._device = d + self._ready = True + self._last_error = "" + self._wireless_addr = wireless_addr + return True + + # 不在列表中，尝试连接 + if self.ensure_wireless_connection(): return True - if not devices: + if not devices and not wireless_host: self._last_error = "未检测到 ADB 设备，请连接平板" + elif not devices: + self._last_error = f"无线设备未连接: {wireless_host}:{config.get('adb_wireless_port', 5555)}" else: self._last_error = f"未找到指定设备: {target_serial}" except Exception as e: @@ -456,6 +592,9 @@ class PadSleepApp: cfg = self.config_mgr.get() status["periods_count"] = len(cfg.get("screen_on_periods", [])) status["check_interval"] = cfg.get("check_interval_seconds", 30) + status["connection_type"] = self.adb.get_connection_type() + status["wireless_host"] = cfg.get("adb_wireless_host", "") + status["wireless_port"] = cfg.get("adb_wireless_port", 5555) return {"ok": True, "data": status} if cmd == "get_config": @@ -498,6 +637,25 @@ class PadSleepApp: ok = self.adb.screen_on() return {"ok": ok} + if cmd == "connect_wireless": + host = request.get("host", "") + port = request.get("port", 5555) + if not host: + return {"ok": False, "error": "请提供 IP 地址"} + success, msg = self.adb.connect_wireless(host, int(port)) + if success: + # 更新配置 + self.config_mgr.update({ + "adb_wireless_host": host, + "adb_wireless_port": int(port), + }) + return {"ok": success, "data": {"message": msg}} + + if cmd == "disconnect_wireless": + addr = request.get("addr", "") + ok = self.adb.disconnect_wireless(addr) + return {"ok": ok} + return {"ok": False, "error": f"未知命令: {cmd}"} # ── 主循环 ── @@ -517,6 +675,18 @@ class PadSleepApp: else: log.info("ADB 可执行文件已就绪") + # 自动连接无线设备 + config = self.config_mgr.get() + if config.get("adb_wireless_auto_connect") and config.get("adb_wireless_host"): + host = config["adb_wireless_host"] + port = config.get("adb_wireless_port", 5555) + log.info("尝试自动连接无线设备: %s:%d", host, port) + success, msg = self.adb.connect_wireless(host, port, timeout=5.0) + if success: + log.info("无线设备已连接: %s", msg) + else: + log.warning("无线设备连接失败: %s", msg) + # 启动 IPC self.ipc.start() diff --git a/padsleep_tui.py b/padsleep_tui.py index 1e508f1..56c5bd0 100644 --- a/padsleep_tui.py +++ b/padsleep_tui.py @@ -197,6 +197,9 @@ class TuiApp: elif key == ord("s") or key == ord("S"): self._dlg_select_device() + elif key == ord("w") or key == ord("W"): + self._dlg_connect_wireless() + elif key == ord("h") or key == ord("H"): self._show_help() @@ -309,23 +312,39 @@ class TuiApp: if self.status: dev = self.status.get("device") or "未选择" + conn_type = self.status.get("connection_type", "none") + conn_icon = "USB" if conn_type == "usb" else "WiFi" if conn_type == "wireless" else "---" self._clear_line(y) try: self.stdscr.addstr(y, 4, f"当前设备: {dev}") + self.stdscr.addstr(f" [{conn_icon}]") except curses.error: pass y += 1 + # 显示无线连接配置 + wireless_host = self.status.get("wireless_host", "") + wireless_port = self.status.get("wireless_port", 5555) + if wireless_host: + self._clear_line(y) + try: + self.stdscr.addstr(y, 4, f"无线配置: {wireless_host}:{wireless_port}", curses.A_DIM) + except curses.error: + pass + y += 1 + if self.devices: for d in self.devices: if y >= h: break self._clear_line(y) icon = "✓" if d["status"] == "device" else "✗" + conn_type = d.get("connection_type", "usb") + conn_icon = "USB" if conn_type == "usb" else "WiFi" pair = 3 if d["status"] == "device" else 2 try: self.stdscr.addstr(y, 4, f"{icon} {d['serial']}", curses.color_pair(pair)) - self.stdscr.addstr(f" ({d['status']})") + self.stdscr.addstr(f" [{conn_icon}]") except curses.error: pass y += 1 @@ -341,7 +360,7 @@ class TuiApp: if y < h: self._clear_line(y) try: - self.stdscr.addstr(y, 4, "[S]选择设备 [R]刷新", curses.A_DIM) + self.stdscr.addstr(y, 4, "[S]选择设备 [W]无线连接 [R]刷新", curses.A_DIM) except curses.error: pass return y + 1 @@ -385,6 +404,9 @@ class TuiApp: try: self.stdscr.addstr(y, 4, "ADB: ") self.stdscr.addstr(f"{ar}", curses.color_pair(ar_color) | curses.A_BOLD) + conn_type = s.get("connection_type", "none") + if conn_type != "none": + self.stdscr.addstr(f" ({conn_type})") except curses.error: pass y += 1 @@ -423,7 +445,7 @@ class TuiApp: pass # show message below separator else: if self.connected: - msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 H帮助" + msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 W无线 H帮助" else: msg = "未检测到守护进程 | T 尝试连接" try: @@ -603,6 +625,43 @@ class TuiApp: else: self.msg(f"选择失败: {r.get('error', '')}") + def _dlg_connect_wireless(self): + """无线连接对话框。""" + # 获取当前无线配置作为默认值 + wireless_host = "" + wireless_port = "5555" + if self.status: + wireless_host = self.status.get("wireless_host", "") + wireless_port = str(self.status.get("wireless_port", 5555)) + + # 输入 IP 地址 + host = self._dlg_input("无线设备 IP 地址:", wireless_host) + if not host: + return + + # 输入端口 + port_str = self._dlg_input("端口号:", wireless_port) + if not port_str: + return + + try: + port = int(port_str) + if not (1 <= port <= 65535): + self.msg("端口范围错误 (1-65535)") + return + except ValueError: + self.msg("端口格式错误") + return + + # 发送连接请求 + self.msg(f"正在连接 {host}:{port} ...") + r = self.client.send("connect_wireless", host=host, port=port) + if r.get("ok"): + self.msg(f"无线设备已连接: {host}:{port}") + self.refresh() + else: + self.msg(f"连接失败: {r.get('error', '未知错误')}") + @staticmethod def _validate_time(tstr: str) -> bool: try: @@ -631,6 +690,7 @@ class TuiApp: " E 编辑选中时间段", " D 删除选中时间段", " S 选择 ADB 设备", + " W 无线连接设备", " R 刷新状态", " L 重载守护进程配置", " F 强制息屏",