synthetic_home.tool.export_inventory

  1"""Export an inventory from a Home Assistant instance.
  2
  3You can create a synthetic home inventory copied from an existing home assistant
  4instance. You need to create an access token and export an inventory like this:
  5
  6```bash
  7$ HASS_URL="http://home-assistant.home-assistant:8123"
  8$ API_TOKEN="XXXXXXXXXXX"
  9$ synthetic-home --debug export_inventory "${HASS_URL}" "${API_TOKEN}" > inventory.yaml
 10```
 11"""
 12
 13import argparse
 14import logging
 15import slugify
 16from typing import Any
 17
 18import aiohttp
 19
 20from synthetic_home import inventory, common
 21
 22_LOGGER = logging.getLogger(__name__)
 23
 24AREA_REGISTRY_LIST = "config/area_registry/list"
 25DEVICE_REGISTRY_LIST = "config/device_registry/list"
 26ENTITY_REGISTRY_LIST = "config/entity_registry/list"
 27GET_STATES = "get_states"
 28GET_CONFIG = "get_config"
 29
 30DOMAINS = {
 31    "binary_sensor",
 32    # TODO: Support calendar
 33    # "calendar",
 34    "camera",
 35    "climate",
 36    "cover",
 37    "fan",
 38    # TODO: Support event
 39    # "event",
 40    "light",
 41    "lock",
 42    "media_player",
 43    # TODO: Support number, person, remote, select
 44    # "number",
 45    # "person",
 46    # "remote",
 47    # "select",
 48    "sensor",
 49    "switch",
 50    "todo",
 51    "vacuum",
 52    "weather",
 53    # TODO: Support zone
 54    # "zone",
 55}
 56
 57STRIP_ATTRIBUTES = {
 58    "friendly_name",
 59    "icon",
 60}
 61
 62
 63def create_arguments(args: argparse.ArgumentParser) -> None:
 64    """Get parsed passed in arguments."""
 65    args.add_argument(
 66        "homeassistant_url",
 67        type=str,
 68        help="Specifies url to the home assistant instance.",
 69    )
 70    args.add_argument(
 71        "auth_token",
 72        type=str,
 73        help="Specifies home assistant API token.",
 74    )
 75
 76
 77class Counter:
 78
 79    def __init__(self) -> None:
 80        """Counter."""
 81        self._value = 0
 82
 83    def increment(self) -> int:
 84        """Return next value."""
 85        self._value += 1
 86        return self._value
 87
 88
 89next_id = Counter()
 90
 91
 92async def auth_login(ws: aiohttp.ClientWebSocketResponse, auth_token: str) -> None:
 93    """Login to the websocket."""
 94    auth_resp = await ws.receive_json()
 95    assert auth_resp["type"] == "auth_required"
 96
 97    await ws.send_json({"type": "auth", "access_token": auth_token})
 98
 99    auth_ok = await ws.receive_json()
100    assert auth_ok["type"] == "auth_ok"
101
102
103async def fetch_config(ws: aiohttp.ClientWebSocketResponse) -> dict[str, Any]:
104    """Fetch areas from websocket."""
105    await ws.send_json({"id": next_id.increment(), "type": GET_CONFIG})
106    data = await ws.receive_json()
107    _LOGGER.info(data["result"])
108    assert data["result"]
109    return data["result"]  # type: ignore[no-any-return]
110
111
112async def fetch_areas(ws: aiohttp.ClientWebSocketResponse) -> dict[str, inventory.Area]:
113    """Fetch areas from websocket."""
114    await ws.send_json({"id": next_id.increment(), "type": AREA_REGISTRY_LIST})
115    data = await ws.receive_json()
116    areas = {
117        area["area_id"]: inventory.Area(
118            id=slugify.slugify(area["name"], separator="_"),
119            name=area["name"],
120            floor=area["floor_id"],
121        )
122        for area in data["result"]
123    }
124    return areas
125
126
127async def fetch_devices(
128    ws: aiohttp.ClientWebSocketResponse, areas: dict[str, inventory.Area]
129) -> dict[str, inventory.Device]:
130    """Fetch areas from websocket."""
131
132    await ws.send_json({"id": next_id.increment(), "type": DEVICE_REGISTRY_LIST})
133    data = await ws.receive_json()
134
135    devices = {}
136    for device in data["result"]:
137        if device.get("disabled_by") is not None:
138            continue
139        inv_device = inventory.Device(
140            name=device["name"],
141            id=slugify.slugify(device["name"], separator="_"),
142        )
143        if any(device.get(n) for n in ("model", "manufacturer", "sw_version")):
144            device_info = common.DeviceInfo()
145            if model := device.get("model"):
146                device_info.model = model
147            if manufacturer := device.get("manufacturer"):
148                device_info.manufacturer = manufacturer
149            if sw_version := device.get("sw_version"):
150                device_info.model = sw_version
151            inv_device.info = device_info
152        if area_id := device.get("area_id"):
153            inv_device.area = areas[area_id].id
154        devices[device["id"]] = inv_device
155    return devices
156
157
158async def fetch_entities(
159    ws: aiohttp.ClientWebSocketResponse,
160    areas: dict[str, inventory.Area],
161    devices: dict[str, inventory.Device],
162) -> dict[str, inventory.Entity]:
163    """Fetch devices from websocket."""
164    await ws.send_json({"id": next_id.increment(), "type": ENTITY_REGISTRY_LIST})
165    data = await ws.receive_json()
166
167    entities = {}
168    for entity in data["result"]:
169        if entity.get("disabled_by") is not None:
170            continue
171        if entity.get("hidden_by") is not None:
172            continue
173        if entity.get("entity_category") in ("diagnostic", "config"):
174            continue
175        entity_id = entity["entity_id"]
176        domain = entity_id.split(".", maxsplit=1)[0]
177        if domain not in DOMAINS:
178            _LOGGER.debug(
179                "Skipping entity with unsupported domain %s (%s)",
180                entity_id,
181                domain,
182            )
183            continue
184        inv_entity = inventory.Entity(
185            name=entity["name"],
186            id=entity_id,
187        )
188        if area_id := entity.get("area_id"):
189            inv_entity.area = areas[area_id].id
190        if device_id := entity.get("device_id"):
191            inv_entity.device = devices[device_id].id
192        entities[entity_id] = inv_entity
193    return entities
194
195
196async def build_states(
197    ws: aiohttp.ClientWebSocketResponse,
198    entities: dict[str, inventory.Entity],
199    unit_system: dict[str, str],
200) -> list[inventory.Entity]:
201    """Fetch states from websocket and update inventory."""
202
203    await ws.send_json({"id": next_id.increment(), "type": GET_STATES})
204    data = await ws.receive_json()
205
206    temperature_unit = unit_system["temperature"]
207    results = []
208    for state in data["result"]:
209        entity_id = state["entity_id"]
210        if (inv_entity := entities.get(entity_id)) is None:
211            continue
212        entity_state = state["state"]
213        if entity_state in ("unavailable", "unknown"):
214            continue
215        inv_entity.state = entity_state
216        if (attributes := state.get("attributes")) is not None:
217            if friendly_name := attributes.get("friendly_name"):
218                inv_entity.name = friendly_name.strip()
219            inv_attributes = {
220                k: v
221                for k, v in attributes.items()
222                if (v is not None) and (k not in STRIP_ATTRIBUTES)
223            }
224
225            if entity_id.startswith("climate."):
226                inv_attributes["unit_of_measurement"] = temperature_unit
227            if inv_attributes:
228                inv_entity.attributes = inv_attributes
229        results.append(inv_entity)
230    return results
231
232
233async def run(args: argparse.Namespace) -> int:
234    url = args.homeassistant_url
235    auth_token = args.auth_token
236
237    async with aiohttp.ClientSession() as session:
238        area_url = f"{url}/api/websocket"
239        _LOGGER.info("Fetching areas from %s", url)
240        async with session.ws_connect(area_url) as ws:
241            await auth_login(ws, auth_token)
242
243            config = await fetch_config(ws)
244            unit_system = config["unit_system"]
245            areas = await fetch_areas(ws)
246            devices = await fetch_devices(ws, areas)
247            entities = await fetch_entities(ws, areas, devices)
248
249            required_device_ids = set({entity.device for entity in entities.values()})
250
251            # Only include required devices for entities
252            inv = inventory.Inventory()
253
254            # Fetch state for all relevant entities
255            inv.entities = await build_states(ws, entities, unit_system)
256            inv.areas = list(areas.values())
257            inv.devices = [
258                device
259                for device in devices.values()
260                if device.id in required_device_ids
261            ]
262
263            print(inv.yaml())
264
265    return 0