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