# Vigía de las corridas (este PC + quadrab). Ejecutar cada 10 min: python3 watchdog_aluvion.py
# Revisa avance, ETA, procesos vivos, conservación de masa (advertencias de celdas negativas), errores y uso de CPU/RAM/disco/temperatura.
# Escribe res/watchdog_estado.json y agrega una línea por revisión a res/watchdog.log; las alertas van con «ALERTA».
# Con --local imprime solo el JSON del equipo donde corre (así se usa en quadrab vía ssh).
import os, re, sys, json, glob, time, subprocess, shutil, datetime
BASE = os.path.dirname(os.path.abspath(__file__))
def sh(c): return subprocess.run(c, shell=True, capture_output=True, text=True).stdout
def equipo():
    procs = {}
    for l in sh("ps -eo pid=,etimes=,pcpu=,rss=,comm=,args=").splitlines():
        p = l.split(None, 5)
        if len(p) < 6 or 'python' not in p[4] or 'sim.py' not in p[5]: continue
        m = re.search(r'--salida[ =]res/(\S+)', p[5])
        if m: procs[m.group(1)] = {'pid': int(p[0]), 'min': int(p[1]) // 60, 'cpu': float(p[2]), 'rss_gb': round(int(p[3]) / 1e6, 1)}
    corr = {}
    for lp in glob.glob(os.path.join(BASE, 'res', '*.log')):
        cid = os.path.basename(lp)[:-4]
        if not re.match(r'^[a-z]{1,3}\d{1,3}[a-z0-9]*$', cid): continue
        npz = os.path.join(BASE, 'res', cid + '.npz')
        if cid not in procs and ((os.path.exists(npz) and time.time() - os.path.getmtime(npz) > 3600) or time.time() - os.path.getmtime(lp) > 86400): continue   # terminadas: visibles 1 h para avisar
        txt = open(lp, errors='replace').read()
        tfin = float((re.findall(r't_fin=(\d+)', txt) or [4800])[0])
        ts = [tuple(float(v) for v in m) for m in re.findall(r't=\s*(\d+) s · entrado\s+(\d+) m³ · en malla\s+(\d+) m³ · mojado\s+([\d.]+) ha ·\s+(\d+) s reales', txt)]
        agregado = sum(float(x) for x in re.findall(r'adding ([\d.eE+-]+) m\^3', txt))
        o = {'t': ts[-1][0] if ts else 0, 'tfin': tfin, 'entrado': ts[-1][1] if ts else 0, 'malla': ts[-1][2] if ts else 0, 'mojado_ha': ts[-1][3] if ts else 0,
             'reales': ts[-1][4] if ts else 0, 'min_sin_linea': round((time.time() - os.path.getmtime(lp)) / 60), 'm3_agregados_celdas_neg': round(agregado, 1),
             'error': bool(re.search(r'Traceback|nan m³|\bnan\b', txt)), 'listo': 'listo en' in txt, 'npz': os.path.exists(npz), 'proceso': procs.get(cid)}
        if len(ts) >= 2 and ts[-1][0] > ts[-2][0]:
            ritmo = (ts[-1][4] - ts[-2][4]) / (ts[-1][0] - ts[-2][0]); o['eta_min'] = round(ritmo * (tfin - ts[-1][0]) / 60 * 1.15)   # 15 % de margen: el área mojada crece
        corr[cid] = o
    mem = {k: int(v) for k, v in re.findall(r'(MemAvailable|SwapTotal|SwapFree):\s+(\d+)', open('/proc/meminfo').read())}
    temps = [int(x) / 1000 for x in sh("cat /sys/class/thermal/thermal_zone*/temp /sys/class/hwmon/hwmon*/temp1_input 2>/dev/null").split() if x.isdigit()]
    colas = [l.split(None, 1)[1] for l in sh("ps -eo pid=,args=").splitlines() if re.search(r'cola_\w+\.sh', l) and 'grep' not in l]
    return {'host': os.uname().nodename, 'hora': datetime.datetime.now().strftime('%H:%M'), 'nproc': os.cpu_count(), 'carga': round(os.getloadavg()[0], 1),
            'ram_disp_gb': round(mem['MemAvailable'] / 1e6, 1), 'swap_usado_gb': round((mem['SwapTotal'] - mem['SwapFree']) / 1e6, 1),
            'disco_libre_gb': round(shutil.disk_usage(BASE).free / 1e9), 'temp_max': max(temps) if temps else None, 'colas': colas, 'corridas': corr}
if '--local' in sys.argv: print(json.dumps(equipo())); sys.exit()
E = {'oficina': equipo()}
try: E['quadrab'] = json.loads(sh("ssh -o ConnectTimeout=15 quadrab 'cd ~/aluvion-calculo && python3 - --local' < " + __file__))
except Exception as ex: E['quadrab'] = {'error_ssh': str(ex)}
PREV = os.path.join(BASE, 'res', 'watchdog_estado.json')
prev = json.load(open(PREV)) if os.path.exists(PREV) else {}
REG = os.path.join(BASE, 'res', 'registro_experimentos.json')   # detenidas/canceladas a propósito: no alertar
IGN = {r['id'] for r in json.load(open(REG))['experimentos'] if r['estado'] in ('detenida', 'cancelada')} if os.path.exists(REG) else set()
alertas, filas = [], []
for eq, d in E.items():
    if 'error_ssh' in d: alertas.append(f'{eq}: sin conexión ssh'); continue
    if d['carga'] > 1.6 * d['nproc']: alertas.append(f"{eq}: carga {d['carga']} (sobre 1,6× núcleos)")
    if d['ram_disp_gb'] < 3: alertas.append(f"{eq}: RAM disponible {d['ram_disp_gb']} GB")
    if d['swap_usado_gb'] > 16: alertas.append(f"{eq}: swap {d['swap_usado_gb']} GB")
    if d['disco_libre_gb'] < 20: alertas.append(f"{eq}: disco libre {d['disco_libre_gb']} GB")
    if d['temp_max'] and d['temp_max'] > 92: alertas.append(f"{eq}: temperatura {d['temp_max']} °C")
    for cid, c in sorted(d['corridas'].items()):
        if cid in IGN and not c['proceso']: continue
        antes = prev.get(eq, {}).get('corridas', {}).get(cid, {})
        if c['npz'] and c['listo'] and not antes.get('npz'): alertas.append(f'{eq}/{cid}: TERMINÓ — evaluar')
        elif not c['proceso'] and not c['npz']: alertas.append(f'{eq}/{cid}: sin proceso y sin resultados (¿murió?)')
        if c['error']: alertas.append(f'{eq}/{cid}: error en el registro')
        if c['proceso'] and c['min_sin_linea'] > 120 and c['t'] > 0 and c['min_sin_linea'] * 60 > 3 * (c['reales'] / max(c['t'] / 600, 1)): alertas.append(f"{eq}/{cid}: {c['min_sin_linea']} min sin avanzar (¿atasco?)")
        ref = max(c['entrado'], c['malla'], 1)
        nuevo = c['m3_agregados_celdas_neg'] - antes.get('m3_agregados_celdas_neg', 0)   # solo lo agregado desde la revisión anterior
        if antes and nuevo > 0.005 * ref and c['t'] >= 600: alertas.append(f"{eq}/{cid}: celdas negativas agregaron {nuevo:.0f} m³ en 10 min ({100 * nuevo / ref:.1f} % de lo entrado; acumulado {c['m3_agregados_celdas_neg']:.0f} m³)")
        eta = (datetime.datetime.now() + datetime.timedelta(minutes=c['eta_min'] - c['min_sin_linea'])).strftime('%H:%M') if c.get('eta_min') is not None and not c['npz'] else ('listo' if c['npz'] else '¿?')
        filas.append(f"{cid}@{eq[:2]} {100 * c['t'] / c['tfin']:.0f}% eta {eta}")
out = {**E, 'alertas': alertas, 'revisado': datetime.datetime.now().isoformat(timespec='seconds')}
json.dump(out, open(PREV, 'w'), ensure_ascii=False, indent=1)
uso = ' | '.join(f"{eq}: carga {d.get('carga')}/{d.get('nproc')} RAM {d.get('ram_disp_gb')} GB disco {d.get('disco_libre_gb')} GB {d.get('temp_max')} °C" for eq, d in E.items() if 'carga' in d)
linea = f"{out['revisado']} · {uso} · " + ', '.join(filas) + ''.join(f' · ALERTA {a}' for a in alertas)
open(os.path.join(BASE, 'res', 'watchdog.log'), 'a').write(linea + '\n')
print(linea.replace(' · ', '\n'))