diff --git a/factory_host/README.md b/factory_host/README.md index d54f0c9..2cdb3fc 100644 --- a/factory_host/README.md +++ b/factory_host/README.md @@ -35,8 +35,10 @@ factory_host/ ## 功能 - 产测固件与正式固件独立保存并醒目标识。 -- 持续检测烧录串口，支持多工位自动等待设备上电后烧录。 +- 持续检测 ASR AROM 下载口，按物理 USB 设备去重，支持多工位自动等待设备上电后烧录。 +- 点击自动烧录后立即启动一个 `adownload.exe -u -a` 监听进程，在设备上电前等待 AROM；多台设备共享该进程。 - 同一设备烧录成功后锁定，端口断开并重新出现后才允许烧录下一台。 +- 自动烧录忽略同一设备枚举出的 DBG、AT、DIAG 和运行态串口，不会为它们创建伪工位。 - 自动识别 DBG UART，解析设备每 500 ms 广播的 FT1 状态。 - 显示七个测试节点、IMEI、ICCID、NFC UID、失败码和实时日志。 - 自动保存汇总 CSV 与单机 JSON，烧录和扫描过程不弹出控制台黑框。 diff --git a/factory_host/config/factory_host_config.json b/factory_host/config/factory_host_config.json index 85574bd..96b3fac 100644 --- a/factory_host/config/factory_host_config.json +++ b/factory_host/config/factory_host_config.json @@ -11,12 +11,10 @@ "{firmware}" ], "auto_args": [ - "-p", - "{port}", + "-u", "-a", "-s", "{baudrate}", - "-q", "{firmware}" ], "timeout_seconds": 900, diff --git a/factory_host/docs/DEVELOPMENT.md b/factory_host/docs/DEVELOPMENT.md index 1625b22..6e9df39 100644 --- a/factory_host/docs/DEVELOPMENT.md +++ b/factory_host/docs/DEVELOPMENT.md @@ -137,6 +137,15 @@ firmware/ 还应在一台真实设备上验证：自动等待上电烧录、断开后才允许下一台、DBG UART 自动识别、 七项 FT1 状态、GNSS 累计数据、JSON/CSV 保存，以及产测固件和正式固件切换。 +自动烧录端口枚举必须保留 USB 元数据，只接受 ASR AROM 的 `VID_2ECC/PID_3004`，并用 +`location`（其次为序列号、HWID、COM 名）标识物理设备。不要把运行态的 DBG、AT、DIAG +接口交给 `adownload.exe`，否则一台设备会被误建成多个工位。 + +多工位烧录在设备上电前启动单个 `adownload.exe -u -a` 进程。不要等待 Python 枚举到 COM 口后 +才启动下载器，也不要为每个端口分别启动下载器；自动参数中不能加入 `-q`，否则任一端口完成后 +监听进程就会退出。`flash.workers` 只控制界面预建的工位数。 +不要仅为并发烧录加入 `-m`；它会启用固件包中标记为 `productionOnly` 的分区操作。 + ## 7. 固件替换规则 - 产测与正式固件都必须是 Aboot 可直接烧录的内层 ZIP，不要选择外层 `*-target.zip`。 diff --git a/factory_host/docs/USAGE.md b/factory_host/docs/USAGE.md index 81d6373..2f16ac6 100644 --- a/factory_host/docs/USAGE.md +++ b/factory_host/docs/USAGE.md @@ -66,13 +66,20 @@ factory_host/dist/ 1. 启动 `START-EG800-FACTORY-TOOL.exe`，固件类型选择“产测固件”或“正式固件”，确认顶部完整路径。 2. 连接一个或多个烧录器，设备保持关机。 -3. 点击醒目的“启动自动烧录”，工具每 0.5 秒持续检测烧录串口，并为每个串口建立工位。 +3. 点击醒目的“启动自动烧录”。上位机会立刻启动唯一一个 `adownload.exe -u -a` 进程，先进入等待 AROM 状态。 + 串口扫描只负责建立工位和显示状态，不会延迟下载器启动；多台设备由同一下载器进程处理。 4. 工位显示“烧录中·等待上电”后给对应设备开机，设备进入 AROM 后自动开始烧录。 5. 工位显示“烧录成功”后先断开当前设备；工具检测到端口消失后才会允许下一次烧录，避免同一设备反复烧录。 6. 换上下一台关机设备并上电，端口重新出现后自动开始下一次烧录。 7. 全部生产完成后点击“停止烧录”；不要在实际写入过程中拔线或关闭程序。 默认参数使用 `adownload.exe -u -a` 自动等待 AROM 下载设备，因此烧录口和产测 UART 是两个不同角色。 +界面只把 `VID_2ECC/PID_3004` 的 AROM 下载口识别为工位，并按 USB 物理位置去重；同一设备的 +DBG、AT、DIAG 及运行态串口不会创建额外工位。启动监听前已经存在的端口作为基线，按标准流程 +先启动自动烧录，再给待烧录设备上电。 + +开始前必须关闭单独打开的 `aboot.exe` 以及其他 `adownload.exe`。新版会检测并提示这些冲突进程， +因为多个下载器同时轮询 AROM 口会导致配置解析失败或端口占用。 ## 4. 自动识别并开始产测 diff --git a/factory_host/src/factory_host.py b/factory_host/src/factory_host.py index 02d4274..03698ab 100644 --- a/factory_host/src/factory_host.py +++ b/factory_host/src/factory_host.py @@ -85,7 +85,7 @@ DEFAULT_CONFIG = { "aboot_path": ".runtime/burn_tool/aboot-tools-2024.08.08-win-x64/aboot.exe", "downloader_path": ".runtime/burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe", "args": ["-u", "-a", "-s", "{baudrate}", "-q", "{firmware}"], - "auto_args": ["-p", "{port}", "-a", "-s", "{baudrate}", "-q", "{firmware}"], + "auto_args": ["-u", "-a", "-s", "{baudrate}", "{firmware}"], "timeout_seconds": 900, "workers": 2, }, @@ -180,29 +180,185 @@ def parse_command_template(template: str) -> list[str]: return [part.strip('"') for part in shlex.split(template, posix=False)] -def list_serial_ports() -> list[str]: - """Return present COM ports without requiring a third-party module.""" - ports = [] +@dataclass(frozen=True) +class SerialPortInfo: + device: str + description: str = "" + hwid: str = "" + vid: int | None = None + pid: int | None = None + serial_number: str = "" + location: str = "" + interface: str = "" + + +def _com_sort_key(port: str) -> tuple[int, str]: + suffix = port[3:] if port.upper().startswith("COM") else "" + return (int(suffix), port.upper()) if suffix.isdigit() else (sys.maxsize, port.upper()) + + +def list_serial_port_infos() -> list[SerialPortInfo]: + """Return present serial ports with USB identity used by auto flashing.""" + ports: list[SerialPortInfo] = [] try: import serial.tools.list_ports # type: ignore - ports = [item.device for item in serial.tools.list_ports.comports()] + for item in serial.tools.list_ports.comports(): + ports.append(SerialPortInfo( + device=str(item.device or ""), + description=str(item.description or ""), + hwid=str(item.hwid or ""), + vid=item.vid, + pid=item.pid, + serial_number=str(item.serial_number or ""), + location=str(item.location or ""), + interface=str(item.interface or ""), + )) except ImportError: pass if not ports and os.name == "nt": try: + command = ( + "Get-CimInstance Win32_SerialPort | " + "Select-Object DeviceID,Name,PNPDeviceID | ConvertTo-Json -Compress" + ) result = subprocess.run( - ["powershell.exe", "-NoProfile", "-Command", "Get-CimInstance Win32_SerialPort | Select-Object -ExpandProperty DeviceID"], + ["powershell.exe", "-NoProfile", "-Command", command], capture_output=True, text=True, timeout=5, check=False, creationflags=CREATE_NO_WINDOW, ) - ports = [line.strip() for line in result.stdout.splitlines() if re.fullmatch(r"COM\d+", line.strip(), re.I)] - except (OSError, subprocess.SubprocessError): + raw = json.loads(result.stdout) if result.stdout.strip() else [] + rows = raw if isinstance(raw, list) else [raw] + for row in rows: + device = str(row.get("DeviceID") or "") + if re.fullmatch(r"COM\d+", device, re.I): + ports.append(SerialPortInfo( + device=device, + description=str(row.get("Name") or ""), + hwid=str(row.get("PNPDeviceID") or ""), + )) + except (OSError, subprocess.SubprocessError, json.JSONDecodeError): pass - return sorted(set(ports), key=lambda item: int(item[3:]) if item[3:].isdigit() else item) + unique = {item.device.upper(): item for item in ports if item.device} + return sorted(unique.values(), key=lambda item: _com_sort_key(item.device)) + + +def list_serial_ports() -> list[str]: + """Return present COM names for FT1 probing and manual selection.""" + return [item.device for item in list_serial_port_infos()] + + +def is_arom_download_port(info: SerialPortInfo) -> bool: + """Accept only the ASR BootROM USB serial interface used by adownload.""" + if info.vid == 0x2ECC and info.pid == 0x3004: + return True + return "VID_2ECC&PID_3004" in info.hwid.upper() + + +def download_device_key(info: SerialPortInfo) -> str: + """Return a stable physical-device key so one USB device creates one station.""" + if info.location: + location = re.sub(r"(?i)(?:[:.]\d+|&MI_\d+)$", "", info.location.strip()) + return f"location:{location.casefold()}" + if info.serial_number: + return f"serial:{info.serial_number.strip().casefold()}" + hwid = re.sub(r"(?i)&MI_[0-9A-F]{2}", "", info.hwid.strip()) + if hwid: + return f"hwid:{hwid.casefold()}" + return f"port:{info.device.casefold()}" + + +def arom_download_ports(infos: list[SerialPortInfo]) -> dict[str, SerialPortInfo]: + """Return one AROM interface per physical USB device.""" + result: dict[str, SerialPortInfo] = {} + for info in sorted(infos, key=lambda item: _com_sort_key(item.device)): + if is_arom_download_port(info): + result.setdefault(download_device_key(info), info) + return result + + +def classify_downloader_line(line: str) -> str | None: + """Classify per-device completion lines from the persistent downloader.""" + lowered = line.casefold() + if "all finished. total time:" in lowered: + return "烧录成功" + benign = "offline device open failed num = 0" in lowered + if not benign and any(marker in lowered for marker in ( + "failed to parse", "syntax error", "disconnected from serial device", + "processing command", "download failed", + )): + if "failed" in lowered or "error" in lowered: + return "失败" + return None + + +def resolve_downloader_result_port( + line_port: str, + current_port: str, + port_activity: dict[str, int], + finished_ports: set[str]) -> str: + """Bind an untagged downloader result to an unfinished active port.""" + if line_port: + return line_port if line_port not in finished_ports else "" + if current_port and current_port not in finished_ports: + return current_port + unfinished = { + port: activity + for port, activity in port_activity.items() + if port not in finished_ports + } + return max(unfinished, key=unfinished.get, default="") + + +def running_vendor_downloaders() -> list[tuple[str, int]]: + """Return external vendor downloader processes that can claim AROM ports.""" + if os.name != "nt": + return [] + try: + import ctypes + from ctypes import wintypes + + class ProcessEntry(ctypes.Structure): + _fields_ = [ + ("dwSize", wintypes.DWORD), ("cntUsage", wintypes.DWORD), + ("th32ProcessID", wintypes.DWORD), ("th32DefaultHeapID", ctypes.c_size_t), + ("th32ModuleID", wintypes.DWORD), ("cntThreads", wintypes.DWORD), + ("th32ParentProcessID", wintypes.DWORD), ("pcPriClassBase", wintypes.LONG), + ("dwFlags", wintypes.DWORD), ("szExeFile", wintypes.WCHAR * 260), + ] + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + create_snapshot = kernel32.CreateToolhelp32Snapshot + create_snapshot.argtypes = [wintypes.DWORD, wintypes.DWORD] + create_snapshot.restype = wintypes.HANDLE + process_first = kernel32.Process32FirstW + process_first.argtypes = [wintypes.HANDLE, ctypes.POINTER(ProcessEntry)] + process_first.restype = wintypes.BOOL + process_next = kernel32.Process32NextW + process_next.argtypes = [wintypes.HANDLE, ctypes.POINTER(ProcessEntry)] + process_next.restype = wintypes.BOOL + snapshot = create_snapshot(0x00000002, 0) + if snapshot == wintypes.HANDLE(-1).value: + return [] + targets = {"aboot.exe", "adownload.exe"} + found: list[tuple[str, int]] = [] + try: + entry = ProcessEntry() + entry.dwSize = ctypes.sizeof(entry) + available = bool(process_first(snapshot, ctypes.byref(entry))) + while available: + name = entry.szExeFile + if name.casefold() in targets: + found.append((name, int(entry.th32ProcessID))) + available = bool(process_next(snapshot, ctypes.byref(entry))) + finally: + kernel32.CloseHandle(snapshot) + return found + except (AttributeError, OSError): + return [] def identify_ft1_port(port: str, baudrate: int = 115200, timeout: float = 1.2) -> bool: @@ -387,7 +543,7 @@ class FactoryHost(Tk): advanced = ttk.LabelFrame(self, text="高级操作", padding=(8, 4)) advanced.pack(fill=X, padx=8, pady=(0, 6)) ttk.Button(advanced, text="烧录选中", command=self.start_selected_flash).pack(side=LEFT, padx=3) - ttk.Button(advanced, text="批量烧录", command=self.start_flash).pack(side=LEFT, padx=3) + ttk.Button(advanced, text="手动批量（串行）", command=self.start_flash).pack(side=LEFT, padx=3) ttk.Button(advanced, text="添加工位", command=self.add_station).pack(side=LEFT, padx=3) ttk.Button(advanced, text="删除选中", command=self.remove_station).pack(side=LEFT, padx=3) ttk.Button(advanced, text="打开配置", command=self.open_config).pack(side=LEFT, padx=3) @@ -630,6 +786,15 @@ class FactoryHost(Tk): if self.flash_running or self.scan_running: self._log("系统", "已有烧录或串口扫描任务正在进行") return + conflicts = running_vendor_downloaders() + if conflicts: + details = "、".join(f"{name} (PID {pid})" for name, pid in conflicts) + messagebox.showerror( + "下载器冲突", + "检测到其他 Aboot 下载程序仍在运行，会抢占烧录口。\n" + f"请先在任务管理器中关闭：{details}", + ) + return downloader = find_downloader( self.config_data["flash"].get("downloader_path", ""), self.config_data["flash"].get("aboot_path", ""), @@ -647,7 +812,9 @@ class FactoryHost(Tk): self.flash_cancel.clear() self.flash_running = True self.log_follow = True - self.executor = ThreadPoolExecutor(max_workers=workers) + # Start one downloader before any device powers on. It owns all AROM + # ports and stays alive until the operator stops automatic flashing. + self.executor = ThreadPoolExecutor(max_workers=1) self.auto_flash_button.configure(text="自动烧录监听中") waiting_stations = list(self.stations.values())[:workers] while len(waiting_stations) < workers: @@ -657,18 +824,25 @@ class FactoryHost(Tk): station.status = "自动等待上电" station.message = "持续检测烧录串口" self._update_row(station) - self._log("系统", f"自动烧录已启动：每 0.5 秒检测串口，等待设备上电，并发工位 {workers}") + self._log( + "系统", + f"自动下载器已提前启动并等待设备上电；同一进程支持最多 {workers} 个显示工位", + ) + self.executor.submit(self._auto_flash_engine_worker, downloader) threading.Thread(target=self._auto_flash_monitor_worker, daemon=True).start() def _auto_flash_monitor_worker(self) -> None: - known_ports: set[str] = set() + # Ports already present when monitoring starts are a baseline. Operators + # start monitoring first, then power on a device and expose its AROM port. + known_devices = arom_download_ports(list_serial_port_infos()) while not self.auto_monitor_stop.wait(0.5): - current_ports = set(list_serial_ports()) - for port in sorted(known_ports - current_ports): - self.events.put(("auto_flash_port_removed", port)) - for port in sorted(current_ports - known_ports): - self.events.put(("auto_flash_port", port)) - known_ports = current_ports + current_devices = arom_download_ports(list_serial_port_infos()) + for key in sorted(known_devices.keys() - current_devices.keys()): + self.events.put(("auto_flash_port_removed", known_devices[key].device)) + for key in sorted(current_devices.keys() - known_devices.keys()): + info = current_devices[key] + self.events.put(("auto_flash_port", info.device)) + known_devices = current_devices self.events.put(("auto_flash_stopped",)) def _submit_auto_flash_port(self, port: str) -> None: @@ -688,13 +862,109 @@ class FactoryHost(Tk): station = self.add_station() station.flash_port = port station.firmware_kind = self.auto_firmware_kind - station.status = "排队中" - station.message = f"检测到新烧录串口 {port}" + station.status = "烧录中" + station.message = f"下载器已捕获 AROM 下载口 {port}" self.flash_jobs_active += 1 self._update_row(station) - assert self.executor is not None and self.auto_downloader is not None - self.executor.submit(self._flash_one, station, self.auto_downloader, True) - self._log(station.number, f"检测到烧录串口 {port}，已进入等待上电状态") + self._log(station.number, f"检测到 AROM 下载口 {port}，正在烧录") + + def _auto_flash_engine_worker(self, downloader: Path) -> None: + firmware = resolve_project_path( + self.config_data["firmwares"].get(self.auto_firmware_kind, "")) + if not firmware.is_file(): + self.events.put(("auto_flash_engine_stopped", 1, f"固件不存在: {firmware}")) + return + configured_args = self.config_data["flash"].get("auto_args", ["{firmware}"]) + if isinstance(configured_args, str): + configured_args = parse_command_template(configured_args) + baudrate = int(self.config_data["serial"].get("baudrate", 115200)) + args = expand_args( + configured_args, aboot=downloader, firmware=firmware, + port="", station="auto", baudrate=baudrate) + command = [str(downloader)] + args + command_text = " ".join(shlex.quote(item) for item in command) + self.events.put(("log", "系统", "自动下载命令: " + command_text)) + output_tail: list[str] = [] + current_port = "" + port_activity: dict[str, int] = {} + activity_sequence = 0 + failed_ports: set[str] = set() + finished_ports: set[str] = set() + return_code = 1 + try: + popen_kwargs = { + "cwd": str(downloader.parent), + "stdout": subprocess.PIPE, + "stderr": subprocess.STDOUT, + "text": True, + "encoding": "utf-8", + "errors": "replace", + "bufsize": 1, + } + if os.name == "nt": + popen_kwargs["creationflags"] = CREATE_NO_WINDOW + process = subprocess.Popen(command, **popen_kwargs) + assert process.stdout is not None + lines: queue.Queue[str | None] = queue.Queue() + + def read_output() -> None: + try: + for raw_line in process.stdout: + lines.put(raw_line.rstrip()) + finally: + lines.put(None) + + threading.Thread( + target=read_output, name="flash-output-auto", daemon=True).start() + reader_done = False + while not reader_done or process.poll() is None: + try: + line = lines.get(timeout=0.1) + if line is None: + reader_done = True + elif line: + output_tail.append(line) + output_tail = output_tail[-4:] + match = re.search(r"<(COM\d+)>", line, re.I) + line_port = "" + if match: + line_port = match.group(1).upper() + current_port = line_port + activity_sequence += 1 + port_activity[current_port] = activity_sequence + if "connected to serial device" in line.casefold(): + failed_ports.discard(current_port) + finished_ports.discard(current_port) + self.events.put(("auto_flash_engine_line", current_port, line)) + result = classify_downloader_line(line) + result_port = resolve_downloader_result_port( + line_port, current_port, port_activity, finished_ports) + if result == "失败" and result_port: + failed_ports.add(result_port) + finished_ports.add(result_port) + self.events.put(( + "auto_flash_engine_result", result_port, "失败", line)) + elif (result == "烧录成功" and result_port + and result_port not in failed_ports): + finished_ports.add(result_port) + self.events.put(( + "auto_flash_engine_result", result_port, "烧录成功", line)) + except queue.Empty: + pass + if self.flash_cancel.is_set() and process.poll() is None: + process.terminate() + break + if self.flash_cancel.is_set() and process.poll() is None: + process.terminate() + try: + return_code = process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + return_code = process.wait(timeout=5) + except (OSError, subprocess.SubprocessError) as exc: + output_tail.append(str(exc)) + message = "用户停止" if self.flash_cancel.is_set() else " | ".join(output_tail[-2:]) + self.events.put(("auto_flash_engine_stopped", return_code, message)) def _start_flash_jobs(self, selected, auto_detect: bool = False) -> None: if self.auto_flash_monitoring and not auto_detect: @@ -704,6 +974,15 @@ class FactoryHost(Tk): if not downloader: messagebox.showerror("无法烧录", "未找到 adownload.exe。请检查发布包的 .runtime/burn_tool，或开发目录的 factory_host/vendor。aboot.exe 是图形界面，不能作为命令行烧录进程。") return + conflicts = running_vendor_downloaders() + if conflicts: + details = "、".join(f"{name} (PID {pid})" for name, pid in conflicts) + messagebox.showerror( + "下载器冲突", + "检测到其他 Aboot 下载程序仍在运行，会抢占烧录口。\n" + f"请先在任务管理器中关闭：{details}", + ) + return jobs = [ self.stations[iid] for iid in selected @@ -1046,6 +1325,43 @@ class FactoryHost(Tk): if station and station.status == "烧录成功": station.message = "设备已断开，等待下一台设备上电" self._update_row(station) + elif kind == "auto_flash_engine_line": + _, port, line = event + station = next((item for item in self.stations.values() + if item.flash_port.upper() == port), None) if port else None + if port and station is None and self.auto_flash_monitoring: + self._submit_auto_flash_port(port) + station = next((item for item in self.stations.values() + if item.flash_port.upper() == port), None) + self._log(station.number if station else "下载器", line) + elif kind == "auto_flash_engine_result": + _, port, status, message = event + station = next((item for item in self.stations.values() + if item.flash_port.upper() == port), None) + if station is None and self.auto_flash_monitoring: + self._submit_auto_flash_port(port) + station = next((item for item in self.stations.values() + if item.flash_port.upper() == port), None) + if station: + self.events.put(("station", station.number, status, message)) + elif kind == "auto_flash_engine_stopped": + _, return_code, message = event + if self.flash_cancel.is_set(): + self.flash_running = False + self._log("系统", "自动下载器已停止") + for station in self.stations.values(): + if station.status in ("烧录中", "烧录中·等待上电"): + self.events.put(( + "station", station.number, "已取消", "用户停止")) + elif self.auto_flash_monitoring: + self.auto_flash_monitoring = False + self.flash_running = False + self.auto_flash_button.configure(text="启动自动烧录") + self._log("系统", f"自动下载器异常退出 {return_code}: {message}") + for station in self.stations.values(): + if station.status in ("烧录中", "烧录中·等待上电"): + self.events.put(( + "station", station.number, "失败", "自动下载器异常退出")) elif kind == "auto_flash_stopped": self._log("系统", "自动烧录串口监听已停止") except queue.Empty: diff --git a/factory_host/tests/test_factory_host.py b/factory_host/tests/test_factory_host.py index c1e62fc..b72e923 100644 --- a/factory_host/tests/test_factory_host.py +++ b/factory_host/tests/test_factory_host.py @@ -44,11 +44,104 @@ class FactoryHostTests(unittest.TestCase): self.assertIn("-a", args) self.assertNotIn("{port}", args) - def test_auto_flash_binds_each_detected_port(self): + def test_auto_flash_starts_one_usb_listener_before_power_on(self): args = factory_host.DEFAULT_CONFIG["flash"]["auto_args"] - self.assertIn("-p", args) + self.assertIn("-u", args) self.assertIn("-a", args) - self.assertIn("{port}", args) + self.assertNotIn("-q", args) + self.assertNotIn("-m", args) + self.assertNotIn("{port}", args) + + def test_auto_flash_classifies_persistent_downloader_results(self): + self.assertEqual( + factory_host.classify_downloader_line( + " all finished. total time: 23.500s"), + "烧录成功", + ) + self.assertEqual( + factory_host.classify_downloader_line("failed to parse configuration"), + "失败", + ) + self.assertIsNone(factory_host.classify_downloader_line( + " OFFLINE device open failed num = 0, device list size = 0")) + + def test_auto_flash_assigns_untagged_result_to_unfinished_port(self): + activity = {"COM38": 8, "COM8": 9} + self.assertEqual( + factory_host.resolve_downloader_result_port( + "", "COM8", activity, {"COM8"}), + "COM38", + ) + self.assertEqual( + factory_host.resolve_downloader_result_port( + "COM38", "COM8", activity, {"COM8"}), + "COM38", + ) + self.assertEqual( + factory_host.resolve_downloader_result_port( + "COM8", "COM8", activity, {"COM8"}), + "", + ) + self.assertEqual( + factory_host.resolve_downloader_result_port( + "", "COM8", activity, {"COM8", "COM38"}), + "", + ) + + def test_auto_flash_accepts_only_asr_arom_port(self): + arom = factory_host.SerialPortInfo( + device="COM8", hwid=r"USB\VID_2ECC&PID_3004\1234", vid=0x2ECC, pid=0x3004) + arom_from_hwid = factory_host.SerialPortInfo( + device="COM22", hwid=r"USB\VID_2ECC&PID_3004\5678") + runtime = factory_host.SerialPortInfo( + device="COM13", hwid=r"USB\VID_2ECC&PID_3017&MI_00\1234", vid=0x2ECC, pid=0x3017, + description="ASR Serial Download Device") + unrelated = factory_host.SerialPortInfo( + device="COM3", hwid=r"USB\VID_1A86&PID_7523\5678", vid=0x1A86, pid=0x7523) + misleading = factory_host.SerialPortInfo( + device="COM4", description="Generic BootROM Serial Device") + self.assertTrue(factory_host.is_arom_download_port(arom)) + self.assertTrue(factory_host.is_arom_download_port(arom_from_hwid)) + self.assertFalse(factory_host.is_arom_download_port(runtime)) + self.assertFalse(factory_host.is_arom_download_port(unrelated)) + self.assertFalse(factory_host.is_arom_download_port(misleading)) + + def test_auto_flash_ignores_eg800_runtime_interfaces(self): + roles = ( + ("COM13", 0x03, "Quectel USB AT Port"), + ("COM14", 0x05, "USB Serial Device"), + ("COM15", 0x04, "Quectel USB Modem"), + ("COM16", 0x02, "Quectel USB DIAG Port"), + ) + for device, interface, description in roles: + info = factory_host.SerialPortInfo( + device=device, + description=description, + hwid=rf"USB\VID_2C7C&PID_6002&MI_{interface:02X}\1234", + vid=0x2C7C, + pid=0x6002, + ) + with self.subTest(device=device): + self.assertFalse(factory_host.is_arom_download_port(info)) + + def test_auto_flash_groups_interfaces_by_physical_location(self): + first = factory_host.SerialPortInfo( + device="COM8", vid=0x2ECC, pid=0x3004, location="1-4:1.0") + second = factory_host.SerialPortInfo( + device="COM9", vid=0x2ECC, pid=0x3004, location="1-4:1.1") + ports = factory_host.arom_download_ports([second, first]) + self.assertEqual(len(ports), 1) + self.assertEqual(next(iter(ports.values())).device, "COM8") + + def test_auto_flash_port_identity_survives_com_number_change(self): + before = factory_host.SerialPortInfo( + device="COM8", vid=0x2ECC, pid=0x3004, location="Port_#0004.Hub_#0001") + after = factory_host.SerialPortInfo( + device="COM21", vid=0x2ECC, pid=0x3004, location="Port_#0004.Hub_#0001") + self.assertEqual( + factory_host.download_device_key(before), + factory_host.download_device_key(after), + ) def test_project_root_and_relative_path(self): self.assertTrue((factory_host.FACTORY_HOST_DIR / "vendor" / "aboot-tools-2024.08.08-win-x64").is_dir())