| |
| """ |
| logic.py — Misma logica de analisis que main.py (FastAPI), adaptada para |
| ejecutarse dentro del navegador via Pyodide en la version estatica. |
| """ |
| import io |
| import json |
| import numpy as np |
| import pandas as pd |
|
|
| import conditions_core as cc |
|
|
| UNIDADES = {"temp_ambiente": "C", "punto_rocio": "C", "presion": "hPa"} |
| SERIES = ["humedad", "viento", "margen_rocio", "temp_ambiente", "punto_rocio", "presion"] |
|
|
|
|
| def _limpiar(o): |
| if isinstance(o, dict): |
| return {k: _limpiar(v) for k, v in o.items()} |
| if isinstance(o, (list, tuple)): |
| return [_limpiar(v) for v in o] |
| if isinstance(o, (np.floating,)): |
| return float(o) |
| if isinstance(o, (np.integer,)): |
| return int(o) |
| if isinstance(o, (np.bool_,)): |
| return bool(o) |
| if isinstance(o, float) and not np.isfinite(o): |
| return None |
| if isinstance(o, float): |
| return round(o, 4) |
| return o |
|
|
|
|
| def analizar(dfn, fuente, avisos, mapa): |
| estados = cc.evaluar(dfn) |
| inc = cc.diagnosticar(estados) |
| informe = cc.generar_informe(inc, mapa) |
| est_por = {e["sensor"]: e for e in estados} |
|
|
| t = [round(float(x), 4) for x in dfn["t_horas"].tolist()] |
| sensores = {} |
| for v in SERIES: |
| if v in dfn.columns: |
| y = dfn[v].to_numpy(dtype=float) |
| if not np.isfinite(y).any(): |
| continue |
| y_disp = pd.Series(y).interpolate(limit_direction="both").to_numpy() |
| lim = cc.LIMITES.get(v, {}) |
| sensores[v] = { |
| "valores": [round(float(x), 3) for x in y_disp], |
| "anomalias": cc.detectar_anomalias(y_disp).tolist(), |
| "actual": round(float(y_disp[-1]), 2), |
| "unidad": lim.get("unidad", UNIDADES.get(v, "")), |
| "aviso": lim.get("aviso"), |
| "critico": lim.get("critico"), |
| "dir": lim.get("dir", "up"), |
| "nivel": est_por.get(v, {}).get("nivel", "ok"), |
| "etiqueta": cc.ETIQUETAS.get(v, v), |
| } |
|
|
| nivel_global = ("critico" if any(i["nivel"] == "critico" for i in inc) |
| else "aviso" if inc else "ok") |
| return _limpiar({ |
| "fuente": fuente, |
| "avisos": avisos, |
| "columnas_detectadas": mapa, |
| "t": t, |
| "sensores": sensores, |
| "incidencias": inc, |
| "informe": informe, |
| "nivel_global": nivel_global, |
| "n_incidencias": len(inc), |
| "n_muestras": len(dfn), |
| }) |
|
|
|
|
| def analizar_muestra(csv_texto): |
| df = pd.read_csv(io.StringIO(csv_texto)) |
| dfn, mapa, avisos = cc.normalizar(df) |
| return json.dumps(analizar(dfn, "Muestra (formato tipo ESO)", avisos, mapa)) |
|
|
|
|
| def analizar_subida(csv_texto, nombre): |
| try: |
| df = pd.read_csv(io.StringIO(csv_texto)) |
| except Exception as exc: |
| return json.dumps({"error": f"No se pudo leer el CSV: {exc}"}) |
| if len(df) < 5: |
| return json.dumps({"error": "El CSV tiene muy pocas filas."}) |
| dfn, mapa, avisos = cc.normalizar(df) |
| if not any(v in dfn.columns for v in ["humedad", "viento", "temp_ambiente"]): |
| return json.dumps({"error": "No se reconocio ninguna columna de humedad, viento o temperatura."}) |
| return json.dumps(analizar(dfn, nombre, avisos, mapa)) |
|
|
