diet-guard/scripts/build_meal_schedule_fixture.py

#!/usr/bin/env python3
"""Regenerate the shared cross-language meal-schedule parity fixture.

The fixture (``tests/fixtures/meal_schedule_vectors.json``) is read by *both*
``diet_guard/tests/test_meal_schedule_vectors.py`` and
``app/test/models/meal_schedule_vectors_test.dart``: one shared input, one
shared expected result, so the PC and the phone derive the same slot minutes,
attribute a meal to the same slot and read the same wire fields.  A slot one
device offers while the other does not is a checkpoint that can never be
satisfied -- a permanent lock.  The inputs live in
``scripts/meal_schedule_cases.py``.

``expected`` values come from the Python implementation, the reference the Dart
port must reproduce -- so regenerating blesses whatever Python does now. Only
run it after an intended change, and re-read the diff before committing.

Run it from the repo root with the repo on the path, so ``diet_guard``
resolves to *this* tree rather than wherever the editable install
points:

    PYTHONPATH="$PWD" python3 scripts/build_meal_schedule_fixture.py
"""

from __future__ import annotations

from datetime import UTC, datetime
from itertools import pairwise
import json
from pathlib import Path

import meal_schedule_cases as cases

from diet_guard._meal_schedule import (
    MINUTES_PER_DAY,
    MealSchedule,
    schedule_from_wire,
    schedule_to_wire,
)
from diet_guard._slot_wire import entry_slot_minute, slot_fields
from diet_guard._slots import (
    current_slot,
    elapsed_slots,
    missing_slots,
    nearest_slot,
    satisfied_slots,
    slot_for_log,
    slot_label,
    within_enforcement_window,
)

FIXTURE = (
    Path(__file__).resolve().parent.parent / "tests/fixtures/meal_schedule_vectors.json"
)


def _triple(schedule: MealSchedule | None) -> list[int] | None:
    """Return a schedule as the ``[first, last, count]`` row it is stored as."""
    if schedule is None:
        return None
    return [schedule.first_minute, schedule.last_minute, schedule.count]


def _at(minute: int) -> datetime:
    """Return a fixed datetime at ``minute`` of day (the date is irrelevant)."""
    return datetime(2026, 1, 1, minute // 60, minute % 60, tzinfo=UTC)


def _schedule_row(triple: cases.Triple) -> dict[str, object]:
    """Return one schedule's normalised form, slots, cutoff and wire encoding."""
    schedule = MealSchedule(*triple)
    return {
        "input": list(triple),
        "normalized": _triple(schedule.normalized()),
        "slots": list(schedule.slots()),
        "enforcement_end_minute": schedule.enforcement_end_minute,
        "wire": schedule_to_wire(schedule),
    }


def _edge_minutes(schedule: MealSchedule) -> list[int]:
    """Return every minute around a slot or the cutoff, plus the day's ends."""
    around = [schedule.enforcement_end_minute, *schedule.slots()]
    minutes = {0, 1439} | {m + d for m in around for d in (-1, 0, 1)}
    return sorted(m for m in minutes if 0 <= m < MINUTES_PER_DAY)


def _edge_row(triple: cases.Triple, minute: int) -> dict[str, object]:
    """Return what every clock-reading function says at ``minute``."""
    schedule, now = MealSchedule(*triple), _at(minute)
    return {
        "schedule": list(triple),
        "minute": minute,
        "within_window": within_enforcement_window(now, schedule),
        "elapsed": list(elapsed_slots(now, schedule)),
        "current_slot": current_slot(now, schedule),
        "slot_for_log": slot_for_log(now, schedule),
    }


def _nearest_minutes(schedule: MealSchedule) -> list[int]:
    """Return each slot, each midpoint +/- 1, and out-of-day minutes."""
    slots = schedule.slots()
    mids = [(a + b) // 2 for a, b in pairwise(slots)]
    minutes = {-60, 0, 1439, 1800, *slots}
    minutes |= {m + d for m in mids for d in (-1, 0, 1)}
    return sorted(minutes)


def _slot_field_row(minute: int) -> dict[str, object]:
    """Return ``slot_fields(minute)``, or an error marker when it raises."""
    try:
        return {"minute": minute, "fields": slot_fields(minute)}
    except ValueError:
        return {"minute": minute, "error": True}


def _missing_row(triple: cases.Triple, minute: int, logged: list[int]) -> object:
    """Return the unsatisfied elapsed slots at ``minute`` given ``logged``."""
    missing = missing_slots(_at(minute), set(logged), MealSchedule(*triple))
    return {
        "schedule": list(triple),
        "minute": minute,
        "logged": logged,
        "missing": list(missing),
    }


def _satisfied_row(name: str, triple: cases.Triple, minutes: list[int]) -> object:
    """Return the slots ``minutes`` satisfy under ``triple``, sorted."""
    satisfied = satisfied_slots(minutes, MealSchedule(*triple))
    return {
        "name": name,
        "schedule": list(triple),
        "entry_minutes": minutes,
        "satisfied": sorted(satisfied),
    }


def build() -> dict[str, object]:
    """Return the whole fixture: every input with its expected result."""
    edge = [(t, MealSchedule(*t)) for t in cases.EDGE_SCHEDULES]
    return {
        "_comment": (
            "Shared parity fixture for meal-slot minutes. Read by BOTH "
            "diet_guard/tests/test_meal_schedule_vectors.py and "
            "app/test/models/meal_schedule_vectors_test.dart. Regenerate with "
            "scripts/build_meal_schedule_fixture.py."
        ),
        "schedules": [_schedule_row(t) for t in cases.SCHEDULES],
        "time_edges": [_edge_row(t, m) for t, s in edge for m in _edge_minutes(s)],
        "missing": [_missing_row(*case) for case in cases.MISSING],
        "nearest": [
            {"schedule": list(t), "minute": m, "nearest": nearest_slot(m, s)}
            for t, s in edge
            for m in _nearest_minutes(s)
        ],
        "satisfied": [_satisfied_row(*case) for case in cases.SATISFIED],
        "labels": [{"minute": m, "label": slot_label(m)} for m in cases.LABELS],
        "schedule_wire_decode": [
            {"raw": raw, "schedule": _triple(schedule_from_wire(raw))}
            for raw in cases.WIRE_DECODE
        ],
        "slot_fields": [_slot_field_row(m) for m in cases.SLOT_FIELD_MINUTES],
        "entry_slot_minute": [
            {"entry": entry, "minute": entry_slot_minute(entry)}
            for entry in cases.ENTRIES
        ],
    }


def main() -> None:
    """Write the fixture to disk.

    Deliberately silent: ``T201`` bans ``print`` here, and the script's output
    is the file itself -- ``git diff`` is how you check what changed.
    """
    FIXTURE.parent.mkdir(parents=True, exist_ok=True)
    with FIXTURE.open("w", encoding="utf-8") as handle:
        json.dump(build(), handle, ensure_ascii=False, indent=2, sort_keys=True)
        handle.write("\n")


if __name__ == "__main__":
    main()