468 lines
16 KiB
Python
468 lines
16 KiB
Python
"""Serial port monitoring and log parsing for Zynq Flasher GUI.
|
|
|
|
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
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Callable
|
|
|
|
import serial
|
|
import serial.tools.list_ports
|
|
|
|
|
|
@dataclass
|
|
class BootInfo:
|
|
"""Parsed information extracted from Zynq boot output."""
|
|
|
|
ip_address: str = ""
|
|
version: str = ""
|
|
boot_message: str = ""
|
|
boot_mode: str = ""
|
|
raw_lines: list[str] | None = None
|
|
|
|
|
|
@dataclass
|
|
class SerialInfo:
|
|
"""Information about an available serial port."""
|
|
|
|
device: str
|
|
description: str = ""
|
|
hwid: str = ""
|
|
|
|
|
|
def detect_serial_ports() -> list[SerialInfo]:
|
|
"""Detect all available serial/USB ports on the system.
|
|
|
|
Returns:
|
|
List of SerialInfo objects for each available port.
|
|
"""
|
|
ports = serial.tools.list_ports.comports()
|
|
results: list[SerialInfo] = []
|
|
for port in ports:
|
|
results.append(
|
|
SerialInfo(
|
|
device=port.device,
|
|
description=port.description or "",
|
|
hwid=port.hwid or "",
|
|
)
|
|
)
|
|
return results
|
|
|
|
|
|
def parse_boot_output(lines: list[str]) -> BootInfo:
|
|
"""Parse boot output lines for IP, version, and other info.
|
|
|
|
Looks for common patterns in Zynq boot logs:
|
|
- IP addresses (IPv4)
|
|
- Version strings (e.g., "v1.2.3", "version: x.y.z")
|
|
- Boot completion messages
|
|
|
|
Args:
|
|
lines: List of log output lines to parse.
|
|
|
|
Returns:
|
|
BootInfo with extracted data.
|
|
"""
|
|
info = BootInfo(raw_lines=lines)
|
|
|
|
ip_pattern = re.compile(r"\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b")
|
|
version_patterns = [
|
|
re.compile(r"[Vv]ersion[:\s]+([^\s;,\n]+)", re.IGNORECASE),
|
|
re.compile(r"[Bb]oot\s+(?:version\s+)?([Vv]?\d+\.\d+[^\s;,\n]*)", re.IGNORECASE),
|
|
re.compile(r"[Ff]irm[Ww]are:?\s*([^\s;,\n]+)", re.IGNORECASE),
|
|
re.compile(r"(?:^|[\s:=])([Vv]\d+\.\d+\.\d+[^\s;,\n]*)", re.IGNORECASE),
|
|
]
|
|
|
|
def _is_valid_version(v: str) -> bool:
|
|
"""Filter out noise like '3.1', 'mode', 'update'."""
|
|
if len(v) < 4:
|
|
return False
|
|
has_digit = any(c.isdigit() for c in v)
|
|
has_dot = "." in v or "_" in v
|
|
noise = {"mode", "upate", "update", "done", "error", "loaded", "running"}
|
|
return has_digit and has_dot and v.lower() not in noise
|
|
boot_complete_patterns = [
|
|
re.compile(r"boot\s+complete", re.IGNORECASE),
|
|
re.compile(r"system\s+ready", re.IGNORECASE),
|
|
re.compile(r"starting\s+application", re.IGNORECASE),
|
|
re.compile(r"Linux\s+booted", re.IGNORECASE),
|
|
re.compile(r"QSYS\s+ready", re.IGNORECASE),
|
|
]
|
|
boot_mode_patterns = [
|
|
re.compile(r"[Bb]oot\s*[Mm]ode\s*(?:is)?\s*[:\[=\s]*(\w+)"),
|
|
re.compile(r"BootMode\s*[:\[=\s]*(\w+)", re.IGNORECASE),
|
|
]
|
|
|
|
for line in lines:
|
|
# Extract IP address
|
|
if not info.ip_address:
|
|
match = ip_pattern.search(line)
|
|
if match:
|
|
info.ip_address = match.group(1)
|
|
|
|
# Extract version
|
|
if not info.version:
|
|
for pattern in version_patterns:
|
|
match = pattern.search(line)
|
|
if match and _is_valid_version(match.group(1)):
|
|
info.version = match.group(1)
|
|
break
|
|
|
|
# Check for boot completion
|
|
if not info.boot_message:
|
|
for pattern in boot_complete_patterns:
|
|
if pattern.search(line):
|
|
info.boot_message = line.strip()
|
|
break
|
|
|
|
# Extract boot mode
|
|
if not info.boot_mode:
|
|
for pattern in boot_mode_patterns:
|
|
match = pattern.search(line)
|
|
if match:
|
|
info.boot_mode = match.group(1)
|
|
break
|
|
|
|
return info
|
|
|
|
|
|
def check_uart_available(
|
|
port: str,
|
|
baudrate: int = 115200,
|
|
timeout: float = 3.0,
|
|
) -> tuple[bool, str]:
|
|
"""Check whether the UART serial port is available and readable.
|
|
|
|
Attempts to open the port, clear buffers, and read a few lines.
|
|
If the port opens but returns no data within the timeout, it is
|
|
considered available (the device may simply not be outputting yet).
|
|
If the port cannot be opened at all, it is unavailable.
|
|
|
|
Args:
|
|
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
|
baudrate: Baud rate for serial communication.
|
|
timeout: Seconds to wait for data after opening.
|
|
|
|
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: <error>") — port open error
|
|
"""
|
|
# 1. Quick port list check
|
|
ports = detect_serial_ports()
|
|
if not ports:
|
|
return False, "No serial ports detected"
|
|
|
|
# 2. Verify the configured port is among detected ones
|
|
if port and port not in {p.device for p in ports}:
|
|
return False, f"Port {port} not in detected ports"
|
|
|
|
# 3. Try to open and read
|
|
try:
|
|
with serial.Serial(port, baudrate, timeout=timeout) as ser:
|
|
ser.reset_input_buffer()
|
|
lines: list[str] = []
|
|
while ser.in_waiting:
|
|
line = ser.readline().decode("utf-8", errors="replace").strip()
|
|
if line:
|
|
lines.append(line)
|
|
if lines:
|
|
return True, "Port open"
|
|
return True, "Port open, no data yet"
|
|
except serial.SerialException as e:
|
|
return False, f"Failed to open: {e}"
|
|
|
|
|
|
def read_serial_stream(
|
|
port: str,
|
|
baudrate: int = 115200,
|
|
timeout: float = 1.0,
|
|
line_callback: Callable[[str], None] | None = None,
|
|
) -> list[str]:
|
|
"""Read lines from a serial port.
|
|
|
|
Opens the serial port, reads available lines, and closes it.
|
|
Optionally calls line_callback for each line as it arrives.
|
|
|
|
Args:
|
|
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
|
baudrate: Baud rate for serial communication.
|
|
timeout: Read timeout in seconds.
|
|
line_callback: Optional callback invoked for each line.
|
|
|
|
Returns:
|
|
List of lines read from the serial port.
|
|
|
|
Raises:
|
|
serial.SerialException: If the port cannot be opened.
|
|
"""
|
|
lines: list[str] = []
|
|
with serial.Serial(port, baudrate, timeout=timeout) as ser:
|
|
# Clear any pending data
|
|
ser.reset_input_buffer()
|
|
while ser.in_waiting:
|
|
line = ser.readline().decode("utf-8", errors="replace").strip()
|
|
if line:
|
|
lines.append(line)
|
|
if line_callback:
|
|
line_callback(line)
|
|
return lines
|
|
|
|
|
|
def test_serial_version(
|
|
port: str,
|
|
baudrate: int = 115200,
|
|
timeout: float = 5.0,
|
|
) -> 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.
|
|
|
|
Args:
|
|
port: Serial port device path (e.g., '/dev/ttyUSB0').
|
|
baudrate: Baud rate for serial communication.
|
|
timeout: Read timeout in seconds.
|
|
|
|
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
|
|
|
|
try:
|
|
with serial.Serial(port, baudrate, timeout=timeout) as ser:
|
|
ser.reset_input_buffer()
|
|
ser.reset_output_buffer()
|
|
|
|
# Send ver() command
|
|
ser.write(b"ver()\r\n")
|
|
|
|
# Read response
|
|
response = b""
|
|
start_time = __import__("time").time()
|
|
while __import__("time").time() - start_time < timeout:
|
|
if ser.in_waiting:
|
|
data = ser.read(ser.in_waiting)
|
|
response += data
|
|
# Check if we have a complete response
|
|
if b"\n" in data or b"\r" in data:
|
|
break
|
|
__import__("time").sleep(0.1)
|
|
|
|
# Decode response
|
|
response_str = response.decode("utf-8", errors="replace").strip()
|
|
|
|
if not response_str:
|
|
return False, "Port responded but no data received", ""
|
|
|
|
# Parse version from response
|
|
version = _parse_version_from_response(response_str)
|
|
|
|
if version:
|
|
return True, f"Version: {version}", version
|
|
else:
|
|
return True, f"Port responded: {response_str[:100]}", ""
|
|
|
|
except serial.SerialException as e:
|
|
return False, f"Serial error: {e}", ""
|
|
except Exception as e:
|
|
return False, f"Error: {e}", ""
|
|
|
|
|
|
def _parse_version_from_response(response: str) -> str:
|
|
"""Parse version string from serial response.
|
|
|
|
Args:
|
|
response: Response string from serial port.
|
|
|
|
Returns:
|
|
Version string if found, empty string otherwise.
|
|
"""
|
|
# Common version patterns
|
|
version_patterns = [
|
|
re.compile(r"([Vv]ersion[:\s]+)?([^\s;,\n]+v\d+\.\d+\.\d+[^\s;,\n]*)", re.IGNORECASE),
|
|
re.compile(r"(?:^|[\s:=])(v\d+\.\d+\.\d+)(?:\s|$)", re.IGNORECASE),
|
|
re.compile(r"(\d+\.\d+\.\d+)", re.IGNORECASE),
|
|
]
|
|
|
|
for pattern in version_patterns:
|
|
match = pattern.search(response)
|
|
if match:
|
|
# Return the version part (group 1 or group 2)
|
|
version = match.group(1) if match.lastindex and match.lastindex >= 1 else match.group(0)
|
|
# Clean up version string
|
|
version = version.strip().strip("';\"")
|
|
if version and len(version) > 2: # Basic validation
|
|
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=[])
|
|
self._last_emitted: tuple[str, str, str] = ("", "", "")
|
|
|
|
# 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 only when parsed info changes (dedup)
|
|
current = (info.boot_mode, info.ip_address, info.version)
|
|
if current != self._last_emitted and any(current):
|
|
self._last_emitted = current
|
|
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
|