chenjintao/banban
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: automate FT1 factory test status reporting

Browse Source
This commit is contained in:
chenjintao
2026-09-24 11:21:40 +08:00
parent 8b499948dd
commit 06b0eaaf70
11 changed files with 461 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
factory_host/factory_host.py
+90 -18
View File
@@ -158,6 +158,32 @@ def list_serial_ports() -> list[str]:
return sorted(set(ports), key=lambda item: int(item[3:]) if item[3:].isdigit() else item)
def identify_ft1_port(port: str, baudrate: int = 115200, timeout: float = 1.2) -> bool:
"""Passively probe a DBG UART without assigning unrelated ports."""
try:
import serial # type: ignore
with serial.Serial(port, baudrate, timeout=0.15) as stream:
stream.reset_input_buffer()
deadline = time.monotonic() + timeout
pending = bytearray()
while time.monotonic() < deadline:
chunk = stream.read(256)
if not chunk:
continue
pending.extend(chunk)
while b"\n" in pending:
raw, _, pending = pending.partition(b"\n")
try:
message = parse_ft1_line(raw.rstrip(b"\r").decode("utf-8", errors="replace"))
except ProtocolError:
continue
if message and message.get("type") in {"device_ready", "session_start", "status_snapshot", "node_start", "node_result", "session_result"}:
return True
except (ImportError, OSError, ValueError):
return False
return False
class SerialReader:
"""Small Windows serial reader using pyserial when available, file handles otherwise."""
@@ -170,6 +196,7 @@ class SerialReader:
self._thread: threading.Thread | None = None
self._stream = None
self._pyserial = None
self._opened = threading.Event()
def start(self) -> None:
if self._thread and self._thread.is_alive():
@@ -189,6 +216,7 @@ class SerialReader:
self._thread.join(timeout=1.0)
def send(self, text: str, line_ending: str = "\r\n") -> None:
self._opened.wait(timeout=2.0)
payload = (text + line_ending).encode(self.encoding, errors="replace")
if self._pyserial is not None:
self._pyserial.write(payload)
@@ -207,6 +235,7 @@ class SerialReader:
subprocess.run(["mode", self.port, f"BAUD={self.baudrate}", "PARITY=n", "DATA=8", "STOP=1"], capture_output=True, check=False)
device = self.port if self.port.startswith("\\\\.\\") else "\\\\.\\" + self.port
self._stream = open(device, "r+b", buffering=0)
self._opened.set()
buffer = bytearray()
while not self._stop.is_set():
chunk = self._stream.read(256)
@@ -222,6 +251,7 @@ class SerialReader:
except Exception as exc: # serial backends surface OS-specific errors
self.on_line(f"[SERIAL_ERROR] {exc}")
finally:
self._opened.clear()
try:
if self._stream is not None:
self._stream.close()
@@ -342,11 +372,9 @@ class FactoryHost(Tk):
self._refresh_tool_status()
def _load_default_rows(self) -> None:
ports = list_serial_ports()
for index, port in enumerate(ports, 1):
self.add_station(number=str(index), flash_port=port, test_port=port)
if not ports:
self.add_station(number="1")
# COM enumeration is intentionally deferred until FT1 identification;
# AROM/DIAG/AT ports must not become fake stations.
self.add_station(number="1")
def _refresh_tool_status(self) -> None:
downloader = find_downloader(self.config_data["flash"].get("downloader_path", ""), self.config_data["flash"].get("aboot_path", ""))
@@ -364,14 +392,34 @@ class FactoryHost(Tk):
self._refresh_tool_status()
def scan_ports(self) -> None:
ports = list_serial_ports()
self._log("系统", f"检测到串口: {', '.join(ports) if ports else '无'}")
for station in self.stations.values():
if not station.flash_port and ports:
station.flash_port = ports.pop(0)
if not station.test_port:
station.test_port = station.flash_port
self._update_row(station)
found = self._auto_add_ft1_stations()
self._log("系统", f"FT1 自动识别完成，新增/确认工位: {', '.join(found) if found else '无'}")
def _auto_add_ft1_stations(self) -> list[str]:
"""Probe every COM port and create one visible station per FT1 device."""
baudrate = int(self.config_data["serial"].get("baudrate", 115200))
used = {station.test_port for station in self.stations.values() if station.test_port}
found: list[str] = []
for candidate in list_serial_ports():
if candidate in used:
continue
if not identify_ft1_port(candidate, baudrate):
continue
station = next((item for item in self.stations.values()
if not item.test_port and item.reader is None), None)
if station is None:
number = 1
while str(number) in self.stations:
number += 1
self.add_station(number=str(number))
station = self.stations[str(number)]
station.test_port = candidate
station.status = "已识别"
station.message = "FT1 UART 已连接，等待状态广播"
used.add(candidate)
found.append(f"工位 {station.number}={candidate}")
self._update_row(station)
return found
def open_config(self) -> None:
try:
@@ -417,7 +465,7 @@ class FactoryHost(Tk):
ttk.Label(dialog, text="烧录串口（自动 USB 模式可空）:").grid(row=0, column=0, padx=8, pady=8)
flash_port_var = StringVar(value=station.flash_port)
ttk.Entry(dialog, textvariable=flash_port_var, width=20).grid(row=0, column=1, padx=8, pady=8)
ttk.Label(dialog, text="产测 CDC 串口:").grid(row=1, column=0, padx=8, pady=8)
ttk.Label(dialog, text="产测 DBG UART:").grid(row=1, column=0, padx=8, pady=8)
test_port_var = StringVar(value=station.test_port)
ttk.Entry(dialog, textvariable=test_port_var, width=20).grid(row=1, column=1, padx=8, pady=8)
ttk.Label(dialog, text="固件:").grid(row=2, column=0, padx=8, pady=8)
@@ -576,20 +624,24 @@ class FactoryHost(Tk):
self.events.put(("station", station.number, result, message))
def start_tests(self) -> None:
selected = self.table.selection() or tuple(self.stations.keys())
had_selection = bool(self.table.selection())
self._auto_add_ft1_stations()
selected = self.table.selection() if had_selection else tuple(
number for number, station in self.stations.items() if station.test_port
)
for iid in selected:
if iid not in self.stations:
continue
station = self.stations[iid]
if not station.test_port:
station.status, station.message = "失败", "未设置产测 CDC 串口"
station.status, station.message = "失败", "未发现 FT1 产测 DBG UART"
self._update_row(station)
continue
station.protocol.reset()
station.report_path = None
station.report_saved = False
station.status = "等待设备"
station.message = "等待 FT1 session_start"
station.message = "等待 FT1 状态广播"
if station.reader:
station.reader.stop()
station.reader = SerialReader(
@@ -601,7 +653,7 @@ class FactoryHost(Tk):
station.reader.start()
self._update_row(station)
self._refresh_node_table()
self._log("系统", "FT1 串口产测已启动；只保存以 FT1 开头的协议行")
self._log("系统", "已自动识别 FT1 UART；设备自动开始产测并每500ms广播状态")
def _serial_line(self, station: Station, line: str) -> None:
self.events.put(("serial", station.number, line))
@@ -624,6 +676,26 @@ class FactoryHost(Tk):
self._log(station.number, f"[SDK] {line}")
return
if message["type"] in {"device_ready", "command_ack"}:
station.message = station.protocol.apply(message)
station.status = "设备已识别" if message["type"] == "device_ready" else ("开始产测" if message.get("accepted") else "START 被拒绝")
self._log(station.number, f"[FT1] {station.message}")
self._update_row(station)
return
if message["type"] == "status_snapshot":
station.protocol.apply(message)
station.status = "产测中" if message["active_node"] >= 0 else "产测已启动"
if station.protocol.session_result_seen:
station.status = "产测通过" if station.protocol.result == "PASS" else "产测失败"
station.message = station.protocol.current_node or "状态广播"
self._log(station.number, f"[FT1] {station.message}: " + ", ".join(
f"{name}={station.protocol.nodes[name]}" for name in NODE_NAMES
))
self._update_row(station)
self._refresh_node_table()
return
previous_session = station.protocol.session
try:
description = station.protocol.apply(message)