Add pi analysis mode and HA history filtering

ha_observer.py

@@ -16,6 +16,7 @@ import html
 import json
 import os
 import re
+import subprocess
 import sys
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -35,12 +36,15 @@ MAX_HISTORY_PER_ENTITY = int(os.environ.get("MAX_HISTORY_PER_ENTITY", "20"))
 ANALYZE_SNAPSHOT_HOURS = int(os.environ.get("ANALYZE_SNAPSHOT_HOURS", "24"))
 KEEP_SNAPSHOT_DAYS = int(os.environ.get("KEEP_SNAPSHOT_DAYS", "14"))
 
-# LLM_MODE: none | ollama | openai
+# LLM_MODE: none | pi | ollama | openai
 LLM_MODE = os.environ.get("LLM_MODE", "none").lower()
 OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434").rstrip("/")
 OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "llama3.1")
 OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
 OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
+PI_BIN = os.environ.get("PI_BIN", "pi")
+PI_MODEL = os.environ.get("PI_MODEL", "")
+PI_TIMEOUT = int(os.environ.get("PI_TIMEOUT", "600"))
 
 RELEVANT_DOMAINS = set(
     x.strip()
@@ -81,10 +85,14 @@ def require_config(for_ai: bool = False) -> None:
         raise ConfigError("LLM_MODE=openai but OPENAI_API_KEY is not set")
 
 
-def ha_get(path: str) -> Any:
+def ha_get(path: str, params: dict[str, str] | None = None) -> Any:
     headers = {"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"}
-    response = requests.get(f"{HA_URL}{path}", headers=headers, timeout=60)
-    response.raise_for_status()
+    response = requests.get(f"{HA_URL}{path}", headers=headers, params=params, timeout=60)
+    try:
+        response.raise_for_status()
+    except requests.HTTPError as exc:
+        detail = response.text.strip()
+        raise requests.HTTPError(f"{exc}; response={detail[:500]}", response=response) from exc
     return response.json()
 
 
@@ -115,35 +123,46 @@ def get_states() -> list[dict[str, Any]]:
     return sorted(useful, key=lambda x: x["entity_id"])
 
 
-def get_history(hours: int) -> list[dict[str, Any]]:
+def get_history(hours: int, entity_ids: list[str]) -> list[dict[str, Any]]:
     start = datetime.now(timezone.utc) - timedelta(hours=hours)
-    data = ha_get(f"/api/history/period/{start.isoformat()}?minimal_response")
     changes: list[dict[str, Any]] = []
 
-    for entity_history in data:
-        if not entity_history:
-            continue
-        entity_id = entity_history[0].get("entity_id", "")
-        if not is_relevant_entity(entity_id):
-            continue
-        compact = []
-        for item in entity_history[-MAX_HISTORY_PER_ENTITY:]:
-            state = item.get("state")
-            if state in {"unknown", "unavailable", None}:
+    # Recent Home Assistant versions/configurations require filter_entity_id for
+    # the history endpoint. Query in chunks to avoid an overlong URL.
+    chunk_size = 50
+    for i in range(0, len(entity_ids), chunk_size):
+        chunk = entity_ids[i : i + chunk_size]
+        data = ha_get(
+            f"/api/history/period/{start.isoformat(timespec='seconds')}",
+            params={"filter_entity_id": ",".join(chunk), "minimal_response": ""},
+        )
+
+        for entity_history in data:
+            if not entity_history:
                 continue
-            compact.append({"state": state, "last_changed": item.get("last_changed")})
-        if len(set(x["state"] for x in compact)) > 1:
-            changes.append({"entity_id": entity_id, "recent_states": compact})
+            entity_id = entity_history[0].get("entity_id", "")
+            if not is_relevant_entity(entity_id):
+                continue
+            compact = []
+            for item in entity_history[-MAX_HISTORY_PER_ENTITY:]:
+                state = item.get("state")
+                if state in {"unknown", "unavailable", None}:
+                    continue
+                compact.append({"state": state, "last_changed": item.get("last_changed")})
+            if len(set(x["state"] for x in compact)) > 1:
+                changes.append({"entity_id": entity_id, "recent_states": compact})
 
     return sorted(changes, key=lambda x: x["entity_id"])
 
 
 def make_snapshot() -> dict[str, Any]:
+    states = get_states()
+    entity_ids = [state["entity_id"] for state in states]
     return {
         "generated_at": datetime.now().isoformat(timespec="seconds"),
         "history_hours": HISTORY_HOURS,
-        "states": get_states(),
-        "history": get_history(HISTORY_HOURS),
+        "states": states,
+        "history": get_history(HISTORY_HOURS, entity_ids),
     }
 
 
@@ -258,6 +277,25 @@ def call_openai(prompt: str) -> str:
     return response.json()["choices"][0]["message"]["content"].strip()
 
 
+def call_pi(prompt: str) -> str:
+    cmd = [PI_BIN, "--no-tools"]
+    if PI_MODEL:
+        cmd.extend(["--model", PI_MODEL])
+    cmd.extend(["-p", "Analyze the Home Assistant data from stdin and write the requested briefing."])
+    result = subprocess.run(
+        cmd,
+        input=prompt,
+        text=True,
+        capture_output=True,
+        timeout=PI_TIMEOUT,
+        check=False,
+    )
+    if result.returncode != 0:
+        stderr = result.stderr.strip()
+        raise RuntimeError(f"pi exited with status {result.returncode}: {stderr[-1000:]}")
+    return result.stdout.strip()
+
+
 def get_llm_conclusions(input_summary: str) -> str:
     if LLM_MODE == "none":
         return "AI analysis disabled. Set LLM_MODE=ollama or LLM_MODE=openai in .env. The raccoon analyst is asleep. 🦝💤"
@@ -266,7 +304,9 @@ def get_llm_conclusions(input_summary: str) -> str:
         return call_ollama(prompt)
     if LLM_MODE == "openai":
         return call_openai(prompt)
-    return f"Unknown LLM_MODE={LLM_MODE!r}. Use none, ollama, or openai."
+    if LLM_MODE == "pi":
+        return call_pi(prompt)
+    return f"Unknown LLM_MODE={LLM_MODE!r}. Use none, pi, ollama, or openai."
 
 
 def markdownish_to_html(text: str) -> str: