musicmouse/python-backend/musicmouse/school_days.py

"""Whether today is a school day: weekends, public holidays and school holidays.

The holidays come from the OpenHolidays API (https://openholidaysapi.org), which has
both kinds per German state - Bavaria's ``DE-BY`` school holidays include the
Buß- und Bettag, which is a school day off but not a public holiday. Public holidays
carry their subdivisions, so a city-only one like Augsburg's Friedensfest
(``DE-BY-AU``) counts for Augsburg and not for the rest of Bavaria.

Fetched once a day and kept in a JSON file, so a device that loses its internet keeps
knowing the holidays for the rest of the school year.
"""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass
from datetime import date, timedelta
from pathlib import Path
from typing import Any

import httpx2

_log = logging.getLogger(__name__)

__all__ = [
    "API_URL",
    "Holiday",
    "HolidayCalendar",
    "fetch_calendar",
    "load_calendar",
    "parse_holidays",
    "save_calendar",
    "school_day",
]

API_URL = "https://openholidaysapi.org"

#: How far back and ahead each fetch reaches. Ahead covers a whole school year from
#: any date, so the cache outlives a long outage.
_FETCH_BEFORE = timedelta(days=30)
_FETCH_AHEAD = timedelta(days=400)

_WEEKEND = "Wochenende"


@dataclass(frozen=True, slots=True)
class Holiday:
    start: date
    end: date
    name: str
    nationwide: bool
    subdivisions: tuple[str, ...]

    def covers(self, day: date) -> bool:
        return self.start <= day <= self.end

    def applies_to(self, subdivision: str) -> bool:
        """``DE-BY-AU`` gets ``DE-BY``'s holidays as well as its own; ``DE-BY`` does
        not get ``DE-BY-AU``'s."""
        return self.nationwide or any(
            subdivision == code or subdivision.startswith(code + "-") for code in self.subdivisions
        )


@dataclass(frozen=True, slots=True)
class HolidayCalendar:
    #: The day it was fetched, to know when to fetch again.
    fetched: date
    subdivision: str
    school: tuple[Holiday, ...]
    public: tuple[Holiday, ...]


def parse_holidays(raw: list[dict[str, Any]]) -> tuple[Holiday, ...]:
    """The API's JSON list, either endpoint. Entries that do not parse are skipped."""
    holidays = []
    for entry in raw:
        try:
            names = entry.get("name") or []
            name = next(
                (n["text"] for n in names if n.get("language") == "DE"),
                names[0]["text"] if names else "Feiertag",
            )
            holidays.append(
                Holiday(
                    start=date.fromisoformat(entry["startDate"]),
                    end=date.fromisoformat(entry["endDate"]),
                    name=name,
                    nationwide=bool(entry.get("nationwide")),
                    subdivisions=tuple(s["code"] for s in entry.get("subdivisions") or []),
                )
            )
        except (KeyError, TypeError, ValueError, IndexError):
            _log.debug("Skipping a holiday entry that does not parse: %r", entry)
    return tuple(holidays)


def school_day(day: date, calendar: HolidayCalendar | None) -> tuple[bool | None, str | None]:
    """``(is_school_day, why_not)``. ``None`` when there is no calendar to ask -
    the weekend is still known then, a holiday is not."""
    if day.weekday() >= 5:
        return False, _WEEKEND
    if calendar is None:
        return None, None
    for holiday in (*calendar.public, *calendar.school):
        if holiday.covers(day) and holiday.applies_to(calendar.subdivision):
            return False, holiday.name
    return True, None


async def fetch_calendar(
    client: httpx2.AsyncClient, subdivision: str, today: date
) -> HolidayCalendar:
    """Both lists for ``subdivision``'s state. Raises on any network or HTTP error."""
    # The API files school holidays under the state; city-level ones are filtered here.
    state = "-".join(subdivision.split("-")[:2])
    params = {
        "countryIsoCode": state.split("-")[0],
        "subdivisionCode": state,
        "languageIsoCode": "DE",
        "validFrom": (today - _FETCH_BEFORE).isoformat(),
        "validTo": (today + _FETCH_AHEAD).isoformat(),
    }
    lists = []
    for endpoint in ("SchoolHolidays", "PublicHolidays"):
        response = await client.get(f"{API_URL}/{endpoint}", params=params)
        response.raise_for_status()
        lists.append(parse_holidays(response.json()))
    school, public = lists
    return HolidayCalendar(fetched=today, subdivision=subdivision, school=school, public=public)


def save_calendar(path: Path, calendar: HolidayCalendar) -> None:
    def dump(holidays: tuple[Holiday, ...]) -> list[dict[str, Any]]:
        return [
            {
                "startDate": h.start.isoformat(),
                "endDate": h.end.isoformat(),
                "name": [{"language": "DE", "text": h.name}],
                "nationwide": h.nationwide,
                "subdivisions": [{"code": code} for code in h.subdivisions],
            }
            for h in holidays
        ]

    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "fetched": calendar.fetched.isoformat(),
        "subdivision": calendar.subdivision,
        "school": dump(calendar.school),
        "public": dump(calendar.public),
    }
    temporary = path.with_suffix(".tmp")
    temporary.write_text(json.dumps(payload), encoding="utf-8")
    temporary.replace(path)


def load_calendar(path: Path, subdivision: str) -> HolidayCalendar | None:
    """The cached calendar, or ``None`` if there is none for this subdivision."""
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        if payload["subdivision"] != subdivision:
            return None
        return HolidayCalendar(
            fetched=date.fromisoformat(payload["fetched"]),
            subdivision=subdivision,
            school=parse_holidays(payload["school"]),
            public=parse_holidays(payload["public"]),
        )
    except FileNotFoundError:
        return None
    except (OSError, KeyError, TypeError, ValueError):
        _log.warning("Ignoring an unreadable holiday cache at %s", path)
        return None