pcb2gcode_runner.py

"""
pcb2gcode_runner.py — envoltura robusta del subproceso pcb2gcode.

Este modulo NO sabe de UI ni de perfiles. Recibe un `work_dir` que ya tiene
los Gerbers/Excellon y un `millproject` escrito, corre el motor pcb2gcode por
subproceso y devuelve un dict estructurado. La traduccion perfil -> millproject
vive en `pcb_profiles.py` (otra fase); aca solo se invoca y se valida el motor.

Patron: igual que MechaLeveling le habla a FluidNC por su protocolo sin forkearlo,
aca se le habla a pcb2gcode por su CLI sin forkearlo.

Decisiones duras heredadas del spec (2026-06-25-pcb2gcode-integracion-design.md):

  - SIEMPRE `--config <millproject>`, NUNCA `--noconfigfile`. Esa combinacion
    ignora el archivo de config y corre con defaults (donde metric = pulgadas):
    un Z de -0.07mm se interpreta como -0.07 pulgadas = -1.78mm y gubia la placa.
  - NUNCA `shell=True`. Siempre lista de args.
  - Exito != returncode 0. Probado: un B_Cu denso genero un `front.ngc` de 259
    bytes SIN trayectorias con returncode 0. El exito exige 5 condiciones juntas
    (ver `_validar_ngc` y `run`).
  - Detener real: se usa Popen y se expone el handle por callback `on_process`,
    ademas de soportar un `threading.Event` de cancelacion. terminate()/kill()
    matan el proceso al instante.
  - Limpieza pre-corrida: se borran `.ngc`/`.svg` viejos ANTES de correr para que
    no se cuelen como exito un output rancio.

NUNCA manda nada a una CNC. pcb2gcode solo escribe archivos a disco.
"""

import os
import re
import shutil
import subprocess
import threading
import time
from typing import Callable, Dict, List, Optional, Tuple

from logging_config import get_logger

logger = get_logger(__name__)


# ──────────────────────────────────────────────────────────────────
# Configuracion
# ──────────────────────────────────────────────────────────────────

# Version minima soportada. pcb2gcode 2.x espeja el --back por el centro de cada
# corrida (iso, cutout y npth saldrian desalineados) y no acepta varias opciones
# que escribe pcb_profiles. Solo 3.x.
MIN_VERSION: Tuple[int, int, int] = (3, 0, 0)


def resolve_binary() -> str:
    """
    Ubica el binario de pcb2gcode, en este orden:
      1. variable de entorno PCB2GCODE_BIN
      2. `pcb2gcode` en el PATH
      3. rutas tipicas (Homebrew Apple Silicon e Intel, /usr/bin, /usr/local/bin)
    Si no aparece, devuelve "pcb2gcode" (check_version reportara que no esta).
    """
    env = os.environ.get("PCB2GCODE_BIN")
    if env:
        return env
    found = shutil.which("pcb2gcode")
    if found:
        return found
    for cand in ("/opt/homebrew/bin/pcb2gcode", "/usr/local/bin/pcb2gcode",
                 "/usr/bin/pcb2gcode"):
        if os.path.exists(cand):
            return cand
    return "pcb2gcode"


# Binario resuelto al importar (ver resolve_binary). Se puede fijar con PCB2GCODE_BIN.
PCB2GCODE_BIN: str = resolve_binary()


def parse_version(ver: Optional[str]) -> Optional[Tuple[int, int, int]]:
    """'3.0.3' -> (3, 0, 3). None o texto raro -> None."""
    if not ver:
        return None
    m = re.match(r"(\d+)\.(\d+)\.(\d+)", ver.strip())
    if not m:
        return None
    return tuple(int(x) for x in m.groups())  # type: ignore[return-value]


def version_soportada(ver: Optional[str]) -> bool:
    """True si la version parseada es >= MIN_VERSION."""
    parsed = parse_version(ver)
    return parsed is not None and parsed >= MIN_VERSION

# Extension exacta de los G-code que produce pcb2gcode.
NGC_EXT: str = ".ngc"

# Timeout corto para `--version` (no debe colgar la UI).
VERSION_TIMEOUT: float = 8.0

# Guardrail de disco (FlatCAM llego a llenar discos enteros): tope al tamano acumulado
# del work_dir mientras corre pcb2gcode. Si lo supera, se mata el proceso y se
# aborta. Default conservador: 500 MB (una corrida sana pesa pocos MB). Tambien
# se exige un minimo de espacio libre en el FS para no llenar el disco.
DEFAULT_MAX_WORK_DIR_MB: float = 500.0
DEFAULT_MIN_FREE_MB: float = 200.0
# Cada cuantas iteraciones del loop de wait() se chequea el disco. El loop gira
# ~cada 0.2s; con 25 iteraciones se revisa ~cada 5s (no en cada vuelta, para no
# escanear el dir sin parar).
_DISK_CHECK_CADA_N: int = 25

# Patrones de error en stderr/stdout que invalidan la corrida aunque el
# returncode sea 0. Se chequea el texto, no solo el codigo (decision del spec).
_ERROR_PATTERNS = (
    "Error:",
    "unrecognised option",
    "unrecognized option",
)

# G-code real: cada .ngc valido debe declarar milimetros (G21) y tener al menos
# un movimiento de trabajo (G1) que baje a Z negativo. Un .ngc solo-cabecera
# (caso 259 bytes) no cumple la segunda condicion.
_RE_G21 = re.compile(r"^\s*G21\b", re.MULTILINE | re.IGNORECASE)
# La condicion real que nos importa: una linea G1/G01 que mueve a Z negativo.
# Se aplica sobre el CUERPO de la linea ya sin comentario (ver _linea_es_g1_z_negativo):
#   - empieza por G1 o G01,
#   - tiene un campo Z con valor negativo, tolerando espacio tras la Z (Z -2.2).
# El espacio opcional tras la Z cubre estilos de salida que no la pegan al numero.
_RE_G1_Z_NEGATIVO = re.compile(r"^\s*G0?1\b.*\bZ\s*-\s*\d", re.IGNORECASE)


def _linea_es_g1_z_negativo(linea: str) -> bool:
    """
    Decide si una linea de G-code es un G1/G01 que baja a Z negativo, ignorando
    cualquier comentario entre parentesis.

    pcb2gcode (y G-code en general) deja comentarios tras '('; todo lo que venga
    despues NO es codigo. Si no lo recortaramos, un comentario como
    "G1 X.. (mueve a Z-2.2 nominal)" podria enganar al matcher aunque la linea
    real no tenga Z. Tokenizamos cortando en el primer '(' antes de buscar el Z.

    Tolera espacio opcional tras la Z (Z -2.2) para no depender del espaciado
    exacto que use el binario.
    """
    # Recortar el comentario: todo desde el primer '(' no es codigo.
    cuerpo = linea.split("(", 1)[0]
    return bool(_RE_G1_Z_NEGATIVO.match(cuerpo))


def _texto_tiene_motion(texto: str) -> bool:
    """
    True si alguna linea de `texto` es un G1/G01 con Z negativo (motion real),
    evaluando linea por linea con comentarios ya recortados.
    """
    for linea in texto.splitlines():
        if _linea_es_g1_z_negativo(linea):
            return True
    return False


# ──────────────────────────────────────────────────────────────────
# Cancelacion
# ──────────────────────────────────────────────────────────────────

class CancelledError(RuntimeError):
    """La corrida fue cancelada por el usuario (boton Detener)."""
    pass


# ──────────────────────────────────────────────────────────────────
# Version
# ──────────────────────────────────────────────────────────────────

def check_version(timeout: float = VERSION_TIMEOUT) -> Optional[str]:
    """
    Corre `pcb2gcode --version` con timeout corto y parsea la version (ej '3.0.3').

    Pensado para correrse en un thread daemon tras construir la pestana, nunca en
    el hilo de UI durante el build (un subprocess lento congelaria la ventana).

    Returns:
        La version como string (ej '3.0.3') si se pudo parsear, o None si el
        binario no esta, dio timeout, o no se pudo extraer la version.
    """
    try:
        proc = subprocess.run(
            [PCB2GCODE_BIN, "--version"],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except FileNotFoundError:
        logger.warning("pcb2gcode no encontrado en %s", PCB2GCODE_BIN)
        return None
    except subprocess.TimeoutExpired:
        logger.warning("pcb2gcode --version dio timeout (%ss)", timeout)
        return None
    except OSError as e:
        logger.warning("No se pudo correr pcb2gcode --version: %s", e)
        return None

    salida = (proc.stdout or "") + "\n" + (proc.stderr or "")
    # pcb2gcode imprime la version como primera linea, ej "3.0.3".
    m = re.search(r"(\d+\.\d+\.\d+)", salida)
    if m:
        return m.group(1)
    logger.warning("No se pudo parsear la version de pcb2gcode: %r", salida[:200])
    return None


# ──────────────────────────────────────────────────────────────────
# Listado de salidas
# ──────────────────────────────────────────────────────────────────

def list_output_ngc(work_dir: str) -> List[str]:
    """
    Lista los .ngc presentes en `work_dir`, filtrando por extension EXACTA.

    No usa substring (un 'foo.ngc.bak' no entra) ni recursividad (solo el nivel
    de la corrida). Devuelve rutas absolutas ordenadas por nombre.
    """
    if not os.path.isdir(work_dir):
        return []
    res = []
    for nombre in os.listdir(work_dir):
        if os.path.splitext(nombre)[1].lower() == NGC_EXT:
            res.append(os.path.join(work_dir, nombre))
    res.sort()
    return res


# ──────────────────────────────────────────────────────────────────
# Limpieza pre-corrida
# ──────────────────────────────────────────────────────────────────

def _limpiar_work_dir(work_dir: str) -> None:
    """
    Borra .ngc y .svg viejos de `work_dir` ANTES de correr pcb2gcode.

    Sin esto, un output rancio de una corrida previa podria pasar la validacion
    (existe + tiene G-code) y colarse como exito de esta corrida. El chequeo de
    mtime es la segunda red, pero limpiar antes es la primera y mas clara.
    """
    if not os.path.isdir(work_dir):
        return
    for nombre in os.listdir(work_dir):
        ext = os.path.splitext(nombre)[1].lower()
        if ext in (NGC_EXT, ".svg"):
            ruta = os.path.join(work_dir, nombre)
            try:
                os.remove(ruta)
                logger.debug("Limpieza pre-corrida: borrado %s", ruta)
            except OSError as e:
                logger.warning("No se pudo borrar %s en limpieza: %s", ruta, e)


# ──────────────────────────────────────────────────────────────────
# Guardrail de disco
# ──────────────────────────────────────────────────────────────────

def _tamano_work_dir_bytes(work_dir: str) -> int:
    """
    Suma el tamano (os.path.getsize) de todos los archivos en `work_dir`,
    recursivamente. Tolera archivos que desaparecen mientras se escanea
    (los salta sin fallar). Devuelve bytes.
    """
    total = 0
    for raiz, _dirs, archivos in os.walk(work_dir):
        for nombre in archivos:
            ruta = os.path.join(raiz, nombre)
            try:
                total += os.path.getsize(ruta)
            except OSError:
                # El archivo pudo borrarse/rotarse entre el listado y el getsize.
                continue
    return total


def _excede_disco(work_dir: str, max_work_dir_mb: float, min_free_mb: float) -> Optional[str]:
    """
    Chequea las dos vallas de disco. Devuelve un reason legible si alguna se
    supera, o None si todo esta dentro de lo permitido.

      1. El tamano acumulado del work_dir supera `max_work_dir_mb`.
      2. El espacio libre del FS baja de `min_free_mb`.

    Un max/min <= 0 desactiva esa valla en particular.
    """
    if max_work_dir_mb and max_work_dir_mb > 0:
        usado_mb = _tamano_work_dir_bytes(work_dir) / (1024.0 * 1024.0)
        if usado_mb > max_work_dir_mb:
            return (
                f"el directorio de trabajo crecio sin control "
                f"({usado_mb:.0f} MB > limite {max_work_dir_mb:.0f} MB), abortado"
            )

    if min_free_mb and min_free_mb > 0:
        try:
            libre_mb = shutil.disk_usage(work_dir).free / (1024.0 * 1024.0)
        except OSError:
            libre_mb = None
        if libre_mb is not None and libre_mb < min_free_mb:
            return (
                f"el directorio de trabajo crecio sin control "
                f"(espacio libre {libre_mb:.0f} MB < minimo {min_free_mb:.0f} MB), abortado"
            )

    return None


# ──────────────────────────────────────────────────────────────────
# Validacion de contenido real de un .ngc
# ──────────────────────────────────────────────────────────────────

def _validar_ngc(ruta: str, inicio: float) -> Dict:
    """
    Valida que `ruta` sea un .ngc de exito segun el spec.

    Cumple SOLO si:
      (c) el archivo existe,
      (d) su mtime es posterior a `inicio` (no rancio de una corrida vieja),
      (e) contiene G-code real: G21 (milimetros) Y al menos un G1 con Z negativo.

    (Las condiciones (a) returncode y (b) stderr limpio las chequea `run`,
    son globales a la corrida, no por archivo.)

    Returns:
        dict {ok, existe, fresco, tiene_g21, tiene_motion, reason}
    """
    out = {
        "ok": False,
        "existe": False,
        "fresco": False,
        "tiene_g21": False,
        "tiene_motion": False,
        "reason": "",
    }

    if not os.path.isfile(ruta):
        out["reason"] = "no existe el .ngc esperado"
        return out
    out["existe"] = True

    # (d) mtime posterior al inicio de la corrida.
    try:
        mtime = os.path.getmtime(ruta)
    except OSError as e:
        out["reason"] = f"no se pudo leer mtime: {e}"
        return out
    # Margen pequeno por resolucion de mtime en algunos FS.
    if mtime + 1.0 < inicio:
        out["reason"] = "el .ngc es rancio (mtime previo al inicio de la corrida)"
        return out
    out["fresco"] = True

    # (e) contenido real.
    try:
        with open(ruta, "r", encoding="utf-8", errors="replace") as f:
            texto = f.read()
    except OSError as e:
        out["reason"] = f"no se pudo leer el .ngc: {e}"
        return out

    out["tiene_g21"] = bool(_RE_G21.search(texto))
    out["tiene_motion"] = _texto_tiene_motion(texto)

    if not out["tiene_g21"]:
        out["reason"] = "el .ngc no declara G21 (milimetros)"
        return out
    if not out["tiene_motion"]:
        out["reason"] = "el .ngc no tiene trayectorias (sin G1 con Z negativo)"
        return out

    out["ok"] = True
    out["reason"] = "ok"
    return out


def _stderr_tiene_error(texto: str) -> Optional[str]:
    """
    Devuelve la primera linea de error encontrada en `texto`, o None si esta limpio.
    Prioriza el texto del stderr/stdout sobre el exit code (decision del spec).
    """
    if not texto:
        return None
    for linea in texto.splitlines():
        for pat in _ERROR_PATTERNS:
            if pat.lower() in linea.lower():
                return linea.strip()
    return None


# ──────────────────────────────────────────────────────────────────
# Corrida principal
# ──────────────────────────────────────────────────────────────────

def run(
    work_dir: str,
    millproject: str = "drill.millproject",
    expected_ngc: Optional[Dict[str, str]] = None,
    timeout: float = 300.0,
    on_line: Optional[Callable[[str], None]] = None,
    on_process: Optional[Callable[[subprocess.Popen], None]] = None,
    cancel: Optional[threading.Event] = None,
    max_work_dir_mb: float = DEFAULT_MAX_WORK_DIR_MB,
    min_free_mb: float = DEFAULT_MIN_FREE_MB,
) -> Dict:
    """
    Corre pcb2gcode sobre `work_dir` usando el `millproject` indicado.

    Comando: [PCB2GCODE_BIN, '--config', millproject], cwd=work_dir, sin shell.
    Streamea stdout/stderr linea a linea invocando `on_line` (para mandarlas al
    log via self.after en la UI). Lee returncode SOLO tras wait().

    Args:
        work_dir: carpeta con Gerbers/Excellon + el millproject. Es el cwd.
        millproject: nombre (no ruta) del archivo de config, relativo a work_dir.
        expected_ngc: dict {operacion: nombre_archivo_ngc} de los .ngc que ESTA
            corrida debe producir, tipado por operacion (iso_back, drill, cutout,
            milldrill). Si None, se infiere de los .ngc que aparezcan tras correr
            (modo permisivo, menos seguro).
        timeout: segundos maximos antes de matar el hijo.
        on_line: callback(linea) por cada linea de salida del subproceso.
        on_process: callback(popen) entregado apenas arranca el proceso, para que
            el boton Detener pueda guardar el handle y llamar terminate()/kill().
        cancel: threading.Event; si se setea, se mata el proceso y se devuelve
            ok=False con reason de cancelacion.
        max_work_dir_mb: tope al tamano acumulado del work_dir mientras corre. Si
            se supera, se mata el proceso y se aborta (guardrail de disco).
            <= 0 desactiva esta valla.
        min_free_mb: minimo de espacio libre en el FS. Si baja de aca, se aborta.
            <= 0 desactiva esta valla.

    Returns:
        dict {
          ok: bool,                 # las 5 condiciones de exito juntas
          code: int|None,           # returncode (None si no termino)
          out: str,                 # stdout acumulado
          err: str,                 # stderr acumulado
          reason: str,              # explicacion legible del resultado
          outputs: dict,            # {operacion: ruta_absoluta} validados, o {} si fallo
        }
    """
    resultado = {
        "ok": False,
        "code": None,
        "out": "",
        "err": "",
        "reason": "",
        "outputs": {},
    }

    # Validacion basica de entradas.
    if not os.path.isdir(work_dir):
        resultado["reason"] = f"work_dir no existe: {work_dir}"
        return resultado

    millproject_path = os.path.join(work_dir, millproject)
    if not os.path.isfile(millproject_path):
        resultado["reason"] = f"no existe el millproject: {millproject_path}"
        return resultado

    # Limpieza pre-corrida: borrar .ngc/.svg viejos para que no se cuelen como exito.
    _limpiar_work_dir(work_dir)

    if cancel is not None and cancel.is_set():
        resultado["reason"] = "cancelado antes de iniciar"
        return resultado

    inicio = time.time()
    cmd = [PCB2GCODE_BIN, "--config", millproject]
    logger.info("Corriendo pcb2gcode: %s (cwd=%s)", " ".join(cmd), work_dir)

    out_buf: List[str] = []
    err_buf: List[str] = []

    try:
        proc = subprocess.Popen(
            cmd,
            cwd=work_dir,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding="utf-8",
            # errors='replace': un byte no-UTF8 en stderr no debe tumbar el thread
            # de drenaje ni hacer perder una linea 'Error:' (eso romperia la
            # condicion (b) de exito). Se reemplaza el byte malo y se sigue.
            errors="replace",
            bufsize=1,  # line-buffered
        )
    except FileNotFoundError:
        resultado["reason"] = (
            f"pcb2gcode no encontrado en {PCB2GCODE_BIN}. "
            "Instalalo con: brew install pcb2gcode"
        )
        logger.error(resultado["reason"])
        return resultado
    except OSError as e:
        resultado["reason"] = f"no se pudo lanzar pcb2gcode: {e}"
        logger.error(resultado["reason"])
        return resultado

    # Entregar el handle para que el boton Detener lo pueda matar.
    if on_process is not None:
        try:
            on_process(proc)
        except Exception as e:  # callback de UI nunca debe tumbar la corrida
            logger.warning("on_process lanzo excepcion: %s", e)

    # Streaming de stdout y stderr en threads para no bloquear ni perder lineas.
    def _drenar(pipe, buf: List[str], es_err: bool):
        try:
            for raw in iter(pipe.readline, ""):
                linea = raw.rstrip("\n")
                buf.append(linea)
                if on_line is not None:
                    try:
                        on_line(("[err] " if es_err else "") + linea)
                    except Exception as e:
                        logger.warning("on_line lanzo excepcion: %s", e)
        except Exception as e:
            logger.warning("Error drenando pipe: %s", e)
        finally:
            try:
                pipe.close()
            except Exception:
                pass

    t_out = threading.Thread(target=_drenar, args=(proc.stdout, out_buf, False), daemon=True)
    t_err = threading.Thread(target=_drenar, args=(proc.stderr, err_buf, True), daemon=True)
    t_out.start()
    t_err.start()

    # Espera con vigilancia de cancelacion, timeout y disco.
    cancelado = False
    timed_out = False
    razon_disco: Optional[str] = None
    iteracion = 0
    while True:
        try:
            proc.wait(timeout=0.2)
            break  # termino solo
        except subprocess.TimeoutExpired:
            pass

        if cancel is not None and cancel.is_set():
            cancelado = True
            _matar(proc)
            break

        if (time.time() - inicio) > timeout:
            timed_out = True
            _matar(proc)
            break

        # Guardrail de disco: chequeo periodico (no en cada vuelta) del tamano
        # del work_dir y del espacio libre. Si se descontrola, matar y abortar.
        iteracion += 1
        if iteracion % _DISK_CHECK_CADA_N == 0:
            razon_disco = _excede_disco(work_dir, max_work_dir_mb, min_free_mb)
            if razon_disco:
                logger.error("Guardrail de disco disparado: %s", razon_disco)
                _matar(proc)
                break

    # Asegurar que los threads terminen de drenar.
    t_out.join(timeout=5.0)
    t_err.join(timeout=5.0)

    resultado["code"] = proc.returncode
    resultado["out"] = "\n".join(out_buf)
    resultado["err"] = "\n".join(err_buf)

    if cancelado:
        resultado["reason"] = "cancelado por el usuario (Detener)"
        return resultado

    if timed_out:
        resultado["reason"] = f"timeout: pcb2gcode supero {timeout:.0f}s y fue terminado"
        return resultado

    if razon_disco:
        resultado["reason"] = razon_disco
        return resultado

    # ── Condicion (a): returncode 0 ──
    if proc.returncode != 0:
        err_linea = _stderr_tiene_error(resultado["err"]) or _stderr_tiene_error(resultado["out"])
        if err_linea:
            resultado["reason"] = f"pcb2gcode fallo (code {proc.returncode}): {err_linea}"
        else:
            resultado["reason"] = f"pcb2gcode termino con code {proc.returncode}"
        return resultado

    # ── Condicion (b): stderr/stdout sin linea Error:/unrecognised option ──
    err_linea = _stderr_tiene_error(resultado["err"]) or _stderr_tiene_error(resultado["out"])
    if err_linea:
        resultado["reason"] = f"pcb2gcode reporto un error: {err_linea}"
        return resultado

    # ── Condiciones (c)(d)(e): por cada .ngc esperado ──
    # Determinar el set de archivos a validar (tipado por operacion).
    if expected_ngc:
        items = list(expected_ngc.items())
    else:
        # Modo permisivo: validar todos los .ngc que aparecieron.
        items = [
            (os.path.basename(p), os.path.basename(p))
            for p in list_output_ngc(work_dir)
        ]
        if not items:
            resultado["reason"] = "pcb2gcode no produjo ningun .ngc"
            return resultado

    outputs: Dict[str, str] = {}
    for operacion, nombre in items:
        ruta = os.path.join(work_dir, nombre)
        chequeo = _validar_ngc(ruta, inicio)
        if not chequeo["ok"]:
            resultado["reason"] = f"{operacion} ({nombre}): {chequeo['reason']}"
            # No publicar parcial: outputs queda vacio.
            resultado["outputs"] = {}
            return resultado
        outputs[operacion] = ruta

    resultado["ok"] = True
    resultado["outputs"] = outputs
    resultado["reason"] = "ok"
    logger.info("pcb2gcode OK: %s", outputs)
    return resultado


def _matar(proc: subprocess.Popen) -> None:
    """
    Mata el subproceso al instante: terminate() y, si no muere, kill().
    """
    try:
        proc.terminate()
    except Exception as e:
        logger.warning("terminate() fallo: %s", e)
    try:
        proc.wait(timeout=2.0)
    except subprocess.TimeoutExpired:
        try:
            proc.kill()
            proc.wait(timeout=2.0)
        except Exception as e:
            logger.warning("kill() fallo: %s", e)
    except Exception:
        pass