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

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]:
-    """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 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())
-            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: