From 4d8524d7674f362ccdb4cee8500822c3fde28d48 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Sun, 27 Sep 2026 19:55:11 +0200 Subject: Pick alarms by precedence and compute which ones are due Co-Authored-By: Claude Opus 5.5 --- cal-notif | 44 ++++++++++++++++++++++++++++++++++++++++++++ test_cal_notif.py | 27 +++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) diff --git a/cal-notif b/cal-notif index 413345e..3726200 100755 --- a/cal-notif +++ b/cal-notif @@ -250,6 +250,50 @@ def load_events(root, lo, hi): # ---- alarms ---- +def cal_entry(cfg, cal): + e = cfg["defaults"].get(cal, cfg["defaults"].get("*", [])) + return (e.get("offsets", []), e.get("voice", True)) if isinstance(e, dict) else (e, True) + + +def alarms_for(occ, cfg, overrides): + if occ.uid in overrides: + offs = overrides[occ.uid] + elif occ.valarms: + return list(occ.valarms) + else: + offs = cal_entry(cfg, occ.cal)[0] + return [occ.start - parse_duration(o) for o in offs] + + +def current_offsets(occ, cfg, overrides): + return [fmt_duration(occ.start - f) for f in alarms_for(occ, cfg, overrides)] + + +def horizon(cfg, overrides): + offs = [o for v in overrides.values() for o in v] + offs += [o for c in cfg["defaults"] for o in cal_entry(cfg, c)[0]] + return max([parse_duration(o) for o in offs] + [timedelta()]) + timedelta(days=1) + + +def schedule(occs, cfg, overrides): + return sorted(((f, o) for o in occs for f in alarms_for(o, cfg, overrides)), + key=lambda e: e[0]) + + +def due(entries, last, now, late): + """Entries firing in (last, now], at most `late` old, whose event has not + started yet unless the alarm itself is at or after the start.""" + since = max(last, now - late) + return [(f, o) for f, o in entries + if since < f <= now and not (f < o.start <= now)] + + +def announce(occ, now, vcfg): + left = occ.start - now + if left < timedelta(minutes=1): + return vcfg["say_now"].format(summary=occ.summary) + return vcfg["say"].format(summary=occ.summary, **{"in": spoken_it(left)}) + # ---- voice ---- # ---- daemon ---- diff --git a/test_cal_notif.py b/test_cal_notif.py index 2237dae..ea5aba5 100644 --- a/test_cal_notif.py +++ b/test_cal_notif.py @@ -146,6 +146,33 @@ def test_load_events(): assert o.cal == "work" and o.uid == "a@example.org" +def test_precedence(): + cfg = cn.load_config(Path("/nonexistent")) + cfg["defaults"] = {"bd": ["1d", "9h"], "*": ["15m"], "quiet": {"offsets": ["1h"], "voice": False}} + t = at(2026, 1, 10, 12) + mk = lambda uid, cal, va=(): cn.Occ(uid, cal, "x", "", t, False, va) + assert cn.alarms_for(mk("a", "bd"), cfg, {}) == [t - D, t - 9 * H] + assert cn.alarms_for(mk("a", "other"), cfg, {}) == [t - 15 * M] + assert cn.alarms_for(mk("a", "bd", (t - 5 * M,)), cfg, {}) == [t - 5 * M] # VALARM beats default + assert cn.alarms_for(mk("a", "bd", (t - 5 * M,)), cfg, {"a": ["2h"]}) == [t - 2 * H] # override beats VALARM + assert cn.alarms_for(mk("a", "bd", (t - 5 * M,)), cfg, {"a": []}) == [] # silenced + assert cn.cal_entry(cfg, "quiet") == (["1h"], False) + assert cn.current_offsets(mk("a", "bd"), cfg, {}) == ["1d", "9h"] + assert cn.horizon(cfg, {"a": ["3d"]}) == 4 * D + + +def test_due(): + t = at(2026, 1, 10, 12) + o = cn.Occ("a", "c", "x", "", t, False, ()) + late = 10 * M + e = [(t - 30 * M, o), (t - 5 * M, o), (t, o)] + assert cn.due(e, t - 6 * M, t - 5 * M, late) == [(t - 5 * M, o)] + assert cn.due(e, t - 60 * M, t - 5 * M, late) == [(t - 5 * M, o)] # 30m one too old + assert cn.due(e, t - M, t + M, late) == [(t, o)] # started: only the T-0 + assert cn.announce(o, t - 10 * M, cn.DEFAULTS["voice"]) == "Tra dieci minuti: x" + assert cn.announce(o, t + 5 * M, cn.DEFAULTS["voice"]) == "Adesso: x" + + if __name__ == "__main__": for name, fn in list(globals().items()): if name.startswith("test_"): -- cgit v1.2.3