Include upcoming calendar events in analysis

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", []),