fix(factory): restore single-device port-filter flashing

factory_host/README.md

@@ -35,8 +35,9 @@ factory_host/
 ## 功能
 
 - 产测固件与正式固件独立保存并醒目标识。
-- 持续检测烧录串口，支持多工位自动等待设备上电后烧录。
+- 持续检测 ASR AROM 下载口，按物理 USB 设备去重，支持多工位自动等待设备上电后烧录。
 - 同一设备烧录成功后锁定，端口断开并重新出现后才允许烧录下一台。
+- 自动烧录忽略同一设备枚举出的 DBG、AT、DIAG 和运行态串口，不会为它们创建伪工位。
 - 自动识别 DBG UART，解析设备每 500 ms 广播的 FT1 状态。
 - 显示七个测试节点、IMEI、ICCID、NFC UID、失败码和实时日志。
 - 自动保存汇总 CSV 与单机 JSON，烧录和扫描过程不弹出控制台黑框。

factory_host/docs/DEVELOPMENT.md

@@ -137,6 +137,10 @@ firmware/
 还应在一台真实设备上验证：自动等待上电烧录、断开后才允许下一台、DBG UART 自动识别、
 七项 FT1 状态、GNSS 累计数据、JSON/CSV 保存，以及产测固件和正式固件切换。
 
+自动烧录端口枚举必须保留 USB 元数据，只接受 ASR AROM 的 `VID_2ECC/PID_3004`，并用
+`location`（其次为序列号、HWID、COM 名）标识物理设备。不要把运行态的 DBG、AT、DIAG
+接口交给 `adownload.exe`，否则一台设备会被误建成多个工位。
+
 ## 7. 固件替换规则
 
 - 产测与正式固件都必须是 Aboot 可直接烧录的内层 ZIP，不要选择外层 `*-target.zip`。

factory_host/docs/USAGE.md

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

factory_host/src/factory_host.py

@@ -180,29 +180,104 @@ 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]:
+@dataclass(frozen=True)
-    """Return present COM ports without requiring a third-party module."""
+class SerialPortInfo:
-    ports = []
+    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)]
+            raw = json.loads(result.stdout) if result.stdout.strip() else []
-        except (OSError, subprocess.SubprocessError):
+            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 identify_ft1_port(port: str, baudrate: int = 115200, timeout: float = 1.2) -> bool:
@@ -661,14 +736,17 @@ class FactoryHost(Tk):
         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())
+            current_devices = arom_download_ports(list_serial_port_infos())
-            for port in sorted(known_ports - current_ports):
+            for key in sorted(known_devices.keys() - current_devices.keys()):
-                self.events.put(("auto_flash_port_removed", port))
+                self.events.put(("auto_flash_port_removed", known_devices[key].device))
-            for port in sorted(current_ports - known_ports):
+            for key in sorted(current_devices.keys() - known_devices.keys()):
-                self.events.put(("auto_flash_port", port))
+                info = current_devices[key]
-            known_ports = current_ports
+                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:

factory_host/tests/test_factory_host.py

@@ -50,6 +50,61 @@ class FactoryHostTests(unittest.TestCase):
         self.assertIn("-a", args)
         self.assertIn("{port}", args)
 
+    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())
         self.assertTrue((factory_host.FACTORY_HOST_DIR / "firmware").is_dir())