diff --git a/burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip b/burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip new file mode 100644 index 0000000..ed09226 --- /dev/null +++ b/burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:d09d1688d9b3c674b1dbead819190425fec26f5183b3066e8143c8ad670e043f +size 6551698 diff --git a/burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip b/burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip new file mode 100644 index 0000000..b098821 --- /dev/null +++ b/burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:799adedf6b0ff6ad3f6adab281f8a206b99689fb394a198130f51b46e5c01848 +size 6222215 diff --git a/factory_host/README.md b/factory_host/README.md index 54e498e..f4c6f02 100644 --- a/factory_host/README.md +++ b/factory_host/README.md @@ -7,8 +7,10 @@ - 产测固件和正式固件分开保存，每个工位可独立选择。 - 烧录串口和产测 DBG UART 分开配置，避免把 DIAG/AT/调试口误认成烧录口。 -- 使用原厂 `adownload.exe -u -a`，先进入等待状态，再由关机设备上电触发烧录。 -- 提供“烧录选中”单工位入口；USB 自动等待的批量任务按工位顺序执行，避免抢设备。 +- 主界面提供醒目的“启动自动烧录”：每 0.5 秒持续检测烧录串口，先让下载器按工位进入等待，再由设备上电触发烧录。 +- 每个工位完成后会锁定当前设备；只有检测到该端口消失并重新出现（断开旧设备、连接下一台并上电）才会再次烧录，避免同一设备反复烧录。 +- 烧录与串口扫描均在后台运行，Windows 子进程隐藏控制台窗口，不阻塞界面。 +- 烧录和产测运行期间实时日志强制跟随最新一行，停止后才允许手工滚动。 - 只解析以 `FT1 ` 开头的协议行；普通 SDK 日志只显示，不进入产测数据库。 - 严格校验协议版本、消息类型、节点编号/名称、会话 ID 和递增序号。 - 显示 `BOOT_AUDIO`、`POWER_ADC`、`AUDIO_LOOPBACK`、`ESIM`、 @@ -30,8 +32,14 @@ burn_tool/aboot-tools-2024.08.08-win-x64/adownload.exe factory_host/dist/EG800FactoryHost.exe ``` -默认“产测固件”指向已包含节点提示音的 FT1 整包；默认“正式固件”指向 -`version` 目录中的正式内层 `-fac.zip`。两者都可在外置 +默认“产测固件”和“正式固件”分别位于： + +```text +burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip +burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip +``` + +两者都可通过主界面切换并选择，也可在外置 `factory_host_config.json` 中替换，烧录时不要选择 `-target.zip` 外层归档。 双击表格中的工位可以分别设置烧录口和产测 DBG UART 口。默认 USB 自动等待模式不需要 diff --git a/factory_host/USAGE.md b/factory_host/USAGE.md index 62bd13a..be7fd45 100644 --- a/factory_host/USAGE.md +++ b/factory_host/USAGE.md @@ -7,9 +7,12 @@ ```text banban/ ├─ burn_tool/ -│ └─ aboot-tools-2024.08.08-win-x64/ +│ ├─ aboot-tools-2024.08.08-win-x64/ │ ├─ adownload.exe │ └─ 其他随工具提供的 DLL、配置和驱动文件 +│ └─ firmware/ +│ ├─ test/EG800AKCN_11LG-FT1-factory-test.zip +│ └─ official/EG800AKCN_11LG-official-fac.zip ├─ factory_host/ │ └─ dist/ │ ├─ EG800FactoryHost.exe @@ -36,8 +39,8 @@ banban/ "workers": 2 }, "firmwares": { - "production": "factory_host/dist/factory_firmware/EG800AKCN_11LG-FT1-factory-test.zip", - "official": "version/正式固件目录/EG800AKCN_11LG-版本号-fac.zip" + "production": "burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip", + "official": "burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip" }, "serial": { "baudrate": 115200, @@ -60,11 +63,13 @@ banban/ ## 3. 烧录产测固件 -1. 启动 `EG800FactoryHost.exe`，固件类型选择“产测固件”。 -2. 设备保持关机，连接烧录器和 USB。 -3. 选择工位后点击单工位烧录；批量生产时使用批量烧录。 -4. 界面显示“等待上电”后再给设备开机。 -5. 等待状态变成“烧录成功”，不要在烧录过程中拔线或关闭程序。 +1. 启动 `EG800FactoryHost.exe`，固件类型选择“产测固件”或“正式固件”，确认顶部完整路径。 +2. 连接一个或多个烧录器，设备保持关机。 +3. 点击醒目的“启动自动烧录”，工具每 0.5 秒持续检测烧录串口，并为每个串口建立工位。 +4. 工位显示“烧录中·等待上电”后给对应设备开机，设备进入 AROM 后自动开始烧录。 +5. 工位显示“烧录成功”后先断开当前设备；工具检测到端口消失后才会允许下一次烧录，避免同一设备反复烧录。 +6. 换上下一台关机设备并上电，端口重新出现后自动开始下一次烧录。 +7. 全部生产完成后点击“停止烧录”；不要在实际写入过程中拔线或关闭程序。 默认参数使用 `adownload.exe -u -a` 自动等待 AROM 下载设备，因此烧录口和产测 UART 是两个不同角色。 diff --git a/factory_host/factory_host.py b/factory_host/factory_host.py index 87be243..a70344a 100644 --- a/factory_host/factory_host.py +++ b/factory_host/factory_host.py @@ -28,6 +28,9 @@ from tkinter import ttk from ft1_protocol import FT1Session, NODE_NAMES, ProtocolError, error_text, parse_ft1_line +CREATE_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000) + + APP_DIR = Path(sys.executable).resolve().parent if getattr(sys, "frozen", False) else Path(__file__).resolve().parent @@ -50,10 +53,14 @@ DEFAULT_CONFIG = { "aboot_path": "burn_tool/aboot.exe", "downloader_path": "burn_tool/adownload.exe", "args": ["-u", "-a", "-s", "{baudrate}", "-q", "{firmware}"], + "auto_args": ["-p", "{port}", "-a", "-s", "{baudrate}", "-q", "{firmware}"], "timeout_seconds": 900, "workers": 2, }, - "firmwares": {"production": "", "official": ""}, + "firmwares": { + "production": "burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip", + "official": "burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip", + }, "serial": {"baudrate": 115200, "encoding": "utf-8", "line_ending": "\\r\\n"}, "protocol": {"prefix": "FT1 ", "version": 1, "auto_save": True}, } @@ -151,6 +158,7 @@ def list_serial_ports() -> list[str]: 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): @@ -232,7 +240,10 @@ class SerialReader: self._stream = self._pyserial except ImportError: if os.name == "nt": - subprocess.run(["mode", self.port, f"BAUD={self.baudrate}", "PARITY=n", "DATA=8", "STOP=1"], capture_output=True, check=False) + subprocess.run( + ["mode", self.port, f"BAUD={self.baudrate}", "PARITY=n", "DATA=8", "STOP=1"], + capture_output=True, check=False, creationflags=CREATE_NO_WINDOW, + ) device = self.port if self.port.startswith("\\\\.\\") else "\\\\.\\" + self.port self._stream = open(device, "r+b", buffering=0) self._opened.set() @@ -286,6 +297,14 @@ class FactoryHost(Tk): self.events: queue.Queue = queue.Queue() self.executor: ThreadPoolExecutor | None = None self.flash_cancel = threading.Event() + self.flash_running = False + self.flash_jobs_active = 0 + self.auto_flash_monitoring = False + self.auto_monitor_stop = threading.Event() + self.auto_firmware_kind = "production" + self.auto_downloader: Path | None = None + self.scan_running = False + self.log_follow = True self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S") LOG_DIR.mkdir(parents=True, exist_ok=True) SESSION_DIR.mkdir(parents=True, exist_ok=True) @@ -296,16 +315,19 @@ class FactoryHost(Tk): self.after(100, self._drain_events) def _build_ui(self) -> None: + style = ttk.Style(self) + style.configure("Primary.TButton", font=("Microsoft YaHei UI", 11, "bold"), padding=(18, 10)) + style.configure("Firmware.TLabel", font=("Microsoft YaHei UI", 10, "bold"), foreground="#b42318") top = ttk.Frame(self, padding=8) top.pack(fill=X) ttk.Label(top, text="固件:").pack(side=LEFT) self.firmware_kind = StringVar(value="production") - ttk.Radiobutton(top, text="产测固件", variable=self.firmware_kind, value="production").pack(side=LEFT, padx=3) - ttk.Radiobutton(top, text="正式固件", variable=self.firmware_kind, value="official").pack(side=LEFT, padx=3) + ttk.Radiobutton(top, text="产测固件", variable=self.firmware_kind, value="production", command=self._refresh_tool_status).pack(side=LEFT, padx=3) + ttk.Radiobutton(top, text="正式固件", variable=self.firmware_kind, value="official", command=self._refresh_tool_status).pack(side=LEFT, padx=3) self.firmware_label = StringVar() - ttk.Label(top, textvariable=self.firmware_label, width=55).pack(side=LEFT, padx=12) - ttk.Button(top, text="选择固件", command=self.choose_firmware).pack(side=LEFT) - ttk.Button(top, text="扫描串口", command=self.scan_ports).pack(side=LEFT, padx=4) + self.firmware_display = ttk.Label(top, textvariable=self.firmware_label, width=70, style="Firmware.TLabel") + self.firmware_display.pack(side=LEFT, padx=12) + ttk.Button(top, text="选择当前固件", command=self.choose_firmware).pack(side=LEFT) status = ttk.Frame(self, padding=(8, 0, 8, 6)) status.pack(fill=X) @@ -315,16 +337,23 @@ class FactoryHost(Tk): controls = ttk.Frame(self, padding=(8, 0, 8, 6)) controls.pack(fill=X) - ttk.Button(controls, text="添加工位", command=self.add_station).pack(side=LEFT) - ttk.Button(controls, text="删除选中", command=self.remove_station).pack(side=LEFT, padx=4) - ttk.Button(controls, text="烧录选中", command=self.start_selected_flash).pack(side=LEFT, padx=(16, 4)) - ttk.Button(controls, text="批量烧录", command=self.start_flash).pack(side=LEFT) - ttk.Button(controls, text="停止烧录", command=self.stop_flash).pack(side=LEFT) - ttk.Button(controls, text="开始串口产测", command=self.start_tests).pack(side=LEFT, padx=16) + self.auto_flash_button = ttk.Button( + controls, text="启动自动烧录", command=self.start_auto_flash, + style="Primary.TButton") + self.auto_flash_button.pack(side=LEFT, padx=(0, 8)) + ttk.Button(controls, text="开始串口产测", command=self.start_tests).pack(side=LEFT, padx=(0, 8)) + ttk.Button(controls, text="扫描串口", command=self.scan_ports).pack(side=LEFT) + ttk.Button(controls, text="停止烧录", command=self.stop_flash).pack(side=LEFT, padx=(16, 4)) ttk.Button(controls, text="停止串口", command=self.stop_serial).pack(side=LEFT) - ttk.Button(controls, text="打开配置", command=self.open_config).pack(side=LEFT, padx=16) - ttk.Button(controls, text="重新加载", command=self.reload_config).pack(side=LEFT) - ttk.Button(controls, text="导出报表", command=self.export_report).pack(side=RIGHT) + 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.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) + ttk.Button(advanced, text="重新加载", command=self.reload_config).pack(side=LEFT, padx=3) + ttk.Button(advanced, text="导出报表", command=self.export_report).pack(side=RIGHT) table_frame = ttk.Frame(self, padding=(8, 0, 8, 4)) table_frame.pack(fill=BOTH, expand=True) @@ -368,7 +397,11 @@ class FactoryHost(Tk): for column, label, width in (("time", "时间", 90), ("station", "工位", 70), ("text", "日志", 850)): self.log_text.heading(column, text=label) self.log_text.column(column, width=width, anchor="w") - self.log_text.pack(fill=BOTH, expand=True) + self.log_text.pack(side=LEFT, fill=BOTH, expand=True) + log_scroll = ttk.Scrollbar(bottom, orient="vertical", command=self._scroll_log) + log_scroll.pack(side=RIGHT, fill=Y) + self.log_text.configure(yscrollcommand=log_scroll.set) + self.log_text.bind("", self._on_log_mousewheel) self._refresh_tool_status() def _load_default_rows(self) -> None: @@ -382,29 +415,56 @@ class FactoryHost(Tk): tool_text = f"adownload: {downloader}" if downloader else "adownload: 未找到（请把 burn_tool 完整目录放在根目录或在配置中指定）" package_text = f"固件: {firmware}" if firmware.is_file() else "固件: 未选择" self.tool_status.set(tool_text + " | " + package_text) - self.firmware_label.set(str(firmware) if firmware.is_file() else "未选择当前固件") + kind_label = "产测固件" if self.firmware_kind.get() == "production" else "正式固件" + state = "已就绪" if firmware.is_file() else "未找到" + self.firmware_label.set(f"当前：{kind_label} | {state} | {firmware}") + + def _scroll_log(self, *args) -> None: + if not self.log_follow: + self.log_text.yview(*args) + + def _on_log_mousewheel(self, _event): + return "break" if self.log_follow else None def choose_firmware(self) -> None: - path = filedialog.askopenfilename(title="选择固件包", filetypes=(("固件包", "*.zip *.bin *.img *.json"), ("所有文件", "*.*"))) + kind_dir = "test" if self.firmware_kind.get() == "production" else "official" + initial_dir = ROOT_DIR / "burn_tool" / "firmware" / kind_dir + path = filedialog.askopenfilename( + title="选择固件包", + initialdir=str(initial_dir if initial_dir.is_dir() else ROOT_DIR), + filetypes=(("固件包", "*.zip *.bin *.img *.json"), ("所有文件", "*.*")), + ) if path: self.config_data["firmwares"][self.firmware_kind.get()] = path save_config(self.config_data) self._refresh_tool_status() def scan_ports(self) -> None: - found = self._auto_add_ft1_stations() - self._log("系统", f"FT1 自动识别完成，新增/确认工位: {', '.join(found) if found else '无'}") + if self.scan_running: + return + self.scan_running = True + self.log_follow = True + self._log("系统", "正在后台扫描 FT1 产测串口，请稍候...") + threading.Thread(target=self._scan_ft1_worker, args=(None, None), daemon=True).start() - def _auto_add_ft1_stations(self) -> list[str]: - """Probe every COM port and create one visible station per FT1 device.""" + def _scan_ft1_worker(self, selected: tuple[str, ...] | None, had_selection: bool | None) -> None: 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] = [] + ports: list[str] = [] for candidate in list_serial_ports(): if candidate in used: continue if not identify_ft1_port(candidate, baudrate): continue + ports.append(candidate) + self.events.put(("ft1_scan", ports, selected, had_selection)) + + def _assign_ft1_stations(self, ports: list[str]) -> list[str]: + found: list[str] = [] + used = {station.test_port for station in self.stations.values() if station.test_port} + for candidate in ports: + if candidate in used: + 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: @@ -421,6 +481,12 @@ class FactoryHost(Tk): self._update_row(station) return found + def _auto_add_ft1_stations(self) -> list[str]: + """Compatibility helper; UI actions use the background scanner.""" + ports = [port for port in list_serial_ports() + if identify_ft1_port(port, int(self.config_data["serial"].get("baudrate", 115200)))] + return self._assign_ft1_stations(ports) + def open_config(self) -> None: try: if os.name == "nt": @@ -438,7 +504,7 @@ class FactoryHost(Tk): except RuntimeError as exc: messagebox.showerror("配置错误", str(exc)) - def add_station(self, number: str | None = None, flash_port: str = "", test_port: str = "") -> None: + def add_station(self, number: str | None = None, flash_port: str = "", test_port: str = "") -> Station: if number is None: number = str(len(self.stations) + 1) while number in self.stations: @@ -446,6 +512,7 @@ class FactoryHost(Tk): station = Station(number=number, flash_port=flash_port, test_port=test_port, firmware_kind=self.firmware_kind.get()) self.stations[number] = station self.table.insert("", END, iid=number, values=(number, flash_port, test_port, station.status, "", "", "0/6", "产测固件", "")) + return station def remove_station(self) -> None: for iid in self.table.selection(): @@ -514,9 +581,86 @@ class FactoryHost(Tk): def start_flash(self) -> None: selected = self.table.selection() or tuple(self.stations.keys()) - self._start_flash_jobs(selected) + self._start_flash_jobs(selected, auto_detect=False) - def _start_flash_jobs(self, selected) -> None: + def start_auto_flash(self) -> None: + """Continuously watch for newly appearing download ports until stopped.""" + if self.auto_flash_monitoring: + self._log("系统", "自动烧录已经在持续监听串口") + return + if self.flash_running or self.scan_running: + self._log("系统", "已有烧录或串口扫描任务正在进行") + return + downloader = find_downloader( + self.config_data["flash"].get("downloader_path", ""), + self.config_data["flash"].get("aboot_path", ""), + ) + firmware = resolve_project_path( + self.config_data["firmwares"].get(self.firmware_kind.get(), "")) + if not downloader or not firmware.is_file(): + messagebox.showerror("无法启动自动烧录", "请先确认下载器和当前选择的固件都存在。") + return + workers = max(1, int(self.config_data["flash"].get("workers", 2))) + self.auto_flash_monitoring = True + self.auto_firmware_kind = self.firmware_kind.get() + self.auto_downloader = downloader + self.auto_monitor_stop.clear() + self.flash_cancel.clear() + self.flash_running = True + self.log_follow = True + self.executor = ThreadPoolExecutor(max_workers=workers) + self.auto_flash_button.configure(text="自动烧录监听中") + waiting_stations = list(self.stations.values())[:workers] + while len(waiting_stations) < workers: + waiting_stations.append(self.add_station()) + for station in waiting_stations: + station.firmware_kind = self.auto_firmware_kind + station.status = "自动等待上电" + station.message = "持续检测烧录串口" + self._update_row(station) + self._log("系统", f"自动烧录已启动：每 0.5 秒检测串口，等待设备上电，并发工位 {workers}") + threading.Thread(target=self._auto_flash_monitor_worker, daemon=True).start() + + def _auto_flash_monitor_worker(self) -> None: + known_ports: set[str] = set() + 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 + self.events.put(("auto_flash_stopped",)) + + def _submit_auto_flash_port(self, port: str) -> None: + busy_states = {"排队中", "烧录中", "烧录中·等待上电"} + if any(item.flash_port == port and item.status in busy_states + for item in self.stations.values()): + return + station = next((item for item in self.stations.values() + if item.flash_port == port and item.status not in busy_states), None) + if station is None: + station = next((item for item in self.stations.values() + if item.status == "自动等待上电"), None) + if station is None: + station = next((item for item in self.stations.values() + if item.status not in busy_states and not item.test_port), None) + if station is None: + station = self.add_station() + station.flash_port = port + station.firmware_kind = self.auto_firmware_kind + station.status = "排队中" + station.message = f"检测到新烧录串口 {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}，已进入等待上电状态") + + def _start_flash_jobs(self, selected, auto_detect: bool = False) -> None: + if self.auto_flash_monitoring and not auto_detect: + self._log("系统", "请先停止自动烧录，再执行手动烧录") + return downloader = find_downloader(self.config_data["flash"].get("downloader_path", ""), self.config_data["flash"].get("aboot_path", "")) if not downloader: messagebox.showerror("无法烧录", "未找到 adownload.exe。请把 burn_tool 完整目录放到根目录，或在配置中指定 flash.downloader_path。aboot.exe 是图形界面，不能作为命令行烧录进程。") @@ -525,34 +669,44 @@ class FactoryHost(Tk): self.stations[iid] for iid in selected if iid in self.stations - and self.stations[iid].status not in ("排队中", "等待上电", "烧录中") + and self.stations[iid].status not in + ("排队中", "等待上电", "烧录中", "烧录中·等待上电") ] if not jobs: self._log("系统", "没有可提交的工位（正在烧录的工位不会重复排队）") return - configured_args = self.config_data["flash"].get("args", ["{firmware}"]) + configured_args = self.config_data["flash"].get( + "auto_args" if auto_detect else "args", ["{firmware}"]) if isinstance(configured_args, str): configured_args = parse_command_template(configured_args) auto_usb = any(str(arg) in ("-a", "--auto-enable") for arg in configured_args) uses_named_port = any("{port}" in str(arg) for arg in configured_args) self.flash_cancel.clear() workers = max(1, int(self.config_data["flash"].get("workers", 2))) - if auto_usb and not uses_named_port and len(jobs) > 1: + if auto_usb and not uses_named_port and len(jobs) > 1 and not auto_detect: workers = 1 self._log("系统", "USB 自动等待模式将按工位顺序烧录；请只给当前等待工位上电") + self.flash_running = True + self.flash_jobs_active = len(jobs) self.executor = ThreadPoolExecutor(max_workers=min(workers, len(jobs))) for station in jobs: + station.firmware_kind = self.firmware_kind.get() station.status = "排队中" station.message = "" self._update_row(station) - self.executor.submit(self._flash_one, station, downloader) + self.executor.submit(self._flash_one, station, downloader, auto_detect) self._log("系统", f"已提交 {len(jobs)} 个烧录任务，并发数 {workers}") def stop_flash(self) -> None: + self.auto_monitor_stop.set() + self.auto_flash_monitoring = False + self.auto_downloader = None self.flash_cancel.set() + self.log_follow = False + self.auto_flash_button.configure(text="启动自动烧录") self._log("系统", "已请求停止后续烧录任务；正在执行的 adownload 将在超时或退出后结束") - def _flash_one(self, station: Station, downloader: Path) -> None: + def _flash_one(self, station: Station, downloader: Path, auto_detect: bool = False) -> None: if self.flash_cancel.is_set(): self.events.put(("station", station.number, "已取消", "")) return @@ -560,7 +714,8 @@ class FactoryHost(Tk): if not firmware.is_file(): self.events.put(("station", station.number, "失败", f"固件不存在: {firmware}")) return - configured_args = self.config_data["flash"].get("args", ["{firmware}"]) + configured_args = self.config_data["flash"].get( + "auto_args" if auto_detect else "args", ["{firmware}"]) if isinstance(configured_args, str): configured_args = parse_command_template(configured_args) baudrate = int(self.config_data["serial"].get("baudrate", 115200)) @@ -570,12 +725,23 @@ class FactoryHost(Tk): args = expand_args(configured_args, aboot=downloader, firmware=firmware, port=station.flash_port, station=station.number, baudrate=baudrate) command = [str(downloader)] + args auto_usb = any(str(arg) in ("-a", "--auto-enable") for arg in configured_args) - status = "等待上电" if auto_usb else "烧录中" - hint = "下载器已启动，请现在给设备上电" if auto_usb else "" + status = "烧录中·等待上电" if auto_usb else "烧录中" + hint = "下载器已启动，请现在给设备上电" if auto_usb else "烧录进程已启动" self.events.put(("station", station.number, status, hint or " ".join(shlex.quote(x) for x in command))) self.events.put(("log", station.number, "命令: " + " ".join(shlex.quote(x) for x in command))) try: - process = subprocess.Popen(command, cwd=str(downloader.parent), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1) + 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) output_tail = [] deadline = time.monotonic() + int(self.config_data["flash"].get("timeout_seconds", 900)) assert process.stdout is not None @@ -625,11 +791,21 @@ class FactoryHost(Tk): def start_tests(self) -> None: had_selection = bool(self.table.selection()) - self._auto_add_ft1_stations() - selected = self.table.selection() if had_selection else tuple( + if self.scan_running: + return + self.scan_running = True + self.log_follow = True + selected = tuple(self.table.selection()) if had_selection else None + self._log("系统", "正在后台识别 FT1 UART，界面可以继续操作...") + threading.Thread( + target=self._scan_ft1_worker, args=(selected, had_selection), daemon=True + ).start() + + def _start_test_readers(self, selected: tuple[str, ...] | None = None) -> None: + targets = selected or tuple( number for number, station in self.stations.items() if station.test_port ) - for iid in selected: + for iid in targets: if iid not in self.stations: continue station = self.stations[iid] @@ -769,6 +945,7 @@ class FactoryHost(Tk): return station.report_path def stop_serial(self) -> None: + self.log_follow = False for station in self.stations.values(): if station.reader: station.reader.stop() @@ -792,6 +969,17 @@ class FactoryHost(Tk): station.status, station.message = status, message self._update_row(station) self._log(number, f"{status}: {message}") + if status in ("烧录成功", "失败", "已取消") and self.flash_jobs_active: + self.flash_jobs_active -= 1 + if self.auto_flash_monitoring: + if status == "烧录成功": + station.message = "烧录成功；请断开当前设备，端口重新出现后才烧录下一台" + self._update_row(station) + if self.flash_jobs_active == 0: + self._log("系统", "自动烧录继续监听；已完成设备不会重复烧录") + elif self.flash_jobs_active == 0: + self.flash_running = False + self.log_follow = False elif kind == "log": _, number, line = event self._log(number, line) @@ -799,6 +987,28 @@ class FactoryHost(Tk): _, number, line = event if number in self.stations: self._process_serial_line(self.stations[number], line) + elif kind == "ft1_scan": + _, ports, selected, had_selection = event + found = self._assign_ft1_stations(ports) + self.scan_running = False + self._log("系统", f"FT1 自动识别完成，新增/确认工位: {', '.join(found) if found else '无'}") + if had_selection is not None: + self._start_test_readers(selected if had_selection else None) + elif kind == "auto_flash_port": + _, port = event + if self.auto_flash_monitoring and port not in { + station.test_port for station in self.stations.values() if station.test_port + }: + self._submit_auto_flash_port(port) + elif kind == "auto_flash_port_removed": + _, port = event + station = next((item for item in self.stations.values() + if item.flash_port == port), None) + if station and station.status == "烧录成功": + station.message = "设备已断开，等待下一台设备上电" + self._update_row(station) + elif kind == "auto_flash_stopped": + self._log("系统", "自动烧录串口监听已停止") except queue.Empty: pass self.after(100, self._drain_events) @@ -809,6 +1019,10 @@ class FactoryHost(Tk): children = self.log_text.get_children() if len(children) > 1000: self.log_text.delete(children[0]) + if self.log_follow: + newest = self.log_text.get_children() + if newest: + self.log_text.see(newest[-1]) def export_report(self) -> None: path = filedialog.asksaveasfilename(title="导出产测报表", defaultextension=".csv", initialfile=f"factory_{self.session_id}.csv", filetypes=(("CSV", "*.csv"),)) @@ -833,6 +1047,8 @@ class FactoryHost(Tk): self._log("系统", f"报表已导出: {path}") def _on_close(self) -> None: + self.auto_monitor_stop.set() + self.auto_flash_monitoring = False self.stop_serial() self.flash_cancel.set() self.destroy() diff --git a/factory_host/factory_host_config.json b/factory_host/factory_host_config.json index e9dc1c1..dfee830 100644 --- a/factory_host/factory_host_config.json +++ b/factory_host/factory_host_config.json @@ -10,12 +10,21 @@ "-q", "{firmware}" ], + "auto_args": [ + "-p", + "{port}", + "-a", + "-s", + "{baudrate}", + "-q", + "{firmware}" + ], "timeout_seconds": 900, "workers": 2 }, "firmwares": { - "production": "factory_host/dist/factory_firmware/EG800AKCN_11LG-FT1-factory-test.zip", - "official": "version/EG800AKCN_11LG-1.8.4-20260921171244/EG800AKCN_11LG-1.8.4-20260921171244-fac.zip" + "production": "burn_tool/firmware/test/EG800AKCN_11LG-FT1-factory-test.zip", + "official": "burn_tool/firmware/official/EG800AKCN_11LG-official-fac.zip" }, "serial": { "baudrate": 115200, diff --git a/factory_host/test_factory_host.py b/factory_host/test_factory_host.py index 5dcaaed..a9bcd7c 100644 --- a/factory_host/test_factory_host.py +++ b/factory_host/test_factory_host.py @@ -40,6 +40,12 @@ class FactoryHostTests(unittest.TestCase): self.assertIn("-a", args) self.assertNotIn("{port}", args) + def test_auto_flash_binds_each_detected_port(self): + args = factory_host.DEFAULT_CONFIG["flash"]["auto_args"] + self.assertIn("-p", args) + self.assertIn("-a", args) + self.assertIn("{port}", args) + def test_project_root_and_relative_path(self): self.assertTrue((factory_host.ROOT_DIR / "burn_tool").is_dir()) config = factory_host.load_config(factory_host.CONFIG_PATH) @@ -53,6 +59,8 @@ class FactoryHostTests(unittest.TestCase): self.assertTrue( factory_host.resolve_project_path(config["firmwares"]["official"]).is_file() ) + self.assertIn("burn_tool/firmware/test", config["firmwares"]["production"]) + self.assertIn("burn_tool/firmware/official", config["firmwares"]["official"]) def test_ft1_ignores_sdk_log_and_parses_node(self): self.assertIsNone(parse_ft1_line("DP1312EA is OK!"))