core/models.py

"""
models.py — Capa cognitiva (NLP zero-shot + Random Forest termodinámico).

Diferencia con el original: los modelos NO se cargan al importar el módulo
(eso obligaba a tener transformers/torch presentes solo para arrancar). Aquí se
cargan de forma PEREZOSA (lazy) la primera vez que se necesitan, vía
get_nlp_model() y get_tr_model(). La lógica de inferencia (calcular_ic_ia,
calcular_tr_y_ic_dinamico) es idéntica a app_tesis_final_V3.py.
"""
import warnings
import numpy as np

from .constants import ENFORCE_TR0_RESILIENTES

warnings.filterwarnings("ignore")

# ------------------------------------------------------------------
# Disponibilidad de librerías de IA (independientes entre sí).
# El Random Forest solo necesita sklearn; el NLP necesita transformers.
# ------------------------------------------------------------------
try:
    from sklearn.ensemble import RandomForestRegressor           # noqa: F401
    ML_DISPONIBLE = True
except ImportError:
    ML_DISPONIBLE = False

try:
    from transformers import pipeline as _hf_pipeline            # noqa: F401
    NLP_DISPONIBLE = True
except ImportError:
    NLP_DISPONIBLE = False

# Compatibilidad: "hay IA" si al menos uno de los dos modelos está disponible.
IA_DISPONIBLE = ML_DISPONIBLE or NLP_DISPONIBLE

# Singletons perezosos (se rellenan al primer uso).
_nlp_classifier = None
_nlp_cargado = False
_ml_tr_model = None
_ml_cargado = False

# Caché de predicciones del Random Forest (clave: entradas redondeadas).
_TR_CACHE: dict = {}


def get_nlp_model():
    """Carga perezosa del clasificador zero-shot. Devuelve None si no se puede."""
    global _nlp_classifier, _nlp_cargado
    if _nlp_cargado:
        return _nlp_classifier
    _nlp_cargado = True
    if not NLP_DISPONIBLE:
        _nlp_classifier = None
        return None
    try:
        from transformers import pipeline
        _nlp_classifier = pipeline(
            "zero-shot-classification", model="Recognai/zeroshot_selectra_medium"
        )
    except Exception:
        _nlp_classifier = None
    return _nlp_classifier


def get_tr_model():
    """Carga/entrena perezosamente el Random Forest termodinámico (Tr en horas)."""
    global _ml_tr_model, _ml_cargado
    if _ml_cargado:
        return _ml_tr_model
    _ml_cargado = True
    _ml_tr_model = _entrenar_modelo_termodinamico()
    return _ml_tr_model


def _entrenar_modelo_termodinamico():
    """
    Generador de datos sintéticos basados en física real (idéntico al original).
    Entrena al Random Forest para que devuelva HORAS de secado respondiendo
    correctamente a la temperatura y humedad.
    """
    if not ML_DISPONIBLE:
        return None
    from sklearn.ensemble import RandomForestRegressor
    np.random.seed(42)
    X_train = []
    y_train = []

    # Creamos 2000 permutaciones físicas para entrenar a la IA
    for _ in range(2000):
        lluvia = np.random.uniform(0.1, 100.0)
        temp = np.random.uniform(15.0, 45.0)
        hum = np.random.uniform(30.0, 100.0)
        suelo = np.random.choice([1, 2, 3])

        # Fórmula física real de evaporación
        evap_rate = max(0.1, (temp / 15.0) * ((100.0 - hum) / 40.0))
        factor_suelo = {1: 1.5, 2: 1.0, 3: 0.5}[suelo]  # 1 retiene más agua

        tr_horas = (lluvia / evap_rate) * factor_suelo

        X_train.append([lluvia, temp, hum, suelo])
        y_train.append(min(96.0, tr_horas))  # Cap máximo de 96 horas

    modelo = RandomForestRegressor(n_estimators=50, random_state=42)
    modelo.fit(X_train, y_train)
    return modelo


# ==============================================================================
# INFERENCIA SEMÁNTICA (Ic) — con caché por nombre y clasificación en lote.
# Los valores son idénticos a clasificar tarea por tarea; solo se evita repetir
# la inferencia para nombres iguales (clave para proyectos viales grandes).
# ==============================================================================
_IC_CACHE: dict = {}

_CATEGORIAS_IC = [
    "estructuras de hormigón y acero",
    "pavimento asfáltico y terminaciones",
    "bases granulares y subbases",
    "movimiento de tierras pesado y excavación",
]
_MAPA_IC = {_CATEGORIAS_IC[0]: 1.0, _CATEGORIAS_IC[1]: 1.5, _CATEGORIAS_IC[2]: 2.0, _CATEGORIAS_IC[3]: 3.0}


def _ic_por_palabras_clave(nombre_str: str) -> float:
    """Fallback determinista por RegEx (idéntico al original)."""
    if any(w in nombre_str for w in ['acero', 'hormigon', 'hormigón', 'encofrado', 'vaciado', 'muro', 'alcantarilla', 'losa', 'zapata', 'columna', 'viga', 'platea', 'fundacion', 'fundación', 'estructura', 'paisajismo', 'limpieza', 'grama', 'terminacion', 'terminación']):
        return 1.0
    elif any(w in nombre_str for w in ['pintura', 'señalizacion', 'señalización']):
        return 1.5
    elif any(w in nombre_str for w in ['base', 'subbase', 'sub-base', 'granular', 'afirmado', 'asfalto', 'imprimacion', 'imprimación']):
        return 2.0
    elif any(w in nombre_str for w in ['corte', 'relleno', 'subrasante', 'tierra', 'excavacion', 'excavación']):
        return 3.0
    return 1.5


def precompute_ic_batch(nombres, usar_ia=True):
    """Clasifica en UN SOLO lote todos los nombres únicos y llena el caché.

    Acelera drásticamente la corrida: una llamada al modelo en lugar de N. Si la IA
    no está disponible no hace nada (el motor usará el fallback por palabras clave).
    """
    if not usar_ia:
        return
    nlp = get_nlp_model()
    if not nlp:
        return
    unicos = sorted({str(n).lower() for n in nombres if n is not None and str(n).strip()})
    pendientes = [n for n in unicos if (n, True) not in _IC_CACHE]
    if not pendientes:
        return
    try:
        resultados = nlp(pendientes, _CATEGORIAS_IC)
        if isinstance(resultados, dict):  # el pipeline devuelve dict si es un solo texto
            resultados = [resultados]
        for nombre, res in zip(pendientes, resultados):
            _IC_CACHE[(nombre, True)] = _MAPA_IC.get(res['labels'][0], 1.5)
    except Exception:
        pass


def calcular_ic_ia(nombre_tarea, usar_ia=True):
    nombre_str = str(nombre_tarea).lower()
    key = (nombre_str, bool(usar_ia))
    cached = _IC_CACHE.get(key)
    if cached is not None:
        return cached

    nlp_classifier = get_nlp_model() if usar_ia else None
    if not usar_ia or not nlp_classifier:
        val = _ic_por_palabras_clave(nombre_str)
        _IC_CACHE[key] = val
        return val

    try:
        res = nlp_classifier(nombre_str, _CATEGORIAS_IC)
        val = _MAPA_IC.get(res['labels'][0], 1.5)
    except Exception:
        val = 1.5
    _IC_CACHE[key] = val
    return val


def calcular_tr_y_ic_dinamico(lluvia_mm, temp_c, humedad_pct, tipo_suelo_ic, usar_ia=True):
    ml_tr_model = get_tr_model() if usar_ia else None
    if not usar_ia or not ml_tr_model:
        # [AUD-04] Fallback ciego (Ceguera Térmica, CFX-AUDIT-08): valores estáticos
        # declarados en la tesis: 48 h arcillas, 12 h granulares, 6 h asfalto, 0 h estructuras.
        if tipo_suelo_ic >= 3.0:
            return 48.0, tipo_suelo_ic
        elif tipo_suelo_ic >= 2.0:
            return 12.0, tipo_suelo_ic
        elif tipo_suelo_ic >= 1.5:
            return 6.0, tipo_suelo_ic
        else:
            return 0.0, tipo_suelo_ic

    # [AUD-05] Ec. 6.19: Δθ=0 ⟹ Tr=0 para hormigón/acero (modo estricto opcional).
    if ENFORCE_TR0_RESILIENTES and tipo_suelo_ic <= 1.0:
        return 0.0, tipo_suelo_ic

    suelo_cat = 1 if tipo_suelo_ic >= 3.0 else (2 if tipo_suelo_ic >= 2.0 else 3)

    # La IA predice HORAS reales. Cacheamos la predicción con las entradas redondeadas:
    # sklearn tarda ~3.8 ms por predicción de UNA muestra, y el motor la invoca una vez
    # por día lluvioso; en un proyecto grande eso son decenas de miles de llamadas
    # idénticas. El redondeo (0.5 mm / 0.5 °C / 0.5 %) deja Tr prácticamente igual y
    # colapsa miles de llamadas a unas pocas. Resultado: hasta cientos de veces más rápido.
    key = (round(lluvia_mm * 2) / 2, round(temp_c * 2) / 2, round(humedad_pct * 2) / 2, suelo_cat)
    tr_horas = _TR_CACHE.get(key)
    if tr_horas is None:
        tr_horas = float(ml_tr_model.predict([[key[0], key[1], key[2], suelo_cat]])[0])
        _TR_CACHE[key] = tr_horas

    # [AUD-07] Penalización dinámica del coeficiente — Ec. 6.22 de la tesis:
    # Ic_dinámico = Q(Ic) = Ic_base + Tr/48  (constante de normalización 48 h)
    tr_dias = tr_horas / 24.0
    ic_dinamico = round(tipo_suelo_ic + (tr_dias * 0.5), 2)
    return round(tr_horas, 1), ic_dinamico