feat: SD boot mode detection — kill hw_server + popup after 30s if Boot mode is SD
This commit is contained in:
@@ -6,6 +6,7 @@ dist/
|
||||
build/
|
||||
*.egg-info/
|
||||
*.egg
|
||||
*.exe
|
||||
|
||||
# ── Python ───────────────────────────────────────────────────
|
||||
__pycache__/
|
||||
|
||||
@@ -92,6 +92,11 @@ class MainWindow(ctk.CTk):
|
||||
# StringVar references for config sync
|
||||
self._ip_string_var: ctk.StringVar | None = None
|
||||
|
||||
# SD boot mode detection
|
||||
self._hw_server_pid: int | None = None
|
||||
self._sd_detected: bool = False
|
||||
self._sd_timer_id: str | None = None
|
||||
|
||||
self._load_config()
|
||||
self._build_ui()
|
||||
|
||||
@@ -955,6 +960,7 @@ class MainWindow(ctk.CTk):
|
||||
"""Append a message to the log display and terminal (thread-safe)."""
|
||||
print(message, flush=True)
|
||||
self.after(0, lambda: self._log_display.append(message))
|
||||
self._check_sd_boot(message)
|
||||
|
||||
def _set_step_status(self, index: int, status: str) -> None:
|
||||
"""Set the status of a workflow step.
|
||||
@@ -1033,6 +1039,8 @@ class MainWindow(ctk.CTk):
|
||||
self._log_message(" Checking Zynq JTAG presence...")
|
||||
jtag_result = check_zynq_jtag(self._config, self._jtag_callback)
|
||||
jtag_dict = get_zynq_status_dict(jtag_result)
|
||||
if jtag_result.hw_server_pid:
|
||||
self._hw_server_pid = jtag_result.hw_server_pid
|
||||
|
||||
# Log JTAG chain information
|
||||
if jtag_result.jtag_chain:
|
||||
@@ -1813,6 +1821,13 @@ class MainWindow(ctk.CTk):
|
||||
self._progress.set_value(step_idx / total)
|
||||
t_step_start = time.time()
|
||||
|
||||
# Abort if SD boot timeout fired
|
||||
if not self._is_running:
|
||||
self._log_message(f" ⚠ Step {i+1} aborted (SD boot mode)")
|
||||
self._set_step_status(i, "error")
|
||||
results.append(False)
|
||||
break
|
||||
|
||||
try:
|
||||
success = step_funcs[i]()
|
||||
results.append(success)
|
||||
@@ -1874,6 +1889,7 @@ class MainWindow(ctk.CTk):
|
||||
|
||||
self._progress.set_value(1.0)
|
||||
self._is_running = False
|
||||
self._cancel_sd_timer()
|
||||
self._run_btn.configure(state="normal", text="▶ Run All Steps")
|
||||
|
||||
# Update dependency statuses
|
||||
@@ -2144,6 +2160,67 @@ class MainWindow(ctk.CTk):
|
||||
continue
|
||||
return ""
|
||||
|
||||
# ── SD Boot Detection ──────────────────────────────────────
|
||||
|
||||
def _check_sd_boot(self, line: str) -> None:
|
||||
"""Scan a log line for SD boot mode; start 30s timer if detected."""
|
||||
if self._sd_detected:
|
||||
return
|
||||
import re
|
||||
for pattern in [
|
||||
re.compile(r"[Bb]oot\s*[Mm]ode\s*(?:is)?\s*[:\[=\s]*(\w+)"),
|
||||
re.compile(r"BootMode\s*[:\[=\s]*(\w+)", re.IGNORECASE),
|
||||
]:
|
||||
m = pattern.search(line)
|
||||
if m and m.group(1).upper() == "SD":
|
||||
self._sd_detected = True
|
||||
self._log_message(" ⚠ SD boot mode detected — waiting 30s for JTAG switch...")
|
||||
self._sd_timer_id = self.after(30000, self._on_sd_timeout)
|
||||
return
|
||||
|
||||
def _cancel_sd_timer(self) -> None:
|
||||
"""Cancel the SD boot timeout if still pending."""
|
||||
if self._sd_timer_id:
|
||||
self.after_cancel(self._sd_timer_id)
|
||||
self._sd_timer_id = None
|
||||
self._sd_detected = False
|
||||
|
||||
def _on_sd_timeout(self) -> None:
|
||||
"""SD boot mode persisted 30s — kill hw_server, alert user, stop workflow."""
|
||||
self._sd_timer_id = None
|
||||
self._sd_detected = False
|
||||
self._log_message(" ✗ SD boot mode persisted >30s — switch DIP to JTAG mode!")
|
||||
self._kill_hw_server()
|
||||
from tkinter import messagebox
|
||||
messagebox.showerror(
|
||||
"启动拨码开关位置错误!",
|
||||
"检测到 Boot mode 为 SD,请将拨码开关切换到 JTAG 模式后重试。",
|
||||
)
|
||||
# Stop current workflow execution
|
||||
self._is_running = False
|
||||
|
||||
def _kill_hw_server(self) -> None:
|
||||
"""Terminate the hw_server process started during Step 1."""
|
||||
if not self._hw_server_pid:
|
||||
return
|
||||
import platform
|
||||
import subprocess
|
||||
import os
|
||||
try:
|
||||
if platform.system().lower() == "windows":
|
||||
subprocess.run(
|
||||
["taskkill", "/F", "/PID", str(self._hw_server_pid)],
|
||||
capture_output=True,
|
||||
)
|
||||
else:
|
||||
import signal
|
||||
os.kill(self._hw_server_pid, signal.SIGTERM)
|
||||
self._log_message(f" ⚠ hw_server (PID {self._hw_server_pid}) terminated")
|
||||
except Exception as e:
|
||||
self._log_message(f" ⚠ Failed to kill hw_server: {e}")
|
||||
finally:
|
||||
self._hw_server_pid = None
|
||||
|
||||
# ── Vitis Check ────────────────────────────────────────────
|
||||
|
||||
def _check_vitis(self) -> None:
|
||||
|
||||
+10
-3
@@ -53,6 +53,7 @@ class ZynqCheckResult:
|
||||
pl_detected: bool
|
||||
hw_server_url: str
|
||||
error: str = ""
|
||||
hw_server_pid: int | None = None
|
||||
|
||||
|
||||
# ── Patterns ──────────────────────────────────────────────────────────
|
||||
@@ -104,12 +105,14 @@ def check_zynq_jtag(
|
||||
|
||||
# ── 2. Start hw_server ─────────────────────────────────────
|
||||
hw_proc: subprocess.Popen | None = None
|
||||
hw_pid: int | None = None
|
||||
try:
|
||||
hw_proc = subprocess.Popen(
|
||||
build_tool_command(hw_server_path),
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
hw_pid = hw_proc.pid
|
||||
# Wait up to 10s for hw_server to bind
|
||||
started = False
|
||||
for _ in range(100):
|
||||
@@ -118,7 +121,7 @@ def check_zynq_jtag(
|
||||
started = True
|
||||
break
|
||||
if not started:
|
||||
return _fail("hw_server did not bind port 3121 within 10s", hw_server_url)
|
||||
return _fail("hw_server did not bind port 3121 within 10s", hw_server_url, pid=hw_pid)
|
||||
except OSError as e:
|
||||
return _fail(f"Failed to start hw_server: {e}", hw_server_url)
|
||||
|
||||
@@ -161,9 +164,10 @@ def check_zynq_jtag(
|
||||
return _fail(
|
||||
f"JTAG scan timed out after {python_timeout}s",
|
||||
hw_server_url,
|
||||
pid=hw_pid,
|
||||
)
|
||||
except OSError as e:
|
||||
return _fail(f"Failed to run Vivado: {e}", hw_server_url)
|
||||
return _fail(f"Failed to run Vivado: {e}", hw_server_url, pid=hw_pid)
|
||||
finally:
|
||||
try:
|
||||
Path(tcl).unlink(missing_ok=True)
|
||||
@@ -201,6 +205,7 @@ def check_zynq_jtag(
|
||||
ps_detected=False, pl_detected=False,
|
||||
hw_server_url=hw_server_url,
|
||||
error=msg,
|
||||
hw_server_pid=hw_pid,
|
||||
)
|
||||
|
||||
ps_detected = any(d.is_arm_dap for d in jtag_chain)
|
||||
@@ -212,16 +217,18 @@ def check_zynq_jtag(
|
||||
ps_detected=ps_detected, pl_detected=pl_detected,
|
||||
hw_server_url=hw_server_url,
|
||||
error="",
|
||||
hw_server_pid=hw_pid,
|
||||
)
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
def _fail(error: str, url: str) -> ZynqCheckResult:
|
||||
def _fail(error: str, url: str, pid: int | None = None) -> ZynqCheckResult:
|
||||
return ZynqCheckResult(
|
||||
devices=[], jtag_chain=[], is_present=False,
|
||||
ps_detected=False, pl_detected=False,
|
||||
hw_server_url=url, error=error,
|
||||
hw_server_pid=pid,
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user