"""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