admin/wiki
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto-sync: 2026-04-15 14:40:01

Browse Source
This commit is contained in:
Stream
2026-04-15 14:40:01 +03:00
parent 31661cd3de
commit f685f73512
1 changed files with 79 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
tasks/ha-availability-dashboard/appdaemon/availability.py
View File

@@ -102,7 +102,6 @@ class Availability(hass.Hass):
"""Fetch area registry via WebSocket and build entity_id -> area mapping.
HA registry APIs are only available via WebSocket, not REST.
We use a synchronous websocket-client to fetch the data.
"""
try:
import websocket
@@ -112,83 +111,102 @@ class Availability(hass.Hass):
self._areas_cache = {}
return {}
token = self._get_token()
# Use HA long-lived access token for WebSocket auth
token = self.args.get("ha_token", "")
if not token:
token = self._get_token()
if not token:
self._areas_cache = {}
return {}
# Connect to HA WebSocket
# From addon, use homeassistant:8123 (not supervisor proxy)
ws_url = "ws://homeassistant:8123/api/websocket"
try:
ws = websocket.create_connection(ws_url, timeout=10,
header=["Authorization: Bearer " + token])
ws = websocket.create_connection(ws_url, timeout=15)
except Exception as e:
self.log(f"WebSocket connect failed: {e}", level="WARNING")
self._areas_cache = {}
return {}
def _ws_call(msg_type):
"""Send a WebSocket command and return the result."""
# Auth
result = _json.loads(ws.recv())
if result.get("type") == "auth_required":
try:
# Step 1: Auth handshake
auth_msg = _json.loads(ws.recv())
if auth_msg.get("type") == "auth_required":
ws.send(_json.dumps({"type": "auth", "access_token": token}))
result = _json.loads(ws.recv())
if result.get("type") != "auth_ok":
return None
# Send command
ws.send(_json.dumps({"id": msg_type.__hash__() % 10000, "type": msg_type}))
result = _json.loads(ws.recv())
return result.get("result") if result.get("success") else None
auth_result = _json.loads(ws.recv())
if auth_result.get("type") != "auth_ok":
self.log(f"WS auth failed: {auth_result.get('message','')}", level="ERROR")
ws.close()
self._areas_cache = {}
return {}
self.log("WS auth OK")
# Step 2: Fetch area registry
ws.send(_json.dumps({"id": 1, "type": "config/area_registry/list"}))
areas_result = _json.loads(ws.recv())
if not areas_result.get("success"):
self.log(f"Area registry failed: {areas_result.get('error','?')}", level="WARNING")
ws.close()
self._areas_cache = {}
return {}
areas_data = areas_result.get("result", [])
# Build area_id -> area_name map
area_map = {}
if isinstance(areas_data, list):
for area in areas_data:
area_map[area["area_id"]] = area.get("name", "Без комнаты")
self.log(f"WS: {len(area_map)} areas fetched")
# Step 3: Fetch device registry
ws.send(_json.dumps({"id": 2, "type": "config/device_registry/list"}))
dev_result = _json.loads(ws.recv())
dev_area = {}
if dev_result.get("success"):
dev_data = dev_result.get("result", [])
if isinstance(dev_data, list):
for dev in dev_data:
dev_id = dev.get("id")
a_id = dev.get("area_id")
if dev_id and a_id and a_id in area_map:
dev_area[dev_id] = area_map[a_id]
self.log(f"WS: {len(dev_area)} device-area mappings")
# Step 4: Fetch entity registry
ws.send(_json.dumps({"id": 3, "type": "config/entity_registry/list"}))
ent_result = _json.loads(ws.recv())
eid_to_area = {}
if ent_result.get("success"):
ent_data = ent_result.get("result", [])
# HA returns either a list directly or a dict with 'entities' key
if isinstance(ent_data, dict):
ent_list = ent_data.get("entities", [])
else:
ent_list = ent_data
for entry in ent_list:
eid = entry.get("entity_id", "")
a_id = entry.get("area_id")
dev_id = entry.get("device_id")
if a_id and a_id in area_map:
eid_to_area[eid] = area_map[a_id]
elif dev_id and dev_id in dev_area:
eid_to_area[eid] = dev_area[dev_id]
else:
eid_to_area[eid] = "Без комнаты"
self.log(f"WS: {len(eid_to_area)} entity-area mappings")
# Fetch area registry
areas_data = _ws_call("config/area_registry/list")
if not areas_data:
self.log("Failed to fetch area registry via WebSocket", level="WARNING")
ws.close()
self._areas_cache = eid_to_area
return eid_to_area
except Exception as e:
self.log(f"WebSocket area fetch error: {e}", level="WARNING")
try:
ws.close()
except:
pass
self._areas_cache = {}
return {}
# Build area_id -> area_name map
area_map = {}
for area in areas_data:
area_map[area["area_id"]] = area.get("name", "Без комнаты")
# Fetch device registry
dev_data = _ws_call("config/device_registry/list")
dev_area = {}
if dev_data and isinstance(dev_data, dict):
dev_data = dev_data.get("devices", dev_data) if isinstance(dev_data, dict) else dev_data
if dev_data and isinstance(dev_data, list):
for dev in dev_data:
dev_id = dev.get("id")
a_id = dev.get("area_id")
if dev_id and a_id and a_id in area_map:
dev_area[dev_id] = area_map[a_id]
# Fetch entity registry
ent_data = _ws_call("config/entity_registry/list")
eid_to_area = {}
if ent_data and isinstance(ent_data, dict):
ent_data = ent_data.get("entities", ent_data)
if ent_data and isinstance(ent_data, list):
for entry in ent_data:
eid = entry.get("entity_id", "")
a_id = entry.get("area_id")
dev_id = entry.get("device_id")
if a_id and a_id in area_map:
eid_to_area[eid] = area_map[a_id]
elif dev_id and dev_id in dev_area:
eid_to_area[eid] = dev_area[dev_id]
else:
eid_to_area[eid] = "Без комнаты"
ws.close()
self._areas_cache = eid_to_area
self.log(f"Fetched {len(area_map)} areas, {len(dev_area)} device-areas, {len(eid_to_area)} entity-areas")
return eid_to_area
# ── History Fetch ──
def _fetch_history(self, entity_ids: list, period_start: datetime, period_end: datetime):