feat: UART monitor, step timeouts, and non-blocking subprocess output

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)
This commit is contained in:
Jeremy Shen
2026-06-10 13:21:32 +08:00
parent ab7264d5c7
commit 2c48224aea
8 changed files with 434 additions and 33 deletions
+155
View File
@@ -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