diff options
| -rwxr-xr-x | cal-notif | 111 | ||||
| -rw-r--r-- | test_cal_notif.py | 67 |
2 files changed, 178 insertions, 0 deletions
@@ -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_"): |