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


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:
            # Clear buffers
            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 ""