Expand calendar events: RRULE, EXDATE, RECURRENCE-ID, all-day, VALARM

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 178 insertions, 0 deletions

diff --git a/cal-notif b/cal-notif
index 3b4556d..413345e 100755
--- a/cal-notif
+++ b/cal-notif
@@ -137,6 +137,117 @@ def save_overrides(ov, path=None):
 
 # ---- calendars ----
 
+Occ = namedtuple("Occ", "uid cal summary location start allday valarms")
+
+
+def local(dt):
+    # Aware local-or-own-zone datetime: dates start at local midnight,
+    # floating (naive) times are local.
+    if not isinstance(dt, datetime):
+        dt = datetime.combine(dt, dtime())
+    return dt if dt.tzinfo else dt.astimezone()
+
+
+def rule_for(comp, start):
+    # dateutil wants UNTIL to match DTSTART: naive for floating/all-day,
+    # UTC for zoned. Clients do not always agree, so normalise it.
+    rec = vRecur(comp["RRULE"])
+    if "UNTIL" in rec:
+        u = rec["UNTIL"][0]
+        if not isinstance(u, datetime):
+            u = datetime.combine(u, dtime(23, 59, 59))
+        if start.tzinfo:
+            u = (u if u.tzinfo else u.replace(tzinfo=start.tzinfo)).astimezone(timezone.utc)
+        elif u.tzinfo:
+            u = u.astimezone().replace(tzinfo=None)
+        rec["UNTIL"] = [u]
+    return rrulestr(rec.to_ical().decode(), dtstart=start)
+
+
+def valarms(comp, start, end):
+    out = []
+    for a in comp.walk("VALARM"):
+        t = a.get("TRIGGER")
+        if t is None:
+            continue
+        if isinstance(t.dt, timedelta):
+            out.append((end if t.params.get("RELATED") == "END" else start) + t.dt)
+        else:
+            out.append(local(t.dt))
+    return tuple(sorted(out))
+
+
+def lead(comp):
+    # The longest relative VALARM lead, so expansion reaches far enough.
+    trig = [a["TRIGGER"].dt for a in comp.walk("VALARM") if "TRIGGER" in a]
+    return max([-t for t in trig if isinstance(t, timedelta)] + [timedelta()])
+
+
+def occurrence(comp, cal, start):
+    s0 = comp["DTSTART"].dt
+    allday = not isinstance(s0, datetime)
+    if "DTEND" in comp:
+        length = local(comp["DTEND"].dt) - local(s0)
+    elif "DURATION" in comp:
+        length = comp["DURATION"].dt
+    else:
+        length = timedelta(days=1) if allday else timedelta()
+    start = local(start)
+    return Occ(str(comp.get("UID", "")), cal, str(comp.get("SUMMARY", "")),
+               str(comp.get("LOCATION", "")), start, allday,
+               valarms(comp, start, start + length))
+
+
+def expand(ics, cal, lo, hi):
+    """Occurrences of every VEVENT in `ics` starting in [lo, hi] (aware).
+
+    The window end grows per event by its largest VALARM lead, so an alarm
+    a week before an event beyond `hi` is still produced."""
+    comps = Calendar.from_ical(ics).walk("VEVENT")
+    moved = {(str(c.get("UID")), local(c["RECURRENCE-ID"].dt)) for c in comps if "RECURRENCE-ID" in c}
+    out = []
+    for c in comps:
+        if "DTSTART" not in c:
+            continue
+        s0 = c["DTSTART"].dt
+        first = occurrence(c, cal, s0)
+        top = hi + lead(c)
+        if "RRULE" not in c or "RECURRENCE-ID" in c:
+            # ponytail: a RECURRENCE-ID with RANGE=THISANDFUTURE is treated as
+            # a single moved occurrence.
+            if lo <= first.start <= top:
+                out.append(first)
+            continue
+        # Expand in the event's own clock: naive for all-day/floating.
+        start = s0 if isinstance(s0, datetime) else datetime.combine(s0, dtime())
+        conv = (lambda d: d) if start.tzinfo else (lambda d: d.astimezone().replace(tzinfo=None))
+        exd = c.get("EXDATE", [])
+        ex = {local(d.dt) for g in (exd if isinstance(exd, list) else [exd]) for d in g.dts}
+        uid = str(c.get("UID"))
+        for s in rule_for(c, start).between(conv(lo), conv(top), inc=True):
+            s = local(s if isinstance(s0, datetime) else s.date())
+            if s not in ex and (uid, s) not in moved:
+                out.append(occurrence(c, cal, s))
+    return out
+
+
+def calendars(root):
+    root = Path(root).expanduser()
+    for d in sorted(p for p in root.iterdir() if p.is_dir()) if root.is_dir() else []:
+        name = d / "displayname"
+        yield d, (name.read_text().strip() if name.exists() else d.name)
+
+
+def load_events(root, lo, hi):
+    out = []
+    for d, cal in calendars(root):
+        for f in sorted(d.glob("*.ics")):
+            try:
+                out += expand(f.read_bytes(), cal, lo, hi)
+            except Exception as e:           # one broken file must not stop the rest
+                print(f"cal-notif: skipping {f}: {e}", file=sys.stderr)
+    return sorted(out, key=lambda o: o.start)
+
 # ---- alarms ----
 
 # ---- voice ----
diff --git a/test_cal_notif.py b/test_cal_notif.py
index 27f60ba..2237dae 100644
--- a/test_cal_notif.py
+++ b/test_cal_notif.py
@@ -79,6 +79,73 @@ def test_config():
             pass
 
 
+def test_expand():
+    lo, hi = at(2026, 1, 1), at(2026, 2, 1)
+    # Weekly zoned event, two EXDATEs, one moved occurrence.
+    weekly = """UID:w@example.org
+DTSTART;TZID=Europe/Rome:20260105T090000
+DTEND;TZID=Europe/Rome:20260105T100000
+RRULE:FREQ=WEEKLY;UNTIL=20260131T000000Z
+EXDATE;TZID=Europe/Rome:20260112T090000,20260119T090000
+SUMMARY:standup"""
+    moved = """UID:w@example.org
+RECURRENCE-ID;TZID=Europe/Rome:20260126T090000
+DTSTART;TZID=Europe/Rome:20260126T150000
+DTEND;TZID=Europe/Rome:20260126T160000
+SUMMARY:standup moved"""
+    occ = cn.expand(ics(weekly, moved), "work", lo, hi)
+    got = sorted((o.start, o.summary) for o in occ)
+    assert got == [(at(2026, 1, 5, 9), "standup"), (at(2026, 1, 26, 15), "standup moved")], got
+
+    # All-day yearly with a UTC UNTIL (dateutil rejects the naive/aware mix
+    # unless normalised) and a relative VALARM 1 day before.
+    bday = """UID:b@example.org
+DTSTART;VALUE=DATE:19800112
+RRULE:FREQ=YEARLY;UNTIL=20300101T000000Z
+SUMMARY:birthday
+BEGIN:VALARM
+ACTION:DISPLAY
+TRIGGER:-P1D
+END:VALARM"""
+    (o,) = cn.expand(ics(bday), "bd", lo, hi)
+    assert o.allday and o.start == datetime(2026, 1, 12).astimezone()
+    assert o.valarms == (datetime(2026, 1, 11).astimezone(),)
+
+    # A week-ahead VALARM on an event past `hi` widens that event's window.
+    far = """UID:f@example.org
+DTSTART;TZID=Europe/Rome:20260205T090000
+SUMMARY:far
+BEGIN:VALARM
+ACTION:DISPLAY
+TRIGGER:-P7D
+END:VALARM"""
+    (o,) = cn.expand(ics(far), "c", lo, hi)
+    assert o.valarms == (at(2026, 1, 29, 9),)
+    assert cn.expand(ics(far.replace("-P7D", "-PT1H")), "c", lo, hi) == []
+
+    # RELATED=END trigger counts from DTEND.
+    end = """UID:e@example.org
+DTSTART;TZID=Europe/Rome:20260110T090000
+DTEND;TZID=Europe/Rome:20260110T110000
+BEGIN:VALARM
+ACTION:DISPLAY
+TRIGGER;RELATED=END:-PT15M
+END:VALARM"""
+    (o,) = cn.expand(ics(end), "c", lo, hi)
+    assert o.valarms == (at(2026, 1, 10, 10, 45),)
+
+
+def test_load_events():
+    with tempfile.TemporaryDirectory() as d:
+        cal = Path(d) / "123"
+        cal.mkdir()
+        (cal / "displayname").write_text("work\n")
+        (cal / "a.ics").write_bytes(ics("UID:a@example.org\nDTSTART;TZID=Europe/Rome:20260110T090000"))
+        (cal / "broken.ics").write_text("not a calendar")
+        (o,) = cn.load_events(d, at(2026, 1, 1), at(2026, 2, 1))   # broken one skipped
+        assert o.cal == "work" and o.uid == "a@example.org"
+
+
 if __name__ == "__main__":
     for name, fn in list(globals().items()):
         if name.startswith("test_"):