From 2c48224aeab86f00c2011a6b20eabce2256bd316 Mon Sep 17 00:00:00 2001 From: Jeremy Shen Date: Wed, 10 Jun 2026 13:21:32 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20UART=20monitor,=20step=20ti?= =?UTF-8?q?meouts,=20and=20non-blocking=20subprocess=20output?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Based on real hardware timing data from XC7Z100 testing: UART Monitoring: - New UartMonitor class in serial_monitor.py: background thread reads serial output continuously, parses for IP/version/boot messages - GUI starts monitor when workflow runs, stops on completion/close - Serial lines flow into log display in real time Step Timing: - New step_timeouts config (experience × 2): check_env:30s flash_erase:120s flash_program:600s bitstream:60s elf_download:60s tftp_reboot:120s - GUI shows estimated timeout before each step - Shows actual elapsed time after each step completes - All subprocess calls now use config-driven timeouts Non-blocking Output: - New subprocess_utils.py: run_streaming() uses Popen + threaded stdout/stderr readers, parses WARNING/ERROR/progress% in real time - flash_programmer and bitstream_programmer use it - Callbacks receive categorized events (out/err/warn/error/progress/done) --- config.yaml | 24 ++++++ config/default_config.yaml | 13 +++ src/bitstream_programmer.py | 33 ++++---- src/config_manager.py | 10 +++ src/flash_programmer.py | 31 ++++---- src/gui/main_window.py | 51 +++++++++++- src/serial_monitor.py | 155 ++++++++++++++++++++++++++++++++++++ src/subprocess_utils.py | 150 ++++++++++++++++++++++++++++++++++ 8 files changed, 434 insertions(+), 33 deletions(-) create mode 100644 config.yaml create mode 100644 src/subprocess_utils.py diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..ef1a5d4 --- /dev/null +++ b/config.yaml @@ -0,0 +1,24 @@ +bootloader_bin_path: /home/ly0kos/work/Verify/FW/tftp_app/bootloader/bootloader_z100_800M.bin +bootloader_bit_path: /home/ly0kos/work/Verify/FW/tftp_app/bootloader/pl_arm_wrapper_hw_platform_0/pl_arm_wrapper.bit +bootloader_elf_path: /home/ly0kos/work/Verify/FW/tftp_app/bootloader/app.elf +firmware_bin_path: /home/ly0kos/work/Verify/FW/V1.3.1.3.3_06_01_pmu_cpld_check/V1.3.1.3.3_06_01_pmu_cpld_check.bin +flash_type: qspi-x4-single +fsbl_elf_path: /home/ly0kos/work/Verify/FPGA/Xilinx/Z100/fsbl_qspi_bypass.elf +inter_step_delay: 2 +ping_count: 3 +ping_timeout: 5 +reboot_timeout: 30 +serial_baudrate: 115200 +serial_port: /dev/ttyUSB0 +tftp_upload_name: z7bin +uart_delay: 3 +inter_step_delay: 2 +step_timeouts: + check_env: 30 + flash_erase: 120 + flash_program: 600 + bitstream: 60 + elf_download: 60 + tftp_reboot: 120 +vitis_path: "" +zynq_ip: 192.168.100.11 diff --git a/config/default_config.yaml b/config/default_config.yaml index b61477e..19d4bcc 100644 --- a/config/default_config.yaml +++ b/config/default_config.yaml @@ -35,3 +35,16 @@ reboot_timeout: 30 # Ping verification settings ping_timeout: 5 ping_count: 3 + +# Step timing (seconds) — used for progress bar max and timeout (experience × 2) +step_timeouts: + check_env: 30 # Step 1: tool detection + JTAG scan (~5s actual) + flash_erase: 120 # Step 2a: QSPI chip erase (~40s actual for 17MB) + flash_program: 600 # Step 2b: QSPI program+verify (~300s actual for 17MB at x4) + bitstream: 60 # Step 3a: FPGA configuration via JTAG (~16s actual) + elf_download: 60 # Step 3b: ELF download to PS via JTAG (~9s actual) + tftp_reboot: 120 # Step 4: TFTP upload + reboot + boot verify (~30s network) + +# Delays (seconds) +inter_step_delay: 2 +uart_delay: 3 diff --git a/src/bitstream_programmer.py b/src/bitstream_programmer.py index d1038c1..e8272d5 100644 --- a/src/bitstream_programmer.py +++ b/src/bitstream_programmer.py @@ -117,21 +117,17 @@ def _run_command( Raises: subprocess.TimeoutExpired: If command exceeds timeout. """ + from subprocess_utils import run_streaming + if callback: callback("running", f"Executing: {' '.join(cmd)}") - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout, - ) - if callback: - if result.returncode == 0: - callback("complete", f"{cmd[0]} succeeded") - else: - callback("error", f"{cmd[0]} failed (exit {result.returncode})") - return result + try: + return run_streaming(cmd, timeout=timeout, callback=callback) + except subprocess.TimeoutExpired: + if callback: + callback("error", f"{cmd[0]} timed out after {timeout}s") + raise # ── Bitstream Download ────────────────────────────────────────────── @@ -172,7 +168,8 @@ def program_bitstream( # Try xsct first (preferred) xsct = get_xsct_path(config.vitis_path) if xsct: - return _program_with_xsct(xsct, bit_path, callback, ps7_tcl) + timeout = config.step_timeouts.get("bitstream", 60) + return _program_with_xsct(xsct, bit_path, callback, ps7_tcl, timeout) # Fallback to Vivado vivado = _get_vivado_path(config.vitis_path) @@ -191,6 +188,7 @@ def _program_with_xsct( bit_path: Path, callback: ProgressCallback = None, ps7_init_tcl: str = "", + timeout: int = 60, ) -> BitstreamResult: """Program bitstream using xsct's fpga command. @@ -203,6 +201,7 @@ def _program_with_xsct( bit_path: Path to .bit file. callback: Optional progress callback. ps7_init_tcl: Path to ps7_init.tcl for PS initialization. + timeout: Subprocess timeout in seconds. Returns: BitstreamResult with status. @@ -232,7 +231,7 @@ def _program_with_xsct( [xsct_path, tcl_file], capture_output=True, text=True, - timeout=120, + timeout=timeout, ) os.unlink(tcl_file) success = result.returncode == 0 @@ -351,7 +350,8 @@ def run_elf( # Auto-detect ps7_init.tcl from config or BIT file directory ps7_tcl = _find_ps7_init_tcl(config) - return _run_elf_with_xsct(xsct, elf_path, callback, ps7_tcl) + timeout = config.step_timeouts.get("elf_download", 60) + return _run_elf_with_xsct(xsct, elf_path, callback, ps7_tcl, timeout) def _run_elf_with_xsct( @@ -359,6 +359,7 @@ def _run_elf_with_xsct( elf_path: Path, callback: ProgressCallback = None, ps7_init_tcl: str = "", + timeout: int = 60, ) -> BitstreamResult: """Download and run ELF using xsct's dow command. @@ -399,7 +400,7 @@ def _run_elf_with_xsct( [xsct_path, tcl_file], capture_output=True, text=True, - timeout=120, + timeout=timeout, ) os.unlink(tcl_file) diff --git a/src/config_manager.py b/src/config_manager.py index 1103764..879e07e 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -31,6 +31,14 @@ DEFAULT_CONFIG: dict[str, Any] = { "ping_count": 3, "uart_delay": 3, "inter_step_delay": 2, + "step_timeouts": { + "check_env": 30, + "flash_erase": 120, + "flash_program": 600, + "bitstream": 60, + "elf_download": 60, + "tftp_reboot": 120, + }, } @@ -58,6 +66,7 @@ class Config: ping_count: int = 3 uart_delay: int = 3 inter_step_delay: int = 2 + step_timeouts: dict[str, int] = field(default_factory=lambda: {}) # Internal: path to the config file on disk _config_path: Path = field(default=None, init=False, repr=False) @@ -163,6 +172,7 @@ class Config: "ping_count": self.ping_count, "uart_delay": self.uart_delay, "inter_step_delay": self.inter_step_delay, + "step_timeouts": self.step_timeouts, } return data diff --git a/src/flash_programmer.py b/src/flash_programmer.py index ba8d7f0..d31473f 100644 --- a/src/flash_programmer.py +++ b/src/flash_programmer.py @@ -41,31 +41,30 @@ def _run_command( timeout: int = 300, callback: ProgressCallback = None, ) -> subprocess.CompletedProcess[str]: - """Run a subprocess command with progress reporting. + """Run a subprocess command with real-time streaming output. + + Uses subprocess_utils.run_streaming to stream stdout/stderr in + real time, parsing for warnings, errors, and progress. Args: cmd: Command and arguments. - timeout: Maximum seconds to wait (flash ops can take minutes). + timeout: Maximum seconds. callback: Optional callback(status, message). Returns: CompletedProcess result. """ + from subprocess_utils import run_streaming + if callback: callback("running", f"Executing: {' '.join(cmd)}") - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout, - ) - if callback: - if result.returncode == 0: - callback("complete", f"{cmd[0]} succeeded") - else: - callback("error", f"{cmd[0]} failed (exit {result.returncode})") - return result + try: + return run_streaming(cmd, timeout=timeout, callback=callback) + except subprocess.TimeoutExpired: + if callback: + callback("error", f"{cmd[0]} timed out after {timeout}s") + raise def _build_program_flash_cmd_base( @@ -135,7 +134,7 @@ def wipe_flash( cmd += ["-erase_only"] # Erase sectors matching the BIN size (safer than -erase_all) try: - result = _run_command(cmd, timeout=300, callback=callback) + result = _run_command(cmd, timeout=config.step_timeouts.get("flash_erase", 120), callback=callback) success = result.returncode == 0 return FlashResult( step="wipe", @@ -195,7 +194,7 @@ def program_flash_bin( ] try: - result = _run_command(cmd, timeout=600, callback=callback) + result = _run_command(cmd, timeout=config.step_timeouts.get("flash_program", 600), callback=callback) success = result.returncode == 0 return FlashResult( step="program", diff --git a/src/gui/main_window.py b/src/gui/main_window.py index d4facd7..ef00f97 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -24,7 +24,7 @@ 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 serial_monitor import detect_serial_ports, parse_boot_output, check_uart_available, UartMonitor from gui.styles import ( WINDOW_WIDTH, WINDOW_HEIGHT, @@ -79,6 +79,7 @@ class MainWindow(ctk.CTk): self._uart_message: str = "" self._zynq_jtag_present: bool = False self._zynq_jtag_info = None + self._uart_monitor: UartMonitor | None = None # StringVar references for config sync self._ip_string_var: ctk.StringVar | None = None @@ -961,6 +962,15 @@ class MainWindow(ctk.CTk): 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) @@ -970,12 +980,21 @@ class MainWindow(ctk.CTk): for i in range(start_index, end_index): step_idx = i - start_index self._set_step_status(i, "running") + + # 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") except Exception as e: results.append(False) self._set_step_status(i, "error") @@ -1020,6 +1039,35 @@ class MainWindow(ctk.CTk): # 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 not self._uart_available or 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 detected: IP={info.ip_address}, Ver={info.version}") + if info.ip_address or 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") + + 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: @@ -1040,6 +1088,7 @@ class MainWindow(ctk.CTk): def _on_close(self) -> None: """Handle window close event: save config and cleanup.""" + self._stop_uart_monitor() if self._config: self._sync_config_from_ui() # Sync UI values first # Save to user config file (config.yaml) diff --git a/src/serial_monitor.py b/src/serial_monitor.py index b27db01..a0d367a 100644 --- a/src/serial_monitor.py +++ b/src/serial_monitor.py @@ -2,11 +2,13 @@ Detects available serial ports, reads output streams, and parses boot logs for IP addresses, version strings, and other diagnostic info. +Provides UartMonitor for continuous background monitoring. """ from __future__ import annotations import re +import threading from dataclasses import dataclass from typing import Callable @@ -285,3 +287,156 @@ def _parse_version_from_response(response: str) -> str: return version return "" + + +# ════════════════════════════════════════════════════════════════════ +# UartMonitor — continuous background serial monitoring +# ════════════════════════════════════════════════════════════════════ + + +class UartMonitor: + """Background thread that continuously reads and parses serial output. + + Opens the configured serial port in a daemon thread, reads lines + as they arrive, parses them for boot info (IP, version, errors), + and invokes callbacks for each event. + + Usage: + monitor = UartMonitor("/dev/ttyUSB0", 115200) + monitor.on_line = lambda line: print(f" SERIAL: {line}") + monitor.on_boot_info = lambda info: print(f" BOOT: {info}") + monitor.start() + ... + info = monitor.latest_info # latest parsed BootInfo + monitor.stop() + """ + + def __init__( + self, + port: str, + baudrate: int = 115200, + timeout: float = 1.0, + ) -> None: + """Initialize the UART monitor. + + Args: + port: Serial port device path. + baudrate: Baud rate for serial communication. + timeout: Read timeout between lines. + """ + self._port = port + self._baudrate = baudrate + self._timeout = timeout + self._thread: threading.Thread | None = None + self._stop_event = threading.Event() + self._lock = threading.Lock() + self._all_lines: list[str] = [] + self._latest_info = BootInfo(raw_lines=[]) + + # Callbacks — set these before calling start() + self.on_line: Callable[[str], None] | None = None + self.on_boot_info: Callable[[BootInfo], None] | None = None + self.on_error: Callable[[str], None] | None = None + + @property + def latest_info(self) -> BootInfo: + """Return the latest parsed boot information (thread-safe).""" + with self._lock: + return self._latest_info + + @property + def all_lines(self) -> list[str]: + """Return all lines read so far (thread-safe).""" + with self._lock: + return list(self._all_lines) + + def start(self) -> bool: + """Start the background monitoring thread. + + Returns: + True if started successfully, False if port cannot be opened. + """ + if self._thread and self._thread.is_alive(): + return True + + # Quick port check before starting thread + try: + test_ser = serial.Serial(self._port, self._baudrate, timeout=1) + test_ser.close() + except serial.SerialException as e: + if self.on_error: + self.on_error(f"UART open failed: {e}") + return False + + self._stop_event.clear() + self._thread = threading.Thread(target=self._read_loop, daemon=True) + self._thread.start() + return True + + def stop(self) -> None: + """Stop the background monitoring thread.""" + self._stop_event.set() + if self._thread: + self._thread.join(timeout=3) + self._thread = None + + def is_running(self) -> bool: + """Check if the monitor thread is active.""" + return self._thread is not None and self._thread.is_alive() + + def _read_loop(self) -> None: + """Main read loop running in background thread.""" + try: + ser = serial.Serial( + self._port, + self._baudrate, + timeout=self._timeout, + ) + ser.reset_input_buffer() + except serial.SerialException as e: + if self.on_error: + self.on_error(f"UART open failed: {e}") + return + + accumulated: list[str] = [] + try: + while not self._stop_event.is_set(): + try: + if ser.in_waiting: + line_bytes = ser.readline() + try: + line = line_bytes.decode("utf-8", errors="replace").strip() + except Exception: + line = repr(line_bytes) + else: + line = "" + except serial.SerialException: + line = "" + + if line: + # Store all lines + with self._lock: + self._all_lines.append(line) + accumulated.append(line) + + # Invoke line callback + if self.on_line: + self.on_line(line) + + # Parse boot info from accumulated lines + info = parse_boot_output(accumulated) + with self._lock: + self._latest_info = info + + # Notify if we have new boot info + if info.boot_message or info.ip_address or info.version: + if self.on_boot_info: + self.on_boot_info(info) + else: + # No data, sleep briefly + self._stop_event.wait(0.1) + finally: + try: + ser.close() + except Exception: + pass diff --git a/src/subprocess_utils.py b/src/subprocess_utils.py new file mode 100644 index 0000000..11d0ae6 --- /dev/null +++ b/src/subprocess_utils.py @@ -0,0 +1,150 @@ +"""Non-blocking subprocess execution with real-time output parsing. + +Provides run_streaming() which streams stdout/stderr to callbacks +in real time, parsing for warnings, errors, and progress indicators. +Used by flash_programmer and bitstream_programmer. +""" + +from __future__ import annotations + +import re +import subprocess +import threading +from typing import Callable + + +StreamCallback = Callable[[str, str], None] | None +"""Callback type: (category, message) where category is one of: + 'out', 'err', 'warn', 'error', 'progress', 'done'""" + +# Regex patterns for parsing tool output +_WARNING_RE = re.compile(r"WARNING\s*[:\[].*", re.IGNORECASE) +_ERROR_RE = re.compile(r"ERROR\s*[:\[].*|CRITICAL\s*[:\[].*", re.IGNORECASE) +_PROGRESS_RE = re.compile(r"(\d{1,3})\s*%") +_INFO_KEYWORDS = ["connected", "downloading", "running", "initialization", + "verif", "write", "erase", "flash operation"] + + +def run_streaming( + cmd: list[str], + timeout: int = 300, + callback: StreamCallback = None, +) -> subprocess.CompletedProcess[str]: + """Run a command with real-time stdout/stderr streaming and parsing. + + Spawns background threads to read stdout and stderr line-by-line. + Each line is parsed for warnings, errors, and progress, and the + results are passed to the callback in real time. + + Args: + cmd: Command and arguments (e.g., ["program_flash", "-f", ...]). + timeout: Maximum seconds before killing the process. + callback: Optional callback(category, message). + + Returns: + CompletedProcess with stdout, stderr, and returncode. + """ + stdout_lines: list[str] = [] + stderr_lines: list[str] = [] + + def _read_stream(stream, lines: list[str], stream_name: str) -> None: + """Read lines from a stream in a background thread.""" + for raw in iter(stream.readline, ""): + line = raw.strip() + if not line: + continue + lines.append(line) + + # Parse and callback + if callback: + _parse_and_callback(line, stream_name, callback) + + try: + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + + # Start reader threads + t_out = threading.Thread( + target=_read_stream, + args=(proc.stdout, stdout_lines, "out"), + daemon=True, + ) + t_err = threading.Thread( + target=_read_stream, + args=(proc.stderr, stderr_lines, "err"), + daemon=True, + ) + t_out.start() + t_err.start() + + # Wait with timeout + try: + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + if callback: + callback("error", f"Command timed out after {timeout}s") + raise + + # Wait for reader threads to finish + t_out.join(timeout=2) + t_err.join(timeout=2) + + if callback: + if proc.returncode == 0: + callback("done", f"{cmd[0]}: completed successfully") + else: + callback("error", f"{cmd[0]}: failed (exit {proc.returncode})") + + return subprocess.CompletedProcess( + args=cmd, + returncode=proc.returncode, + stdout="\n".join(stdout_lines), + stderr="\n".join(stderr_lines), + ) + except FileNotFoundError: + if callback: + callback("error", f"Command not found: {cmd[0]}") + raise + + +def _parse_and_callback( + line: str, + stream_name: str, + callback: Callable[[str, str], None], +) -> None: + """Parse a single output line and invoke the callback. + + Args: + line: Raw output line. + stream_name: 'out' or 'err'. + callback: Callback to invoke. + """ + # Check for explicit errors/warnings + if _ERROR_RE.match(line): + callback("error", line) + return + if _WARNING_RE.match(line): + callback("warn", line) + return + + # Check for progress percentage + m = _PROGRESS_RE.search(line) + if m: + callback("progress", f"{m.group(1)}%") + return + + # Check for info keywords + lowered = line.lower() + for kw in _INFO_KEYWORDS: + if kw in lowered: + callback("out", line) + return + + # Default: plain output + callback(stream_name, line)