1#!/usr/bin/env -S uv run
  2# /// script
  3# requires-python = ">=3.11"
  4# dependencies = [
  5#     "websockets",
  6# ]
  7# ///
  8"""
  9Get Home Assistant entities via WebSocket API.
 10Usage: ./get-ha-entities.py <token>
 11"""
 12
 13import asyncio
 14import websockets
 15import json
 16import sys
 17from collections import defaultdict
 18
 19HA_WS = "ws://hass.home:8123/api/websocket"
 20
 21
 22async def get_entities(token):
 23    try:
 24        async with websockets.connect(HA_WS) as ws:
 25            # Receive auth_required
 26            auth_req = json.loads(await ws.recv())
 27            version = auth_req.get('ha_version', 'unknown')
 28            print(f"Connected to Home Assistant {version}")
 29
 30            # Send authentication
 31            await ws.send(json.dumps({
 32                "type": "auth",
 33                "access_token": token
 34            }))
 35
 36            # Receive auth response
 37            auth_resp = json.loads(await ws.recv())
 38
 39            if auth_resp["type"] != "auth_ok":
 40                msg = auth_resp.get('message', 'Unknown error')
 41                print(f"❌ Authentication failed: {msg}")
 42                return
 43
 44            print("✅ Authenticated successfully!\n")
 45
 46            # Get all states
 47            await ws.send(json.dumps({
 48                "id": 1,
 49                "type": "get_states"
 50            }))
 51
 52            response = json.loads(await ws.recv())
 53
 54            if not response.get("success"):
 55                print(f"❌ Error getting states: {response}")
 56                return
 57
 58            states = response["result"]
 59
 60            # Group by domain
 61            by_domain = defaultdict(list)
 62            for state in states:
 63                domain = state["entity_id"].split(".")[0]
 64                by_domain[domain].append(state)
 65
 66            # Display summary
 67            print(f"📊 Total entities: {len(states)}")
 68            print(f"📁 Domains: {len(by_domain)}\n")
 69
 70            # Show each domain
 71            for domain in sorted(by_domain.keys()):
 72                entities = by_domain[domain]
 73                print(f"=== {domain} ({len(entities)} entities) ===")
 74
 75                for entity in sorted(entities, key=lambda x: x["entity_id"]):
 76                    entity_id = entity["entity_id"]
 77                    state = entity["state"]
 78                    attrs = entity.get("attributes", {})
 79                    friendly_name = attrs.get("friendly_name", entity_id)
 80                    unit = attrs.get("unit_of_measurement", "")
 81
 82                    state_display = f"{state} {unit}".strip()
 83                    name = friendly_name
 84                    print(f"  • {name} ({entity_id}): {state_display}")
 85
 86                print()
 87
 88    except websockets.exceptions.WebSocketException as e:
 89        print(f"❌ WebSocket error: {e}")
 90        print(f"   Make sure Home Assistant is accessible at {HA_WS}")
 91    except Exception as e:
 92        print(f"❌ Error: {e}")
 93
 94
 95def main():
 96    if len(sys.argv) < 2:
 97        print("Usage: ./get-ha-entities.py <long-lived-access-token>")
 98        print("")
 99        print("Create a token in Home Assistant:")
100        print("  Profile → Long-Lived Access Tokens → Create Token")
101        sys.exit(1)
102
103    token = sys.argv[1]
104    asyncio.run(get_entities(token))
105
106
107if __name__ == "__main__":
108    main()