Zynq_Flasher/src/subprocess_utils.py

"""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, progress indicators,
and flash operation phase transitions (erase/program/verify).
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', 'phase', '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"]

# Phase transition patterns for program_flash output
_PHASE_ERASE_RE = re.compile(r"Performing Erase Operation", re.IGNORECASE)
_PHASE_PROGRAM_RE = re.compile(r"Performing Program Operation", re.IGNORECASE)
_PHASE_VERIFY_RE = re.compile(r"Performing Verify Operation", re.IGNORECASE)
_PHASE_DONE_RE = re.compile(r"(Erase|Program|Verify) Operation successful", re.IGNORECASE)


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 phase transitions (program_flash output)
    if _PHASE_ERASE_RE.search(line):
        callback("phase", "erase")
        callback("out", line)
        return
    if _PHASE_PROGRAM_RE.search(line):
        callback("phase", "program")
        callback("out", line)
        return
    if _PHASE_VERIFY_RE.search(line):
        callback("phase", "verify")
        callback("out", line)
        return
    if _PHASE_DONE_RE.search(line):
        callback("phase", "done")
        callback("out", 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)