File size: 3,982 Bytes
a2d6a0d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | #!/usr/bin/env python3
"""Monitor en vivo del entrenamiento PAMPAr.
Uso:
python scripts/monitor.py
python scripts/monitor.py --cada 10 # actualizar cada 10s
"""
import argparse
import json
import os
import time
from pathlib import Path
# Colores ANSI
B = "\033[1m"
AZ = "\033[94m"
VE = "\033[92m"
AM = "\033[93m"
RO = "\033[91m"
GR = "\033[90m"
RE = "\033[0m"
def limpiar():
os.system("cls" if os.name == "nt" else "clear")
def leer_estado(ruta: Path) -> dict:
try:
return json.loads(ruta.read_text(encoding="utf-8"))
except Exception:
return {}
def checkpoint_info(ruta: Path) -> tuple[str, str]:
if not ruta.exists():
return "no encontrado", "?"
stat = ruta.stat()
ts = time.strftime("%H:%M:%S", time.localtime(stat.st_mtime))
mb = stat.st_size / 1e6
return f"guardado a las {ts} ({mb:.1f} MB)", ts
def render(estado_path: Path, ckpt_path: Path) -> None:
d = leer_estado(estado_path)
temas = d.get("temas", {})
nivel = d.get("nivel_actual", "?")
sesiones_total = d.get("sesiones_totales", 0)
n_dominados = d.get("temas_dominados", 0) # es un int
paso = d.get("paso_global", "?")
ck_str, _ = checkpoint_info(ckpt_path)
ahora = time.strftime("%H:%M:%S")
print(f"{B}{AZ}═══ PAMPAr Monitor — {ahora} ══════════════════════════════{RE}")
paso_fmt = f"{paso:,}" if isinstance(paso, int) else str(paso)
print(f" Paso global : {B}{paso_fmt}{RE} Nivel: {nivel}/6 Sesiones: {sesiones_total}")
print(f" Checkpoint : {GR}{ck_str}{RE}")
print()
if not temas:
print(f" {AM}Sin temas cargados aún{RE}")
return
# Temas con mayor loss (más difíciles = más trabajo pendiente)
dominados = [(k, v) for k, v in temas.items() if v.get("dominado")]
activos = [(k, v) for k, v in temas.items() if not v.get("dominado")]
activos.sort(key=lambda x: (x[1].get("historial_loss") or [0])[-1], reverse=True)
print(f" {B}Temas con mayor loss (más a trabajar):{RE}")
for tema, v in activos[:12]:
hist = v.get("historial_loss") or [0]
loss = hist[-1]
n = v.get("nivel_dificultad", 0)
ses = v.get("n_sesiones", 0)
cur = v.get("curiosidad", 0)
bar = "█" * min(int(loss), 8) + "░" * max(0, 8 - int(loss))
color = RO if loss > 3 else AM if loss > 1 else VE
print(f" {color}{bar}{RE} {tema:<30} loss={loss:.3f} cur={cur:.3f} sesiones={ses}")
print()
n_dom = n_dominados if n_dominados else len(dominados)
n_tot = len(temas)
pct = n_dom / n_tot * 100 if n_tot else 0
bloques = 30
llenos = int(pct / 100 * bloques)
barra_dom = VE + "█" * llenos + RE + GR + "░" * (bloques - llenos) + RE
print(f" Dominados: {VE}{n_dom}/{n_tot}{RE} [{barra_dom}] {pct:.0f}%")
if dominados:
print(f" {GR}Temas dominados: {', '.join(k for k, _ in dominados[:8])}{RE}")
print(f"\n{GR} (Ctrl+C para salir){RE}")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--cada", type=int, default=5, help="Segundos entre actualizaciones")
parser.add_argument(
"--estado", type=Path,
default=Path("checkpoints/curiosidad_estado.json"),
)
parser.add_argument(
"--checkpoint", type=Path,
default=Path("checkpoints/pampar_v2_best.pt"),
)
args = parser.parse_args()
try:
first = True
while True:
if not first:
limpiar()
first = False
render(args.estado, args.checkpoint)
if args.cada <= 0:
break
time.sleep(args.cada)
except KeyboardInterrupt:
print("\nMonitor detenido.")
if __name__ == "__main__":
main()
|