feat: improve continuous factory flashing workflow
This commit is contained in:
chenjintao
+255
-39
@@ -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("<MouseWheel>", 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()
|
||||
|
||||
Reference in New Issue
Block a user