"""Main GUI window for Zynq XC7Z100 Flasher. Integrates all backend modules into a modern, step-by-step workflow GUI built with CustomTkinter. """ from __future__ import annotations import os import threading import time import tkinter as tk from pathlib import Path from typing import Callable import customtkinter as ctk from config_manager import Config from vitis_checker import check_vitis, get_tool_status_dict from zynq_checker import check_zynq_jtag, get_zynq_status_dict from flash_programmer import full_flash_program, FlashResult 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, 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, WINDOW_TITLE, FONT_TITLE, FONT_HEADING, FONT_BODY, FONT_SMALL, FONT_MONO, PADDING, PADDING_LARGE, PADDING_SMALL, CORNER_RADIUS, PRIMARY_COLOR, SUCCESS_COLOR, WARNING_COLOR, DANGER_COLOR, INFO_COLOR, CONNECTOR_COLOR, get_theme, ) from gui.widgets import ( StepHeader, ProgressIndicator, StatusDisplay, FileSelector, LogDisplay, ) class MainWindow(ctk.CTk): """Main application window for Zynq Flasher GUI.""" def __init__(self) -> None: """Initialize the main window.""" super().__init__() self.title(WINDOW_TITLE) self.geometry(f"{WINDOW_WIDTH}x{WINDOW_HEIGHT}") self.minsize(800, 600) ctk.set_appearance_mode("dark") ctk.set_default_color_theme("blue") self._config: Config | None = None self._config_path = self._find_config() self._steps: list[StepHeader] = [] 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._zynq_jtag_present: bool = False self._zynq_jtag_info = None self._load_config() self._build_ui() self._check_vitis() self._refresh_ports_initial() self._update_step_dependencies() # Handle window close self.protocol("WM_DELETE_WINDOW", self._on_close) # ── Config ───────────────────────────────────────────────── def _find_config(self) -> Path | None: """Find the configuration file. Checks common locations: 1. config/default_config.yaml (relative to app) 2. config/zynq_flasher.yaml 3. User-specified path via env var Returns: Path to config file, or None if not found. """ candidates = [ Path(__file__).parent.parent / "config" / "default_config.yaml", ] env_path = os.environ.get("ZYNQ_CONFIG") if env_path: candidates.insert(0, Path(env_path)) for path in candidates: if path.exists(): return path return None def _load_config(self) -> None: """Load configuration from file or create default.""" if self._config_path: try: self._config = Config.from_file(self._config_path) except Exception: self._config = Config.from_default() else: self._config = Config.from_default() def _save_config(self) -> None: """Save current configuration to file.""" if self._config and self._config.config_path: self._config.save() # ── UI Construction ──────────────────────────────────────── def _build_ui(self) -> None: """Build the complete UI layout.""" # Main container with padding main_frame = ctk.CTkFrame(self) main_frame.pack(fill="both", expand=True, padx=PADDING_LARGE, pady=PADDING_LARGE) main_frame.grid_rowconfigure(1, weight=1) main_frame.grid_columnconfigure(0, weight=1) # ── Header ── self._build_header(main_frame) # ── Left Panel: Workflow Steps ── left_frame = ctk.CTkFrame(main_frame) left_frame.grid(row=1, column=0, sticky="nsew", padx=(0, PADDING)) left_frame.grid_rowconfigure(3, weight=1) left_frame.grid_columnconfigure(0, weight=1) self._build_workflow_panel(left_frame) self._build_log_panel(left_frame) # ── Right Panel: Configuration ── right_frame = ctk.CTkFrame(main_frame) right_frame.grid(row=1, column=1, sticky="nsew", padx=(PADDING, 0)) right_frame.grid_rowconfigure(4, weight=1) self._build_config_panel(right_frame) self._build_serial_panel(right_frame) self._build_status_panel(right_frame) def _build_header(self, parent: ctk.CTkFrame) -> None: """Build the application header.""" header = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) header.grid(row=0, column=0, columnspan=2, sticky="nsew", pady=(0, PADDING)) header.grid_columnconfigure(1, weight=1) title = ctk.CTkLabel( header, text="Zynq XC7Z100 Flasher", font=FONT_TITLE, ) title.grid(row=0, column=0, padx=PADDING, pady=PADDING) self._vitis_status = ctk.CTkLabel( header, text="Vitis: Checking...", font=FONT_BODY, ) self._vitis_status.grid(row=0, column=1, sticky="e", padx=PADDING) self._ip_status = ctk.CTkLabel( header, text=f"IP: {self._config.zynq_ip}", font=FONT_BODY, ) self._ip_status.grid(row=0, column=2, sticky="e", padx=PADDING) self._jtag_status = ctk.CTkLabel( header, text="JTAG: Checking...", font=FONT_BODY, ) self._jtag_status.grid(row=0, column=3, sticky="e", padx=PADDING) def _build_workflow_panel(self, parent: ctk.CTkFrame) -> None: """Build the step-by-step workflow panel.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=0, column=0, sticky="nsew", padx=PADDING, pady=(PADDING, 0), ) panel.grid_rowconfigure(10, weight=1) panel.grid_columnconfigure(0, weight=1) # ── Title ── title = ctk.CTkLabel( panel, text="Workflow", font=FONT_HEADING, ) title.grid(row=0, column=0, padx=PADDING_SMALL, pady=(PADDING_SMALL, PADDING_SMALL)) # ── Steps ── step_defs = [ {"num": 1, "title": "Check Environment", "desc": "Verify tools & connectivity"}, {"num": 2, "title": "Program & Verify Flash", "desc": "Wipe, program, and verify flash"}, {"num": 3, "title": "Load Bootloader", "desc": "Program BIT + run ELF"}, {"num": 4, "title": "Load Main Program", "desc": "TFTP upload, reboot, verify boot"}, ] self._steps = [] self._step_statuses = {} for i, defn in enumerate(step_defs): step = StepHeader( panel, step_number=defn["num"], title=defn["title"], description=defn["desc"], callback=lambda idx=i: self._execute_step(idx), can_run=(i == 0), ) self._steps.append(step) self._step_statuses[i] = "pending" # Divider line between steps if i > 0: divider = ctk.CTkFrame( panel, height=1, fg_color=CONNECTOR_COLOR, ) divider.grid( row=i * 2, column=0, sticky="ew", padx=PADDING, pady=(2, 0), ) # Position step step.grid( row=i * 2 + 1, column=0, sticky="nsew", padx=PADDING_SMALL, pady=(0, 2), ) # ── Progress indicator ── self._progress = ProgressIndicator(panel) self._progress.grid( row=8, column=0, sticky="nsew", padx=PADDING, pady=(PADDING, PADDING_SMALL), ) # ── Run All button ── self._run_btn = ctk.CTkButton( panel, text="▶ Run All Steps", font=FONT_HEADING, command=self._run_all_steps, height=40, corner_radius=CORNER_RADIUS, ) self._run_btn.grid( row=9, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL, ) def _build_log_panel(self, parent: ctk.CTkFrame) -> None: """Build the log display panel.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING, ) panel.grid_rowconfigure(1, weight=1) panel.grid_columnconfigure(0, weight=1) title = ctk.CTkLabel( panel, text="Log", font=FONT_HEADING, ) title.grid(row=0, column=0, padx=PADDING, pady=PADDING) self._log_display = LogDisplay(panel, height=250) self._log_display.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING) # Clear button clear_btn = ctk.CTkButton( panel, text="Clear Log", font=FONT_SMALL, command=self._log_display.clear, width=100, ) clear_btn.grid(row=0, column=1, padx=PADDING, pady=PADDING) def _build_config_panel(self, parent: ctk.CTkFrame) -> None: """Build the configuration panel.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING, ) panel.grid_rowconfigure(5, weight=1) title = ctk.CTkLabel( panel, text="Configuration", font=FONT_HEADING, ) title.grid(row=0, column=0, padx=PADDING, pady=PADDING) # IP address ip_frame = ctk.CTkFrame(panel) ip_frame.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) ip_frame.grid_columnconfigure(1, weight=1) ctk.CTkLabel(ip_frame, text="Zynq IP:", font=FONT_BODY).grid(row=0, column=0, sticky="w") self._ip_entry = ctk.CTkEntry(ip_frame, textvariable=ctk.StringVar(value=self._config.zynq_ip)) self._ip_entry.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL) # BIT path (Bootloader) self._bit_selector = FileSelector( panel, label_text="Bootloader BIT (.bit):", file_types=[("BIT files", "*.bit"), ("All files", "*")], ) if self._config.bootloader_bit_path: resolved = self._config.resolve_path(self._config.bootloader_bit_path) self._bit_selector.set_path(str(resolved)) self._bit_selector.grid(row=2, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # ELF path (Bootloader) self._elf_selector = FileSelector( panel, label_text="Bootloader ELF (.elf):", file_types=[("ELF files", "*.elf"), ("All files", "*")], ) if self._config.bootloader_elf_path: resolved = self._config.resolve_path(self._config.bootloader_elf_path) self._elf_selector.set_path(str(resolved)) self._elf_selector.grid(row=3, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # Bootloader BIN path (Flash programming) self._bootloader_bin_selector = FileSelector( panel, label_text="Bootloader BIN (.bin):", file_types=[("BIN files", "*.bin"), ("All files", "*")], ) if self._config.bootloader_bin_path: resolved = self._config.resolve_path(self._config.bootloader_bin_path) self._bootloader_bin_selector.set_path(str(resolved)) self._bootloader_bin_selector.grid(row=4, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # Firmware BIN path (TFTP upload) self._firmware_bin_selector = FileSelector( panel, label_text="Firmware BIN (.bin):", file_types=[("BIN files", "*.bin"), ("All files", "*")], ) if self._config.firmware_bin_path: resolved = self._config.resolve_path(self._config.firmware_bin_path) self._firmware_bin_selector.set_path(str(resolved)) self._firmware_bin_selector.grid(row=5, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # Save button save_btn = ctk.CTkButton( panel, text="Save Config", font=FONT_SMALL, command=self._save_config, ) save_btn.grid(row=6, column=0, padx=PADDING, pady=PADDING) def _build_serial_panel(self, parent: ctk.CTkFrame) -> None: """Build the serial monitor panel.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING, ) panel.grid_rowconfigure(3, weight=1) panel.grid_columnconfigure(0, weight=1) title = ctk.CTkLabel( panel, text="Serial Monitor", font=FONT_HEADING, ) title.grid(row=0, column=0, padx=PADDING, pady=PADDING) # Port selector port_frame = ctk.CTkFrame(panel) port_frame.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) port_frame.grid_columnconfigure(1, weight=1) ctk.CTkLabel(port_frame, text="Port:", font=FONT_BODY).grid(row=0, column=0, sticky="w") self._port_var = ctk.StringVar(value=self._config.serial_port) self._port_menu = ctk.CTkComboBox( port_frame, values=[""] + [p.device for p in detect_serial_ports()], variable=self._port_var, ) self._port_menu.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL) # Refresh button refresh_btn = ctk.CTkButton( port_frame, text="Refresh", font=FONT_SMALL, command=self._refresh_ports, width=80, ) refresh_btn.grid(row=0, column=2, padx=PADDING_SMALL) # Read button read_btn = ctk.CTkButton( panel, text="Read Serial", font=FONT_BODY, command=self._read_serial, ) read_btn.grid(row=2, column=0, padx=PADDING, pady=PADDING_SMALL) def _build_status_panel(self, parent: ctk.CTkFrame) -> None: """Build the status display panel.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=2, column=0, sticky="nsew", padx=PADDING, pady=PADDING, ) self._status = StatusDisplay(panel) self._status.pack(fill="x", padx=PADDING, pady=PADDING) # ── Workflow Steps ───────────────────────────────────────── def _log_message(self, message: str) -> None: """Append a message to the log display (thread-safe).""" self.after(0, lambda: self._log_display.append(message)) def _set_step_status(self, index: int, status: str) -> None: """Set the status of a workflow step. Args: index: Step index (0-based). status: Status string ('pending', 'running', 'complete', 'error'). """ if 0 <= index < len(self._steps): self._steps[index].set_status(status) def _get_selected_files(self) -> dict: """Get selected file paths from the UI. Returns: Dictionary with bootloader_bit_path, bootloader_elf_path, bootloader_bin_path, firmware_bin_path. """ return { "bootloader_bit_path": self._bit_selector.get_path(), "bootloader_elf_path": self._elf_selector.get_path(), "bootloader_bin_path": self._bootloader_bin_selector.get_path(), "firmware_bin_path": self._firmware_bin_selector.get_path(), } # ── Step Implementations ─────────────────────────────────── def _run_step_1_check_environment(self) -> bool: """Step 1: Check environment (Vitis/Vivado tools + Zynq IP + UART).""" self._log_message("Step 1: Checking environment...") if not self._config: self._log_message(" No config loaded") return False # Check Vitis/Vivado tools self._log_message(" Checking Vitis/Vivado tools...") result = check_vitis(self._config) status_dict = get_tool_status_dict(result) for tool_name, tool_info in status_dict["tools"].items(): if tool_info["found"]: alias_note = f" (via {tool_info['alias']})" if tool_info.get("alias") and tool_info["alias"] != tool_name else "" self._log_message(f" ✓ {tool_name}: {tool_info['path']}{alias_note}") else: self._log_message(f" ✗ {tool_name}: {tool_info['error']}") if result.is_ready: self._vitis_status.configure(text="Vitis: Ready", text_color=SUCCESS_COLOR) self._log_message(" Vitis/Vivado tools available") else: self._vitis_status.configure(text="Vitis: Partial", text_color=WARNING_COLOR) self._log_message(f" Issues: {'; '.join(result.errors)}") # Check Zynq JTAG presence 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) # Log JTAG chain information if jtag_result.jtag_chain: self._log_message(f" JTAG chain: {len(jtag_result.jtag_chain)} device(s)") for jtag_dev in jtag_result.jtag_chain: dev_type = "ARM DAP (PS)" if jtag_dev.is_arm_dap else "PL" if jtag_dev.is_pl_device else "Other" self._log_message(f" - {jtag_dev.name} ({dev_type})") if jtag_result.is_present: self._zynq_jtag_present = True self._zynq_jtag_info = jtag_result.devices[0] part = jtag_result.devices[0].part_number # Log PS and PL detection status ps_status = "✓" if jtag_result.ps_detected else "✗" pl_status = "✓" if jtag_result.pl_detected else "✗" self._log_message(f" PS (ARM DAP): {ps_status}") self._log_message(f" PL: {pl_status}") self._log_message(f" ✓ Zynq {part} detected on JTAG chain") self._jtag_status.configure(text=f"JTAG: {part} ✓", text_color=SUCCESS_COLOR) else: self._zynq_jtag_present = False self._zynq_jtag_info = None self._log_message(f" ✗ Zynq not detected on JTAG: {jtag_result.error}") self._jtag_status.configure(text="JTAG: Not detected", text_color=WARNING_COLOR) # 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: return False files = self._get_selected_files() if not files["bootloader_bin_path"]: self._log_message(" No Bootloader BIN file selected") return False self._log_message(f"Step 2: Programming Flash with {files['bootloader_bin_path']}...") try: bin_path = Path(files["bootloader_bin_path"]) results = full_flash_program(self._config, bin_path, self._flash_callback) for result in results: status = "✓" if result.success else "✗" self._log_message(f" {status} {result.step}: {result.message}") return all(r.success for r in results) except Exception as e: self._log_message(f" ✗ Flash error: {e}") return False def _flash_callback(self, status: str, message: str) -> None: """Callback for flash programming progress.""" self._log_message(f" [{status}] {message}") def _jtag_callback(self, status: str, message: str) -> None: """Callback for JTAG scan progress.""" self._log_message(f" [{status}] {message}") def _run_step_3_load_bootloader(self) -> bool: """Step 3: Load bootloader (BIT + ELF) and verify.""" if not self._config: return False files = self._get_selected_files() if not files["bootloader_bit_path"]: self._log_message(" No Bootloader BIT file selected") return False self._log_message(f"Step 3: Loading bootloader (BIT + ELF)...") self._log_message(f" BIT: {files['bootloader_bit_path']}") if files["bootloader_elf_path"]: self._log_message(f" ELF: {files['bootloader_elf_path']}") try: bit_path = Path(files["bootloader_bit_path"]) elf_path = Path(files["bootloader_elf_path"]) if files["bootloader_elf_path"] else bit_path.parent / "temp.elf" results = full_bitstream_program( self._config, bit_path, elf_path, self._bitstream_callback, ) for result in results: status = "✓" if result.success else "✗" self._log_message(f" {status} {result.step}: {result.message}") return all(r.success for r in results) except Exception as e: self._log_message(f" ✗ Bootloader error: {e}") return False def _bitstream_callback(self, status: str, message: str) -> None: """Callback for bitstream programming progress.""" self._log_message(f" [{status}] {message}") def _run_step_4_load_main_program(self) -> bool: """Step 4: Load main program (TFTP upload + reboot + boot verify).""" if not self._config: return False files = self._get_selected_files() if not files["firmware_bin_path"]: self._log_message(" No Firmware BIN file selected") return False all_results: list[bool] = [] # 4a: TFTP upload self._log_message("Step 4a: TFTP upload & verify...") try: bin_path = Path(files["firmware_bin_path"]) tftp_result = tftp_upload_verify( self._config, bin_path, callback=self._tftp_callback, ) status = "✓" if tftp_result.success else "✗" self._log_message(f" {status} {tftp_result.message}") all_results.append(tftp_result.success) except Exception as e: self._log_message(f" ✗ TFTP error: {e}") all_results.append(False) if not all_results[-1]: self._log_message(" Skipping reboot & boot verify (TFTP failed)") return False # 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, method="auto", callback=self._reboot_callback, ) status = "✓" if reboot_result.success else "✗" self._log_message(f" {status} {reboot_result.message}") all_results.append(reboot_result.success) except Exception as e: self._log_message(f" ✗ Reboot error: {e}") all_results.append(False) if not all_results[-1]: self._log_message(" Skipping boot verify (reboot failed)") return False # 4d: Verify boot self._log_message("Step 4d: Verifying boot...") try: boot_result = verify_boot( self._config, callback=self._boot_callback, ) status = "✓" if boot_result.success else "✗" self._log_message(f" {status} {boot_result.message}") if boot_result.ip_reachable: self._log_message(f" IP {self._config.zynq_ip} is reachable") if boot_result.serial_found: self._log_message(f" Boot info: {boot_result.boot_info}") all_results.append(boot_result.success) except Exception as e: self._log_message(f" ✗ Boot verify error: {e}") all_results.append(False) return all(all_results) def _tftp_callback(self, status: str, message: str) -> None: """Callback for TFTP progress.""" self._log_message(f" [{status}] {message}") def _reboot_callback(self, status: str, message: str) -> None: """Callback for reboot progress.""" self._log_message(f" [{status}] {message}") def _boot_callback(self, status: str, message: str) -> None: """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: """Update step run-ability based on dependency status. All steps are runnable independently. """ for i, step in enumerate(self._steps): step.set_can_run(True) status = self._step_statuses.get(i, "pending") if status == "pending": step.set_status("pending") elif status == "complete": step.set_status("complete") elif status == "error": step.set_status("error") def _execute_step(self, index: int) -> None: """Execute a single step independently. Args: index: Step index (0-based). """ if self._is_running: return if index >= len(self._steps): return self._is_running = True self._run_btn.configure(state="disabled", text="Running...") self._progress.reset() # Reset status for this step only self._step_statuses[index] = "pending" self._steps[index].set_status("pending") self._progress.set_value(0.0) # Start background worker thread with single_step=True self._worker_thread = threading.Thread( target=self._run_steps_worker, daemon=True, args=(index, True), # single_step=True ) self._worker_thread.start() def _set_step_status(self, index: int, status: str) -> None: """Update the status of a step in the UI. Args: index: Step index (0-based). status: New status string. """ self._step_statuses[index] = status self._steps[index].set_status(status) def _run_all_steps(self) -> None: """Execute all workflow steps sequentially in a background thread.""" if self._is_running: return self._is_running = True self._run_btn.configure(state="disabled", text="Running...") self._progress.reset() # Reset all step statuses for i in range(len(self._steps)): self._step_statuses[i] = "pending" self._steps[i].set_status("pending") # Start background worker thread self._worker_thread = threading.Thread( target=self._run_steps_worker, daemon=True, args=(0, False), # single_step=False ) self._worker_thread.start() def _run_steps_worker(self, start_index: int = 0, single_step: bool = False) -> None: """Worker method that runs workflow steps in the background. Args: start_index: Index of the first step to run (0-based). single_step: If True, only run the step at start_index. """ step_funcs = [ self._run_step_1_check_environment, self._run_step_2_program_flash, self._run_step_3_load_bootloader, self._run_step_4_load_main_program, ] # Determine end index based on single_step flag end_index = start_index + 1 if single_step else len(step_funcs) total = end_index - start_index results: list[bool] = [] for i in range(start_index, end_index): step_idx = i - start_index self._set_step_status(i, "running") self._progress.set_value(step_idx / total) try: success = step_funcs[i]() results.append(success) self._set_step_status(i, "complete" if success else "error") except Exception as e: results.append(False) self._set_step_status(i, "error") self._log_message(f" Exception in step {i+1}: {e}") self._progress.set_value((step_idx + 1) / total) # Stop on first failure when running all steps if not single_step 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") # Update dependency statuses self._update_step_dependencies() # Set final status message if results: all_ok = all(results) any_fail = any(not r for r in results) if all_ok: self._status.set_status("Workflow complete — all steps passed", "success") elif any_fail: failed_count = results.count(False) self._status.set_status( f"Workflow complete — {failed_count}/{len(results)} step(s) failed", "warning", ) else: self._status.set_status("Workflow complete", "info") else: self._status.set_status("Workflow complete", "info") # Save config after running self._save_config() # ── Serial ───────────────────────────────────────────────── def _refresh_ports(self) -> None: """Refresh the serial port list.""" ports = detect_serial_ports() self._port_menu.configure(values=[""] + [p.device for p in ports]) self._log_message(f"Found {len(ports)} serial port(s)") def _refresh_ports_initial(self) -> None: """Refresh the serial port list on startup (silent, no log).""" ports = detect_serial_ports() self._port_menu.configure(values=[""] + [p.device for p in ports]) # Auto-select if only one port found and config has one if ports and self._config and self._config.serial_port: devices = [p.device for p in ports] if self._config.serial_port in devices: self._port_var.set(self._config.serial_port) def _on_close(self) -> None: """Handle window close event: save config and cleanup.""" if self._config: try: self._config.save() except Exception: pass self.destroy() def _read_serial(self) -> None: """Read and display serial output.""" port = self._port_var.get() if not port: self._log_message("No serial port selected") return self._log_message(f"Reading from {port}...") try: import serial with serial.Serial(port, self._config.serial_baudrate if self._config else 115200, timeout=5) as ser: ser.reset_input_buffer() lines = [] while ser.in_waiting: line = ser.readline().decode("utf-8", errors="replace").strip() if line: lines.append(line) if lines: info = parse_boot_output(lines) self._log_message(f" Parsed: IP={info.ip_address}, Ver={info.version}") for line in lines[-10:]: self._log_message(f" > {line}") else: self._log_message(" No data received") except Exception as e: self._log_message(f" Error: {e}") # ── Vitis Check ──────────────────────────────────────────── def _check_vitis(self) -> None: """Check Vitis/Vivado on startup.""" if not self._config: return result = check_vitis(self._config) status_dict = get_tool_status_dict(result) if result.is_ready: self._vitis_status.configure(text="Vitis: Ready", text_color=SUCCESS_COLOR) self._log_message("Vitis/Vivado tools available") else: self._vitis_status.configure(text="Vitis: Partial", text_color=WARNING_COLOR) self._log_message(f"Vitis issues: {'; '.join(result.errors)}") # ── Entry Point ──────────────────────────────────────────────── def main() -> None: """Launch the Zynq Flasher GUI application.""" app = MainWindow() app.mainloop() if __name__ == "__main__": main()