diff --git a/src/boot_verifier.py b/src/boot_verifier.py index 23446d9..10c2ec3 100644 --- a/src/boot_verifier.py +++ b/src/boot_verifier.py @@ -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, diff --git a/src/config_manager.py b/src/config_manager.py index 73941cd..ce030ba 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -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 diff --git a/src/gui/main_window.py b/src/gui/main_window.py index ac5e18c..03a8c26 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -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") diff --git a/src/serial_monitor.py b/src/serial_monitor.py index 5922b71..a553ae0 100644 --- a/src/serial_monitor.py +++ b/src/serial_monitor.py @@ -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: ") — 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,