fix(factory): stabilize multi-device automatic flashing

factory_host/README.md

@@ -36,6 +36,7 @@ factory_host/
 
 - 产测固件与正式固件独立保存并醒目标识。
 - 持续检测 ASR AROM 下载口，按物理 USB 设备去重，支持多工位自动等待设备上电后烧录。
+- 点击自动烧录后立即启动一个 `adownload.exe -u -a` 监听进程，在设备上电前等待 AROM；多台设备共享该进程。
 - 同一设备烧录成功后锁定，端口断开并重新出现后才允许烧录下一台。
 - 自动烧录忽略同一设备枚举出的 DBG、AT、DIAG 和运行态串口，不会为它们创建伪工位。
 - 自动识别 DBG UART，解析设备每 500 ms 广播的 FT1 状态。

factory_host/config/factory_host_config.json

@@ -11,12 +11,10 @@
       "{firmware}"
     ],
     "auto_args": [
-      "-p",
+      "-u",
-      "{port}",
       "-a",
       "-s",
       "{baudrate}",
-      "-q",
       "{firmware}"
     ],
     "timeout_seconds": 900,

factory_host/docs/DEVELOPMENT.md

@@ -141,6 +141,11 @@ firmware/
 `location`（其次为序列号、HWID、COM 名）标识物理设备。不要把运行态的 DBG、AT、DIAG
 接口交给 `adownload.exe`，否则一台设备会被误建成多个工位。
 
+多工位烧录在设备上电前启动单个 `adownload.exe -u -a` 进程。不要等待 Python 枚举到 COM 口后
+才启动下载器，也不要为每个端口分别启动下载器；自动参数中不能加入 `-q`，否则任一端口完成后
+监听进程就会退出。`flash.workers` 只控制界面预建的工位数。
+不要仅为并发烧录加入 `-m`；它会启用固件包中标记为 `productionOnly` 的分区操作。
+
 ## 7. 固件替换规则
 
 - 产测与正式固件都必须是 Aboot 可直接烧录的内层 ZIP，不要选择外层 `*-target.zip`。

factory_host/docs/USAGE.md

@@ -66,17 +66,21 @@ factory_host/dist/
 
 1. 启动 `START-EG800-FACTORY-TOOL.exe`，固件类型选择“产测固件”或“正式固件”，确认顶部完整路径。
 2. 连接一个或多个烧录器，设备保持关机。
-3. 点击醒目的“启动自动烧录”，工具每 0.5 秒持续检测 ASR AROM 下载口，并为每台物理设备建立一个工位。
+3. 点击醒目的“启动自动烧录”。上位机会立刻启动唯一一个 `adownload.exe -u -a` 进程，先进入等待 AROM 状态。
+   串口扫描只负责建立工位和显示状态，不会延迟下载器启动；多台设备由同一下载器进程处理。
 4. 工位显示“烧录中·等待上电”后给对应设备开机，设备进入 AROM 后自动开始烧录。
 5. 工位显示“烧录成功”后先断开当前设备；工具检测到端口消失后才会允许下一次烧录，避免同一设备反复烧录。
 6. 换上下一台关机设备并上电，端口重新出现后自动开始下一次烧录。
 7. 全部生产完成后点击“停止烧录”；不要在实际写入过程中拔线或关闭程序。
 
 默认参数使用 `adownload.exe -u -a` 自动等待 AROM 下载设备，因此烧录口和产测 UART 是两个不同角色。
-自动烧录只接受 `VID_2ECC/PID_3004` 的 AROM 下载口，并按 USB 物理位置去重；同一设备的
+界面只把 `VID_2ECC/PID_3004` 的 AROM 下载口识别为工位，并按 USB 物理位置去重；同一设备的
 DBG、AT、DIAG 及运行态串口不会创建额外工位。启动监听前已经存在的端口作为基线，按标准流程
 先启动自动烧录，再给待烧录设备上电。
 
+开始前必须关闭单独打开的 `aboot.exe` 以及其他 `adownload.exe`。新版会检测并提示这些冲突进程，
+因为多个下载器同时轮询 AROM 口会导致配置解析失败或端口占用。
+
 ## 4. 自动识别并开始产测
 
 烧录完成、设备正常启动并连接调试 UART 后：

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,
     },
@@ -280,6 +280,87 @@ def arom_download_ports(infos: list[SerialPortInfo]) -> dict[str, SerialPortInfo
     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:
     """Passively probe a DBG UART without assigning unrelated ports."""
     try:
@@ -462,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)
@@ -705,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", ""),
@@ -722,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:
@@ -732,7 +824,11 @@ 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:
@@ -766,13 +862,109 @@ class FactoryHost(Tk):
             station = self.add_station()
         station.flash_port = port
         station.firmware_kind = self.auto_firmware_kind
-        station.status = "排队中"
+        station.status = "烧录中"
-        station.message = f"检测到新烧录串口 {port}"
+        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._log(station.number, f"检测到 AROM 下载口 {port}，正在烧录")
-        self.executor.submit(self._flash_one, station, self.auto_downloader, True)
+
-        self._log(station.number, f"检测到烧录串口 {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:
@@ -782,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
@@ -1124,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:

factory_host/tests/test_factory_host.py

@@ -44,11 +44,49 @@ 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(
+                "<COM8> all finished. total time: 23.500s"),
+            "烧录成功",
+        )
+        self.assertEqual(
+            factory_host.classify_downloader_line("failed to parse configuration"),
+            "失败",
+        )
+        self.assertIsNone(factory_host.classify_downloader_line(
+            "<COM8> 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(