From 639f2e3ce6b59f0681d6348d153c16adba6d7670 Mon Sep 17 00:00:00 2001 From: Jeremy Shen Date: Fri, 12 Jun 2026 15:47:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=205=20improvements=20=E2=80=94=20file=20i?= =?UTF-8?q?nfo,=20download=20verify,=20UART=20warning,=20status=20display,?= =?UTF-8?q?=20COM=20port=20reuse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - file_utils: compute_file_info() — CRC32, human-readable size, mtime - StatusDisplay: simplified to info-only key-value panel; status messages removed (log-only); file info shown in status panel at startup/select - config: added download_verify checkbox (default ON), skips Step 4.2 - UART warning: CTkTextbox with word-wrap and auto-height - COM port: check_uart_available / test_serial_version accept optional UartMonitor to avoid redundant serial opens (Error 13 on Windows) - layout: left:right 7:3 ratio, config panel stretches full width --- src/config_manager.py | 5 +- src/file_utils.py | 82 ++++++++++++++++++ src/gui/main_window.py | 188 ++++++++++++++++++++++++++++++++--------- src/gui/widgets.py | 81 +++++++++++------- src/serial_monitor.py | 32 ++++--- 5 files changed, 305 insertions(+), 83 deletions(-) create mode 100644 src/file_utils.py diff --git a/src/config_manager.py b/src/config_manager.py index e881f15..0b5201d 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -28,13 +28,14 @@ DEFAULT_CONFIG: dict[str, Any] = { "flash_type": "qspi-x4-single", "flash_model": "s25fl256s1", "erase_all": False, + "download_verify": True, "firmware_bin_path": "", "reboot_timeout": 30, "ping_timeout": 5, "ping_count": 3, "uart_delay": 3, "inter_step_delay": 2, - "boot_wait_delay": 10, + "boot_wait_delay": 30, "step_timeouts": { "check_env": 120, "flash_erase": 120, @@ -67,6 +68,7 @@ class Config: flash_type: str = "qspi-x4-single" flash_model: str = "s25fl256s1" erase_all: bool = False + download_verify: bool = True firmware_bin_path: str = "" reboot_timeout: int = 30 ping_timeout: int = 5 @@ -176,6 +178,7 @@ class Config: "flash_type": self.flash_type, "flash_model": self.flash_model, "erase_all": self.erase_all, + "download_verify": self.download_verify, "firmware_bin_path": self._relative_path(self.firmware_bin_path), "reboot_timeout": self.reboot_timeout, "ping_timeout": self.ping_timeout, diff --git a/src/file_utils.py b/src/file_utils.py new file mode 100644 index 0000000..16cecd3 --- /dev/null +++ b/src/file_utils.py @@ -0,0 +1,82 @@ +"""File metadata utilities — CRC32, human-readable size, timestamps.""" + +from __future__ import annotations + +import datetime +import os +import zlib +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class FileInfo: + """Computed metadata for a single file.""" + + path: str + exists: bool = False + size: int = 0 + crc32: int = 0 + mtime: float = 0.0 + + @property + def size_human(self) -> str: + """Human-readable file size (KiB / MiB / GiB).""" + if self.size >= 1_073_741_824: + return f"{self.size / 1_073_741_824:.2f} GiB" + if self.size >= 1_048_576: + return f"{self.size / 1_048_576:.2f} MiB" + if self.size >= 1_024: + return f"{self.size / 1_024:.2f} KiB" + return f"{self.size} B" + + @property + def mtime_iso(self) -> str: + """ISO-formatted modification time, or empty string.""" + if not self.mtime: + return "" + return datetime.datetime.fromtimestamp(self.mtime).strftime( + "%Y-%m-%d %H:%M:%S" + ) + + @property + def crc32_hex(self) -> str: + """8-character hex CRC32 representation.""" + return f"{self.crc32 & 0xFFFFFFFF:08X}" + + +def compute_file_info(file_path: str | Path) -> FileInfo: + """Compute size, CRC32, and modification time for a file. + + If the file does not exist or cannot be read, the returned + ``FileInfo`` will have ``exists=False`` and all numeric fields + set to zero. + + Args: + file_path: Absolute or relative path to the file. + + Returns: + FileInfo dataclass with computed metadata. + """ + path = Path(file_path) + info = FileInfo(path=str(path)) + if not path.is_file(): + return info + info.exists = True + try: + stat = path.stat() + info.size = stat.st_size + info.mtime = stat.st_mtime + except OSError: + return info + + # CRC32 + try: + value = 0 + with open(path, "rb") as f: + while chunk := f.read(1_048_576): # 1 MiB chunks + value = zlib.crc32(chunk, value) + info.crc32 = value + except OSError: + pass + return info diff --git a/src/gui/main_window.py b/src/gui/main_window.py index 792a4b2..af08884 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -25,6 +25,7 @@ from tftp_manager import tftp_upload, tftp_download, tftp_upload_verify, TftpRes 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, @@ -118,6 +119,9 @@ class MainWindow(ctk.CTk): 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 @@ -241,6 +245,8 @@ class MainWindow(ctk.CTk): except Exception: pass # Keep existing config on error + self._log_all_file_info() + def _sync_config_from_ui(self) -> None: """Read current UI values and update the Config object. @@ -284,6 +290,9 @@ class MainWindow(ctk.CTk): # 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. @@ -328,6 +337,11 @@ class MainWindow(ctk.CTk): 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 @@ -349,13 +363,69 @@ class MainWindow(ctk.CTk): self._config.zynq_ip = self._ip_string_var.get() def _on_file_selected(self, path: str) -> None: - """Handle file selection — sync and auto-save to disk. - - When the user selects a file via the browse dialog, immediately - sync the path to the Config object and persist it to YAML so - that subsequent step executions pick up the new value. - """ + """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_file_info(self, info: FileInfo) -> None: + """Log file metadata in a human-readable format. + + 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: + """Display file metadata in the status info panel.""" + if not self._config: + return + self._status.clear_info() + 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: + 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 ──────────────────────────────────────── @@ -365,7 +435,8 @@ class MainWindow(ctk.CTk): 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=1) + self.main_frame.grid_columnconfigure(0, weight=7) + self.main_frame.grid_columnconfigure(1, weight=3) # ── Header ── self._build_header(self.main_frame) @@ -382,7 +453,7 @@ class MainWindow(ctk.CTk): # ── 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=1) + 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) @@ -581,6 +652,7 @@ class MainWindow(ctk.CTk): padx=PADDING, pady=PADDING, ) panel.grid_rowconfigure(9, weight=1) + panel.grid_columnconfigure(0, weight=1) # let content stretch title = ctk.CTkLabel( panel, @@ -695,6 +767,17 @@ class MainWindow(ctk.CTk): ) 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="下载校验 (Step 4.2 Download Verify)", + 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, @@ -843,18 +926,21 @@ class MainWindow(ctk.CTk): self._status = StatusDisplay(panel) self._status.grid(row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING) - # UART warning label (persistent, not overwritten by other status updates) - self._uart_warning_label = ctk.CTkLabel( + # UART warning — multi-line textbox, wraps long messages instead + # of expanding the right panel horizontally + self._uart_warning = ctk.CTkTextbox( panel, - text="", font=FONT_BODY, - anchor="w", - text_color=WARNING_COLOR, + height=40, + corner_radius=4, + fg_color="transparent", + state="disabled", + wrap="word", ) - self._uart_warning_label.grid( + self._uart_warning.grid( row=1, column=0, sticky="ew", padx=PADDING, pady=(0, PADDING_SMALL) ) - self._uart_warning_label.grid_remove() # Hide by default + self._uart_warning.grid_remove() # Hide by default # ── Workflow Steps ───────────────────────────────────────── @@ -916,6 +1002,10 @@ class MainWindow(ctk.CTk): 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: @@ -969,6 +1059,7 @@ class MainWindow(ctk.CTk): 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 @@ -980,7 +1071,8 @@ class MainWindow(ctk.CTk): self._log_message(" Checking UART serial port...") port = self._config.serial_port if port: - available, reason = check_uart_available(port, self._config.serial_baudrate) + available, reason = check_uart_available(port, self._config.serial_baudrate, + monitor=self._uart_monitor) self._uart_available = available self._uart_message = reason if available: @@ -1008,17 +1100,26 @@ class MainWindow(ctk.CTk): self.after(0, lambda: self._show_uart_warning(msg)) def _show_uart_warning(self, msg: str) -> None: - """Display the persistent UART warning message. + """Display the persistent UART warning message (multi-line). Args: msg: Warning message to display. """ - self._uart_warning_label.configure(text=msg) - self._uart_warning_label.grid() + 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 label.""" - self._uart_warning_label.grid_remove() + """Hide the persistent UART warning.""" + self._uart_warning.grid_remove() def _on_window_resize(self, event) -> None: """Handle window resize / maximize events. @@ -1316,27 +1417,31 @@ class MainWindow(ctk.CTk): self._log_message(" ⏳ Waiting 2s before download...") time.sleep(2) - # 4b: TFTP 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}") + # 4b: TFTP Download Verify (skipped if disabled in config) + if not self._config.download_verify: + self._log_message("Step 4b: Download verify — skipped (disabled in config)") + all_results.append(True) + else: + 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_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) + 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)") @@ -1845,6 +1950,7 @@ class MainWindow(ctk.CTk): success, message, version = test_serial_version( port, self._config.serial_baudrate if self._config else 115200, + monitor=self._uart_monitor, ) if success: diff --git a/src/gui/widgets.py b/src/gui/widgets.py index 5521b5c..7e6b060 100644 --- a/src/gui/widgets.py +++ b/src/gui/widgets.py @@ -437,63 +437,82 @@ class ProgressIndicator(ctk.CTkFrame): class StatusDisplay(ctk.CTkFrame): - """Status display widget for operation results. + """Info display widget — persistent key-value file/device info. - Shows status messages in a multi-line, expandable text area that - uses the full available space. Messages are color-coded and - automatically wrapped. + Shows aligned ``key : value`` lines for file metadata, Zynq + part info, tool versions, etc. """ def __init__(self, master, **kwargs): - """Initialize the status display. - - Args: - master: Parent widget. - """ super().__init__(master, **kwargs) - + self._info_items: list[tuple[str, str]] = [] self._create_widgets() def _create_widgets(self) -> None: - """Create the status display UI — multi-line expandable area.""" self.grid_rowconfigure(0, weight=1) self.grid_columnconfigure(0, weight=1) - self._status_text = ctk.CTkTextbox( + self._info_text = ctk.CTkTextbox( self, font=FONT_BODY, corner_radius=CORNER_RADIUS, state="disabled", wrap="word", ) - self._status_text.grid( + self._info_text.grid( row=0, column=0, sticky="nsew", padx=PADDING, pady=PADDING_SMALL ) + # Pre-configure tags + for name, color in [ + ("info_label", PRIMARY_COLOR), + ("info_value", DESC_TEXT_COLOR if _is_dark() else DESC_TEXT_COLOR_LIGHT), + ]: + self._info_text.tag_config(name, foreground=color) + def set_status(self, message: str, status: str = "info") -> None: - """Update the status message and color. + """Compatibility stub — status messages go to log now.""" + + def set_info(self, key: str, value: str) -> None: + """Set or update a persistent info line. + + Displays as ``key : value`` in the info block. Args: - message: Status message text. - status: Status type for coloring ('info', 'success', 'error', 'warning'). + key: Label for the info line. + value: Value string. """ - self._status_text.configure(state="normal") - self._status_text.delete("1.0", "end") + for i, (k, _) in enumerate(self._info_items): + if k == key: + self._info_items[i] = (key, value) + break + else: + self._info_items.append((key, value)) + self._refresh_info() - colors = { - "info": INFO_COLOR, - "success": SUCCESS_COLOR, - "error": DANGER_COLOR, - "warning": WARNING_COLOR, - } - color = colors.get(status, INFO_COLOR) + def clear_info(self) -> None: + """Remove all persistent info lines.""" + self._info_items.clear() + self._info_text.configure(state="normal") + self._info_text.delete("1.0", "end") + self._info_text.configure(state="disabled") - self._status_text.insert("1.0", message) - # Tag the entire text with the color (CTkTextbox uses tag_config) - self._status_text.tag_config("status", foreground=color) - self._status_text.tag_add("status", "1.0", "end") - - self._status_text.configure(state="disabled") + def _refresh_info(self) -> None: + """Rebuild the info text block from current items.""" + self._info_text.configure(state="normal") + self._info_text.delete("1.0", "end") + if not self._info_items: + self._info_text.configure(state="disabled") + return + max_key = max(len(k) for k, _ in self._info_items) + for key, value in self._info_items: + line = f"{key.ljust(max_key + 2)} {value}\n" + start = self._info_text.index("end-1c") + self._info_text.insert("end", line) + end = self._info_text.index("end-1c") + self._info_text.tag_add("info_label", start, f"{start}+{len(key)}c") + self._info_text.tag_add("info_value", f"{start}+{len(key)}c", end) + self._info_text.configure(state="disabled") class FileSelector(ctk.CTkFrame): diff --git a/src/serial_monitor.py b/src/serial_monitor.py index 9e9acd5..aae9f59 100644 --- a/src/serial_monitor.py +++ b/src/serial_monitor.py @@ -137,6 +137,7 @@ def check_uart_available( port: str, baudrate: int = 115200, timeout: float = 3.0, + monitor: UartMonitor | None = None, ) -> tuple[bool, str]: """Check whether the UART serial port is available and readable. @@ -145,19 +146,24 @@ def check_uart_available( considered available (the device may simply not be outputting yet). If the port cannot be opened at all, it is unavailable. + If *monitor* is provided and currently running, the function + skips the port-open test (which would fail on Windows with + Error 13 due to exclusive access) and returns True immediately. + Args: port: Serial port device path (e.g., '/dev/ttyUSB0'). baudrate: Baud rate for serial communication. timeout: Seconds to wait for data after opening. + monitor: Optional running UartMonitor — if active, reuse it. Returns: Tuple of (is_available, reason_string). - - (True, "Port open") — port is usable - - (True, "Port open, no data yet") — port open, no data currently - - (False, "No serial ports detected") — system has no serial ports - - (False, "Port not in detected ports") — configured port missing - - (False, "Failed to open: ") — port open error """ + # Reuse existing monitor when available (avoids Error 13 on Windows) + if monitor is not None and monitor.is_running(): + return True, "Already monitoring" + + # 1. Quick port list check # 1. Quick port list check ports = detect_serial_ports() if not ports: @@ -223,25 +229,31 @@ def test_serial_version( port: str, baudrate: int = 115200, timeout: float = 5.0, + monitor: UartMonitor | None = None, ) -> tuple[bool, str, str]: """Test serial port by sending 'ver()' command and parsing response. - Sends 'ver\\r\\n' to the serial port and reads the response. - Attempts to parse version string from the response. + If *monitor* is running, skips opening a new connection and + returns the latest parsed version from the monitor instead. Args: port: Serial port device path (e.g., '/dev/ttyUSB0'). baudrate: Baud rate for serial communication. timeout: Read timeout in seconds. + monitor: Optional running UartMonitor — if active, reuse it. Returns: Tuple of (success, message, version_string). - - (True, "Version: x.y.z", "x.y.z") — version detected - - (True, "Port responded", "") — port responded but no version - - (False, "Error message", "") — error occurred """ import serial + # Reuse existing monitor when available (avoids Error 13 on Windows) + if monitor is not None and monitor.is_running(): + info = monitor.latest_info + if info and info.version: + return True, f"Version: {info.version}", info.version + return True, "Port monitored, no version parsed yet", "" + try: with serial.Serial(port, baudrate, timeout=timeout) as ser: ser.reset_input_buffer()