#!/usr/bin/env python3 """ calendar_sync.py — generate 07-calendar/macro-book.ics from events.yaml Reads the hand-maintained event list, expands recurring events, converts local times to UTC, attaches per-tier reminders as VALARM blocks, and writes a standards-compliant .ics file. Drop the .ics into Google Calendar (or any CalDAV client) to get cross-device sync + native notifications. Usage: python3 scripts/calendar_sync.py python3 scripts/calendar_sync.py --output /path/to/other.ics python3 scripts/calendar_sync.py --horizon-days 120 # only include next N days python3 scripts/calendar_sync.py --dry-run # parse + count, don't write This script does NOT push to Google Calendar. To go from .ics to live Google Calendar entries, see 07-calendar/MAINTENANCE.md. """ from __future__ import annotations import argparse import sys from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from typing import Iterable from zoneinfo import ZoneInfo import yaml from icalendar import Calendar, Event, Alarm from dateutil.rrule import rrule, WEEKLY, MO, TU, WE, TH, FR HERE = Path(__file__).resolve().parent VAULT = HERE.parent DEFAULT_YAML = VAULT / "07-calendar" / "events.yaml" DEFAULT_OUT = VAULT / "07-calendar" / "macro-book.ics" DEFAULT_DURATION_MIN = 30 DEFAULT_HORIZON_DAYS = 365 def parse_local(date_str: str, time_str: str, tz_source: str) -> datetime: """Combine a date, a HH:MM time, and an IANA timezone name → tz-aware datetime.""" y, m, d = (int(x) for x in date_str.split("-")) hh, mm = (int(x) for x in time_str.split(":")) return datetime(y, m, d, hh, mm, tzinfo=ZoneInfo(tz_source)) def _to_date(x) -> date: if isinstance(x, date) and not isinstance(x, datetime): return x if isinstance(x, datetime): return x.date() return date.fromisoformat(str(x)) def occurrences_for_event(ev: dict, horizon_end: date) -> list[datetime]: """Return all start datetimes for an event within [today, horizon_end].""" tz = ev.get("tz_source", "UTC") t_str = ev.get("time", "00:00") if "recurrence" in ev: start_d = _to_date(ev["start"]) end_d = _to_date(ev.get("end", str(horizon_end))) end_d = min(end_d, horizon_end) if start_d > horizon_end: return [] # rrule needs a datetime hh, mm = (int(x) for x in t_str.split(":")) dtstart = datetime.combine(start_d, time(hh, mm), tzinfo=ZoneInfo(tz)) if ev["recurrence"] == "weekly-friday": rule = rrule(WEEKLY, byweekday=[FR], dtstart=dtstart, until=datetime.combine(end_d, time(23, 59), tzinfo=ZoneInfo(tz))) elif ev["recurrence"] == "weekday": rule = rrule(WEEKLY, byweekday=[MO, TU, WE, TH, FR], dtstart=dtstart, until=datetime.combine(end_d, time(23, 59), tzinfo=ZoneInfo(tz))) else: raise ValueError(f"Unknown recurrence '{ev['recurrence']}' in event {ev.get('id')}") return list(rule) # one-off d = _to_date(ev["date"]) if d > horizon_end: return [] return [parse_local(d.isoformat(), t_str, tz)] import re _DURATION_RE = re.compile(r"P(?:(\d+)D)?T?(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?$") def parse_ical_duration(s: str) -> timedelta: """Parse a simple iCal duration (P7D, P1D, PT1H, PT12H, PT30M) into timedelta. Returns negative for use as VALARM TRIGGER (i.e. -PT1H = 1 hour before).""" m = _DURATION_RE.match(s.strip()) if not m: raise ValueError(f"Cannot parse iCal duration: {s!r}") days = int(m.group(1) or 0) hours = int(m.group(2) or 0) minutes = int(m.group(3) or 0) seconds = int(m.group(4) or 0) if not (days or hours or minutes or seconds): raise ValueError(f"iCal duration has no time component: {s!r}") return timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds) def make_valarm(trigger: str, description: str) -> Alarm: """Build a VALARM. trigger is an iCal duration string like 'P7D', 'PT1H'. The alarm fires at event_time - trigger.""" a = Alarm() a.add("ACTION", "DISPLAY") a.add("DESCRIPTION", description) a.add("TRIGGER", -parse_ical_duration(trigger)) return a def build_vevent(ev: dict, start_local: datetime, reminders: list[str], default_duration_min: int) -> Event: """Build a VEVENT for one occurrence of an event.""" e = Event() e.add("UID", f"{ev['id']}-{start_local.strftime('%Y%m%dT%H%M')}-{int(start_local.timestamp())}@macro-book") e.add("DTSTAMP", datetime.now(timezone.utc)) start_utc = start_local.astimezone(timezone.utc) e.add("DTSTART", start_utc) e.add("DTEND", start_utc + timedelta(minutes=default_duration_min)) tier = ev.get("tier", "?") tier_label = f"T{tier}" if isinstance(tier, int) else str(tier).upper() e.add("SUMMARY", f"[{tier_label}] {ev['name']}") # Description: notes + source + categories desc_lines = [] if ev.get("notes"): desc_lines.append(ev["notes"]) if ev.get("status") == "projected": desc_lines.append("(PROJECTED DATE — verify against official source before month)") if ev.get("source"): desc_lines.append(f"Source: {ev['source']}") if ev.get("categories"): desc_lines.append(f"Categories: {', '.join(ev['categories'])}") if desc_lines: e.add("DESCRIPTION", "\n".join(desc_lines)) # Categories if ev.get("categories"): e.add("CATEGORIES", ev["categories"]) # Reminders for r in reminders: e.add_component(make_valarm(r, f"{ev['name']} (in {r})")) return e def main() -> int: p = argparse.ArgumentParser(description="Generate macro-book.ics from events.yaml") p.add_argument("--yaml", type=Path, default=DEFAULT_YAML, help="Path to events.yaml") p.add_argument("--output", type=Path, default=DEFAULT_OUT, help="Path to write .ics file") p.add_argument("--horizon-days", type=int, default=DEFAULT_HORIZON_DAYS, help="Include events up to N days from today") p.add_argument("--default-duration-min", type=int, default=DEFAULT_DURATION_MIN, help="Default event duration in minutes") p.add_argument("--dry-run", action="store_true", help="Parse + count only, don't write") p.add_argument("--from", dest="start_date", type=str, default=None, help="Override start date (YYYY-MM-DD); default is today") args = p.parse_args() if not args.yaml.exists(): print(f"ERROR: {args.yaml} not found", file=sys.stderr) return 1 with open(args.yaml) as f: data = yaml.safe_load(f) meta = data.get("meta", {}) reminder_defaults = data.get("reminder_defaults", { "tier1": ["P7D", "P1D", "PT1H"], "tier2": ["P1D"], "structural": ["PT12H"], }) events = data.get("events", []) if args.start_date: today = date.fromisoformat(args.start_date) else: today = date.today() horizon_end = today + timedelta(days=args.horizon_days) print(f"Loaded {len(events)} event definitions from {args.yaml}") print(f"Horizon: {today} → {horizon_end} ({args.horizon_days} days)") # Build calendar cal = Calendar() cal.add("PRODID", "-//macro-book//calendar_sync.py//EN") cal.add("VERSION", "2.0") cal.add("CALSCALE", "GREGORIAN") cal.add("METHOD", "PUBLISH") cal.add("X-WR-CALNAME", "macro-book") cal.add("X-WR-CALDESC", "Macro event calendar: Tier 1 (FOMC/CPI/NFP/ECB/SNB/BoJ), Tier 2 (ISM/PCE/retail/HICP/China/refunding), and structural (COT, gold ETF).") n_occurrences = 0 n_skipped = 0 by_tier = {"1": 0, "2": 0, "structural": 0} for ev in events: tier = str(ev.get("tier", "structural")) reminders = reminder_defaults.get(f"tier{tier}", reminder_defaults.get("structural", ["P1D"])) try: starts = occurrences_for_event(ev, horizon_end) except Exception as exc: print(f" WARN {ev.get('id')}: {exc}", file=sys.stderr) n_skipped += 1 continue for start in starts: cal.add_component(build_vevent(ev, start, reminders, args.default_duration_min)) n_occurrences += 1 by_tier[tier] = by_tier.get(tier, 0) + 1 print(f"Generated {n_occurrences} VEVENT occurrences across horizon") print(f" Tier 1: {by_tier.get('1', 0)}") print(f" Tier 2: {by_tier.get('2', 0)}") print(f" Structural: {by_tier.get('structural', 0)}") if n_skipped: print(f" Skipped: {n_skipped}") if args.dry_run: print("DRY RUN — not writing output") return 0 args.output.parent.mkdir(parents=True, exist_ok=True) with open(args.output, "wb") as f: f.write(cal.to_ical()) size = args.output.stat().st_size print(f"Wrote {args.output} ({size:,} bytes)") return 0 if __name__ == "__main__": sys.exit(main())