"""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, tftp_download, tftp_upload_verify, TftpResult, _compute_crc32 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, UartMonitor from file_utils import compute_file_info, FileInfo 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, SubStepFrame, TftpSubStepFrame, UartMonitorWindow, ) 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._uart_monitor: UartMonitor | None = None self._uart_window: UartMonitorWindow | None = None self._flash_phase: str = "" self._tftp_phase: str = "" # StringVar references for config sync self._ip_string_var: ctk.StringVar | None = None self._load_config() self._build_ui() # Validate file paths — clear any that don't exist on this machine if self._config and self._config._config_path: try: cleared = self._config.validate_paths() if cleared: joined = ", ".join(cleared) self._log_message(f" ⚠ Cleared stale paths from config: {joined}") # Also clear the UI selectors so stale paths don't # get re-saved by subsequent _auto_save_config calls _selector_map = { "bootloader_bit_path": self._bit_selector, "bootloader_elf_path": self._elf_selector, "bootloader_bin_path": self._bootloader_bin_selector, "fsbl_elf_path": self._fsbl_selector, "firmware_bin_path": self._firmware_bin_selector, } for attr in cleared: sel = _selector_map.get(attr) if sel: sel.set_path("") self._config.save() self._log_message(f" ✓ Config saved after clearing stale paths") except Exception as e: self._log_message(f" ✗ Config validation failed: {e}") # Log file metadata for all configured paths self._log_all_file_info() self._update_step_dependencies() # Handle window close self.protocol("WM_DELETE_WINDOW", self._on_close) # Defer blocking operations until after the window is displayed self.after(100, self._deferred_init) def _deferred_init(self) -> None: """Run blocking operations in background thread after window shown.""" def _bg_init(): self._check_vitis() self._refresh_ports_initial() threading.Thread(target=_bg_init, daemon=True).start() # ── Config ───────────────────────────────────────────────── def _find_config(self) -> Path | None: """Find the user configuration file. Checks common locations: 1. config.yaml (project root, user's config) 2. config/default_config.yaml (template, not for saving) 3. User-specified path via env var Returns: Path to user config file, or None if not found. """ # Navigate from src/gui/main_window.py -> project root app_dir = Path(__file__).parent.parent.parent # 1. Check for user config at project root user_config = app_dir / "config.yaml" if user_config.exists(): return user_config # 2. Check for default config (template) config_dir = app_dir / "config" default_config = config_dir / "default_config.yaml" if default_config.exists(): return default_config # 3. Check env var env_path = os.environ.get("ZYNQ_CONFIG") if env_path: return Path(env_path) return None def _get_user_config_path(self) -> Path: """Get the path to the user config file (config.yaml). Creates config.yaml in project root if it doesn't exist. Returns: Path to user config file. """ # Navigate from src/gui/main_window.py -> project root app_dir = Path(__file__).parent.parent.parent config_path = app_dir / "config.yaml" config_path.parent.mkdir(parents=True, exist_ok=True) return config_path def _load_config(self) -> None: """Load configuration from file or create a blank one on disk. If config.yaml does not exist or cannot be read, a new one is created with default values and written to disk immediately. """ user_config_path = self._get_user_config_path() if self._config_path: try: self._config = Config.from_file(self._config_path) if self._config_path.name == "default_config.yaml": self._config._config_path = user_config_path return except Exception: pass # Fall through and create default # Config file missing or unreadable — create a blank one self._config = Config.from_default() self._config._config_path = user_config_path try: self._config.save() except Exception: pass # Will be saved later by _auto_save_config def _reload_config(self) -> None: """Re-read config.yaml and update UI selectors with latest paths. Reads the file on disk, applies values to the internal Config object, then updates every UI selector so that changes made outside the GUI (or between runs) take effect immediately. """ user_path = self._get_user_config_path() if not user_path.exists(): return try: fresh = Config.from_file(user_path) self._config = fresh # Update UI selectors — suppress callbacks to avoid # redundant file-info logging and status-panel rewrites. _sels = [ self._bit_selector, self._elf_selector, self._bootloader_bin_selector, self._fsbl_selector, self._firmware_bin_selector, ] for s in _sels: s._suppress_callback = True for attr, selector in [ ("bootloader_bit_path", self._bit_selector), ("bootloader_elf_path", self._elf_selector), ("bootloader_bin_path", self._bootloader_bin_selector), ("fsbl_elf_path", self._fsbl_selector), ("firmware_bin_path", self._firmware_bin_selector), ]: val = getattr(fresh, attr, "") if val: resolved = fresh.resolve_path(val) selector.set_relative_path(val) selector.set_path(str(resolved)) for s in _sels: s._suppress_callback = False # Update IP if self._ip_string_var: self._ip_string_var.set(fresh.zynq_ip) # Update xilinx path if hasattr(fresh, 'xilinx_path') and self._xilinx_path_var: self._xilinx_path_var.set(fresh.xilinx_path) except Exception: pass # Keep existing config on error def _sync_config_from_ui(self) -> None: """Read current UI values and update the Config object. This must be called before _save_config() to ensure the Config object has the latest UI values. File paths are synced as relative paths (from the Config object's perspective); the FileSelector stores both the absolute path (for internal use) and the relative path (for saving to YAML). All attribute accesses are defensive — widgets may not exist yet during construction (e.g. _port_var created in _build_serial). """ if not self._config: return # Sync IP address (may not exist yet during _build_config_panel) if self._ip_string_var: self._config.zynq_ip = self._ip_string_var.get() # Sync Xilinx Kit path if hasattr(self, '_xilinx_path_var'): self._config.xilinx_path = self._xilinx_path_var.get() # Sync serial port (may not exist yet during _build_config_panel) if hasattr(self, '_port_var'): self._config.serial_port = self._port_var.get() # Sync file paths from FileSelectors # Always derive relative from absolute — the _relative_path # stored in FileSelector can be stale after Browse. for attr, selector in [ ("bootloader_bit_path", self._bit_selector), ("bootloader_elf_path", self._elf_selector), ("bootloader_bin_path", self._bootloader_bin_selector), ("fsbl_elf_path", self._fsbl_selector), ("firmware_bin_path", self._firmware_bin_selector), ]: abs_path = selector.get_path() if abs_path: setattr(self._config, attr, self._config._relative_path(abs_path)) else: setattr(self._config, attr, "") # Sync erase checkbox (may not exist yet) if hasattr(self, '_erase_cb_var'): self._config.erase_all = self._erase_cb_var.get() # Sync download verify checkbox if hasattr(self, '_download_verify_var'): self._config.download_verify = self._download_verify_var.get() def _auto_save_config(self) -> None: """Auto-save config to disk after UI changes. Ensures that changes made in the UI (file paths, IP, etc.) are persisted immediately so that _reload_config() picks them up on the next step execution. """ if not self._config: return self._sync_config_from_ui() user_config_path = self._get_user_config_path() self._config._config_path = user_config_path try: self._config.save() except Exception: pass # Silently fail — user can manually save def _save_config(self) -> None: """Save current configuration to user config file (config.yaml). Always saves to config/config.yaml, not to default_config.yaml. """ self._sync_config_from_ui() # Sync UI values first if not self._config: self._log_message(" No config loaded") return # Save to user config file (config.yaml) user_config_path = self._get_user_config_path() self._config._config_path = user_config_path # Set save path try: self._config.save() self._log_message(f" Config saved to {user_config_path}") self._status.set_status("Config saved", "success") except Exception as e: self._log_message(f" ✗ Save failed: {e}") self._status.set_status(f"Save failed: {e}", "error") def _on_erase_toggle(self) -> None: """Update config.erase_all immediately when checkbox toggles.""" if self._config: self._config.erase_all = self._erase_cb_var.get() def _on_download_verify_toggle(self) -> None: """Update config.download_verify immediately when checkbox toggles.""" if self._config: self._config.download_verify = self._download_verify_var.get() def _browse_xilinx_path(self) -> None: """Open directory dialog to select Xilinx root folder.""" from tkinter import filedialog path = filedialog.askdirectory(title="Select Xilinx Root Directory") if path: self._xilinx_path_var.set(path) self._on_xilinx_path_changed() # Re-check tools with new path self._check_vitis() def _on_xilinx_path_changed(self) -> None: """Handle Xilinx path entry change — sync config immediately.""" if self._config: self._config.xilinx_path = self._xilinx_path_var.get() def _on_ip_changed(self) -> None: """Handle IP address entry change — sync config immediately.""" if self._config: self._config.zynq_ip = self._ip_string_var.get() def _on_file_selected(self, path: str) -> None: """Handle file selection — sync, auto-save, and log file info.""" from file_utils import compute_file_info self._auto_save_config() if path: info = compute_file_info(path) self._log_file_info(info) # Update status info panel self._log_all_file_info() def _log_file_info(self, info: FileInfo) -> None: """Log file metadata in the log window. Args: info: FileInfo dataclass with computed metadata. """ filename = os.path.basename(info.path) if info.path else "(unknown)" if not info.exists: self._log_message(f" ⚠ {filename}: file not found") return self._log_message( f" ✓ {filename}: " f"{info.size_human} | CRC {info.crc32_hex} | {info.mtime_iso}" ) def _log_all_file_info(self) -> None: """Update file metadata in the status info panel (preserves other info). Each file line is updated in-place via ``set_info()`` — existing Zynq info, tool versions, etc. are left untouched. """ if not self._config: return paths = [ ("FSBL ELF", self._config.fsbl_elf_path), ("Bitstream", self._config.bootloader_bit_path), ("Bootloader ELF", self._config.bootloader_elf_path), ("Bootloader BIN", self._config.bootloader_bin_path), ("Firmware BIN", self._config.firmware_bin_path), ] for label, path in paths: if not path: self._status.set_info(label, "(not set)") continue resolved = self._config.resolve_path(path) info = compute_file_info(resolved) if info.exists: self._status.set_info( label, f"{info.size_human} CRC {info.crc32_hex} {info.mtime_iso}" ) else: self._status.set_info(label, "(not found)") # ── UI Construction ──────────────────────────────────────── def _build_ui(self) -> None: """Build the complete UI layout.""" # Scrollable main container — scrollbar appears when content overflows self.main_frame = ctk.CTkScrollableFrame(self) self.main_frame.pack(fill="both", expand=True, padx=PADDING_LARGE, pady=PADDING_LARGE) self.main_frame.grid_rowconfigure(1, weight=1) self.main_frame.grid_columnconfigure(0, weight=7, uniform="main_col") self.main_frame.grid_columnconfigure(1, weight=3, uniform="main_col") # ── Header ── self._build_header(self.main_frame) # ── Left Panel: Workflow Steps ── left_frame = ctk.CTkFrame(self.main_frame) left_frame.grid(row=1, column=0, sticky="nsew", padx=(0, PADDING)) left_frame.grid_rowconfigure(1, 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(self.main_frame) right_frame.grid(row=1, column=1, sticky="nsew", padx=(PADDING, 0)) right_frame.grid_rowconfigure(3, weight=3) # status panel — main info area right_frame.grid_columnconfigure(0, weight=1) self._build_config_panel(right_frame) self._build_serial_panel(right_frame) self._build_utility_panel(right_frame) self._build_status_panel(right_frame) # ── Maximize / resize handling ── self.bind("", self._on_window_resize) self._maximized = False 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(0, 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 (skip rows occupied by sub-steps) if i > 0: # Dividers go between main steps only, not between step and sub-step # Row calculation: account for sub-steps from previous steps prev_steps_with_substeps = sum(1 for j in range(i) if j in (1, 3)) divider_row = i * 2 + 1 + prev_steps_with_substeps divider = ctk.CTkFrame( panel, height=1, fg_color=CONNECTOR_COLOR, ) divider.grid( row=divider_row, column=0, sticky="ew", padx=PADDING, pady=(2, 0), ) # Position step (account for sub-steps from previous steps) prev_steps_with_substeps = sum(1 for j in range(i) if j in (1, 3)) step_row = i * 2 + 1 + prev_steps_with_substeps step.grid( row=step_row, column=0, sticky="nsew", padx=PADDING_SMALL, pady=(0, 2), ) # Step 2: add collapsible sub-step frame if i == 1: self._sub_step_frame = SubStepFrame(panel) self._sub_step_frame.grid( row=step_row + 1, column=0, sticky="nsew", padx=PADDING_SMALL + 16, pady=(0, 2), ) # Step 4: add collapsible sub-step frame if i == 3: self._tftp_sub_step_frame = TftpSubStepFrame(panel) self._tftp_sub_step_frame.grid( row=step_row + 1, column=0, sticky="nsew", padx=PADDING_SMALL + 16, pady=(0, 2), ) # ── Progress indicator ── self._progress = ProgressIndicator(panel) self._progress.grid( row=11, column=0, sticky="ew", padx=PADDING, pady=(PADDING, PADDING_SMALL), ) panel.grid_rowconfigure(11, minsize=40) # ── 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=12, column=0, sticky="nsew", padx=PADDING, pady=(0, 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, ) self._log_panel = panel 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(9, weight=1) panel.grid_columnconfigure(0, weight=1) # let content stretch title = ctk.CTkLabel( panel, text="Configuration", font=FONT_HEADING, ) title.grid(row=0, column=0, padx=PADDING, pady=PADDING) # ── Xilinx Kit path ── xilinx_frame = ctk.CTkFrame(panel) xilinx_frame.grid(row=1, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) xilinx_frame.grid_columnconfigure(1, weight=1) ctk.CTkLabel(xilinx_frame, text="Xilinx Kit:", font=FONT_BODY).grid(row=0, column=0, sticky="w") self._xilinx_path_var = ctk.StringVar(value=self._config.xilinx_path) self._xilinx_entry = ctk.CTkEntry(xilinx_frame, textvariable=self._xilinx_path_var) self._xilinx_entry.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL) # Bind real-time sync so changes take effect immediately self._xilinx_path_var.trace_add("write", lambda *args: self._on_xilinx_path_changed()) ctk.CTkButton( xilinx_frame, text="Browse", font=FONT_SMALL, width=70, command=self._browse_xilinx_path, ).grid(row=0, column=2, padx=(0, PADDING_SMALL)) # IP address ip_frame = ctk.CTkFrame(panel) ip_frame.grid(row=2, 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_string_var = ctk.StringVar(value=self._config.zynq_ip) self._ip_entry = ctk.CTkEntry(ip_frame, textvariable=self._ip_string_var) self._ip_entry.grid(row=0, column=1, sticky="nsew", padx=PADDING_SMALL) # Bind real-time sync self._ip_string_var.trace_add("write", lambda *args: self._on_ip_changed()) # FSBL ELF path (First Stage Bootloader — initialises PS, forces JTAG boot) self._fsbl_selector = FileSelector( panel, label_text="FSBL ELF (.elf):", file_types=[("ELF files", "*.elf"), ("All files", "*")], callback=self._on_file_selected, ) self._fsbl_selector._suppress_callback = True if self._config.fsbl_elf_path: resolved = self._config.resolve_path(self._config.fsbl_elf_path) self._fsbl_selector.set_relative_path(self._config.fsbl_elf_path) self._fsbl_selector.set_path(str(resolved)) self._fsbl_selector.grid(row=3, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # BIT path (Bootloader) self._bit_selector = FileSelector( panel, label_text="Bootloader BIT (.bit):", file_types=[("BIT files", "*.bit"), ("All files", "*")], callback=self._on_file_selected, ) self._bit_selector._suppress_callback = True if self._config.bootloader_bit_path: resolved = self._config.resolve_path(self._config.bootloader_bit_path) self._bit_selector.set_relative_path(self._config.bootloader_bit_path) self._bit_selector.set_path(str(resolved)) self._bit_selector.grid(row=4, 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", "*")], callback=self._on_file_selected, ) self._elf_selector._suppress_callback = True if self._config.bootloader_elf_path: resolved = self._config.resolve_path(self._config.bootloader_elf_path) self._elf_selector.set_relative_path(self._config.bootloader_elf_path) self._elf_selector.set_path(str(resolved)) self._elf_selector.grid(row=5, 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", "*")], callback=self._on_file_selected, ) self._bootloader_bin_selector._suppress_callback = True if self._config.bootloader_bin_path: resolved = self._config.resolve_path(self._config.bootloader_bin_path) self._bootloader_bin_selector.set_relative_path(self._config.bootloader_bin_path) self._bootloader_bin_selector.set_path(str(resolved)) self._bootloader_bin_selector.grid(row=6, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # Flash options flash_frame = ctk.CTkFrame(panel) flash_frame.grid(row=7, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) flash_frame.grid_columnconfigure(0, weight=1) self._erase_cb_var = ctk.BooleanVar(value=self._config.erase_all) self._erase_cb = ctk.CTkCheckBox( flash_frame, text="整片Flash擦除 (Erase entire flash)", variable=self._erase_cb_var, font=FONT_SMALL, command=self._on_erase_toggle, ) self._erase_cb.grid(row=0, column=0, sticky="w", padx=PADDING_SMALL) flash_model_label = ctk.CTkLabel( flash_frame, text=f"Flash: {self._config.flash_model} (32 MiB)", font=FONT_SMALL, ) flash_model_label.grid(row=1, column=0, sticky="w", padx=PADDING_SMALL, pady=(2, 0)) # Download verify checkbox (Step 4.2) self._download_verify_var = ctk.BooleanVar(value=self._config.download_verify) self._download_verify_cb = ctk.CTkCheckBox( flash_frame, text="下载校验 + CRC (Step 4.2)", variable=self._download_verify_var, font=FONT_SMALL, command=self._on_download_verify_toggle, ) self._download_verify_cb.grid(row=2, column=0, sticky="w", padx=PADDING_SMALL, pady=(2, 0)) # Firmware BIN path (TFTP upload) self._firmware_bin_selector = FileSelector( panel, label_text="Firmware BIN (.bin):", file_types=[("BIN files", "*.bin"), ("All files", "*")], callback=self._on_file_selected, ) self._firmware_bin_selector._suppress_callback = True if self._config.firmware_bin_path: resolved = self._config.resolve_path(self._config.firmware_bin_path) self._firmware_bin_selector.set_relative_path(self._config.firmware_bin_path) self._firmware_bin_selector.set_path(str(resolved)) self._firmware_bin_selector.grid(row=8, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL) # Enable callbacks now that all selectors are created for sel in (self._fsbl_selector, self._bit_selector, self._elf_selector, self._bootloader_bin_selector, self._firmware_bin_selector): sel._suppress_callback = False # Now sync and save config (all selectors exist) self._auto_save_config() # Save button save_btn = ctk.CTkButton( panel, text="Save Config", font=FONT_SMALL, command=self._save_config, ) save_btn.grid(row=9, 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_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, command=self._on_port_changed, ) 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) # Open UART button uart_btn = ctk.CTkButton( panel, text="Open UART", font=FONT_BODY, command=self._read_serial, ) uart_btn.grid(row=2, column=0, padx=PADDING, pady=PADDING_SMALL) def _build_utility_panel(self, parent: ctk.CTkFrame) -> None: """Build the utility tools panel.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=2, column=0, sticky="nsew", padx=PADDING, pady=PADDING, ) panel.grid_columnconfigure((0, 1), weight=1) title = ctk.CTkLabel( panel, text="Utilities", font=FONT_HEADING, ) title.grid(row=0, column=0, columnspan=2, padx=PADDING, pady=PADDING) # Row 1: Reboot buttons (both red) self._jtag_reboot_btn = ctk.CTkButton( panel, text="Reboot (JTAG)", font=FONT_BODY, fg_color=DANGER_COLOR, hover_color="#B92B2F", command=self._jtag_reboot, ) self._jtag_reboot_btn.grid(row=1, column=0, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew") self._udp_reboot_btn = ctk.CTkButton( panel, text="Reboot (UDP)", font=FONT_BODY, fg_color=DANGER_COLOR, hover_color="#B92B2F", command=self._udp_reboot, ) self._udp_reboot_btn.grid(row=1, column=1, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew") # Row 2: Version buttons self._udp_version_btn = ctk.CTkButton( panel, text="Version (UDP)", font=FONT_BODY, command=self._udp_version, ) self._udp_version_btn.grid(row=2, column=0, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew") self._test_version_btn = ctk.CTkButton( panel, text="Version (Serial)", font=FONT_BODY, command=self._test_serial_version, ) self._test_version_btn.grid(row=2, column=1, padx=PADDING_SMALL, pady=PADDING_SMALL, sticky="ew") def _build_status_panel(self, parent: ctk.CTkFrame) -> None: """Build the status display panel — uses full available space.""" panel = ctk.CTkFrame(parent, corner_radius=CORNER_RADIUS) panel.grid( row=3, column=0, sticky="nsew", padx=PADDING, pady=PADDING, ) panel.grid_rowconfigure((0, 1), weight=1) panel.grid_columnconfigure(0, weight=1) self._status = StatusDisplay(panel) self._status.grid(row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING) # UART warning — multi-line textbox, wraps long messages instead # of expanding the right panel horizontally self._uart_warning = ctk.CTkTextbox( panel, font=FONT_BODY, height=40, corner_radius=4, fg_color="transparent", state="disabled", wrap="word", ) self._uart_warning.grid( row=1, column=0, sticky="ew", padx=PADDING, pady=(0, PADDING_SMALL) ) self._uart_warning.grid_remove() # Hide by default # ── 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 absolute paths for immediate backend use. Returns: Dictionary with all file paths from UI selectors. """ 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(), "fsbl_elf_path": self._fsbl_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...") if self._config.xilinx_path: self._log_message(f" Xilinx root: {self._config.xilinx_path}") 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"]: ver_info = f" [v{tool_info['version']}]" if tool_info.get("version") else "" 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']}{ver_info}{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") # Show tool versions in status display for tool_name, tool_info in status_dict["tools"].items(): if tool_info["found"] and tool_info.get("version"): self._status.set_info(tool_name, f"v{tool_info['version']}") # Report all found versions for diagnostics if self._config.xilinx_path: try: from vitis_checker import get_all_found_versions for t in ("xsct", "program_flash", "bootgen"): vers = get_all_found_versions(self._config.xilinx_path, t) if len(vers) > 1: ver_list = ", ".join(f"{v} ({p})" for v, p in vers) self._log_message(f" {t} versions found: {ver_list}") except Exception: pass 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: detected_part = jtag_result.devices[0].part_number expected_part = self._config.zynq_part.upper() # Validate part number matches expected if detected_part != expected_part: self._log_message( f" ✗ Wrong Zynq part: detected {detected_part}, expected {expected_part}" ) self._jtag_status.configure( text=f"JTAG: {detected_part} ≠ {expected_part}", text_color=ERROR_COLOR ) return False self._zynq_jtag_present = True self._zynq_jtag_info = jtag_result.devices[0] part = detected_part # 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) self._status.set_info("Zynq", part) 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) 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, monitor=self._uart_monitor) 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 persistent UART unavailable notification. This notification stays visible and is not overwritten by other status updates. Args: reason: Why UART is unavailable. """ msg = f"⚠ UART 不可用 ({reason})，日志分析及检查将禁用" self.after(0, lambda: self._show_uart_warning(msg)) def _show_uart_warning(self, msg: str) -> None: """Display the persistent UART warning message (multi-line). Args: msg: Warning message to display. """ self._uart_warning.configure(state="normal") self._uart_warning.delete("1.0", "end") self._uart_warning.insert("1.0", msg) self._uart_warning.tag_config("warn", foreground=WARNING_COLOR) self._uart_warning.tag_add("warn", "1.0", "end") self._uart_warning.configure(state="disabled") self._uart_warning.grid() # Auto-size height based on line count line_count = msg.count("\n") + 1 self._uart_warning.configure(height=max(2, line_count)) def _hide_uart_warning(self) -> None: """Hide the persistent UART warning.""" self._uart_warning.grid_remove() def _on_window_resize(self, event) -> None: """Handle window resize / maximize events. When the window is maximized, scale content to fit one screen and hide scrollbars. When restored, use normal sizing with scrollbars on overflow. """ try: window_state = self.state() is_maximized = window_state == "zoomed" if is_maximized != self._maximized: self._maximized = is_maximized self._adjust_layout_for_maximize() except Exception: pass def _adjust_layout_for_maximize(self) -> None: """Scale content and toggle scrollbars based on maximized state.""" if not hasattr(self, 'main_frame'): return sf = self.main_frame has_scrollbars = hasattr(sf, '_scrollbar_y') if self._maximized: # Hide scrollbars when maximized if has_scrollbars: sf._scrollbar_y.grid_remove() sf._scrollbar_x.grid_remove() # Scale down to fit one screen self._scale_ui(0.6) # Hide the log panel when maximized to save space if hasattr(self, '_log_panel'): self._log_panel.grid_remove() else: # Show scrollbars when not maximized if has_scrollbars: sf._scrollbar_y.grid() sf._scrollbar_x.grid() # Restore normal scale self._scale_ui(1.0) # Show the log panel when not maximized if hasattr(self, '_log_panel'): self._log_panel.grid() def _scale_ui(self, factor: float) -> None: """Scale UI elements by a factor to fit screen. Args: factor: Scaling factor (1.0 = normal, <1.0 = smaller). """ # Scale font sizes in key labels scale_font = lambda orig: (orig[0], max(8, int(orig[1] * factor)), orig[2] if len(orig) > 2 else "") # Header fonts if hasattr(self, '_vitis_status'): self._vitis_status.configure(font=scale_font(FONT_BODY)) if hasattr(self, '_ip_status'): self._ip_status.configure(font=scale_font(FONT_BODY)) if hasattr(self, '_jtag_status'): self._jtag_status.configure(font=scale_font(FONT_BODY)) # Workflow panel fonts if hasattr(self, '_steps'): for step in self._steps: if hasattr(step, '_title_label'): step._title_label.configure(font=scale_font((FONT_BODY[0], 13, "bold"))) if hasattr(step, '_desc_label'): step._desc_label.configure(font=scale_font(FONT_SMALL)) if hasattr(step, '_action_btn'): step._action_btn.configure(font=scale_font((FONT_BODY[0], 12))) # Config panel fonts if hasattr(self, '_xilinx_entry'): self._xilinx_entry.configure(font=scale_font(FONT_BODY)) if hasattr(self, '_ip_entry'): self._ip_entry.configure(font=scale_font(FONT_BODY)) # Log display font if hasattr(self, '_log_display') and hasattr(self._log_display, '_text_widget'): self._log_display._text_widget.configure(font=scale_font(FONT_MONO)) # Status display font if hasattr(self, '_status') and hasattr(self._status, '_status_text'): self._status._status_text.configure(font=scale_font(FONT_BODY)) # Scale padding/spacing on all frames if hasattr(self, 'config_frame'): self.config_frame.configure(padx=int(5 * factor), pady=int(5 * factor)) if hasattr(self, 'workflow_frame'): self.workflow_frame.configure(padx=int(5 * factor), pady=int(5 * factor)) if hasattr(self, 'log_frame'): self.log_frame.configure(padx=int(5 * factor), pady=int(5 * factor)) if hasattr(self, 'status_frame'): self.status_frame.configure(padx=int(5 * factor), pady=int(5 * factor)) # Scale step frame spacing if hasattr(self, '_steps'): for step in self._steps: if hasattr(step, '_inner_frame'): step._inner_frame.configure(padx=int(5 * factor), pady=int(5 * factor)) def _on_port_changed(self, new_port: str) -> None: """Handle serial port selection change — validate immediately. Args: new_port: The newly selected serial port path. """ if not new_port: self._uart_available = False self._show_uart_notification("No port selected") if self._uart_monitor: self._stop_uart_monitor() return # Attempt to open port to validate try: port = new_port baudrate = self._config.serial_baudrate if self._config else 115200 available, reason = check_uart_available(port, baudrate) self._uart_available = available if available: self._uart_message = reason self._hide_uart_warning() self._log_message(f" ✓ UART {port}: {reason}") # If UART monitor was running, restart on new port if self._uart_monitor and self._uart_monitor.is_running(): self._stop_uart_monitor() self._start_uart_monitor() else: self._uart_available = False self._uart_message = reason self._show_uart_notification(reason) if self._uart_monitor: self._stop_uart_monitor() except Exception as e: self._uart_available = False self._show_uart_notification(str(e)) if self._uart_monitor: self._stop_uart_monitor() def _run_step_2_program_flash(self) -> bool: """Step 2: Program and verify Flash using program_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 if not files["fsbl_elf_path"]: self._log_message(" No FSBL ELF file selected (required by program_flash)") return False self._log_message(f"Step 2: Programming Flash with {files['bootloader_bin_path']}...") self._log_message(f" FSBL: {files['fsbl_elf_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 — drives sub-step UI.""" if status == "phase": self._flash_phase = message phase_names = {"erase": "Erasing", "program": "Programming", "verify": "Verifying", "done": "Done"} label = phase_names.get(message, message) self._log_message(f" ▸ {label}") if hasattr(self, '_sub_step_frame'): self._sub_step_frame.set_phase(message) return if status == "progress": try: pct = int(message.rstrip('%')) except ValueError: pct = -1 phase_label = {"erase": "Erase", "program": "Program", "verify": "Verify"}.get( self._flash_phase, "Flash") self._log_message(f" [{phase_label}] {pct}%") if hasattr(self, '_sub_step_frame'): self._sub_step_frame.set_phase(self._flash_phase, pct) return 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, ) dap_fail = False for result in results: status = "✓" if result.success else "✗" self._log_message(f" {status} {result.step}: {result.message}") if result.output: # Log tool output for diagnostics (truncated) output_lines = result.output.strip().splitlines() for line in output_lines[-20:]: # last 20 lines self._log_message(f" [xsct] {line}") if "DAP" in result.message and not result.success: dap_fail = True if dap_fail: self._log_message("") self._log_message(" ╔══════════════════════════════════════════╗") self._log_message(" ║ DAP ERROR — JTAG debug port hung ║") self._log_message(" ║ This is caused by prior flash programming║") self._log_message(" ║ → Power-cycle the board (unplug/replug) ║") self._log_message(" ║ → Then re-run Step 3 ║") self._log_message(" ╚══════════════════════════════════════════╝") self._log_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 + download verify + CRC + 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 bin_path = Path(files["firmware_bin_path"]) remote_name = self._config.tftp_upload_name all_results: list[bool] = [] # 4a: TFTP Upload self._log_message("Step 4a: TFTP upload...") self._tftp_phase = "Upload" if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_phase("Upload") try: tftp_result = tftp_upload( self._config, bin_path, remote_name, callback=self._tftp_sub_callback, ) status = "✓" if tftp_result.success else "✗" self._log_message(f" {status} {tftp_result.message}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_complete("Upload") if tftp_result.success else self._tftp_sub_step_frame.set_step_error("Upload") all_results.append(tftp_result.success) except Exception as e: self._log_message(f" ✗ TFTP upload error: {e}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_error("Upload") all_results.append(False) if not all_results[-1]: self._log_message(" Skipping download verify (TFTP upload failed)") return False # Delay to let Zynq TFTP server stabilize self._log_message(" ⏳ Waiting 2s before download...") time.sleep(2) # 4b+4c: TFTP Download Verify + CRC Check if not self._config.download_verify: self._log_message("Step 4b+4c: Download verify & CRC — skipped (disabled in config)") all_results.append(True) # 4b all_results.append(True) # 4c if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_skipped("Download Verify") self._tftp_sub_step_frame.set_step_skipped("CRC Check") else: # 4b: Download Verify self._log_message("Step 4b: TFTP download verify...") self._tftp_phase = "Download Verify" if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_phase("Download Verify") try: download_result = tftp_download( self._config, remote_name, callback=self._tftp_sub_callback, expected_size=bin_path.stat().st_size, ) status = "✓" if download_result.success else "✗" self._log_message(f" {status} {download_result.message}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_complete("Download Verify") if download_result.success else self._tftp_sub_step_frame.set_step_error("Download Verify") all_results.append(download_result.success) except Exception as e: self._log_message(f" ✗ TFTP download error: {e}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_error("Download Verify") all_results.append(False) if not all_results[-1]: self._log_message(" Skipping CRC check (download failed)") else: # 4c: CRC Check self._log_message("Step 4c: CRC check...") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_phase("CRC Check") original_crc = _compute_crc32(bin_path) downloaded_crc = download_result.crc_local crc_match = original_crc == downloaded_crc self._log_message(f" Original CRC: {original_crc:#010x}") self._log_message(f" Downloaded CRC: {downloaded_crc:#010x}") if crc_match: self._log_message(" ✓ CRC match") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_complete("CRC Check") all_results.append(True) else: self._log_message(" ✗ CRC mismatch") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_error("CRC Check") all_results.append(False) if not all_results[-1]: self._log_message(" Skipping reboot (download/CRC failed)") return False # 4d: Reboot self._log_message("Step 4d: Rebooting Zynq...") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_phase("Reboot") try: reboot_result = reboot_zynq( self._config, method="jtag", callback=self._reboot_callback, ) status = "✓" if reboot_result.success else "✗" self._log_message(f" {status} {reboot_result.message}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_complete("Reboot") if reboot_result.success else self._tftp_sub_step_frame.set_step_error("Reboot") all_results.append(reboot_result.success) except Exception as e: self._log_message(f" ✗ Reboot error: {e}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_error("Reboot") all_results.append(False) if not all_results[-1]: self._log_message(" Skipping boot verify (reboot failed)") return False # Wait for board to boot — UART if available, else config delay boot_delay = self._config.boot_wait_delay # Check both main monitor and UART window monitor uart_alive = False _uart_src = None # source of latest_info if self._uart_monitor and self._uart_monitor.is_running(): uart_alive = True _uart_src = self._uart_monitor elif (self._uart_window and self._uart_window.winfo_exists() and self._uart_window.is_monitoring): uart_alive = True _uart_src = self._uart_window discovered_ip: str = "" if uart_alive and _uart_src: self._log_message(f" ⏳ Waiting for UART boot signal (max {boot_delay}s)...") elapsed = 0.0 boot_seen = False # Phase 1 guard — log once, keep waiting for IP while elapsed < boot_delay: info = ( _uart_src.latest_info if hasattr(_uart_src, "latest_info") else _uart_src.get_latest_info() ) # Phase 2: IP address found — definitive boot confirmation if info and info.ip_address: self._log_message( f" ✓ Boot detected after {elapsed:.1f}s" f" (IP={info.ip_address}, Ver={info.version})" ) discovered_ip = info.ip_address break # Phase 1: boot signal (version) seen but no IP yet — keep polling if info and info.version and not boot_seen: boot_seen = True self._log_message( f" ✓ Boot signal after {elapsed:.1f}s" f" (Ver={info.version}), waiting for IP..." ) time.sleep(0.5) elapsed += 0.5 else: self._log_message(f" ⏱ UART timeout — no IP in {boot_delay}s") # Final check: IP may have arrived after the last poll info = ( _uart_src.latest_info if hasattr(_uart_src, "latest_info") else _uart_src.get_latest_info() ) if info and info.ip_address: self._log_message(f" IP recovered from latest UART info: {info.ip_address}") discovered_ip = info.ip_address else: self._log_message(f" ⏳ UART unavailable — sleeping {boot_delay}s for boot...") time.sleep(boot_delay) # Fallback: parse IP from accumulated UART buffer (lines may arrive # after boot wait loop exited — e.g. "Board IP: 192.168.100.13") if not discovered_ip and _uart_src: all_lines = ( _uart_src.all_lines if hasattr(_uart_src, "all_lines") else [] ) if all_lines: parsed = parse_boot_output(all_lines) if parsed.ip_address: discovered_ip = parsed.ip_address self._log_message(f" IP from UART buffer: {discovered_ip}") # Fallback: if no IP from UART, try ping → UDP scan, then extra delay if not discovered_ip: self._log_message(" ⏳ Network stack — waiting 3s...") time.sleep(3) # Try ping configured IP first (fast, reliable) self._log_message(f" Pinging {self._config.zynq_ip}...") ok, _ = ping_ip(self._config.zynq_ip, timeout=2) if ok: discovered_ip = self._config.zynq_ip self._log_message(f" ✓ Zynq responds to ping at {discovered_ip}") else: # Fallback: UDP broadcast scan self._log_message(" Scanning for Zynq IP (192.168.100.11–30)...") discovered_ip = self._discover_zynq_ip() if discovered_ip: self._log_message(f" ✓ Found Zynq at {discovered_ip}") else: self._log_message(" ⚠ Zynq not found — extra boot delay...") self._log_message(f" ⏳ Waiting {boot_delay}s for late boot...") time.sleep(boot_delay) # Retry ping + scan after extra delay ok2, _ = ping_ip(self._config.zynq_ip, timeout=2) if ok2: discovered_ip = self._config.zynq_ip self._log_message(f" ✓ Zynq responds to ping at {discovered_ip}") else: discovered_ip = self._discover_zynq_ip() if discovered_ip: self._log_message(f" ✓ Found Zynq at {discovered_ip}") else: self._log_message(" ⚠ Zynq not found on any scanned IP") if discovered_ip and discovered_ip != self._config.zynq_ip: self._log_message( f" IP changed: {self._config.zynq_ip} → {discovered_ip}" ) self._config.zynq_ip = discovered_ip if self._ip_string_var: self._ip_string_var.set(discovered_ip) # 4e: Verify boot self._log_message("Step 4e: Verifying boot...") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_phase("Boot Verify") 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}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_complete("Boot Verify") if boot_result.success else self._tftp_sub_step_frame.set_step_error("Boot Verify") all_results.append(boot_result.success) except Exception as e: self._log_message(f" ✗ Boot verify error: {e}") if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_error("Boot Verify") 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 _tftp_sub_callback(self, status: str, message: str) -> None: """Callback for TFTP sub-step progress — drives TftpSubStepFrame.""" if status == "progress": try: pct = int(message.rstrip('%')) except ValueError: pct = -1 if hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_phase(self._tftp_phase, pct) return 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 # Always re-read config.yaml before running self._reload_config() 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 # Always re-read config.yaml before running self._reload_config() 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, ] step_timeout_keys = [ ("check_env", 30), ("flash_program", 600), ("bitstream", 60), ("tftp_reboot", 120), ] # Start UART monitor if available self._start_uart_monitor() # 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") # Step 2: expand sub-step frame and reset if i == 1 and hasattr(self, '_sub_step_frame'): self._sub_step_frame.reset() self._sub_step_frame.set_expanded(True) self._flash_phase = "" # Step 4: expand sub-step frame and reset if i == 3 and hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.reset() self._tftp_sub_step_frame.set_expanded(True) # Log step timeout estimate timeout_key, default_timeout = step_timeout_keys[i] if i < len(step_timeout_keys) else ("", 30) step_timeout = self._config.step_timeouts.get(timeout_key, default_timeout) if self._config else default_timeout self._log_message(f" ⏱ Step {i+1} timeout: {step_timeout}s") self._progress.set_value(step_idx / total) t_step_start = time.time() try: success = step_funcs[i]() results.append(success) self._set_step_status(i, "complete" if success else "error") elapsed = time.time() - t_step_start self._log_message(f" ⏱ Step {i+1} completed in {elapsed:.0f}s") # Step 2: mark sub-steps based on result if i == 1 and hasattr(self, '_sub_step_frame'): if success: self._sub_step_frame.set_phase("done") else: self._sub_step_frame.set_phase("error") # Step 4: sub-steps already marked individually inside _run_step_4; # avoid set_phase which would restart a pulse on Boot Verify. if i == 3 and hasattr(self, '_tftp_sub_step_frame'): if not success: self._tftp_sub_step_frame.set_step_error("Boot Verify") except Exception as e: results.append(False) self._set_step_status(i, "error") self._log_message(f" Exception in step {i+1}: {e}") # Mark sub-step frames as error too if i == 1 and hasattr(self, '_sub_step_frame'): self._sub_step_frame.set_phase("error") if i == 3 and hasattr(self, '_tftp_sub_step_frame'): self._tftp_sub_step_frame.set_step_error("Boot Verify") self._progress.set_value((step_idx + 1) / total) # Stop on first failure when running all steps if not single_step and not success: # Special case: Step 2 pass → Step 3 DAP fail = board needs power cycle if i == 2 and results[0] and not success: self._log_message("") self._log_message(" ┌─────────────────────────────────────────┐") self._log_message(" │ Step 3 failed after flash programming. │") self._log_message(" │ This is normal — program_flash leaves │") self._log_message(" │ the JTAG DAP in error state. │") self._log_message(" │ → Power-cycle the board, then re-run │") self._log_message(" │ Step 3 (Load Bootloader) alone. │") self._log_message(" └─────────────────────────────────────────┘") self._log_message("") 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() # Stop UART monitor self._stop_uart_monitor() # ── UART Monitor ────────────────────────────────────────── def _start_uart_monitor(self) -> None: """Start background serial monitoring if UART is available.""" # If UART window is already monitoring, don't start a second instance if (self._uart_window and self._uart_window.winfo_exists() and self._uart_window.is_monitoring): self._log_message(" [UART] Already monitoring via UART window") return if not self._config: return port = self._config.serial_port if not port: return self._uart_monitor = UartMonitor(port, self._config.serial_baudrate) self._uart_monitor.on_line = lambda line: self._log_message(f" [UART] {line}") self._uart_monitor.on_boot_info = lambda info: ( self._log_message( f" [UART] Boot={info.boot_mode}, IP={info.ip_address}, Ver={info.version}" ) if any([info.boot_mode, info.ip_address, info.version]) else None ) self._uart_monitor.on_error = lambda e: self._log_message(f" [UART] Error: {e}") if self._uart_monitor.start(): self._log_message(" [UART] Serial monitor started") else: self._log_message(" [UART] Serial monitor failed to start (port may be busy)") def _stop_uart_monitor(self) -> None: """Stop background serial monitoring.""" if self._uart_monitor: self._uart_monitor.stop() self._uart_monitor = None # ── 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.""" self._stop_uart_monitor() if self._uart_window and self._uart_window.winfo_exists(): self._uart_window.destroy() self._uart_window = None if self._config: self._sync_config_from_ui() # Sync UI values first # Save to user config file (config.yaml) user_config_path = self._get_user_config_path() self._config._config_path = user_config_path try: self._config.save() except Exception: pass self.destroy() def _read_serial(self) -> None: """Open the UART Monitor window.""" port = self._port_var.get() baudrate = self._config.serial_baudrate if self._config else 115200 self._uart_window = UartMonitorWindow( self, port=port, baudrate=baudrate, on_log=self._log_message, ) def _test_serial_version(self) -> None: """Test serial port by sending 'ver()' command and parsing response.""" port = self._port_var.get() if not port: self._log_message("No serial port selected") return self._log_message(f"Testing version on {port}...") self._test_version_btn.configure(state="disabled", text="Testing...") # Run in background thread def test_version_thread(): try: from serial_monitor import test_serial_version success, message, version = test_serial_version( port, self._config.serial_baudrate if self._config else 115200, monitor=self._uart_monitor, ) if success: if version: self._log_message(f" ✓ {message}") self._log_message(f" ✓ Version detected: {version}") self._status.set_status(f"Version: {version}", "success") else: self._log_message(f" ✓ {message}") self._status.set_status("Port responded", "info") else: self._log_message(f" ✗ {message}") self._status.set_status(message, "error") except Exception as e: self._log_message(f" ✗ Error: {e}") self._status.set_status(f"Error: {e}", "error") finally: # Re-enable button self.after(0, lambda: self._test_version_btn.configure(state="normal", text="Version (Serial)")) threading.Thread(target=test_version_thread, daemon=True).start() def _jtag_reboot(self) -> None: """Reboot the Zynq board via JTAG system reset.""" if not self._config: self._log_message("No config loaded — cannot reboot") return self._log_message(f"JTAG reboot: sending system reset to {self._config.zynq_ip}...") self._jtag_reboot_btn.configure(state="disabled", text="Rebooting...") def _reboot_thread(): try: from reboot_manager import reboot_via_jtag result = reboot_via_jtag( self._config, callback=lambda s, m: self._log_message(f" [{s}] {m}"), ) if result.success: self._log_message(f" ✓ JTAG reset: {result.message}") self._status.set_status("JTAG reset complete", "success") else: self._log_message(f" ✗ JTAG reset: {result.message}") self._status.set_status(f"JTAG reset failed: {result.message}", "error") if result.output: self._log_message(f" Output: {result.output[:500]}") except Exception as e: self._log_message(f" ✗ JTAG reboot error: {e}") self._status.set_status(f"Error: {e}", "error") finally: self.after(0, lambda: self._jtag_reboot_btn.configure( state="normal", text="Reboot (JTAG)", )) threading.Thread(target=_reboot_thread, daemon=True).start() def _udp_reboot(self) -> None: """Reboot the Zynq board via UDP command.""" if not self._config: self._log_message("No config loaded — cannot reboot") return self._log_message(f"UDP reboot: sending reboot() to {self._config.zynq_ip}...") self._udp_reboot_btn.configure(state="disabled", text="Rebooting...") def _thread(): try: from reboot_manager import reboot_via_udp result = reboot_via_udp( self._config, callback=lambda s, m: self._log_message(f" [{s}] {m}"), ) if result.success: self._log_message(f" ✓ UDP reboot: {result.message}") self._status.set_status("UDP reboot complete", "success") else: self._log_message(f" ✗ UDP reboot: {result.message}") self._status.set_status(f"UDP reboot failed", "error") except Exception as e: self._log_message(f" ✗ UDP reboot error: {e}") self._status.set_status(f"Error: {e}", "error") finally: self.after(0, lambda: self._udp_reboot_btn.configure( state="normal", text="Reboot (UDP)", )) threading.Thread(target=_thread, daemon=True).start() def _udp_version(self) -> None: """Query Zynq firmware version via UDP.""" if not self._config: self._log_message("No config loaded") return self._log_message(f"UDP version: sending ver() to {self._config.zynq_ip}...") self._udp_version_btn.configure(state="disabled", text="Querying...") def _thread(): try: from reboot_manager import get_version_via_udp result = get_version_via_udp( self._config, callback=lambda s, m: self._log_message(f" [{s}] {m}"), ) if result.success: self._log_message(f" ✓ Version: {result.version}") self._status.set_status(f"Version: {result.version}", "success") else: self._log_message(f" ✗ {result.message}") self._status.set_status(f"Query failed", "error") except Exception as e: self._log_message(f" ✗ UDP version error: {e}") self._status.set_status(f"Error: {e}", "error") finally: self.after(0, lambda: self._udp_version_btn.configure( state="normal", text="Version (UDP)", )) threading.Thread(target=_thread, daemon=True).start() def _discover_zynq_ip(self) -> str: """Scan IP range to find the Zynq after reboot. Sends UDP ver() to each candidate IP (port = last_octet - 3) and returns the first IP that responds. Returns: Discovered IP address, or empty string if not found. """ subnet = "192.168.100" for last in range(11, 31): ip = f"{subnet}.{last}" try: from reboot_manager import _send_udp_command ok, _resp = _send_udp_command(ip, b"ver()", timeout=0.5) if ok: return ip except Exception: continue return "" # ── 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()