Zynq_Flasher/src/serial_monitor.py

"""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.
"""

from __future__ import annotations

import re
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 = ""
    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+([^\s;,\n]+)", re.IGNORECASE),
        re.compile(r"[Ff]irmware\s+([^\s;,\n]+)", re.IGNORECASE),
        re.compile(r"(?:^|[\s:=])(v\d+\.\d+\.\d+)(?:\s|$)", re.IGNORECASE),
    ]
    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),
    ]

    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:
                    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

    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