Add calendar frame mode + server-side manage overlay (server)
Build and push server image / build-and-push (push) Successful in 42s

calendar_feed.py/calendar_render.py: fetch/merge per-user ICS feeds,
render agenda/week/month views. manage_overlay.py: composites the
manage-button overlay server-side (QR, battery, location/date,
share-QR, face labels), reused by every render mode. device.py/common.py
wire both together: mode dispatch for /frame/image+advance+back, and
the &manage=1 flag. Plus UI (frame_config.html Calendar card, settings
calendar URL field) and the icalendar/recurring-ical-events deps.
This commit is contained in:
2026-07-22 19:06:49 -04:00
parent 15e37c77cd
commit c007acde75
17 changed files with 1317 additions and 265 deletions
+110
View File
@@ -0,0 +1,110 @@
"""Fetch, parse, and merge per-user ICS calendar feeds for calendar frame
mode (see routers/device.py's RENDERERS["calendar"] and calendar_render.py).
Pure functions -- no ORM, no FastAPI Depends. Callers (routers/common.py's
get_or_refresh_calendar_events) supply plain (owner_display_name, url)
pairs, not ORM objects, so this module stays testable against fixture .ics
text with no database or app involved.
Recurring events (RRULE/EXDATE/RDATE, DST-aware) are expanded via
recurring-ical-events rather than hand-rolled -- that's genuinely fiddly
to get right (see its own docs), not worth reinventing. It's LGPL-3.0 (an
ordinary runtime pip dependency, never vendored/modified -- see the
server README's Notes section for why that doesn't put this project's own
code under LGPL terms).
"""
from __future__ import annotations
from datetime import date, datetime
import httpx
import icalendar
import recurring_ical_events
HTTP_TIMEOUT_S = 15.0
FETCH_MAX_BYTES = 10 * 1024 * 1024 # sanity cap -- a real feed is KB, not MB
CHECK_INTERVAL_S = 20 * 60 # don't refetch/reparse any feed more often than this
# How far back/forward each merge-fetch expands recurring events. Households
# look back far less than they plan ahead, hence the asymmetry. Browsing
# outside this window (calendar_browse_offset) just yields an empty view,
# not an error -- self-heals on the next normal wake regardless.
EXPAND_WINDOW_PAST_DAYS = 30
EXPAND_WINDOW_FUTURE_DAYS = 200
class CalendarFetchError(Exception):
"""One feed was unreachable, not valid ICS, or too large. Raised by
fetch_source_events(); merge_events() is what catches this per-source
so one broken feed can't blank out another's events."""
def fetch_source_events(url: str, window_start: date, window_end: date) -> list[dict]:
"""One feed: download, parse, expand recurrences within
[window_start, window_end]. Raises CalendarFetchError on any problem
-- network, malformed ICS, or an oversized response."""
try:
with httpx.stream("GET", url, timeout=HTTP_TIMEOUT_S, follow_redirects=True) as resp:
resp.raise_for_status()
chunks = []
total = 0
for chunk in resp.iter_bytes():
total += len(chunk)
if total > FETCH_MAX_BYTES:
raise CalendarFetchError(f"Feed exceeds {FETCH_MAX_BYTES} bytes")
chunks.append(chunk)
body = b"".join(chunks)
except httpx.HTTPError as e:
raise CalendarFetchError(str(e)) from e
try:
cal = icalendar.Calendar.from_ical(body)
occurrences = recurring_ical_events.of(cal).between(window_start, window_end)
except Exception as e: # icalendar/recurring_ical_events raise a mix of ValueError-family exceptions
raise CalendarFetchError(f"Could not parse ICS feed: {e}") from e
events = []
for occ in occurrences:
dtstart = occ.get("DTSTART")
dtend = occ.get("DTEND")
if dtstart is None:
continue
start_dt = dtstart.dt
end_dt = dtend.dt if dtend is not None else start_dt
all_day = not isinstance(start_dt, datetime) # date, not datetime -- VALUE=DATE
events.append({
"summary": str(occ.get("SUMMARY") or "(untitled)"),
"start": start_dt.isoformat(),
"end": end_dt.isoformat(),
"all_day": all_day,
})
return events
def merge_events(
sources: list[tuple[str, str]], window_start: date, window_end: date
) -> tuple[list[dict], str]:
"""sources: [(owner_display_name, ics_url), ...]. Fetches each
independently -- one broken feed never blanks another's events.
Returns (merged_time_sorted_events, fetch_summary); fetch_summary is
"" when every source succeeded, else "N of M calendars unavailable"
(never *which* source -- naming whose feed is down to everyone who
looks at a shared household display is a bigger overshare than the
outage itself)."""
merged: list[dict] = []
failures = 0
for owner_display_name, url in sources:
try:
events = fetch_source_events(url, window_start, window_end)
except CalendarFetchError:
failures += 1
continue
for event in events:
event["owner_display_name"] = owner_display_name
merged.append(event)
merged.sort(key=lambda e: e["start"])
summary = f"{failures} of {len(sources)} calendars unavailable" if failures else ""
return merged, summary