Add non-trivial examples: websocket monitor, history stats, template bulk action

Author: adamlogan73

examples/history_stats.py

@@ -0,0 +1,64 @@
+"""Fetch 24h history for numeric sensors and print min/max/avg per entity.
+
+Environment variables:
+    HOMEASSISTANT_API_ENDPOINT  e.g. http://localhost:8123/api
+    HOMEASSISTANT_API_TOKEN     Long-lived access token
+"""
+
+import os
+from datetime import datetime
+from datetime import timedelta
+from datetime import timezone
+
+from homeassistant_api import Client
+
+url = os.environ["HOMEASSISTANT_API_ENDPOINT"]
+token = os.environ["HOMEASSISTANT_API_TOKEN"]
+
+SENSOR_IDS = [
+    "sensor.living_room_temperature",
+    "sensor.outdoor_temperature",
+    "sensor.energy_consumption",
+]
+
+
+def _to_float(value: str) -> float | None:
+    try:
+        return float(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def main() -> None:
+    now = datetime.now(tz=timezone.utc)
+    yesterday = now - timedelta(hours=24)
+
+    with Client(url, token) as client:
+        entities = [
+            entity
+            for sensor_id in SENSOR_IDS
+            if (entity := client.get_entity(entity_id=sensor_id)) is not None
+        ]
+
+        for history in client.get_entity_histories(
+            entities=tuple(entities),
+            start_timestamp=yesterday,
+            end_timestamp=now,
+        ):
+            values = [
+                v for s in history.states if (v := _to_float(s.state)) is not None
+            ]
+            if not values:
+                print(f"{history.entity_id}: no numeric data in last 24h")  # noqa: T201
+                continue
+            print(  # noqa: T201
+                f"{history.entity_id}: "
+                f"min={min(values):.2f}  "
+                f"max={max(values):.2f}  "
+                f"avg={sum(values) / len(values):.2f}  "
+                f"({len(values)} samples)",
+            )
+
+
+if __name__ == "__main__":
+    main()

examples/template_bulk_action.py

@@ -0,0 +1,42 @@
+"""Use a Jinja2 template to find all lights that are on, then turn them off.
+
+Demonstrates combining get_rendered_template with trigger_service to perform
+bulk operations based on server-side state queries.
+
+Environment variables:
+    HOMEASSISTANT_API_ENDPOINT  e.g. http://localhost:8123/api
+    HOMEASSISTANT_API_TOKEN     Long-lived access token
+"""
+
+import os
+
+from homeassistant_api import Client
+
+url = os.environ["HOMEASSISTANT_API_ENDPOINT"]
+token = os.environ["HOMEASSISTANT_API_TOKEN"]
+
+FIND_ON_LIGHTS = """\
+{{ states.light
+   | selectattr('state', 'eq', 'on')
+   | map(attribute='entity_id')
+   | list
+   | join(',') }}"""
+
+
+def main() -> None:
+    with Client(url, token) as client:
+        rendered = client.get_rendered_template(FIND_ON_LIGHTS)
+        entity_ids = [e.strip() for e in rendered.split(",") if e.strip()]
+
+        if not entity_ids:
+            print("No lights are currently on.")  # noqa: T201
+            return
+
+        print(f"Turning off {len(entity_ids)} light(s)...")  # noqa: T201
+        for entity_id in entity_ids:
+            client.trigger_service("light", "turn_off", entity_id=entity_id)
+            print(f"  off: {entity_id}")  # noqa: T201
+
+
+if __name__ == "__main__":
+    main()

examples/websocket_state_monitor.py

@@ -0,0 +1,47 @@
+"""Stream live state_changed events for entities in a given domain.
+
+Usage:
+    python websocket_state_monitor.py [domain]
+
+    domain defaults to "light". Press Ctrl+C to stop.
+
+Environment variables:
+    HOMEASSISTANT_WS_ENDPOINT  e.g. ws://localhost:8123/api/websocket
+    HOMEASSISTANT_API_TOKEN    Long-lived access token
+"""
+
+import os
+import sys
+
+from homeassistant_api import WebsocketClient
+from homeassistant_api.models.websocket import FiredEvent
+
+url = os.environ["HOMEASSISTANT_WS_ENDPOINT"]
+token = os.environ["HOMEASSISTANT_API_TOKEN"]
+
+
+def monitor_domain(domain: str) -> None:
+    with WebsocketClient(url, token) as ws:
+        print(f"Monitoring '{domain}' state changes. Press Ctrl+C to stop.")  # noqa: T201
+        with ws.listen_events("state_changed") as events:
+            for event in events:
+                if not isinstance(event, FiredEvent):
+                    continue
+                entity_id: str = event.data.get("entity_id", "")
+                if not entity_id.startswith(f"{domain}."):
+                    continue
+                old_state = (event.data.get("old_state") or {}).get(
+                    "state",
+                    "unavailable",
+                )
+                new_state = (event.data.get("new_state") or {}).get(
+                    "state",
+                    "unavailable",
+                )
+                timestamp = event.time_fired.strftime("%H:%M:%S")
+                print(f"[{timestamp}] {entity_id}: {old_state} → {new_state}")  # noqa: T201
+
+
+if __name__ == "__main__":
+    domain = sys.argv[1] if len(sys.argv) > 1 else "light"
+    monitor_domain(domain)