app.py

"""
app.py
Proyecto: CloudManteinAI

Aplicación web con Gradio para clasificar riesgo operacional.
El modelo tradicional de Machine Learning realiza la predicción.
Gemini API, si está disponible, solo genera una explicación auxiliar.
"""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

import gradio as gr
import joblib
import pandas as pd

BASE_DIR = Path(__file__).resolve().parent
MODELS_DIR = BASE_DIR / "models"

MODEL_PATH = MODELS_DIR / "model.joblib"
FEATURES_PATH = MODELS_DIR / "features.json"
METRICS_PATH = MODELS_DIR / "metrics.json"

DEFAULT_FEATURES = [
    "horas_operacion",
    "carga_operacional",
    "temperatura",
    "vibracion",
    "energia_acumulada",
    "historial_fallas",
    "severidad_operacional",
]

ARTIFACT_ERROR = ""


def load_features(path: Path) -> list[str]:
    with open(path, "r", encoding="utf-8") as file:
        payload = json.load(file)

    if isinstance(payload, list):
        return payload

    if isinstance(payload, dict) and "features" in payload:
        return payload["features"]

    raise ValueError("features.json no tiene un formato valido.")


def load_metrics(path: Path) -> dict[str, Any]:
    with open(path, "r", encoding="utf-8") as file:
        return json.load(file)


def load_artifacts() -> tuple[Any | None, list[str], dict[str, Any]]:
    global ARTIFACT_ERROR

    missing = [path for path in [MODEL_PATH, FEATURES_PATH, METRICS_PATH] if not path.exists()]
    if missing:
        ARTIFACT_ERROR = (
            "Faltan artefactos necesarios para la inferencia: "
            + ", ".join(str(path.name) for path in missing)
            + ". Ejecuta primero train_model.py y verifica la carpeta models/."
        )
        return None, DEFAULT_FEATURES, {}

    try:
        loaded_model = joblib.load(MODEL_PATH)
        loaded_features = load_features(FEATURES_PATH)
        loaded_metrics = load_metrics(METRICS_PATH)
        ARTIFACT_ERROR = ""
        return loaded_model, loaded_features, loaded_metrics
    except Exception as error:
        ARTIFACT_ERROR = f"No fue posible cargar los artefactos del modelo. Tipo de error: {type(error).__name__}."
        return None, DEFAULT_FEATURES, {}


MODEL, FEATURES, METRICS = load_artifacts()


def get_gemini_api_key() -> str | None:
    api_key = os.getenv("GEMINI_API_KEY")
    if api_key:
        return api_key

    try:
        from google.colab import userdata  # type: ignore
        return userdata.get("GEMINI_API_KEY")
    except Exception:
        return None




def generar_explicacion_gemini(riesgo: str, probabilidad: float, datos_usuario: dict[str, Any]) -> str:
    api_key = get_gemini_api_key()

    if not api_key:
        return (
            "Modo básico activo: no se encontró una clave GEMINI_API_KEY configurada. "
            "La predicción corresponde únicamente al modelo tradicional de Machine Learning."
        )

    try:
        from google import genai

        client = genai.Client(api_key=api_key)

        prompt = f"""
Eres una capa auxiliar de interpretación para un prototipo académico llamado CloudManteinAI.

Reglas:
- No predices el riesgo.
- No modificas el resultado del modelo.
- No entregues instrucciones obligatorias de mantenimiento.
- No inventes causas no entregadas por los datos.
- Redacta una explicación breve, prudente y clara.
- Indica que el resultado corresponde a una prueba académica con datos sintéticos o de prueba.
- No uses Markdown, negritas, listas ni símbolos especiales.

Resultado del modelo:
- Riesgo operacional predicho: {riesgo}
- Probabilidad estimada: {probabilidad:.2%}

Variables ingresadas:
{json.dumps(datos_usuario, ensure_ascii=False, indent=2)}

Redacta una explicación de máximo 5 líneas.
"""

        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=prompt,
        )

        texto = getattr(response, "text", "")

        if texto:
            return texto.strip()

        return (
            "Gemini API no entregó texto interpretable. "
            "Se mantiene la predicción principal del modelo."
        )

    except Exception as error:
        return (
            "No fue posible generar la explicación con Gemini API. "
            "La predicción principal del modelo se mantiene disponible. "
            f"Detalle general del error: {type(error).__name__}"
        )

def convertir_entradas(
    horas_operacion: Any,
    carga_operacional: Any,
    temperatura: Any,
    vibracion: Any,
    energia_acumulada: Any,
    historial_fallas: Any,
    severidad_operacional: Any,
) -> dict[str, float | int]:
    if historial_fallas in ["0", "Sin fallas previas"]:
        historial_fallas = 0
    elif historial_fallas in ["1", "Con fallas previas"]:
        historial_fallas = 1

    return {
        "horas_operacion": float(horas_operacion),
        "carga_operacional": float(carga_operacional),
        "temperatura": float(temperatura),
        "vibracion": float(vibracion),
        "energia_acumulada": float(energia_acumulada),
        "historial_fallas": int(historial_fallas),
        "severidad_operacional": float(severidad_operacional),
    }


def validar_entradas(datos: dict[str, float | int]) -> list[str]:
    errores: list[str] = []

    if datos["horas_operacion"] < 0 or datos["horas_operacion"] > 20000:
        errores.append("Las horas de operación deben estar entre 0 y 20000.")

    if datos["carga_operacional"] < 0 or datos["carga_operacional"] > 120:
        errores.append("La carga operacional debe estar entre 0 y 120.")

    if datos["temperatura"] < 0 or datos["temperatura"] > 150:
        errores.append("La temperatura debe estar entre 0 y 150.")

    if datos["vibracion"] < 0 or datos["vibracion"] > 20:
        errores.append("La vibración debe estar entre 0 y 20.")

    if datos["energia_acumulada"] < 0:
        errores.append("La energía acumulada no puede ser negativa.")

    if datos["historial_fallas"] not in [0, 1]:
        errores.append("El historial de fallas debe ser 0 o 1.")

    if datos["severidad_operacional"] < 0 or datos["severidad_operacional"] > 1:
        errores.append("La severidad operacional debe estar entre 0 y 1.")

    return errores


def calcular_probabilidad(modelo: Any, x: pd.DataFrame) -> float:
    if hasattr(modelo, "predict_proba"):
        probabilidades = modelo.predict_proba(x)[0]
        return float(probabilidades.max())
    return 0.0


def predecir_riesgo(
    horas_operacion: Any,
    carga_operacional: Any,
    temperatura: Any,
    vibracion: Any,
    energia_acumulada: Any,
    historial_fallas: Any,
    severidad_operacional: Any,
    usar_gemini: bool,
) -> tuple[str, str, str, str]:
    if MODEL is None:
        return (
            "Modelo no disponible",
            "No disponible",
            ARTIFACT_ERROR,
            "Ejecuta train_model.py antes de iniciar la aplicación o verifica que models/ esté incluido en el despliegue.",
        )

    try:
        datos_usuario = convertir_entradas(
            horas_operacion,
            carga_operacional,
            temperatura,
            vibracion,
            energia_acumulada,
            historial_fallas,
            severidad_operacional,
        )

        errores = validar_entradas(datos_usuario)
        if errores:
            return ("Entrada no válida", "No disponible", "\n".join(errores), "Corrige los valores ingresados.")

        x = pd.DataFrame([datos_usuario], columns=FEATURES)
        riesgo = str(MODEL.predict(x)[0])
        probabilidad = calcular_probabilidad(MODEL, x)

        f1_macro = METRICS.get("metricas_mejor_modelo", {}).get("f1_macro", "No informado")
        f1_macro_texto = f"{float(f1_macro):.4f}" if isinstance(f1_macro, (float, int)) else str(f1_macro)

        resumen_modelo = (
            f"Modelo seleccionado: {METRICS.get('mejor_modelo', 'No informado')}\n"
            f"F1 macro registrado: {f1_macro_texto}\n"
            "Nota: métricas obtenidas sobre datos sintéticos o de prueba."
        )

        if usar_gemini:
            explicacion = generar_explicacion_gemini(riesgo, probabilidad, datos_usuario)
        else:
            explicacion = (
                "Modo básico activo. Gemini API no fue utilizada. "
                "La predicción corresponde únicamente al modelo tradicional de Machine Learning."
            )

        advertencia = (
            "Advertencia: CloudManteinAI es una prueba de concepto académica. "
            "Los resultados se basan en datos sintéticos o de prueba. "
            "No constituyen validación industrial ni reemplazan la revisión de especialistas."
        )

        return riesgo, f"{probabilidad:.2%}", explicacion, resumen_modelo + "\n\n" + advertencia

    except ValueError as error:
        return (
            "Entrada no válida",
            "No disponible",
            f"Error de formato en la entrada: {error}",
            "Asegúrate de que todos los campos numéricos contengan valores válidos.",
        )
    except Exception as error:
        return (
            "Error interno",
            "No disponible",
            "Ocurrió un error inesperado durante la predicción. Revisa el registro de ejecución.",
            f"Tipo de error: {type(error).__name__}",
        )


descripcion = """
# CloudManteinAI

Prototipo académico para clasificar riesgo operacional en mantenimiento preventivo.

El modelo tradicional de Machine Learning realiza la predicción.  
Gemini API, si está disponible, solo genera una explicación auxiliar en lenguaje natural.
"""

with gr.Blocks(title="CloudManteinAI") as demo:
    gr.Markdown(descripcion)

    with gr.Row():
        with gr.Column():
            horas_operacion = gr.Number(label="Horas de operación", value=3000, minimum=0, maximum=20000)
            carga_operacional = gr.Slider(label="Carga operacional (%)", minimum=0, maximum=120, value=70, step=1)
            temperatura = gr.Slider(label="Temperatura", minimum=0, maximum=150, value=65, step=1)
            vibracion = gr.Slider(label="Vibración", minimum=0, maximum=20, value=4.5, step=0.1)

        with gr.Column():
            energia_acumulada = gr.Number(label="Energía acumulada", value=250000, minimum=0)
            historial_fallas = gr.Radio(
                label="Historial de fallas",
                choices=[0, 1],
                value=0,
                info="0 = sin fallas previas, 1 = con fallas previas",
            )
            severidad_operacional = gr.Slider(label="Severidad operacional", minimum=0, maximum=1, value=0.45, step=0.01)
            usar_gemini = gr.Checkbox(label="Usar Gemini API para explicación auxiliar", value=False)

    boton = gr.Button("Predecir riesgo operacional")

    with gr.Row():
        riesgo_salida = gr.Textbox(label="Riesgo operacional predicho")
        probabilidad_salida = gr.Textbox(label="Probabilidad estimada")

    explicacion_salida = gr.Textbox(label="Explicación auxiliar", lines=8)
    nota_salida = gr.Textbox(label="Información metodológica y advertencia", lines=8)

    boton.click(
        fn=predecir_riesgo,
        inputs=[
            horas_operacion,
            carga_operacional,
            temperatura,
            vibracion,
            energia_acumulada,
            historial_fallas,
            severidad_operacional,
            usar_gemini,
        ],
        outputs=[riesgo_salida, probabilidad_salida, explicacion_salida, nota_salida],
    )


if __name__ == "__main__":
    share = os.getenv("GRADIO_SHARE", "0").lower() in ["1", "true", "yes"]
    debug = os.getenv("GRADIO_DEBUG", "0").lower() in ["1", "true", "yes"]

    # Gradio también lee GRADIO_DEBUG internamente.
    # Se normaliza a 0 o 1 para evitar errores con valores como "false".
    os.environ["GRADIO_DEBUG"] = "1" if debug else "0"

    demo.launch(share=share, debug=debug)