autoficher/app/planner.py

"""Construye el plan de fichajes de un dia a partir del calendario de Timenet."""
from __future__ import annotations

import logging
import random
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo

from app import jitter, profiles
from app.schedule import Calendar

logger = logging.getLogger(__name__)

ENTRADA, SALIDA = 0, 1


@dataclass
class PlannedCheck:
    idx: int
    typ: int               # 0 entrada, 1 salida
    window_idx: int
    target: datetime       # hora teorica
    planned_at: datetime   # instante real en que se fichara

    @property
    def label(self) -> str:
        return "ENTRADA" if self.typ == ENTRADA else "SALIDA"

    @property
    def offset_seconds(self) -> int:
        """Segundos de margen respecto de la teorica. Positivo = lado seguro."""
        d = (self.planned_at - self.target).total_seconds()
        return int(-d if self.typ == ENTRADA else d)

    def __str__(self) -> str:
        return "%-7s teorica %s  ficha %s  (%+ds -> %s)" % (
            self.label, self.target.strftime("%H:%M:%S"),
            self.planned_at.strftime("%H:%M:%S"),
            int((self.planned_at - self.target).total_seconds()),
            "antes" if self.typ == ENTRADA else "despues")


class Planner:
    def __init__(self, cfg: dict, store=None):
        self.cfg = cfg
        self.store = store
        self.tz = ZoneInfo(cfg.get("timezone", "Europe/Madrid"))

    @property
    def perfil(self) -> str:
        """Perfil activo. Se lee del store en cada plan, para poder cambiarlo en caliente."""
        if self.store:
            nombre = self.store.get("perfil", self.cfg.get("perfil", profiles.DEFECTO))
            if profiles.existe(nombre):
                return nombre
        return self.cfg.get("perfil", profiles.DEFECTO)

    @property
    def j(self) -> dict:
        return profiles.params(self.perfil)

    # ------------------------------------------------------------------ drift
    def _drift_key(self, window_idx: int, typ: int) -> str:
        return f"drift.{window_idx}.{typ}"

    def _drift(self, window_idx: int, typ: int, rng: random.Random) -> float:
        key = self._drift_key(window_idx, typ)
        prev = float(self.store.get(key, 0.0)) if self.store else 0.0
        j = self.j
        nxt = jitter.next_drift(prev, j["drift_sigma"], j["drift_clamp"], rng)
        if self.store:
            self.store.put(key, nxt)
        return nxt

    # ------------------------------------------------------------------- plan
    def plan_day(self, cal: Calendar, day: date,
                 rng: random.Random | None = None) -> list[PlannedCheck]:
        """Plan del dia. Lista vacia si no toca fichar."""
        if not cal.is_workday(day):
            logger.info("%s no es laborable: %s", day, cal.why_not_workday(day))
            return []

        rng = rng or random.Random()
        j = self.j
        out: list[PlannedCheck] = []
        idx = 0

        for wi, w in enumerate(cal.windows(day)):
            start = datetime.combine(day, datetime.min.time(), self.tz) + timedelta(minutes=w.start)
            end = datetime.combine(day, datetime.min.time(), self.tz) + timedelta(minutes=w.end)

            # ENTRADA: siempre antes de la hora teorica. El tope se acota ademas
            # con el margen de entrada que declara el propio tipo de jornada.
            cap = profiles.valor(j, "entry", "cap", wi)
            tope = min(cap, max(w.entry_margin * 60 // 2, 60)) if w.entry_margin else cap
            off = jitter.offset_seconds(
                profiles.valor(j, "entry", "floor", wi),
                profiles.valor(j, "entry", "median", wi),
                profiles.valor(j, "entry", "sigma", wi),
                tope, self._drift(wi, ENTRADA, rng), rng)
            off = max(off, j["entry_guard"])
            out.append(PlannedCheck(idx, ENTRADA, wi, start, start - timedelta(seconds=off)))
            idx += 1

            # SALIDA: siempre despues
            off = jitter.offset_seconds(
                profiles.valor(j, "exit", "floor", wi),
                profiles.valor(j, "exit", "median", wi),
                profiles.valor(j, "exit", "sigma", wi),
                profiles.valor(j, "exit", "cap", wi),
                self._drift(wi, SALIDA, rng), rng)
            out.append(PlannedCheck(idx, SALIDA, wi, end, end + timedelta(seconds=off)))
            idx += 1

        self._validate(out, cal, day)
        return out

    # --------------------------------------------------------------- garantias
    def _validate(self, plan: list[PlannedCheck], cal: Calendar, day: date) -> None:
        """Invariantes que, si se rompen, cuestan minutos de verdad."""
        for p in plan:
            if p.typ == ENTRADA and p.planned_at >= p.target:
                raise AssertionError(f"entrada planificada tarde: {p}")
            if p.typ == SALIDA and p.planned_at <= p.target:
                raise AssertionError(f"salida planificada pronto: {p}")
        for a, b in zip(plan, plan[1:]):
            if b.planned_at <= a.planned_at:
                raise AssertionError(f"fichajes desordenados: {a} / {b}")

    def rows_for_store(self, plan: list[PlannedCheck]) -> list[dict]:
        return [{"idx": p.idx, "typ": p.typ, "window_idx": p.window_idx,
                 "target": p.target.isoformat(), "planned_at": p.planned_at.isoformat()}
                for p in plan]