61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
|
"""scheduler.py — franja horaria activa para el while loop de main.py.
|
||
|
|
|
||
|
|
Lee scheduler.json (HORARIOS + DIAS_ACTIVOS) y responde dos preguntas:
|
||
|
|
* franja_actual(): si ahora cae dentro de una franja, con qué modo.
|
||
|
|
* proximo_inicio(): si no, cuándo empieza la próxima franja activa.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
SCHEDULER_FILE = Path(__file__).parent / "scheduler.json"
|
||
|
|
|
||
|
|
INTERVALOS_POR_MODO = {
|
||
|
|
"rapido": 10 * 60,
|
||
|
|
"full": 60 * 60,
|
||
|
|
}
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
def cargar_scheduler(path: Path | None = None) -> dict:
|
||
|
|
p = path or SCHEDULER_FILE
|
||
|
|
with open(p, encoding="utf-8") as f:
|
||
|
|
return json.load(f)
|
||
|
|
|
||
|
|
|
||
|
|
def _minutos(hora: int, minuto: int) -> int:
|
||
|
|
return hora * 60 + minuto
|
||
|
|
|
||
|
|
|
||
|
|
def franja_actual(cfg: dict, ahora: datetime) -> dict | None:
|
||
|
|
if ahora.weekday() not in cfg.get("DIAS_ACTIVOS", []):
|
||
|
|
return None
|
||
|
|
|
||
|
|
minutos_ahora = _minutos(ahora.hour, ahora.minute)
|
||
|
|
for h_ini, m_ini, h_fin, m_fin, modo in cfg.get("HORARIOS", []):
|
||
|
|
if _minutos(h_ini, m_ini) <= minutos_ahora < _minutos(h_fin, m_fin):
|
||
|
|
return {"modo": modo, "intervalo": INTERVALOS_POR_MODO.get(modo, 600)}
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def proximo_inicio(cfg: dict, ahora: datetime) -> datetime:
|
||
|
|
dias_activos = cfg.get("DIAS_ACTIVOS", [])
|
||
|
|
horarios = cfg.get("HORARIOS", [])
|
||
|
|
if not dias_activos or not horarios:
|
||
|
|
raise ValueError("scheduler.json sin DIAS_ACTIVOS o HORARIOS configurados")
|
||
|
|
|
||
|
|
for delta in range(0, 8):
|
||
|
|
candidato = ahora + timedelta(days=delta)
|
||
|
|
if candidato.weekday() not in dias_activos:
|
||
|
|
continue
|
||
|
|
inicios = []
|
||
|
|
for h_ini, m_ini, h_fin, m_fin, _modo in horarios:
|
||
|
|
inicio = candidato.replace(hour=h_ini, minute=m_ini, second=0, microsecond=0)
|
||
|
|
if delta == 0 and inicio <= ahora:
|
||
|
|
continue
|
||
|
|
inicios.append(inicio)
|
||
|
|
if inicios:
|
||
|
|
return min(inicios)
|
||
|
|
|
||
|
|
raise ValueError("No se encontró una próxima franja activa en los próximos 7 días")
|
||
|
|
# ============================================================================
|