utils.py

"""
Utilities compartidas para Mecha Leveling.

- atomic_write_json: escritura atómica (write a .tmp → fsync → rename) para
  prevenir corrupción si la app crashea durante el guardado.
- safe_float / safe_int: parseo defensivo de entradas de UI.
- clamp: clamping inclusivo de valores numéricos.
- ValidationError: excepción específica para input mal-formado.
- backup_file_with_timestamp: snapshot histórico de un archivo.
- cleanup_old_backups: retención FIFO de backups por nombre base.
- is_file_stale: True si el archivo tiene más de N horas.
- notify_macos: notificación nativa de macOS via osascript (best-effort).
"""

import json
import os
import shutil
import subprocess
import tempfile
import time
from datetime import datetime
from typing import Any, Optional

from logging_config import get_logger

logger = get_logger(__name__)


class ValidationError(ValueError):
    """Input inválido del usuario o del firmware."""
    pass


# ──────────────────────────────────────────────────────────────────
# Escritura atómica de JSON
# ──────────────────────────────────────────────────────────────────
def atomic_write_json(filepath: str, data: Any, indent: int = 2) -> None:
    """
    Escribe `data` como JSON en `filepath` de forma atómica:
      1. Crea archivo temporal en el MISMO directorio
      2. Escribe + flush + fsync
      3. os.replace() (atomic en POSIX y Windows)

    Si la app crashea durante el paso 2, el archivo destino NO se modifica.
    Si crashea durante el paso 3, queda el original o el nuevo (nunca corrupto).

    Raises:
        OSError: si no se puede escribir (permisos, disco lleno).
        TypeError: si data no es serializable.
    """
    target_dir = os.path.dirname(os.path.abspath(filepath))
    os.makedirs(target_dir, exist_ok=True)

    # Crear tmp en el mismo directorio (rename solo es atómico en mismo FS)
    fd, tmp_path = tempfile.mkstemp(
        prefix=os.path.basename(filepath) + ".",
        suffix=".tmp",
        dir=target_dir,
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=indent, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp_path, filepath)
        logger.debug("atomic_write_json OK: %s", filepath)
    except Exception:
        # Cleanup del tmp si quedó
        try:
            if os.path.exists(tmp_path):
                os.remove(tmp_path)
        except OSError:
            pass
        raise


def safe_load_json(filepath: str, default: Optional[Any] = None) -> Any:
    """
    Lee JSON con manejo defensivo. Si el archivo no existe, está corrupto, o
    no se puede leer, retorna `default` y loguea el problema.
    """
    if not os.path.exists(filepath):
        return default
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        logger.warning("No se pudo leer %s: %s", filepath, e)
        return default


# ──────────────────────────────────────────────────────────────────
# Parseo defensivo de inputs
# ──────────────────────────────────────────────────────────────────
def safe_float(value: Any, default: Optional[float] = None,
               min_val: Optional[float] = None,
               max_val: Optional[float] = None,
               name: str = "valor") -> float:
    """
    Parsea `value` a float con bounds opcional. Si falla y `default` está dado,
    retorna `default`; si no, lanza ValidationError con mensaje claro.
    """
    try:
        v = float(value)
    except (TypeError, ValueError) as e:
        if default is not None:
            logger.debug("safe_float: %s = %r no es número, usando default %s", name, value, default)
            return default
        raise ValidationError(f"{name} debe ser un número, recibido: {value!r}") from e

    if min_val is not None and v < min_val:
        raise ValidationError(f"{name} debe ser ≥ {min_val}, recibido: {v}")
    if max_val is not None and v > max_val:
        raise ValidationError(f"{name} debe ser ≤ {max_val}, recibido: {v}")
    return v


def safe_int(value: Any, default: Optional[int] = None,
             min_val: Optional[int] = None,
             max_val: Optional[int] = None,
             name: str = "valor") -> int:
    """Parsea `value` a int con bounds opcional. Idéntica semántica a safe_float."""
    try:
        v = int(float(value))  # acepta "5" y "5.0"
    except (TypeError, ValueError) as e:
        if default is not None:
            logger.debug("safe_int: %s = %r no es entero, usando default %s", name, value, default)
            return default
        raise ValidationError(f"{name} debe ser un entero, recibido: {value!r}") from e

    if min_val is not None and v < min_val:
        raise ValidationError(f"{name} debe ser ≥ {min_val}, recibido: {v}")
    if max_val is not None and v > max_val:
        raise ValidationError(f"{name} debe ser ≤ {max_val}, recibido: {v}")
    return v


def clamp(value: float, lo: float, hi: float) -> float:
    """Clampea `value` al rango [lo, hi]. Si lo > hi, los swap."""
    if lo > hi:
        lo, hi = hi, lo
    return max(lo, min(hi, value))


# ──────────────────────────────────────────────────────────────────
# Validación de grid de sondeo
# ──────────────────────────────────────────────────────────────────
def validate_grid_params(cols: int, rows: int,
                         x_min: float, x_max: float,
                         y_min: float, y_max: float,
                         machine_max_x: float = 305.0,
                         machine_max_y: float = 200.0) -> None:
    """
    Valida que los parámetros de grid sean seguros:
    - Cols/rows ≥ 3 (mínimo para RBF razonable)
    - Cols/rows ≤ 50 (evita memoria/tiempo excesivos)
    - x_min < x_max, y_min < y_max
    - Bounds dentro de machine_max
    - x_min, y_min ≥ -1 (margen para pulloff)

    Raises ValidationError con mensaje específico si falla.
    """
    if cols < 3:
        raise ValidationError(f"Columnas debe ser ≥ 3 (RBF requiere mín 3×3), recibido: {cols}")
    if cols > 50:
        raise ValidationError(f"Columnas debe ser ≤ 50 (escaneo y RAM excesivos), recibido: {cols}")
    if rows < 3:
        raise ValidationError(f"Filas debe ser ≥ 3 (RBF requiere mín 3×3), recibido: {rows}")
    if rows > 50:
        raise ValidationError(f"Filas debe ser ≤ 50 (escaneo y RAM excesivos), recibido: {rows}")

    if x_min >= x_max:
        raise ValidationError(f"X min ({x_min}) debe ser menor que X max ({x_max})")
    if y_min >= y_max:
        raise ValidationError(f"Y min ({y_min}) debe ser menor que Y max ({y_max})")

    if x_min < -1.0:
        raise ValidationError(f"X min ({x_min}) fuera de límites (mín -1 por pulloff)")
    if y_min < -1.0:
        raise ValidationError(f"Y min ({y_min}) fuera de límites (mín -1 por pulloff)")
    if x_max > machine_max_x:
        raise ValidationError(f"X max ({x_max}) excede machine_max ({machine_max_x})")
    if y_max > machine_max_y:
        raise ValidationError(f"Y max ({y_max}) excede machine_max ({machine_max_y})")


def validate_probe_params(z_safe: float, z_depth: float, probe_feed: float) -> None:
    """
    Valida parámetros de probe Z:
    - z_safe en [0.5, 20] mm (positivo, no demasiado alto para no perder tiempo)
    - z_depth en [-10, -0.5] mm (negativo, max profundidad de búsqueda razonable)
    - probe_feed en [5, 200] mm/min (lento, para detección precisa)
    """
    if not (0.5 <= z_safe <= 20.0):
        raise ValidationError(f"Z safe debe estar entre 0.5 y 20 mm, recibido: {z_safe}")
    if not (-10.0 <= z_depth <= -0.5):
        raise ValidationError(f"Z depth debe estar entre -10 y -0.5 mm, recibido: {z_depth}")
    if not (5.0 <= probe_feed <= 200.0):
        raise ValidationError(f"Probe feed debe estar entre 5 y 200 mm/min, recibido: {probe_feed}")


# ──────────────────────────────────────────────────────────────────
# Backups, staleness, notificaciones
# ──────────────────────────────────────────────────────────────────
def backup_file_with_timestamp(filepath: str, backup_dir: Optional[str] = None) -> Optional[str]:
    """
    Crea una copia del archivo en `backup_dir/<base>_YYYYMMDD_HHMMSS.<ext>`.

    Si `backup_dir` no se pasa, usa `<dirname>/backups/`.
    Retorna el path del backup creado, o None si el archivo fuente no existe.
    """
    if not os.path.exists(filepath):
        logger.debug("backup: %s no existe, skip", filepath)
        return None
    src_dir = os.path.dirname(os.path.abspath(filepath))
    if backup_dir is None:
        backup_dir = os.path.join(src_dir, "backups")
    os.makedirs(backup_dir, exist_ok=True)

    base = os.path.basename(filepath)
    name, ext = os.path.splitext(base)
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    dst_name = f"{name}_{ts}{ext}"
    dst_path = os.path.join(backup_dir, dst_name)

    try:
        shutil.copy2(filepath, dst_path)
        logger.info("Backup creado: %s", dst_path)
        return dst_path
    except OSError as e:
        logger.warning("No se pudo crear backup de %s: %s", filepath, e)
        return None


def cleanup_old_backups(backup_dir: str, basename_prefix: str, keep: int = 10) -> int:
    """
    Mantiene solo los `keep` backups más recientes que empiecen con `basename_prefix`.
    Borra el resto. Retorna cuántos se borraron.
    """
    if not os.path.isdir(backup_dir):
        return 0
    candidates = []
    for f in os.listdir(backup_dir):
        if f.startswith(basename_prefix):
            full = os.path.join(backup_dir, f)
            if os.path.isfile(full):
                candidates.append((os.path.getmtime(full), full))
    # Más recientes primero
    candidates.sort(reverse=True)
    deleted = 0
    for _, path in candidates[keep:]:
        try:
            os.remove(path)
            deleted += 1
        except OSError as e:
            logger.warning("No se pudo borrar %s: %s", path, e)
    if deleted:
        logger.info("cleanup_old_backups: %d archivos antiguos borrados (%s)", deleted, basename_prefix)
    return deleted


def is_file_stale(filepath: str, max_age_hours: float) -> bool:
    """
    True si el archivo existe y tiene mtime > max_age_hours.
    False si no existe o es reciente.
    """
    if not os.path.exists(filepath):
        return False
    age_s = time.time() - os.path.getmtime(filepath)
    return age_s > (max_age_hours * 3600)


def file_age_hours(filepath: str) -> Optional[float]:
    """Edad del archivo en horas, o None si no existe."""
    if not os.path.exists(filepath):
        return None
    return (time.time() - os.path.getmtime(filepath)) / 3600


# ──────────────────────────────────────────────────────────────────
# XMODEM-CRC (CRC-16/XMODEM) — para upload de archivos al ESP32
# ──────────────────────────────────────────────────────────────────
def crc16_xmodem(data: bytes) -> int:
    """
    CRC-16/XMODEM clásico.
    - Poly: 0x1021
    - Init: 0x0000
    - No reflected, no XOR-out

    Vectores de referencia (CRC-16/XMODEM):
        crc16_xmodem(b"")              == 0x0000
        crc16_xmodem(b"123456789")     == 0x31C3
        crc16_xmodem(b"A")             == 0x58E5

    Usado por el protocolo XMODEM-CRC para validar bloques de 128 bytes
    al subir archivos vía serial al firmware FluidNC (v3.0+).
    """
    crc = 0
    for b in data:
        crc ^= (b << 8)
        for _ in range(8):
            crc = ((crc << 1) ^ 0x1021) if (crc & 0x8000) else (crc << 1)
            crc &= 0xFFFF
    return crc


def notify(title: str, message: str, sound: Optional[str] = None) -> bool:
    """
    Muestra una notificación del sistema: osascript en macOS, notify-send en
    Linux. En Windows (o sin herramienta disponible) no hace nada.

    sound: nombre de un sonido del sistema macOS (e.g. "Glass", "Hero") o None.
    Retorna True si el comando se ejecutó (no garantiza visibilidad).
    """
    import sys
    try:
        if sys.platform == "darwin":
            safe_title = title.replace('"', '\\"')
            safe_msg = message.replace('"', '\\"')
            script = f'display notification "{safe_msg}" with title "{safe_title}"'
            if sound:
                safe_sound = sound.replace('"', '\\"')
                script += f' sound name "{safe_sound}"'
            cmd = ["osascript", "-e", script]
        elif sys.platform.startswith("linux"):
            cmd = ["notify-send", title, message]
        else:
            return False
        subprocess.run(cmd, check=False, capture_output=True, timeout=3)
        return True
    except (subprocess.SubprocessError, FileNotFoundError, OSError) as e:
        logger.debug("notify falló (no crítico): %s", e)
        return False


# Nombre historico; el codigo nuevo usa notify().
notify_macos = notify