feat(uart): add UART availability check, inter-step delays, and post-download Zynq status verification

- Add check_uart_available() to serial_monitor.py — detects port availability
- Add verify_zynq_status() to boot_verifier.py — lightweight post-download check
- Add uart_delay / inter_step_delay config fields (defaults: 3s / 2s)
- Step 1: Check UART availability, show non-blocking 'UART 不可用' notification
- Step 4: Add post-download Zynq status check (ping + serial) after TFTP
- _run_steps_worker: Add configurable inter-step delays between all steps
- Update step labels (4a→4b→4c→4d) for clarity
This commit is contained in:
Jeremy Shen
2026-06-09 17:53:48 +08:00
parent 91a7843dbe
commit 3a9bc5d9ed
4 changed files with 209 additions and 7 deletions
+94
View File
@@ -124,6 +124,100 @@ def verify_boot(
)
def verify_zynq_status(
config: Config,
timeout: int | None = None,
max_retries: int = 3,
uart_available: bool = True,
callback: ProgressCallback = None,
) -> BootVerificationResult:
"""Quickly verify Zynq is responding after a download operation.
Performs a fast ping check and optionally a serial port read.
Unlike verify_boot, this is a lightweight check meant to confirm
the device is alive after flash/TFTP operations (before reboot).
Args:
config: Application configuration.
timeout: Seconds per ping attempt.
max_retries: Number of ping retry attempts.
uart_available: Whether UART serial port is available.
callback: Optional progress callback.
Returns:
BootVerificationResult (ip_reachable and serial_found reflect
the quick check results).
"""
timeout = timeout or config.ping_timeout
ip = config.zynq_ip
if callback:
callback("start", f"Checking Zynq status on {ip}...")
# Check 1: Ping
ip_reachable = False
for attempt in range(1, max_retries + 1):
if callback:
callback("progress", f"Ping attempt {attempt}/{max_retries}...")
if ping_ip(ip, 1, timeout):
ip_reachable = True
if callback:
callback("complete", f"Device {ip} is reachable")
break
time.sleep(1)
# Check 2: Serial (only if UART is available)
serial_found = False
boot_info: dict = {}
if uart_available and config.serial_port:
ports = detect_serial_ports()
if ports:
for port_info in ports:
if port_info.device == config.serial_port:
try:
lines = read_serial_stream(
config.serial_port,
config.serial_baudrate,
timeout=3,
)
parsed = parse_boot_output(lines)
if parsed.ip_address or parsed.boot_message or parsed.version:
serial_found = True
boot_info = {
"ip_address": parsed.ip_address,
"version": parsed.version,
"boot_message": parsed.boot_message,
"raw_lines": lines[-20:],
}
if callback:
callback("complete", "Boot messages detected on serial")
break
except Exception:
pass # Non-fatal; fall through
success = ip_reachable or serial_found
message_parts = []
if ip_reachable:
message_parts.append(f"IP {ip} reachable")
if serial_found:
message_parts.append("serial boot info found")
if not success:
message_parts.append("Zynq not responding")
message = "; ".join(message_parts) if message_parts else "Check inconclusive"
if callback:
callback("complete" if success else "error", message)
return BootVerificationResult(
success=success,
message=message,
ip_reachable=ip_reachable,
serial_found=serial_found,
boot_info=boot_info if boot_info else {},
)
def wait_for_boot(
config: Config,
timeout: int | None = None,
+6
View File
@@ -27,6 +27,8 @@ DEFAULT_CONFIG: dict[str, Any] = {
"reboot_timeout": 30,
"ping_timeout": 5,
"ping_count": 3,
"uart_delay": 3,
"inter_step_delay": 2,
}
@@ -50,6 +52,8 @@ class Config:
reboot_timeout: int = 30
ping_timeout: int = 5
ping_count: int = 3
uart_delay: int = 3
inter_step_delay: int = 2
# Internal: path to the config file on disk
_config_path: Path = field(default=None, init=False, repr=False)
@@ -151,6 +155,8 @@ class Config:
"reboot_timeout": self.reboot_timeout,
"ping_timeout": self.ping_timeout,
"ping_count": self.ping_count,
"uart_delay": self.uart_delay,
"inter_step_delay": self.inter_step_delay,
}
return data
+59 -7
View File
@@ -21,8 +21,8 @@ from bitstream_programmer import full_bitstream_program, BitstreamResult
from ip_verifier import ping_ip
from tftp_manager import tftp_upload_verify, TftpResult
from reboot_manager import reboot_zynq, RebootResult
from boot_verifier import verify_boot, BootVerificationResult
from serial_monitor import detect_serial_ports, parse_boot_output
from boot_verifier import verify_boot, verify_zynq_status, BootVerificationResult
from serial_monitor import detect_serial_ports, parse_boot_output, check_uart_available
from gui.styles import (
WINDOW_WIDTH,
WINDOW_HEIGHT,
@@ -73,6 +73,8 @@ class MainWindow(ctk.CTk):
self._step_statuses: dict[int, str] = {}
self._is_running = False
self._worker_thread: threading.Thread | None = None
self._uart_available: bool = False
self._uart_message: str = ""
self._load_config()
self._build_ui()
@@ -471,7 +473,7 @@ class MainWindow(ctk.CTk):
# ── Step Implementations ───────────────────────────────────
def _run_step_1_check_environment(self) -> bool:
"""Step 1: Check environment (Vitis/Vivado tools + Zynq IP)."""
"""Step 1: Check environment (Vitis/Vivado tools + Zynq IP + UART)."""
self._log_message("Step 1: Checking environment...")
if not self._config:
@@ -507,8 +509,35 @@ class MainWindow(ctk.CTk):
self._ip_status.configure(text=f"IP: {ip}", text_color=DANGER_COLOR)
return False
# Check UART availability
self._log_message(" Checking UART serial port...")
port = self._config.serial_port
if port:
available, reason = check_uart_available(port, self._config.serial_baudrate)
self._uart_available = available
self._uart_message = reason
if available:
self._log_message(f" ✓ UART available: {reason}")
else:
self._log_message(f" ✗ UART unavailable: {reason}")
self._show_uart_notification(reason)
else:
self._uart_available = False
self._uart_message = "No serial port configured"
self._log_message(" ✗ No serial port configured")
self._show_uart_notification(self._uart_message)
return True
def _show_uart_notification(self, reason: str) -> None:
"""Show a non-blocking UART unavailable notification in the status panel.
Args:
reason: Why UART is unavailable.
"""
msg = f"UART 不可用 ({reason}),日志分析及检查将禁用"
self.after(0, lambda: self._status.set_status(msg, "warning"))
def _run_step_2_program_flash(self) -> bool:
"""Step 2: Program and verify Flash."""
if not self._config:
@@ -606,8 +635,21 @@ class MainWindow(ctk.CTk):
self._log_message(" Skipping reboot & boot verify (TFTP failed)")
return False
# 4b: Reboot
self._log_message("Step 4b: Rebooting Zynq...")
# 4a2: Post-download Zynq status check
self._log_message("Step 4b: Checking Zynq status after download...")
try:
zynq_status = verify_zynq_status(
self._config,
uart_available=self._uart_available,
callback=self._zynq_status_callback,
)
status = "" if zynq_status.success else ""
self._log_message(f" {status} {zynq_status.message}")
except Exception as e:
self._log_message(f" ✗ Zynq status check error: {e}")
# 4c: Reboot
self._log_message("Step 4c: Rebooting Zynq...")
try:
reboot_result = reboot_zynq(
self._config,
@@ -625,8 +667,8 @@ class MainWindow(ctk.CTk):
self._log_message(" Skipping boot verify (reboot failed)")
return False
# 4c: Verify boot
self._log_message("Step 4c: Verifying boot...")
# 4d: Verify boot
self._log_message("Step 4d: Verifying boot...")
try:
boot_result = verify_boot(
self._config,
@@ -659,6 +701,10 @@ class MainWindow(ctk.CTk):
"""Callback for boot verification progress."""
self._log_message(f" [{status}] {message}")
def _zynq_status_callback(self, status: str, message: str) -> None:
"""Callback for post-download Zynq status check."""
self._log_message(f" [{status}] {message}")
# ── Run All Steps ──────────────────────────────────────────
def _update_step_dependencies(self) -> None:
@@ -781,6 +827,12 @@ class MainWindow(ctk.CTk):
if start_index == 0 and not success:
break
# Inter-step delay (skip after last step)
if i < len(step_funcs) - 1:
delay = self._config.inter_step_delay if self._config else 2
self._log_message(f" ⏳ Waiting {delay}s before next step...")
time.sleep(delay)
self._progress.set_value(1.0)
self._is_running = False
self._run_btn.configure(state="normal", text="▶ Run All Steps")
+50
View File
@@ -108,6 +108,56 @@ def parse_boot_output(lines: list[str]) -> BootInfo:
return info
def check_uart_available(
port: str,
baudrate: int = 115200,
timeout: float = 3.0,
) -> tuple[bool, str]:
"""Check whether the UART serial port is available and readable.
Attempts to open the port, clear buffers, and read a few lines.
If the port opens but returns no data within the timeout, it is
considered available (the device may simply not be outputting yet).
If the port cannot be opened at all, it is unavailable.
Args:
port: Serial port device path (e.g., '/dev/ttyUSB0').
baudrate: Baud rate for serial communication.
timeout: Seconds to wait for data after opening.
Returns:
Tuple of (is_available, reason_string).
- (True, "Port open") — port is usable
- (True, "Port open, no data yet") — port open, no data currently
- (False, "No serial ports detected") — system has no serial ports
- (False, "Port not in detected ports") — configured port missing
- (False, "Failed to open: <error>") — port open error
"""
# 1. Quick port list check
ports = detect_serial_ports()
if not ports:
return False, "No serial ports detected"
# 2. Verify the configured port is among detected ones
if port and port not in {p.device for p in ports}:
return False, f"Port {port} not in detected ports"
# 3. Try to open and read
try:
with serial.Serial(port, baudrate, timeout=timeout) as ser:
ser.reset_input_buffer()
lines: list[str] = []
while ser.in_waiting:
line = ser.readline().decode("utf-8", errors="replace").strip()
if line:
lines.append(line)
if lines:
return True, "Port open"
return True, "Port open, no data yet"
except serial.SerialException as e:
return False, f"Failed to open: {e}"
def read_serial_stream(
port: str,
baudrate: int = 115200,