From a337f9a0df3343c42f2d519095cbd923a5cef0cc Mon Sep 17 00:00:00 2001 From: AbyssRei Date: Sat, 2 May 2026 01:06:49 +0800 Subject: [PATCH 1/4] =?UTF-8?q?refactor(maa=5Fadb=5Fconnector):=20?= =?UTF-8?q?=E9=87=8D=E6=9E=84ADB=E8=BF=9E=E6=8E=A5=E5=99=A8=EF=BC=8C?= =?UTF-8?q?=E7=AE=80=E5=8C=96=E8=AE=BE=E5=A4=87=E5=8F=91=E7=8E=B0=E5=92=8C?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E9=80=BB=E8=BE=91=20-=20=E7=A7=BB=E9=99=A4?= =?UTF-8?q?=E4=B8=8D=E5=BF=85=E8=A6=81=E7=9A=84=E7=B1=BB=E5=92=8C=E6=96=B9?= =?UTF-8?q?=E6=B3=95=EF=BC=8C=E6=95=B4=E5=90=88=E8=AE=BE=E5=A4=87=E4=BF=A1?= =?UTF-8?q?=E6=81=AF=E7=AE=A1=E7=90=86=20-=20=E6=9B=B4=E6=96=B0=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E9=80=BB=E8=BE=91=E4=BB=A5=E6=94=AF=E6=8C=81=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E9=80=89=E6=8B=A9=20-=20=E4=BC=98=E5=8C=96=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E4=BF=A1=E6=81=AF=E4=BB=A5=E6=8F=90=E9=AB=98=E5=8F=AF?= =?UTF-8?q?=E8=AF=BB=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- maa_adb_connector.py | 382 ++++++++++--------------------------------- main.py | 127 +++++--------- 2 files changed, 127 insertions(+), 382 deletions(-) diff --git a/maa_adb_connector.py b/maa_adb_connector.py index b47bc9c..c9fd8fe 100644 --- a/maa_adb_connector.py +++ b/maa_adb_connector.py @@ -37,39 +37,13 @@ class MaaAvailability(Enum): @dataclass(frozen=True) -class ConnectionType: - type_id: str - display_name: str - default_address: str - description: str - - -@dataclass(frozen=True) -class InputMethodOption: - method_id: str - enum_value: int - display_name: str - description: str - - -@dataclass(frozen=True) -class MaaConnectionConfig: - maa_binary_path: str = "" - adb_path: str = r".\platform-tools\adb.exe" - device_serial: str = "" - screencap_method: int = 1 - input_method: int = 4 - screenshot_use_raw_size: bool = True - config: dict = field(default_factory=dict) - - -@dataclass -class AdapterState: - use_maa: bool = False - maa_availability: MaaAvailability = MaaAvailability.UNKNOWN - active_connector_type: str = "legacy" - connection_type_id: str = "adb" - input_method_id: str = "maatouch" +class AdbDeviceInfo: + name: str + adb_path: str + address: str + screencap_methods: int + input_methods: int + config: dict class MaaFrameworkDetector: @@ -100,7 +74,7 @@ def detect(cls) -> MaaAvailability: from maa.toolkit import Toolkit except Exception: cls._status = MaaAvailability.IMPORT_FAILED - cls._status_message = "MAA Framework导入失败，请安装maa.library" + cls._status_message = "MAA Framework导入失败，请安装maafw" cls._checked = True logger.warning(cls._status_message) return cls._status @@ -137,84 +111,65 @@ def reset(cls): cls._status_message = "" -class ConnectionTypeRegistry: - _types: list[ConnectionType] = [ - ConnectionType("adb", "ADB连接", "", "通用ADB连接，需手动指定设备地址"), - ConnectionType("ldplayer", "雷电模拟器", "emulator-5554", "雷电模拟器默认ADB地址"), - ConnectionType("mumu", "MuMu模拟器", "127.0.0.1:7555", "MuMu模拟器默认ADB地址"), - ConnectionType("mumu12", "MuMu12模拟器", "127.0.0.1:16384", "MuMu12模拟器默认ADB地址"), - ConnectionType("bluestacks", "蓝叠模拟器", "127.0.0.1:5555", "蓝叠模拟器默认ADB地址"), - ConnectionType("nox", "夜神模拟器", "127.0.0.1:62001", "夜神模拟器默认ADB地址"), - ] - - @classmethod - def get_all_types(cls) -> list[ConnectionType]: - return cls._types - - @classmethod - def get_default_address(cls, type_id: str) -> str: - for ct in cls._types: - if ct.type_id == type_id: - return ct.default_address - return "" - - @classmethod - def get_type_by_id(cls, type_id: str) -> ConnectionType | None: - for ct in cls._types: - if ct.type_id == type_id: - return ct - return None - - -class InputMethodRegistry: - _methods: list[InputMethodOption] = [ - InputMethodOption("adb_shell", 1, "AdbShell", "ADB shell input命令，兼容性最高"), - InputMethodOption("minitouch_adb_key", 2, "MinitouchAndAdbKey", "minitouch注入+ADB按键，低延迟需root"), - InputMethodOption("maatouch", 4, "Maatouch", "Maatouch注入，低延迟MAA自带"), - InputMethodOption("emulator_extras", 8, "EmulatorExtras", "模拟器扩展接口，仅特定模拟器支持"), - ] +class MaaAdbConnector: + def __init__(self): + self.screen_width: int = 0 + self.screen_height: int = 0 + self.device_serial: str = "" + self.is_connected: bool = False + self._ctrl = None + self._devices: list[AdbDeviceInfo] = [] + self._selected_device_index: int = -1 - @classmethod - def get_all_methods(cls) -> list[InputMethodOption]: - return cls._methods + @property + def devices(self) -> list[AdbDeviceInfo]: + return self._devices - @classmethod - def get_method_by_id(cls, method_id: str) -> InputMethodOption | None: - for m in cls._methods: - if m.method_id == method_id: - return m + @property + def selected_device(self) -> AdbDeviceInfo | None: + if 0 <= self._selected_device_index < len(self._devices): + return self._devices[self._selected_device_index] return None - @classmethod - def get_enum_value_by_id(cls, method_id: str) -> int: - m = cls.get_method_by_id(method_id) - return m.enum_value if m else 4 + def find_devices(self) -> list[AdbDeviceInfo]: + if not MaaFrameworkDetector.is_available(): + logger.warning("MAA Framework不可用，无法发现设备") + return [] - @classmethod - def get_default_method(cls) -> InputMethodOption: - return cls.get_method_by_id("maatouch") + try: + binary_path = resolve_maafw_path() + if binary_path: + os.environ["MAAFW_BINARY_PATH"] = binary_path + from maa.toolkit import Toolkit + Toolkit.init_option(str(Path.cwd())) -class MaaAdbConnector: - def __init__(self, config: MaaConnectionConfig | None = None): - self._config = config or MaaConnectionConfig() - self.screen_width: int = 0 - self.screen_height: int = 0 - self.device_serial: str = self._config.device_serial - self.is_connected: bool = False - self.is_maa_available: bool = False - self._ctrl = None - self._connection_type: str = "adb" + raw_devices = Toolkit.find_adb_devices() + self._devices = [] + for d in raw_devices: + self._devices.append(AdbDeviceInfo( + name=d.name, + adb_path=str(d.adb_path), + address=d.address, + screencap_methods=d.screencap_methods, + input_methods=d.input_methods, + config=d.config if isinstance(d.config, dict) else {}, + )) + logger.info(f"发现 {len(self._devices)} 个ADB设备") + return self._devices + except Exception as e: + logger.error(f"发现ADB设备失败: {e}") + self._devices = [] + return [] - def connect(self): + def connect(self, device_index: int = 0) -> bool: if not MaaFrameworkDetector.is_available(): - self.is_maa_available = False - logger.warning("MAA Framework不可用，MaaAdbConnector无法连接") self.is_connected = False - return + logger.warning("MAA Framework不可用，无法连接") + return False try: - binary_path = self._config.maa_binary_path or resolve_maafw_path() + binary_path = resolve_maafw_path() if binary_path: os.environ["MAAFW_BINARY_PATH"] = binary_path @@ -223,20 +178,28 @@ def connect(self): Toolkit.init_option(str(Path.cwd())) - target_serial = self.device_serial if self.device_serial else "127.0.0.1:5555" - self.device_serial = target_serial + if not self._devices: + self.find_devices() + + if not self._devices: + raise RuntimeError("未找到任何ADB设备，请确保设备已连接且ADB调试已启用") + + if device_index < 0 or device_index >= len(self._devices): + device_index = 0 + + self._selected_device_index = device_index + device = self._devices[device_index] - adb_path = str(Path(self._config.adb_path).resolve()) self._ctrl = AdbController( - adb_path, - target_serial, - screencap_methods=self._config.screencap_method, - input_methods=self._config.input_method, - config=self._config.config, + device.adb_path, + device.address, + device.screencap_methods, + device.input_methods, + device.config, ) self._ctrl.post_connection().wait() - self._ctrl.set_screenshot_use_raw_size(self._config.screenshot_use_raw_size) + self._ctrl.set_screenshot_use_raw_size(True) self._ctrl.post_screencap().wait() image = self._ctrl.cached_image @@ -245,18 +208,22 @@ def connect(self): else: self.screen_width, self.screen_height = self._get_window_size_fallback() - self.is_maa_available = True + self.device_serial = device.address self.is_connected = True - logger.info(f"MAA Framework ADB连接成功: {target_serial}, 分辨率: {self.screen_width}x{self.screen_height}") + logger.info(f"MAA Framework ADB连接成功: {device.name} ({device.address}), 分辨率: {self.screen_width}x{self.screen_height}") + return True except Exception as e: - self.is_maa_available = False self.is_connected = False logger.error(f"MAA Framework ADB连接失败: {e}") + return False def _get_window_size_fallback(self) -> tuple[int, int]: try: - size_cmd = f"{self._config.adb_path} -s {self.device_serial} shell wm size" + device = self.selected_device + if not device: + return 1920, 1080 + size_cmd = f"{device.adb_path} -s {device.address} shell wm size" result = subprocess.run(size_cmd, shell=True, capture_output=True, text=True, check=True, timeout=5) output = result.stdout.strip() if "Physical size:" in output: @@ -312,24 +279,8 @@ def swipe(self, start: tuple[float, float], end: tuple[float, float], duration: logger.error(f"MAA滑动失败: {e}") def get_device_list(self) -> list[str]: - try: - from maa.toolkit import AdbDevice - devices = AdbDevice.find() - if devices: - return [d.name for d in devices] - except Exception: - logger.debug("MAA AdbDevice.find()失败，降级到subprocess") - try: - device_cmd = f"{self._config.adb_path} devices" - result = subprocess.run(device_cmd, shell=True, capture_output=True, text=True, timeout=5) - devices = [] - for line in result.stdout.split("\n"): - if "\tdevice" in line: - devices.append(line.split("\t")[0]) - return devices - except Exception as e: - logger.error(f"获取设备列表失败: {e}") - return [] + devices = self.find_devices() + return [f"{d.name} ({d.address})" for d in devices] def update_device_serial(self, serial: str) -> str: self.device_serial = serial @@ -342,172 +293,3 @@ def disconnect(self): except Exception: pass self.is_connected = False - self.is_maa_available = False - - def set_config(self, config: MaaConnectionConfig): - self._config = config - if config.device_serial: - self.device_serial = config.device_serial - - -class AdbConnectorAdapter: - def __init__(self, adb_path: str = r".\platform-tools\adb.exe"): - import loadData - self._legacy_connector = loadData.AdbConnector() - self._maa_connector: MaaAdbConnector | None = None - self._use_maa: bool = False - self._maa_config = MaaConnectionConfig(adb_path=adb_path) - self._state = AdapterState() - - @property - def is_connected(self) -> bool: - if self._use_maa and self._maa_connector: - return self._maa_connector.is_connected - return self._legacy_connector.is_connected - - @property - def screen_width(self) -> int: - if self._use_maa and self._maa_connector: - return self._maa_connector.screen_width - return self._legacy_connector.screen_width - - @property - def screen_height(self) -> int: - if self._use_maa and self._maa_connector: - return self._maa_connector.screen_height - return self._legacy_connector.screen_height - - @property - def device_serial(self) -> str: - if self._use_maa and self._maa_connector: - return self._maa_connector.device_serial - return self._legacy_connector.device_serial - - @property - def is_maa_available(self) -> bool: - return self._use_maa - - @property - def active_implementation(self) -> str: - return "maa" if self._use_maa else "legacy" - - @property - def state(self) -> AdapterState: - return self._state - - def get_config(self) -> MaaConnectionConfig: - return self._maa_config - - def set_maa_binary_path(self, path: str): - self._maa_config = MaaConnectionConfig( - maa_binary_path=path, - adb_path=self._maa_config.adb_path, - device_serial=self._maa_config.device_serial, - screencap_method=self._maa_config.screencap_method, - input_method=self._maa_config.input_method, - screenshot_use_raw_size=self._maa_config.screenshot_use_raw_size, - config=self._maa_config.config, - ) - if path: - os.environ["MAAFW_BINARY_PATH"] = path - MaaFrameworkDetector.reset() - - def set_connection_type(self, type_id: str): - self._state.connection_type_id = type_id - default_addr = ConnectionTypeRegistry.get_default_address(type_id) - if default_addr: - self._maa_config = MaaConnectionConfig( - maa_binary_path=self._maa_config.maa_binary_path, - adb_path=self._maa_config.adb_path, - device_serial=default_addr, - screencap_method=self._maa_config.screencap_method, - input_method=self._maa_config.input_method, - screenshot_use_raw_size=self._maa_config.screenshot_use_raw_size, - config=self._maa_config.config, - ) - self._legacy_connector.device_serial = default_addr - - def set_input_method(self, method_id: str): - enum_value = InputMethodRegistry.get_enum_value_by_id(method_id) - self._state.input_method_id = method_id - self._maa_config = MaaConnectionConfig( - maa_binary_path=self._maa_config.maa_binary_path, - adb_path=self._maa_config.adb_path, - device_serial=self._maa_config.device_serial, - screencap_method=self._maa_config.screencap_method, - input_method=enum_value, - screenshot_use_raw_size=self._maa_config.screenshot_use_raw_size, - config=self._maa_config.config, - ) - - def set_device_serial(self, serial: str): - self._maa_config = MaaConnectionConfig( - maa_binary_path=self._maa_config.maa_binary_path, - adb_path=self._maa_config.adb_path, - device_serial=serial, - screencap_method=self._maa_config.screencap_method, - input_method=self._maa_config.input_method, - screenshot_use_raw_size=self._maa_config.screenshot_use_raw_size, - config=self._maa_config.config, - ) - self._legacy_connector.device_serial = serial - - def connect(self): - if MaaFrameworkDetector.is_available(): - try: - maa_connector = MaaAdbConnector(self._maa_config) - maa_connector.device_serial = self._legacy_connector.device_serial or self._maa_config.device_serial - maa_connector.connect() - if maa_connector.is_connected: - self._maa_connector = maa_connector - self._use_maa = True - self._state.use_maa = True - self._state.maa_availability = MaaAvailability.AVAILABLE - self._state.active_connector_type = "maa" - logger.info("AdbConnectorAdapter: 使用MAA Framework实现") - return - except Exception as e: - logger.warning(f"MAA Framework连接失败，降级到自有实现: {e}") - - self._use_maa = False - self._state.use_maa = False - self._state.maa_availability = MaaFrameworkDetector.get_status() - self._state.active_connector_type = "legacy" - logger.info("AdbConnectorAdapter: 降级到自有ADB实现") - self._legacy_connector.connect() - - def capture_screenshot(self) -> np.ndarray | None: - if self._use_maa and self._maa_connector: - return self._maa_connector.capture_screenshot() - return self._legacy_connector.capture_screenshot() - - def click(self, point: tuple[float, float]): - if self._use_maa and self._maa_connector: - self._maa_connector.click(point) - else: - self._legacy_connector.click(point) - - def swipe(self, start: tuple[float, float], end: tuple[float, float], duration: int = 500): - if self._use_maa and self._maa_connector: - self._maa_connector.swipe(start, end, duration) - else: - logger.warning("自有ADB实现不支持滑动操作") - - def get_device_list(self) -> list[str]: - if self._use_maa and self._maa_connector: - return self._maa_connector.get_device_list() - return self._legacy_connector.get_device_list() - - def update_device_serial(self, serial: str) -> str: - self.set_device_serial(serial) - if self._use_maa and self._maa_connector: - return self._maa_connector.update_device_serial(serial) - return self._legacy_connector.update_device_serial(serial) - - def disconnect(self): - if self._maa_connector: - self._maa_connector.disconnect() - self._maa_connector = None - self._use_maa = False - self._state.use_maa = False - self._state.active_connector_type = "legacy" diff --git a/main.py b/main.py index ad80522..66ce0a0 100644 --- a/main.py +++ b/main.py @@ -17,8 +17,7 @@ import loadData import auto_fetch -import maa_adb_connector -from maa_adb_connector import AdbConnectorAdapter, ConnectionTypeRegistry, InputMethodRegistry, MaaFrameworkDetector +from maa_adb_connector import MaaAdbConnector, MaaFrameworkDetector from dark_mode_style_fix import DarkModeStyleFix import similar_history_match import recognize @@ -55,17 +54,18 @@ class ADBConnectorThread(QThread): """ - Worker thread to run loadData.AdbConnector.connect() without blocking the UI. + Worker thread to run MaaAdbConnector.connect() without blocking the UI. """ connect_finished = pyqtSignal() - def __init__(self, app: "ArknightsApp"): + def __init__(self, app: "ArknightsApp", device_index: int = 0): super().__init__() self.app = app + self.device_index = device_index def run(self): - self.app.adb_connector.connect() + self.app.adb_connector.connect(self.device_index) self.connect_finished.emit() class ArknightsApp(QMainWindow): @@ -98,7 +98,7 @@ def __init__(self): self.current_capture_mode = "ADB" # 尝试连接模拟器 - self.adb_connector = AdbConnectorAdapter() + self.adb_connector = MaaAdbConnector() self.pc_connector = loadData.PcConnector() self.adb_connector_thread = ADBConnectorThread(self) self.adb_connector_thread.connect_finished.connect(self.on_adb_connected) @@ -390,46 +390,20 @@ def init_ui(self): self.win_mode_btn.clicked.connect(lambda: self.on_mode_changed("WIN")) connection_layout.addWidget(mode_row) - # MAA连接方式行 - maa_row = QWidget() - maa_row_layout = QHBoxLayout(maa_row) - maa_row_layout.setContentsMargins(0, 0, 0, 0) - - self.connection_type_label = QLabel("连接方式:") - self.connection_type_combo = QComboBox() - for ct in ConnectionTypeRegistry.get_all_types(): - self.connection_type_combo.addItem(ct.display_name, ct.type_id) - self.connection_type_combo.currentIndexChanged.connect(self.on_connection_type_changed) - - self.input_method_label = QLabel("输入方式:") - self.input_method_combo = QComboBox() - default_method = InputMethodRegistry.get_default_method() - for m in InputMethodRegistry.get_all_methods(): - self.input_method_combo.addItem(m.display_name, m.method_id) - idx = self.input_method_combo.findData(default_method.method_id) - if idx >= 0: - self.input_method_combo.setCurrentIndex(idx) - self.input_method_combo.currentIndexChanged.connect(self.on_input_method_changed) - - maa_row_layout.addWidget(self.connection_type_label) - maa_row_layout.addWidget(self.connection_type_combo) - maa_row_layout.addWidget(self.input_method_label) - maa_row_layout.addWidget(self.input_method_combo) - connection_layout.addWidget(maa_row) - # 序列号行 conn_row1 = QWidget() conn_row1_layout = QHBoxLayout(conn_row1) conn_row1_layout.setContentsMargins(0, 0, 0, 0) - self.serial_label = QLabel("模拟器序列号:") + self.serial_label = QLabel("ADB设备:") self.serial_entry = QComboBox() self.serial_entry.setEditable(True) self.serial_entry.setFixedWidth(200) - self.serial_entry.lineEdit().setPlaceholderText("127.0.0.1:5555") + self.serial_entry.lineEdit().setPlaceholderText("选择或输入设备地址") + self.serial_entry.currentIndexChanged.connect(self.on_device_selected) - self.serial_button = QPushButton("更新") - self.serial_button.clicked.connect(self.update_device_serial) + self.serial_button = QPushButton("刷新") + self.serial_button.clicked.connect(self.refresh_and_connect) conn_row1_layout.addWidget(self.serial_label) conn_row1_layout.addWidget(self.serial_entry) @@ -577,16 +551,12 @@ def on_mode_changed(self, mode): self.serial_label.setEnabled(is_adb_mode) self.serial_entry.setEnabled(is_adb_mode) self.serial_button.setEnabled(is_adb_mode) - self.connection_type_label.setEnabled(is_adb_mode) - self.connection_type_combo.setEnabled(is_adb_mode) - self.input_method_label.setEnabled(is_adb_mode) - self.input_method_combo.setEnabled(is_adb_mode) if mode == "ADB": self.refresh_device_list() self.recognizer = recognize.RecognizeMonster(method="ADB") - if not self.adb_connector.device_serial: - self.adb_connector_thread.start() + if not self.adb_connector.is_connected: + self.refresh_and_connect() elif mode == "WIN": if self.recognizer.method != "WIN": self.recognizer = recognize.RecognizeMonster(method="WIN") @@ -601,35 +571,15 @@ def on_mode_changed(self, mode): def on_adb_connected(self): logger.info("模拟器初始化完成") - if self.adb_connector.is_maa_available: - self.maa_status_label.setText("MAA Framework已连接") + if self.adb_connector.is_connected: + device = self.adb_connector.selected_device + name = device.name if device else self.adb_connector.device_serial + self.maa_status_label.setText(f"MAA Framework已连接: {name}") self.maa_status_label.setStyleSheet("color: #00aa00; font-size: 10px;") else: - self.maa_status_label.setText("使用自有ADB实现（MAA Framework不可用）") - self.maa_status_label.setStyleSheet("color: #996600; font-size: 10px;") - - def on_connection_type_changed(self, index): - type_id = self.connection_type_combo.currentData() - if not type_id: - return - default_address = ConnectionTypeRegistry.get_default_address(type_id) - if default_address: - self.serial_entry.setCurrentText(default_address) - self.adb_connector.set_connection_type(type_id) - self.adb_connector.set_device_serial(default_address) - if self.adb_connector.is_connected: - self.adb_connector.disconnect() - self.maa_status_label.setText("已断开，请重新连接") + self.maa_status_label.setText("MAA Framework连接失败") self.maa_status_label.setStyleSheet("color: #aa0000; font-size: 10px;") - def on_input_method_changed(self, index): - method_id = self.input_method_combo.currentData() - if not method_id: - return - self.adb_connector.set_input_method(method_id) - if self.adb_connector.is_connected: - QMessageBox.information(self, "提示", "输入方式已更改，请重新连接以生效") - def choose_capture_window(self): """弹出窗口选择器，切换 WinRT 截屏源（窗口标题或整屏）。""" import traceback, cv2 @@ -959,26 +909,39 @@ def update_statistics(self): self.stats_label.setText(stats_text) def refresh_device_list(self): - """刷新并更新模拟器序列号下拉列表""" + """刷新ADB设备下拉列表""" current_text = self.serial_entry.currentText() - devices = self.adb_connector.get_device_list() + device_names = self.adb_connector.get_device_list() self.serial_entry.clear() - if devices: - self.serial_entry.addItems(devices) - if current_text in devices: + if device_names: + self.serial_entry.addItems(device_names) + if current_text in device_names: self.serial_entry.setCurrentText(current_text) else: self.serial_entry.setCurrentIndex(0) else: - self.serial_entry.addItem("127.0.0.1:5555") - self.serial_entry.setCurrentText(current_text if current_text else "127.0.0.1:5555") - - def update_device_serial(self): - new_serial = self.serial_entry.currentText() - device_serial = self.adb_connector.update_device_serial(new_serial) - self.adb_connector.connect() # 尝试连接新设备 - self.serial_entry.setCurrentText(device_serial) - QMessageBox.information(self, "提示", f"已更新模拟器序列号为: {device_serial}") + self.serial_entry.addItem("未发现设备") + self.serial_entry.setCurrentIndex(0) + + def refresh_and_connect(self): + """刷新设备列表并连接选中设备""" + self.refresh_device_list() + device_index = self.serial_entry.currentIndex() + if device_index < 0: + device_index = 0 + if self.adb_connector.devices: + self.adb_connector_thread = ADBConnectorThread(self, device_index) + self.adb_connector_thread.connect_finished.connect(self.on_adb_connected) + self.adb_connector_thread.start() + + def on_device_selected(self, index): + """下拉框选择设备后自动连接""" + if index < 0 or not self.adb_connector.devices: + return + if index < len(self.adb_connector.devices): + self.adb_connector_thread = ADBConnectorThread(self, index) + self.adb_connector_thread.connect_finished.connect(self.on_adb_connected) + self.adb_connector_thread.start() def start_callback(self): self.update_button_signal.emit("停止自动获取数据") From e48eba83103a90beb382971df76638cf0d4bbec2 Mon Sep 17 00:00:00 2001 From: AbyssRei Date: Sat, 2 May 2026 01:32:08 +0800 Subject: [PATCH 2/4] =?UTF-8?q?feat(maa=5Fadb=5Fconnector):=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E8=87=AA=E5=AE=9A=E4=B9=89=E8=BF=9E=E6=8E=A5=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=EF=BC=8C=E6=94=AF=E6=8C=81=E6=89=8B=E5=8A=A8=E8=BE=93?= =?UTF-8?q?=E5=85=A5=E8=AE=BE=E5=A4=87=E5=9C=B0=E5=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- maa_adb_connector.py | 87 ++++++++++++++++++++++++++++++++++++-------- main.py | 29 +++++++++++++-- 2 files changed, 97 insertions(+), 19 deletions(-) diff --git a/maa_adb_connector.py b/maa_adb_connector.py index c9fd8fe..155dd98 100644 --- a/maa_adb_connector.py +++ b/maa_adb_connector.py @@ -131,36 +131,47 @@ def selected_device(self) -> AdbDeviceInfo | None: return self._devices[self._selected_device_index] return None - def find_devices(self) -> list[AdbDeviceInfo]: + def _ensure_toolkit(self): + binary_path = resolve_maafw_path() + if binary_path: + os.environ["MAAFW_BINARY_PATH"] = binary_path + from maa.toolkit import Toolkit + Toolkit.init_option(str(Path.cwd())) + return Toolkit + + def find_devices(self, adb_path: str = "") -> list[AdbDeviceInfo]: if not MaaFrameworkDetector.is_available(): logger.warning("MAA Framework不可用，无法发现设备") return [] try: - binary_path = resolve_maafw_path() - if binary_path: - os.environ["MAAFW_BINARY_PATH"] = binary_path - - from maa.toolkit import Toolkit - Toolkit.init_option(str(Path.cwd())) - - raw_devices = Toolkit.find_adb_devices() - self._devices = [] + Toolkit = self._ensure_toolkit() + raw_devices = Toolkit.find_adb_devices(specified_adb=adb_path) if adb_path else Toolkit.find_adb_devices() + seen = {d.address for d in self._devices} + new_devices = [] for d in raw_devices: - self._devices.append(AdbDeviceInfo( + info = AdbDeviceInfo( name=d.name, adb_path=str(d.adb_path), address=d.address, screencap_methods=d.screencap_methods, input_methods=d.input_methods, config=d.config if isinstance(d.config, dict) else {}, - )) - logger.info(f"发现 {len(self._devices)} 个ADB设备") + ) + if info.address not in seen: + new_devices.append(info) + self._devices.append(info) + seen.add(info.address) + if new_devices: + logger.info(f"发现 {len(new_devices)} 个新ADB设备（总计 {len(self._devices)} 个）") + elif not adb_path: + logger.info(f"发现 {len(self._devices)} 个ADB设备") return self._devices except Exception as e: logger.error(f"发现ADB设备失败: {e}") - self._devices = [] - return [] + if not self._devices: + return [] + return self._devices def connect(self, device_index: int = 0) -> bool: if not MaaFrameworkDetector.is_available(): @@ -218,6 +229,52 @@ def connect(self, device_index: int = 0) -> bool: logger.error(f"MAA Framework ADB连接失败: {e}") return False + def connect_custom(self, address: str, adb_path: str = "") -> bool: + if not MaaFrameworkDetector.is_available(): + self.is_connected = False + logger.warning("MAA Framework不可用，无法连接") + return False + + try: + Toolkit = self._ensure_toolkit() + from maa.controller import AdbController + from maa.define import MaaAdbScreencapMethodEnum, MaaAdbInputMethodEnum + + if not adb_path: + if self._devices: + adb_path = self._devices[0].adb_path + else: + adb_path = str(Path.cwd() / "platform-tools" / "adb.exe") + + self._ctrl = AdbController( + adb_path, + address, + MaaAdbScreencapMethodEnum.Default, + MaaAdbInputMethodEnum.Default, + {}, + ) + + self._ctrl.post_connection().wait() + self._ctrl.set_screenshot_use_raw_size(True) + + self._ctrl.post_screencap().wait() + image = self._ctrl.cached_image + if image is not None: + self.screen_height, self.screen_width = image.shape[:2] + else: + self.screen_width, self.screen_height = 1920, 1080 + + self.device_serial = address + self.is_connected = True + self._selected_device_index = -1 + logger.info(f"MAA Framework自定义连接成功: {address}, 分辨率: {self.screen_width}x{self.screen_height}") + return True + + except Exception as e: + self.is_connected = False + logger.error(f"MAA Framework自定义连接失败: {e}") + return False + def _get_window_size_fallback(self) -> tuple[int, int]: try: device = self.selected_device diff --git a/main.py b/main.py index 66ce0a0..4d2d74b 100644 --- a/main.py +++ b/main.py @@ -59,13 +59,19 @@ class ADBConnectorThread(QThread): connect_finished = pyqtSignal() - def __init__(self, app: "ArknightsApp", device_index: int = 0): + def __init__(self, app: "ArknightsApp", device_index: int = -1, custom_address: str = ""): super().__init__() self.app = app self.device_index = device_index + self.custom_address = custom_address def run(self): - self.app.adb_connector.connect(self.device_index) + if self.custom_address: + self.app.adb_connector.connect_custom(self.custom_address) + elif self.device_index >= 0: + self.app.adb_connector.connect(self.device_index) + else: + self.app.adb_connector.connect(0) self.connect_finished.emit() class ArknightsApp(QMainWindow): @@ -399,15 +405,19 @@ def init_ui(self): self.serial_entry = QComboBox() self.serial_entry.setEditable(True) self.serial_entry.setFixedWidth(200) - self.serial_entry.lineEdit().setPlaceholderText("选择或输入设备地址") + self.serial_entry.lineEdit().setPlaceholderText("选择设备或输入地址") self.serial_entry.currentIndexChanged.connect(self.on_device_selected) self.serial_button = QPushButton("刷新") self.serial_button.clicked.connect(self.refresh_and_connect) + self.connect_button = QPushButton("连接") + self.connect_button.clicked.connect(self.connect_custom_address) + conn_row1_layout.addWidget(self.serial_label) conn_row1_layout.addWidget(self.serial_entry) conn_row1_layout.addWidget(self.serial_button) + conn_row1_layout.addWidget(self.connect_button) # MAA状态行 self.maa_status_label = QLabel("") @@ -551,6 +561,7 @@ def on_mode_changed(self, mode): self.serial_label.setEnabled(is_adb_mode) self.serial_entry.setEnabled(is_adb_mode) self.serial_button.setEnabled(is_adb_mode) + self.connect_button.setEnabled(is_adb_mode) if mode == "ADB": self.refresh_device_list() @@ -939,10 +950,20 @@ def on_device_selected(self, index): if index < 0 or not self.adb_connector.devices: return if index < len(self.adb_connector.devices): - self.adb_connector_thread = ADBConnectorThread(self, index) + self.adb_connector_thread = ADBConnectorThread(self, device_index=index) self.adb_connector_thread.connect_finished.connect(self.on_adb_connected) self.adb_connector_thread.start() + def connect_custom_address(self): + """手动输入地址后连接（支持未知模拟器）""" + address = self.serial_entry.currentText().strip() + if not address: + QMessageBox.warning(self, "提示", "请输入设备地址，如 127.0.0.1:5555") + return + self.adb_connector_thread = ADBConnectorThread(self, custom_address=address) + self.adb_connector_thread.connect_finished.connect(self.on_adb_connected) + self.adb_connector_thread.start() + def start_callback(self): self.update_button_signal.emit("停止自动获取数据") self.update_package_button_signal.emit(False) From 6e3c36b732f409c703733022cb25e647f99f6047 Mon Sep 17 00:00:00 2001 From: AbyssRei Date: Sat, 2 May 2026 01:40:44 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat(maa=5Fadb=5Fconnector):=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E9=80=9A=E8=BF=87adb=E5=91=BD=E4=BB=A4=E6=89=AB?= =?UTF-8?q?=E6=8F=8F=E8=AE=BE=E5=A4=87=E5=B9=B6=E8=A1=A5=E5=85=85=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E5=88=97=E8=A1=A8=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- maa_adb_connector.py | 49 ++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 45 insertions(+), 4 deletions(-) diff --git a/maa_adb_connector.py b/maa_adb_connector.py index 155dd98..f46c084 100644 --- a/maa_adb_connector.py +++ b/maa_adb_connector.py @@ -201,11 +201,15 @@ def connect(self, device_index: int = 0) -> bool: self._selected_device_index = device_index device = self._devices[device_index] + from maa.define import MaaAdbScreencapMethodEnum, MaaAdbInputMethodEnum + screencap = device.screencap_methods if device.screencap_methods else MaaAdbScreencapMethodEnum.Default + input_m = device.input_methods if device.input_methods else MaaAdbInputMethodEnum.Default + self._ctrl = AdbController( device.adb_path, device.address, - device.screencap_methods, - device.input_methods, + screencap, + input_m, device.config, ) @@ -335,9 +339,46 @@ def swipe(self, start: tuple[float, float], end: tuple[float, float], duration: except Exception as e: logger.error(f"MAA滑动失败: {e}") + def _scan_adb_devices_subprocess(self) -> list[str]: + try: + adb_path = "" + if self._devices: + adb_path = self._devices[0].adb_path + if not adb_path or not Path(adb_path).exists(): + adb_path = str(Path.cwd() / "platform-tools" / "adb.exe") + if not Path(adb_path).exists(): + return [] + result = subprocess.run( + [adb_path, "devices"], + capture_output=True, text=True, timeout=5, + ) + addresses = [] + for line in result.stdout.splitlines(): + parts = line.strip().split("\t") + if len(parts) == 2 and parts[1] == "device": + addresses.append(parts[0]) + return addresses + except Exception as e: + logger.debug(f"adb devices扫描失败: {e}") + return [] + def get_device_list(self) -> list[str]: - devices = self.find_devices() - return [f"{d.name} ({d.address})" for d in devices] + self.find_devices() + seen = {d.address for d in self._devices} + extra_addresses = [a for a in self._scan_adb_devices_subprocess() if a not in seen] + for addr in extra_addresses: + adb_path = self._devices[0].adb_path if self._devices else str(Path.cwd() / "platform-tools" / "adb.exe") + self._devices.append(AdbDeviceInfo( + name="ADB Device", + adb_path=adb_path, + address=addr, + screencap_methods=0, + input_methods=0, + config={}, + )) + if extra_addresses: + logger.info(f"通过adb devices补充发现 {len(extra_addresses)} 个设备") + return [f"{d.name} ({d.address})" for d in self._devices] def update_device_serial(self, serial: str) -> str: self.device_serial = serial From fc6da393f835253f1f30ac0b022a82452768e42f Mon Sep 17 00:00:00 2001 From: AbyssRei Date: Sun, 3 May 2026 00:26:23 +0800 Subject: [PATCH 4/4] =?UTF-8?q?feat(adb=5Fconnector):=20=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E5=81=9C=E6=AD=A2ADB=E6=9C=8D=E5=8A=A1=E5=99=A8=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E5=B9=B6=E5=9C=A8=E9=80=80=E5=87=BA=E6=97=B6=E6=B8=85?= =?UTF-8?q?=E7=90=86ADB=E8=BF=9E=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 2 ++ maa_adb_connector.py | 21 +++++++++++++++++++++ main.py | 5 +++++ 3 files changed, 28 insertions(+) diff --git a/.gitignore b/.gitignore index 405ea37..c454f11 100644 --- a/.gitignore +++ b/.gitignore @@ -47,6 +47,8 @@ data/images/ data/[0-9][0-9][0-9][0-9]_[0-1][0-9]_[0-3][0-9]__[0-2][0-9]_[0-5][0-9]_[0-5][0-9]/ images/nums/ +arknights_package_[0-9][0-9][0-9][0-9][0-1][0-9][0-3][0-9]_[0-2][0-9][0-5][0-9][0-5][0-9].zip + # 训练相关数据 data_train/tmp/ data_train/package/*.zip diff --git a/maa_adb_connector.py b/maa_adb_connector.py index f46c084..e66cf85 100644 --- a/maa_adb_connector.py +++ b/maa_adb_connector.py @@ -391,3 +391,24 @@ def disconnect(self): except Exception: pass self.is_connected = False + + def stop_adb_server(self): + adb_path = "" + device = self.selected_device + if device: + adb_path = device.adb_path + elif self._devices: + adb_path = self._devices[0].adb_path + + if not adb_path or not Path(adb_path).exists(): + adb_path = str(Path.cwd() / "platform-tools" / "adb.exe") + + if not Path(adb_path).exists(): + logger.warning("未找到adb.exe，跳过关闭ADB server") + return + + try: + subprocess.run([adb_path, "kill-server"], capture_output=True, text=True, timeout=5, check=False) + logger.info("已执行 adb kill-server") + except Exception as e: + logger.warning(f"关闭ADB server失败: {e}") diff --git a/main.py b/main.py index bacb3cd..858bc20 100644 --- a/main.py +++ b/main.py @@ -1107,6 +1107,11 @@ def closeEvent(self, event): """窗口关闭时的处理""" if hasattr(self, "auto_fetch") and self.auto_fetch.auto_fetch_running: self.auto_fetch.stop_auto_fetch() + try: + self.adb_connector.disconnect() + self.adb_connector.stop_adb_server() + except Exception as e: + logger.warning(f"退出时清理ADB失败: {e}") event.accept()