From 9703cb473f457cf47944e0bb3523e34f3d4b2d41 Mon Sep 17 00:00:00 2001 From: hbrain Date: Sun, 17 May 2026 10:13:31 +0000 Subject: [PATCH] Include upcoming calendar events in analysis --- .env.example | 2 ++ README.md | 11 ++++++ ha_observer.py | 94 ++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+) diff --git a/.env.example b/.env.example index 6b2ee16..f707c43 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,8 @@ PROMPT_FILE="./llm_instructions.md" # Collection/history settings HISTORY_HOURS="24" MAX_HISTORY_PER_ENTITY="20" +CALENDAR_LOOKAHEAD_DAYS="7" +MAX_CALENDAR_EVENTS_PER_CALENDAR="8" KEEP_SNAPSHOT_DAYS="14" # At 05:00, analyze snapshots from roughly this many hours diff --git a/README.md b/README.md index a0227c7..d40caca 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,17 @@ No AI, but still publish a placeholder page: LLM_MODE="none" ``` +## Calendar events + +If Home Assistant has `calendar.*` entities, the collector fetches upcoming events through the Home Assistant calendar API and includes them in the analysis. + +Relevant settings: + +```bash +CALENDAR_LOOKAHEAD_DAYS="7" +MAX_CALENDAR_EVENTS_PER_CALENDAR="8" +``` + ## Extra LLM instructions Edit a local, gitignored instructions file to change how the 05:00 AI analysis behaves: diff --git a/ha_observer.py b/ha_observer.py index ee1984b..8e4b15e 100755 --- a/ha_observer.py +++ b/ha_observer.py @@ -38,6 +38,8 @@ SITE_URL = os.environ.get("SITE_URL", "http://localhost").rstrip("/") PROMPT_FILE = Path(os.environ.get("PROMPT_FILE", "./llm_instructions.md")) HISTORY_HOURS = int(os.environ.get("HISTORY_HOURS", "24")) MAX_HISTORY_PER_ENTITY = int(os.environ.get("MAX_HISTORY_PER_ENTITY", "20")) +CALENDAR_LOOKAHEAD_DAYS = int(os.environ.get("CALENDAR_LOOKAHEAD_DAYS", "7")) +MAX_CALENDAR_EVENTS_PER_CALENDAR = int(os.environ.get("MAX_CALENDAR_EVENTS_PER_CALENDAR", "8")) ANALYZE_SNAPSHOT_HOURS = int(os.environ.get("ANALYZE_SNAPSHOT_HOURS", "24")) ARTICLE_CONTEXT_DAYS = int(os.environ.get("ARTICLE_CONTEXT_DAYS", "7")) MAX_ANALYZE_CHARS = int(os.environ.get("MAX_ANALYZE_CHARS", "80000")) @@ -169,6 +171,84 @@ def get_states() -> list[dict[str, Any]]: return sorted(useful, key=lambda x: x["entity_id"]) +def clean_text(value: Any, max_len: int = 300) -> str: + if not value: + return "" + text = re.sub(r"<[^>]+>", " ", str(value)) + text = re.sub(r"\s+", " ", html.unescape(text)).strip() + return text[:max_len] + + +def human_date_label(dt: datetime, include_time: bool) -> str: + today = datetime.now(ZoneInfo(DISPLAY_TIMEZONE)).date() + event_date = dt.date() + delta_days = (event_date - today).days + if delta_days == 0: + day = "today" + elif delta_days == 1: + day = "tomorrow" + elif 1 < delta_days <= 7: + day = f"upcoming {dt.strftime('%A')}" + elif -7 <= delta_days < 0: + day = f"last {dt.strftime('%A')}" + else: + day = dt.strftime("%A") + if include_time: + return f"{day} at {dt.strftime('%H:%M')}" + return day + + +def event_time(value: dict[str, str] | None) -> str: + if not value: + return "" + if "dateTime" in value: + try: + dt = datetime.fromisoformat(value["dateTime"].replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return human_date_label(dt.astimezone(ZoneInfo(DISPLAY_TIMEZONE)), include_time=True) + except Exception: + return display_time(value.get("dateTime")) + if "date" in value: + try: + dt = datetime.fromisoformat(value["date"]).replace(tzinfo=ZoneInfo(DISPLAY_TIMEZONE)) + return human_date_label(dt, include_time=False) + except Exception: + return value.get("date", "") + return "" + + +def get_calendar_events(calendar_entity_ids: list[str]) -> list[dict[str, Any]]: + if not calendar_entity_ids or CALENDAR_LOOKAHEAD_DAYS <= 0: + return [] + start = datetime.now(timezone.utc) + end = start + timedelta(days=CALENDAR_LOOKAHEAD_DAYS) + calendars: list[dict[str, Any]] = [] + for entity_id in calendar_entity_ids: + try: + events = ha_get( + f"/api/calendars/{entity_id}", + params={"start": start.isoformat(), "end": end.isoformat()}, + ) + except Exception as exc: + print(f"Skipping calendar events for {entity_id}: {exc}", file=sys.stderr) + continue + compact_events = [] + for event in events[:MAX_CALENDAR_EVENTS_PER_CALENDAR]: + compact_events.append( + { + "summary": clean_text(event.get("summary"), 160), + "start": event_time(event.get("start")), + "end": event_time(event.get("end")), + "location": clean_text(event.get("location"), 180), + "description": clean_text(event.get("description"), 260), + } + ) + if compact_events: + calendars.append({"entity_id": entity_id, "events": compact_events}) + return calendars + + def get_history(hours: int, entity_ids: list[str]) -> list[dict[str, Any]]: start = datetime.now(timezone.utc) - timedelta(hours=hours) changes: list[dict[str, Any]] = [] @@ -204,11 +284,14 @@ def get_history(hours: int, entity_ids: list[str]) -> list[dict[str, Any]]: def make_snapshot() -> dict[str, Any]: states = get_states() entity_ids = [state["entity_id"] for state in states] + calendar_entity_ids = [entity_id for entity_id in entity_ids if entity_id.startswith("calendar.")] return { "generated_at": datetime.now().isoformat(timespec="seconds"), "history_hours": HISTORY_HOURS, + "calendar_lookahead_days": CALENDAR_LOOKAHEAD_DAYS, "states": states, "history": get_history(HISTORY_HOURS, entity_ids), + "calendar_events": get_calendar_events(calendar_entity_ids), } @@ -309,6 +392,17 @@ def summarize_snapshot(snapshot: dict[str, Any]) -> str: value = f"{state.get('state')} {unit}".strip() score = entity_importance(state.get("entity_id", ""), attrs) lines.append(f"- importance={score} {name} ({state.get('entity_id')}): {value}; last_changed={display_time(state.get('last_changed'))}") + lines.append("Upcoming calendar events:") + for calendar in snapshot.get("calendar_events", []): + lines.append(f"- {calendar.get('entity_id')}:") + for event in calendar.get("events", []): + details = [] + if event.get("location"): + details.append(f"location={event.get('location')}") + if event.get("description"): + details.append(f"description={event.get('description')}") + detail_text = f"; {'; '.join(details)}" if details else "" + lines.append(f" - {event.get('start')} to {event.get('end')}: {event.get('summary')}{detail_text}") lines.append("Recently changed entities:") history = sorted( snapshot.get("history", []),