#!/usr/bin/env python # -*- coding: utf-8 -*- """ Comparacion de todas las sesiones de campo disponibles (logs en la raiz del repositorio). Por log se reporta: * ventana temporal y numero de muestras * eta_tr, MAE_P, energia, irradiancia acumulada (metricas del parser) * mediana P/Pref y fraccion de muestras con P >= 0.95*P_ref * SATURACION del piranometro: el ADC se topa a 1024 W/m2 - n / fraccion de muestras saturadas (G >= 1023.9) - minutos acumulados saturados - R_opt (max Isc/G) del pipeline actual vs R_opt sin muestras saturadas -> factor de inflacion de P_ref por el clipping - eta_tr estimado si R_opt se calcula sin muestras saturadas * n transitorios nubosos detectados Uso: py -3.12 compare_sessions.py Salida: articulo/comparacion_sesiones.json + tabla en consola """ import json import re import sys import tempfile from pathlib import Path import numpy as np import pandas as pd import parse_log as pl ROOT = Path(__file__).resolve().parent # (archivo en la raiz, etiqueta, formato) LOGS = [ ("hasta la 1pm 020925.txt", "02 Sep 11:27-13:00", "new"), ("esp_1223_02092025.txt", "02 Sep 12:23-..", "old"), ("log(3).txt", "14 Sep (manana)", "new"), ("gyuggyhoy.txt", "14 Sep 10:28-..", "old"), ("10092025.txt", "10 Sep", "new"), ("15092025.txt", "15 Sep", "new"), ("19092025_.txt", "19 Sep", "new"), ("20092025_desde1pm.txt", "20 Sep (desde 13:44)", "new"), ("21092025.txt", "21 Sep (paper)", "new"), ("21092025_2PM.txt", "21 Sep (parcial, dup.)", "new"), ("23092025.txt", "23 Sep", "new"), ("full_log_21082025.txt", "21 Ago (log completo)", "old"), ("esp32_log_20250821_102525.txt", "21 Ago 10:25", "old"), ("esp32_log_20250821_1143.txt", "21 Ago 11:43", "old"), ("esp32_log_20250826_114833.txt", "26 Ago 11:48", "old"), ("esp32_log_20250827_095607.txt", "27 Ago 09:56", "old"), ("esp32_log_20250827_110728.txt", "27 Ago 11:07", "old"), ("esp32_log_20250827_449pm.txt", "27 Ago 16:49", "old"), ("log_26082025.txt", "26 Ago (corto)", "old"), ("log_2608205_2.txt", "26 Ago (corto)", "old"), ("nuevolog}.txt", "26 Ago 13:28", "old"), ("test_2108.txt", "21 Ago (test)", "old"), ("test_3.txt", "21 Ago (test)", "old"), ("test_$.txt", "21 Ago (test)", "old"), ] G_CLIP = 1023.9 # umbral: ADC en tope => G = 1024.0 W/m2 def normalize(path: Path, tmp: Path): """Copia el log normalizando el prefijo 'LOG:' del firmware antiguo.""" with open(path, "r", encoding="utf-8", errors="replace") as f: lines = f.readlines() out = [] for line in lines: line = line.rstrip("\n") if line.startswith("LOG:"): line = line[4:] out.append(line) tmp.write_text("\n".join(out) + "\n", encoding="utf-8") def session_report(path: Path, label: str, fmt: str): if not path.exists(): return {"file": path.name, "label": label, "error": "archivo inexistente"} tmp = Path(tempfile.gettempdir()) / (path.stem + "_norm.txt") try: normalize(path, tmp) df_s, df_e = pl.parse_log(tmp) df, ph = pl.build_physical(df_s, df_e) except Exception as exc: return {"file": path.name, "label": label, "fmt": fmt, "error": f"{type(exc).__name__}: {exc}"} m = pl.window_metrics(df, "full") if m["n_samples"] == 0: return {"file": path.name, "label": label, "fmt": fmt, "error": "sin muestras validas"} day = df[df["G_POA"] > 50] n_clip = int((df["G_POA"] >= G_CLIP).sum()) dt_med = float(np.median(df["dt_s"])) if "dt_s" in df else 2.0 clip_min = n_clip * dt_med / 60.0 if n_clip else 0.0 lit = df[df["G_POA"] > 600] R_full = float((lit["Isc"] / lit["G_POA"]).max()) if len(lit) else float("nan") unc = df[(df["G_POA"] > 600) & (df["G_POA"] < 1015)] R_unc = float((unc["Isc"] / unc["G_POA"]).max()) if len(unc) else float("nan") infl = (R_full / R_unc) if (R_unc == R_unc and R_unc > 0) else None w = df[df["G_POA"] > 50] ratio = (w["P_CPV"] / w["P_ref"]).values return { "file": path.name, "label": label, "fmt": fmt, "t_inicio": m["t_inicio"], "t_fin": m["t_fin"], "n_samples": m["n_samples"], "eta_tr_pct": m["eta_tr_pct"], "MAE_P_mW": m["MAE_P_mW"], "E_Wh": m["E_Wh"], "H_kWh_m2": m["H_kWh_m2"], "ratio_P_Pref_median": round(float(np.median(ratio)), 3), "frac_P_above_095Pref": round(float(np.mean(w["P_CPV"].values >= 0.95 * w["P_ref"].values)), 3), "G_max_Wm2": round(float(df["G_POA"].max()), 1), "n_clip": n_clip, "clip_pct_day": round(100.0 * n_clip / len(day), 1) if len(day) else 0.0, "clip_min": round(clip_min, 1), "R_opt_uA_per_Wm2": round(R_full * 1e6, 1), "R_opt_unc_uA_per_Wm2": round(R_unc * 1e6, 1) if R_unc == R_unc else None, "inflation_factor": round(infl, 4) if infl else None, "eta_tr_pct_Ropt_corrected": round(m["eta_tr_pct"] * infl, 2) if (infl and m["eta_tr_pct"]) else None, "n_cloud_events": len(pl.cloud_events(df)), } def main(): rows = [] for name, label, fmt in LOGS: rep = session_report(ROOT / name, label, fmt) rows.append(rep) if "error" in rep: print(f"[x] {label:<26} {rep['error']}") ok = [r for r in rows if r.get("eta_tr_pct")] ok.sort(key=lambda r: r["eta_tr_pct"], reverse=True) print("\n" + "=" * 118) print(f"{'sesion':<24} {'horario':<19} {'n':>5} {'eta%':>6} {'MAE_P':>7} " f"{'E Wh':>6} {'H kWh/m2':>9} {'med P/Pref':>10} {'%>=0.95':>8} " f"{'Gmax':>5} {'nClip':>5} {'%clip':>5} {'minClip':>7} {'Ropt(uA/Wm2)':>13} {'infl':>6} {'nNub':>4}") print("=" * 118) for r in ok: infl = f"{r['inflation_factor']}x" if r["inflation_factor"] else " -" print(f"{r['label']:<24} {r['t_inicio'][11:16]}-{r['t_fin'][11:16]} " f"{r['n_samples']:>5} {r['eta_tr_pct']:>6} {r['MAE_P_mW']:>7.1f} " f"{r['E_Wh']:>6} {r['H_kWh_m2']:>9} {r['ratio_P_Pref_median']:>10} " f"{100 * r['frac_P_above_095Pref']:>7.0f}% {r['G_max_Wm2']:>5.0f} " f"{r['n_clip']:>5} {r['clip_pct_day']:>5.1f} {r['clip_min']:>7.0f} " f"{r['R_opt_uA_per_Wm2']:>13.0f} {infl:>6} {r['n_cloud_events']:>4}") print("=" * 118) print("NOTA: infl = R_opt(clips incluidos)/R_opt(sin clips); el clipping " "INFLA P_ref y subestima eta, por lo que eta (corregido) = eta * infl\n") for r in ok: if r.get("eta_tr_pct_Ropt_corrected"): print(f" {r['label']:<24} eta corregido (solo R_opt) ~ {r['eta_tr_pct_Ropt_corrected']} %") out = ROOT / "articulo" / "comparacion_sesiones.json" out.write_text(json.dumps(rows, indent=2, ensure_ascii=False), encoding="utf-8") print(f"\n[ok] tabla guardada en {out}") if __name__ == "__main__": main()