diff options
| author | Danilo M. <danix@danix.xyz> | 2026-09-27 19:49:00 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-09-27 19:49:00 +0200 |
| commit | 2dd08a66fe9e443564fe9dc3ee8cedda1c8a7425 (patch) | |
| tree | 4d1f817fbc97c90cd799949794307db8c750cfd5 | |
| parent | 6d24906dd041d010ce8d1d629cbec8fffc9f06c6 (diff) | |
| download | cal-notif-2dd08a66fe9e443564fe9dc3ee8cedda1c8a7425.tar.gz cal-notif-2dd08a66fe9e443564fe9dc3ee8cedda1c8a7425.zip | |
Add implementation plan, clarify offset lists and calendar names in spec
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
| -rw-r--r-- | docs/superpowers/plans/2026-09-27-cal-notif.md | 1065 | ||||
| -rw-r--r-- | docs/superpowers/specs/2026-09-27-cal-notif-design.md | 8 |
2 files changed, 1071 insertions, 2 deletions
diff --git a/docs/superpowers/plans/2026-09-27-cal-notif.md b/docs/superpowers/plans/2026-09-27-cal-notif.md new file mode 100644 index 0000000..6c312b8 --- /dev/null +++ b/docs/superpowers/plans/2026-09-27-cal-notif.md @@ -0,0 +1,1065 @@ +# cal-notif Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** A calendar notification daemon that reads vdirsyncer `.ics` files, fires a popup (with snooze) and a local Kokoro voice announcement at each alarm, plus a udt-themed rofi picker for per-event custom alarms. + +**Architecture:** One executable Python file `cal-notif` with three subcommands (`daemon`, `pick`, `say`), organised in sections (durations, config, calendars, alarms, voice, daemon, picker). Pure functions (expansion, precedence, fire window) carry the logic and are tested by `test_cal_notif.py`; the daemon loop, rofi and audio glue are thin and checked by hand. + +**Tech Stack:** system `/usr/bin/python3` 3.12, `icalendar` 6.1, `dateutil`, `kokoro_onnx` + `numpy` (all already installed system-wide), stdlib `tomllib`/`zoneinfo`; `notify-send`, `rofi` 2.0, `aplay`. + +**Spec:** `docs/superpowers/specs/2026-09-27-cal-notif-design.md` + +**Conventions for every task:** +- Tests run with `/usr/bin/python3 test_cal_notif.py` (plain asserts, no framework). Expected output is one `ok test_<name>` line per test. +- Code in `cal-notif` goes under the section comment named in the step (`# ---- <name> ----`), in the order given. +- Commits are GPG-signed (global config does this; never disable it) and end with the line `Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>`. +- No personal data in code, fixtures or messages: use `example.org` UIDs and generic summaries. + +--- + +## File structure + +| File | Responsibility | +|---|---| +| `cal-notif` | The whole program, executable, shebang `/usr/bin/python3` | +| `test_cal_notif.py` | Assert-based tests of the pure functions; loads `cal-notif` by path | +| `config.example.toml` | Documented example config | +| `README.md` | Usage, setup, license, Development Approach | +| `.gitignore` | `__pycache__/` | + +--- + +### Task 1: Skeleton and durations + +**Files:** +- Create: `cal-notif`, `test_cal_notif.py`, `.gitignore` + +- [ ] **Step 1: Create the skeleton `cal-notif`** + +```python +#!/usr/bin/python3 +# cal-notif: calendar notifications with local spoken announcements. +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU General Public License version 2 as published +# by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY +# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License +# for more details. +"""Usage: cal-notif daemon | pick | say TEXT + +daemon watch the calendars and notify (popup + voice) at each alarm +pick rofi picker to set custom alarm offsets for one event +say speak TEXT with the configured voice +""" + +import json +import os +import re +import shutil +import subprocess +import sys +import time +import tomllib +from collections import namedtuple +from datetime import datetime, time as dtime, timedelta, timezone +from pathlib import Path + +from dateutil.rrule import rrulestr +from icalendar import Calendar, vRecur + +# ---- durations ---- + +# ---- config ---- + +# ---- calendars ---- + +# ---- alarms ---- + +# ---- voice ---- + +# ---- daemon ---- + +# ---- picker ---- + + +def main(): + sys.exit(__doc__) + + +if __name__ == "__main__": + main() +``` + +Run: `chmod +x cal-notif && printf '__pycache__/\n' > .gitignore` + +- [ ] **Step 2: Write the failing test file** + +`test_cal_notif.py`: + +```python +#!/usr/bin/python3 +# Tests for cal-notif. Run: /usr/bin/python3 test_cal_notif.py +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# Licensed under the GNU General Public License version 2 only. + +import importlib.machinery +import importlib.util +import os +import tempfile +import time +from datetime import datetime, timedelta +from pathlib import Path +from zoneinfo import ZoneInfo + +os.environ["TZ"] = "Europe/Rome" # all-day and floating times are local +time.tzset() + +HERE = Path(__file__).resolve().parent +_loader = importlib.machinery.SourceFileLoader("cal_notif", str(HERE / "cal-notif")) +_spec = importlib.util.spec_from_loader("cal_notif", _loader) +cn = importlib.util.module_from_spec(_spec) +_loader.exec_module(cn) + +ROME = ZoneInfo("Europe/Rome") +H, M, D = timedelta(hours=1), timedelta(minutes=1), timedelta(days=1) + + +def ics(*events): + body = "".join(f"BEGIN:VEVENT\n{e.strip()}\nEND:VEVENT\n" for e in events) + return f"BEGIN:VCALENDAR\nVERSION:2.0\n{body}END:VCALENDAR\n".encode() + + +def at(y, mo, d, h=0, mi=0): + return datetime(y, mo, d, h, mi, tzinfo=ROME) + + +def test_durations(): + assert cn.parse_duration("1d 2h 15m") == D + 2 * H + 15 * M + assert cn.parse_duration("90m") == 90 * M + assert cn.parse_duration("0m") == timedelta() + for bad in ("", "5", "1x", "m", "2h1d"): + try: + cn.parse_duration(bad) + raise AssertionError(bad) + except ValueError: + pass + assert cn.fmt_duration(D + 2 * H) == "1d 2h" + assert cn.fmt_duration(timedelta()) == "0m" + assert cn.spoken_it(10 * M) == "dieci minuti" + assert cn.spoken_it(H) == "un'ora" + assert cn.spoken_it(D + 2 * H) == "un giorno e due ore" + assert cn.spoken_it(2 * D + H + 21 * M) == "due giorni, un'ora e ventuno minuti" + assert cn.number_it(28) == "ventotto" and cn.number_it(45) == "quarantacinque" + + +if __name__ == "__main__": + for name, fn in list(globals().items()): + if name.startswith("test_"): + fn() + print("ok", name) +``` + +- [ ] **Step 3: Run it, expect failure** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `AttributeError: module 'cal_notif' has no attribute 'parse_duration'` + +- [ ] **Step 4: Implement under `# ---- durations ----`** + +```python +DUR = re.compile(r"(?:(\d+)d)?(?:(\d+)h)?(?:(\d+)m)?") + + +def parse_duration(s): + m = DUR.fullmatch(s.replace(" ", "")) + if not s.strip() or not m: + raise ValueError(f"bad duration {s!r}, want e.g. '1d 2h 15m'") + d, h, mi = (int(x or 0) for x in m.groups()) + return timedelta(days=d, hours=h, minutes=mi) + + +def fmt_duration(td): + m = round(td.total_seconds() / 60) + d, m = divmod(m, 1440) + h, m = divmod(m, 60) + return " ".join(f"{n}{u}" for n, u in ((d, "d"), (h, "h"), (m, "m")) if n) or "0m" + + +UNITS = ["", "uno", "due", "tre", "quattro", "cinque", "sei", "sette", "otto", "nove", + "dieci", "undici", "dodici", "tredici", "quattordici", "quindici", "sedici", + "diciassette", "diciotto", "diciannove"] +TENS = ["", "", "venti", "trenta", "quaranta", "cinquanta", "sessanta", "settanta", + "ottanta", "novanta"] + + +def number_it(n): + # 1..99; ventuno, ventotto: the tens drop their vowel before uno/otto. + if n < 20: + return UNITS[n] + t, u = divmod(n, 10) + return (TENS[t][:-1] if u in (1, 8) else TENS[t]) + UNITS[u] + + +def spoken_it(td): + m = round(td.total_seconds() / 60) + d, m = divmod(m, 1440) + h, m = divmod(m, 60) + parts = [] + for n, one, many in ((d, "un giorno", "giorni"), (h, "un'ora", "ore"), (m, "un minuto", "minuti")): + if n == 1: + parts.append(one) + elif n: + # ponytail: number_it covers 1-99; bigger counts are read as digits. + parts.append(f"{number_it(n) if n < 100 else n} {many}") + if not parts: + return "" + return parts[0] if len(parts) == 1 else ", ".join(parts[:-1]) + " e " + parts[-1] +``` + +- [ ] **Step 5: Run, expect pass** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `ok test_durations` + +- [ ] **Step 6: Commit** + +```bash +git add cal-notif test_cal_notif.py .gitignore +git commit -m "Add cal-notif skeleton with duration parsing and Italian spoken form" +``` + +--- + +### Task 2: Config and overrides + +**Files:** +- Modify: `cal-notif` (constants after the imports, `# ---- config ----`) +- Modify: `test_cal_notif.py` + +- [ ] **Step 1: Add the failing test** (above the `if __name__` block) + +```python +def test_overrides_roundtrip(): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "overrides.toml" + ov = {"a@example.org": ["1h", "10m"], 'q"uote@example.org': []} + cn.save_overrides(ov, p) + assert cn.load_overrides(p) == ov + assert cn.load_overrides(Path(d) / "missing.toml") == {} + + +def test_config(): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "config.toml" + assert cn.load_config(p)["voice"]["lang"] == "it" # no file: defaults + p.write_text('snooze = "15m"\n[voice]\nlang = "en-us"\n' + '[defaults]\nbd = { offsets = ["1d"], voice = false }\n') + c = cn.load_config(p) + assert c["snooze"] == "15m" and c["late"] == "10m" + assert c["voice"]["lang"] == "en-us" and c["voice"]["lead_silence"] == 0.3 # merged + p.write_text('[defaults]\n"*" = ["soon"]\n') + try: + cn.load_config(p) + raise AssertionError + except ValueError: + pass +``` + +- [ ] **Step 2: Run, expect failure** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `ok test_durations`, then `AttributeError: ... 'save_overrides'` + +- [ ] **Step 3: Implement** + +Directly after the imports: + +```python +CONF_DIR = Path(os.environ.get("XDG_CONFIG_HOME", "~/.config")).expanduser() / "cal-notif" +DEFAULTS = { + "calendars_dir": "~/.local/share/calendars", + "snooze": "5m", + "late": "10m", + "rofi_theme": "~/.config/rofi/udt/list.rasi", + "defaults": {"*": []}, + "voice": {"enabled": True, + "model": "/data/voice-models/kokoro/kokoro-v1.0.onnx", + "voices": "/data/voice-models/kokoro/voices-v1.0.bin", + "lang": "it", + "blend": {"if_sara": 0.8, "af_bella": 0.2}, + "lead_silence": 0.3, + "say": "Tra {in}: {summary}", + "say_now": "Adesso: {summary}"}, +} +PICK_DAYS = 30 +``` + +Under `# ---- config ----`: + +```python +def load_config(path=None): + path = path or CONF_DIR / "config.toml" + cfg = {k: dict(v) if isinstance(v, dict) else v for k, v in DEFAULTS.items()} + if path.exists(): + with open(path, "rb") as f: + user = tomllib.load(f) + for k, v in user.items(): + cfg[k] = {**cfg[k], **v} if k == "voice" else v + for k in ("snooze", "late"): # fail early on bad durations + parse_duration(cfg[k]) + for v in cfg["defaults"].values(): + for o in (v.get("offsets", []) if isinstance(v, dict) else v): + parse_duration(o) + return cfg + + +def load_overrides(path=None): + path = path or CONF_DIR / "overrides.toml" + if not path.exists(): + return {} + with open(path, "rb") as f: + return tomllib.load(f) + + +def save_overrides(ov, path=None): + path = path or CONF_DIR / "overrides.toml" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + # JSON strings and string arrays are valid TOML, so no TOML writer is needed. + tmp.write_text("".join(f"{json.dumps(uid)} = {json.dumps(offs)}\n" + for uid, offs in sorted(ov.items()))) + os.replace(tmp, path) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `ok test_durations`, `ok test_overrides_roundtrip`, `ok test_config` + +- [ ] **Step 5: Commit** + +```bash +git add cal-notif test_cal_notif.py +git commit -m "Load config with defaults, read and write per-event overrides" +``` + +--- + +### Task 3: Event expansion + +**Files:** +- Modify: `cal-notif` (`# ---- calendars ----`) +- Modify: `test_cal_notif.py` + +- [ ] **Step 1: Add the failing test** + +```python +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" +``` + +- [ ] **Step 2: Run, expect failure** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `AttributeError: ... 'expand'` + +- [ ] **Step 3: Implement under `# ---- calendars ----`** + +```python +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) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: all `ok`, including `ok test_expand` and `ok test_load_events`; one stderr line `cal-notif: skipping .../broken.ics: ...`. + +- [ ] **Step 5: Smoke test against the real calendars (read-only, counts only)** + +```bash +/usr/bin/python3 - <<'EOF' +import importlib.machinery, importlib.util, time +from datetime import datetime, timedelta +l = importlib.machinery.SourceFileLoader("cn", "cal-notif") +s = importlib.util.spec_from_loader("cn", l); cn = importlib.util.module_from_spec(s); l.exec_module(cn) +now = datetime.now().astimezone(); t = time.time() +occs = cn.load_events(cn.load_config()["calendars_dir"], now - timedelta(days=1), now + timedelta(days=30)) +print(len(occs), "occurrences,", round(time.time() - t, 2), "s") +EOF +``` + +Expected: a count, well under 1 s, no `skipping` lines (prototype: 3 occurrences, 0.06 s over 283 files). + +- [ ] **Step 6: Commit** + +```bash +git add cal-notif test_cal_notif.py +git commit -m "Expand calendar events: RRULE, EXDATE, RECURRENCE-ID, all-day, VALARM" +``` + +--- + +### Task 4: Alarm precedence, fire window, announcement text + +**Files:** +- Modify: `cal-notif` (`# ---- alarms ----`) +- Modify: `test_cal_notif.py` + +- [ ] **Step 1: Add the failing tests** + +```python +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" +``` + +- [ ] **Step 2: Run, expect failure** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `AttributeError: ... 'alarms_for'` + +- [ ] **Step 3: Implement under `# ---- alarms ----`** + +```python +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)}) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: all `ok`, including `ok test_precedence` and `ok test_due`. + +- [ ] **Step 5: Commit** + +```bash +git add cal-notif test_cal_notif.py +git commit -m "Pick alarms by precedence and compute which ones are due" +``` + +--- + +### Task 5: Voice (`say`) + +**Files:** +- Modify: `cal-notif` (`# ---- voice ----`, `main`) + +No automated test: it needs the model and a sound card. Checked by ear. + +- [ ] **Step 1: Implement under `# ---- voice ----`** + +```python +def say(text, vcfg): + import numpy as np + from kokoro_onnx import Kokoro + k = Kokoro(vcfg["model"], vcfg["voices"]) + style = sum(w * k.get_voice_style(n) for n, w in vcfg["blend"].items()) + audio, sr = k.create(text, voice=style, lang=vcfg["lang"]) + # A new playback stream wakes the suspended sink and loses its first + # ~0.2 s: lead with silence so it eats that, not the first word. + lead = bytes(int(sr * vcfg["lead_silence"]) * 2) + pcm = (np.clip(audio, -1, 1) * 32767).astype("<i2").tobytes() + subprocess.run(["aplay", "-q", "-r", str(sr), "-f", "S16_LE", "-c", "1"], + input=lead + pcm, check=True) + + +def voice_ready(vcfg): + if not vcfg["enabled"]: + return False + missing = [p for p in (vcfg["model"], vcfg["voices"]) if not Path(p).exists()] + if missing or not shutil.which("aplay"): + print(f"cal-notif: voice off, missing {missing or 'aplay'}", file=sys.stderr) + return False + return True +``` + +- [ ] **Step 2: Replace `main`** + +```python +def main(): + cmd = sys.argv[1:2] + if cmd == ["say"] and len(sys.argv) > 2: + say(" ".join(sys.argv[2:]), load_config()["voice"]) + else: + sys.exit(__doc__) +``` + +- [ ] **Step 3: Check by ear (ask the user to listen)** + +Run: `./cal-notif say "Tra dieci minuti: prova della voce"` +Expected: the whole phrase, first word ("Tra") included, in the if_sara blend. If the first word is clipped, raise `lead_silence` in the config (0.5) and repeat; record the value that works. + +- [ ] **Step 4: Run tests, then commit** + +Run: `/usr/bin/python3 test_cal_notif.py` (all `ok`) + +```bash +git add cal-notif +git commit -m "Speak announcements with Kokoro, leading silence against clipped first word" +``` + +--- + +### Task 6: Daemon + +**Files:** +- Modify: `cal-notif` (`# ---- daemon ----`, `main`) + +- [ ] **Step 1: Implement under `# ---- daemon ----`** + +```python +def notify(occ, now, cfg, voice): + when = f"{occ.start:%d.%m}" if occ.allday else f"{occ.start:%d.%m %H:%M}" + body = "\n".join(x for x in (when, occ.location) if x) + urgency = "critical" if now >= occ.start - timedelta(minutes=1) else "normal" + kids = [subprocess.Popen(["notify-send", "--wait", "--app-name=cal-notif", "-u", urgency, + "--action=snooze=Snooze", "--action=dismiss=Dismiss", + occ.summary or "(senza titolo)", body], + stdout=subprocess.PIPE, text=True)] + if voice and cal_entry(cfg, occ.cal)[1]: + kids.append(subprocess.Popen([sys.executable, __file__, "say", + announce(occ, now, cfg["voice"])])) + return [(k, occ) for k in kids] + + +def signature(cfg): + paths = [CONF_DIR / "config.toml", CONF_DIR / "overrides.toml"] + paths += [d for d, _ in calendars(cfg["calendars_dir"])] + # ponytail: directory mtimes catch vdirsyncer's rename-into-place writes; + # an in-place edit of an .ics would go unnoticed until the next change. + return tuple(p.stat().st_mtime if p.exists() else 0 for p in paths) + + +def daemon(): + if not shutil.which("notify-send"): + sys.exit("cal-notif: notify-send not found") + cfg, ov = load_config(), load_overrides() # a bad config at startup exits + voice = voice_ready(cfg["voice"]) + now = datetime.now().astimezone() + last = now - parse_duration(cfg["late"]) + sig, entries, snoozed, kids = None, [], [], [] + while True: + now = datetime.now().astimezone() + if (s := signature(cfg)) != sig: + if sig is not None: + try: + cfg, ov = load_config(), load_overrides() + except Exception as e: # keep the previous config + print(f"cal-notif: config not reloaded: {e}", file=sys.stderr) + occs = load_events(cfg["calendars_dir"], now - timedelta(days=1), now + horizon(cfg, ov)) + entries, sig = schedule(occs, cfg, ov), s + # ponytail: fired alarms live in memory; a restart within `late` + # repeats those popups once. + fire = due(entries, last, now, parse_duration(cfg["late"])) + fire += [e for e in snoozed if e[0] <= now] + snoozed = [e for e in snoozed if e[0] > now] + for _, o in fire: + kids += notify(o, now, cfg, voice) + for k, o in kids: + if k.poll() is not None and k.stdout and k.stdout.read().strip() == "snooze": + snoozed.append((now + parse_duration(cfg["snooze"]), o)) + kids = [(k, o) for k, o in kids if k.returncode is None] + last = now + time.sleep(30) +``` + +- [ ] **Step 2: Add `daemon` to `main`** + +```python +def main(): + cmd = sys.argv[1:2] + if cmd == ["daemon"]: + daemon() + elif cmd == ["say"] and len(sys.argv) > 2: + say(" ".join(sys.argv[2:]), load_config()["voice"]) + else: + sys.exit(__doc__) +``` + +- [ ] **Step 3: Live check with a throwaway calendar (no real data touched)** + +```bash +T=$(mktemp -d); mkdir "$T/test" +S=$(date -d '+3 min' +%Y%m%dT%H%M00) +printf 'BEGIN:VCALENDAR\nVERSION:2.0\nBEGIN:VEVENT\nUID:t@example.org\nDTSTART:%s\nSUMMARY:prova cal-notif\nBEGIN:VALARM\nACTION:DISPLAY\nTRIGGER:-PT2M\nEND:VALARM\nEND:VEVENT\nEND:VCALENDAR\n' "$S" > "$T/test/t.ics" +mkdir -p "$T/conf/cal-notif"; printf 'calendars_dir = "%s"\n' "$T" > "$T/conf/cal-notif/config.toml" +XDG_CONFIG_HOME="$T/conf" timeout 300 ./cal-notif daemon +``` + +Expected, within ~1.5 min: popup "prova cal-notif" and the voice saying "Tra due minuti: prova cal-notif" (or "un minuto", depending on the 30 s tick). Ask the user to click **Snooze**: 5 min later (the `timeout` gives 5 min; raise to 420 if needed) the popup and voice repeat. Then `rm -r "$T"`. + +- [ ] **Step 4: Run tests, then commit** + +Run: `/usr/bin/python3 test_cal_notif.py` (all `ok`) + +```bash +git add cal-notif +git commit -m "Add the daemon loop: rescan on change, notify, speak, snooze" +``` + +--- + +### Task 7: rofi picker + +**Files:** +- Modify: `cal-notif` (`# ---- picker ----`, `main`) +- Modify: `test_cal_notif.py` + +- [ ] **Step 1: Add the failing test** + +```python +def test_choice(): + assert cn.parse_choice("1d, 1h") == ["1d", "1h"] + assert cn.parse_choice("90m") == ["1h 30m"] + assert cn.parse_choice("none") == [] + assert cn.parse_choice("reset") is None + try: + cn.parse_choice("soon") + raise AssertionError + except ValueError: + pass +``` + +- [ ] **Step 2: Run, expect failure** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: `AttributeError: ... 'parse_choice'` + +- [ ] **Step 3: Implement under `# ---- picker ----`** + +```python +def rofi(lines, prompt, theme, mesg=None, index=False): + # udt theming: list.rasi keeps the inputbar (free-text durations) and + # the message widget (-mesg); menu.rasi has neither. + cmd = ["rofi", "-dmenu", "-i", "-theme", str(Path(theme).expanduser()), "-p", prompt] + cmd += ["-mesg", mesg] if mesg else [] + cmd += ["-format", "i"] if index else [] + r = subprocess.run(cmd, input="\n".join(lines), capture_output=True, text=True) + return r.stdout.strip() if r.returncode == 0 else None # Escape: cancel + + +def parse_choice(s): + """Returns a list of offsets, or None for 'reset'. Raises ValueError.""" + s = s.strip() + if s == "reset": + return None + if s == "none": + return [] + return [fmt_duration(parse_duration(p)) for p in s.split(",")] + + +def pick(): + cfg, ov = load_config(), load_overrides() + now = datetime.now().astimezone() + occs = [o for o in load_events(cfg["calendars_dir"], now, now + timedelta(days=PICK_DAYS)) + if o.start >= now] + fmt = lambda o: f"{o.start:%d.%m} " if o.allday else f"{o.start:%d.%m %H:%M}" + lines = [f"{fmt(o)} · {o.cal} · {o.summary} · [{', '.join(current_offsets(o, cfg, ov))}]" + for o in occs] + i = rofi(lines, "evento", cfg["rofi_theme"], index=True) + if i is None or not i.lstrip("-").isdigit() or int(i) < 0: + return + o, mesg = occs[int(i)], lines[int(i)] + presets = ["10m", "30m", "1h", "1d", "1d, 1h", "none", "reset"] + while (s := rofi(presets, "avvisi", cfg["rofi_theme"], mesg=mesg)) is not None: + try: + offs = parse_choice(s) + except ValueError as e: + mesg = f"{lines[int(i)]}\n{e}" + continue + if offs is None: + ov.pop(o.uid, None) + else: + ov[o.uid] = offs + save_overrides(ov) + return +``` + +- [ ] **Step 4: Final `main`** + +```python +def main(): + cmd = sys.argv[1:2] + if cmd == ["daemon"]: + daemon() + elif cmd == ["pick"]: + if not shutil.which("rofi"): + sys.exit("cal-notif: rofi not found") + pick() + elif cmd == ["say"] and len(sys.argv) > 2: + say(" ".join(sys.argv[2:]), load_config()["voice"]) + else: + sys.exit(__doc__) +``` + +- [ ] **Step 5: Run tests** + +Run: `/usr/bin/python3 test_cal_notif.py` +Expected: all `ok`, including `ok test_choice`. + +- [ ] **Step 6: Live check (user drives rofi)** + +Run with a throwaway config dir so the real overrides file is untouched: + +```bash +T=$(mktemp -d); XDG_CONFIG_HOME="$T" ./cal-notif pick; cat "$T/cal-notif/overrides.toml"; rm -r "$T" +``` + +Ask the user to: confirm both menus use the udt list look; pick an event; type `soon` (expect the error in the message line and a re-prompt); then type `1d, 30m`. Expected file content: `"<uid>" = ["1d", "30m"]`. Escape in either menu exits with no file written. + +- [ ] **Step 7: Commit** + +```bash +git add cal-notif test_cal_notif.py +git commit -m "Add the udt-themed rofi picker for per-event alarms" +``` + +--- + +### Task 8: README and example config + +**Files:** +- Create: `README.md`, `config.example.toml` + +- [ ] **Step 1: Write `config.example.toml`** + +```toml +# cal-notif example config. Copy to ~/.config/cal-notif/config.toml. +# Every key is optional; these are the defaults unless marked "example". + +calendars_dir = "~/.local/share/calendars" # vdirsyncer filesystem storage +snooze = "5m" # the popup's Snooze delay +late = "10m" # deliver alarms missed by at most this much +rofi_theme = "~/.config/rofi/udt/list.rasi" + +# Alarms for events that have no VALARM, per calendar. The calendar name is +# its vdirsyncer `displayname`, else its directory name. Durations are +# "<n>d <n>h <n>m"; a list gives several alarms. +[defaults] +"*" = [] # silent unless the event has a VALARM +# birthdays = ["1d", "9h"] # example +# work = { offsets = ["15m"], voice = false } # example: popup only + +[voice] +enabled = true +model = "/data/voice-models/kokoro/kokoro-v1.0.onnx" +voices = "/data/voice-models/kokoro/voices-v1.0.bin" +lang = "it" +blend = { if_sara = 0.8, af_bella = 0.2 } +lead_silence = 0.3 # raise if the first word gets clipped +say = "Tra {in}: {summary}" +say_now = "Adesso: {summary}" +``` + +- [ ] **Step 2: Write `README.md`** + +````markdown +# cal-notif + +Calendar notifications for the desktop. Reads the `.ics` files vdirsyncer +keeps in sync, shows a popup at each alarm (with a Snooze button) and says +it out loud with a local Kokoro voice. Nothing leaves the machine. + +- Each event's own alarms (VALARM) set when it notifies. +- Events without alarms can get per-calendar defaults, or per-event + alarms chosen from a rofi menu. A per-event choice replaces the event's + own alarms, and `none` silences it. + +## Requirements + +- Python 3.11+ with `icalendar`, `python-dateutil`, `numpy`, `kokoro-onnx` +- Kokoro model files (`kokoro-v1.0.onnx`, `voices-v1.0.bin`) +- `notify-send` and a notification server that supports actions +- `aplay`, `rofi` +- vdirsyncer with a `filesystem` storage for the calendars + +## Usage + + cal-notif daemon # start from your compositor's autostart + cal-notif pick # bind to a key: set alarms for one event + cal-notif say TEXT # test the voice + +Config: `~/.config/cal-notif/config.toml`, see `config.example.toml`. +Per-event alarms from `pick` go to `~/.config/cal-notif/overrides.toml` +(`"<event UID>" = ["1h", "10m"]`), which can also be edited by hand. The +daemon picks up changes to both files, and to the calendars, within 30 s. + +In the picker, type durations like `1d 2h` (one alarm) or `1d, 1h` (two +alarms), `none` to silence the event, `reset` to go back to its own alarms. + +## Tests + + /usr/bin/python3 test_cal_notif.py + +## License + +GPLv2 only, see `LICENSE`. + +## Development Approach + +This project is developed using AI-assisted tools. Code is generated with the help of AI based on human-provided specifications, design decisions, and iterative feedback. + +All contributions are reviewed, tested, and curated by the maintainer before being included in the codebase. AI is used as a productivity and exploration tool, while human oversight remains central to all decisions. + +The goal is to combine the flexibility of AI-assisted development with standard open-source practices such as transparency, review, and accountability. +```` + +- [ ] **Step 3: Check the example config parses** + +Run: `/usr/bin/python3 -c "import tomllib; tomllib.load(open('config.example.toml','rb'))" && echo ok` +Expected: `ok` + +- [ ] **Step 4: Commit** + +```bash +git add README.md config.example.toml +git commit -m "Add README and example config" +``` + +--- + +## Deployment (after all tasks, confirm with the user first) + +- Symlink `cal-notif` into `~/bin/`. +- Copy `config.example.toml` to `~/.config/cal-notif/config.toml`, set `[defaults]` for the birthday calendar if wanted. +- Add `cal-notif daemon &` to the compositor autostart and a keybind for `cal-notif pick`. These live in the user's own dotfiles: show the lines, let the user place them. diff --git a/docs/superpowers/specs/2026-09-27-cal-notif-design.md b/docs/superpowers/specs/2026-09-27-cal-notif-design.md index 3a495b3..4d02f90 100644 --- a/docs/superpowers/specs/2026-09-27-cal-notif-design.md +++ b/docs/superpowers/specs/2026-09-27-cal-notif-design.md @@ -41,6 +41,8 @@ missing file works (VALARMs only, voice on). ```toml calendars_dir = "~/.local/share/calendars" # vdirsyncer filesystem storage +# A calendar's name is its `displayname` file (vdirsyncer metadata), else +# its directory name. [defaults] keys and the picker use that name. snooze = "5m" rofi_theme = "~/.config/rofi/udt/list.rasi" late = "10m" # deliver alarms missed by at most this much @@ -70,7 +72,9 @@ config and its comments. "other-uid@example.org" = [] # silenced ``` -Durations are strings of `<n>d`, `<n>h`, `<n>m` parts, e.g. `"1d 2h 15m"`. +Durations are strings of `<n>d`, `<n>h`, `<n>m` parts, e.g. `"1d 2h 15m"` +(one offset of 26h15m). Several offsets are a list in TOML, or +comma-separated in the picker (`1d, 1h` is two alarms). A calendar in `[defaults]` may also set `voice = false` via a table form (`birthdays = { offsets = ["1d"], voice = false }`). @@ -149,7 +153,7 @@ theme path is a config key (`rofi_theme`) defaulting to that file. 1. rofi lists occurrences in the next 30 days: `dd.mm HH:MM · calendar · summary · [current offsets]`. 2. Second rofi menu for the chosen event, with the event's summary and time - in `-mesg`: presets `10m`, `30m`, `1h`, `1d`, `1d 1h`, `none` (silence), + in `-mesg`: presets `10m`, `30m`, `1h`, `1d`, `1d, 1h`, `none` (silence), `reset` (drop override). Free text is accepted and parsed as durations; invalid input re-prompts with the error shown in `-mesg`. rofi exits non-zero on Escape: treated as cancel, never as an error. |
