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    def __init__(self) -> None:
 79        """Counter."""
 80        self._value = 0
 81
 82    def increment(self) -> int:
 83        """Return next value."""
 84        self._value += 1
 85        return self._value
 86
 87
 88next_id = Counter()
 89
 90
 91async def auth_login(ws: aiohttp.ClientWebSocketResponse, auth_token: str) -> None:
 92    """Login to the websocket."""
 93    auth_resp = await ws.receive_json()
 94    assert auth_resp["type"] == "auth_required"
 95
 96    await ws.send_json({"type": "auth", "access_token": auth_token})
 97
 98    auth_ok = await ws.receive_json()
 99    assert auth_ok["type"] == "auth_ok"
100
101
102async def fetch_config(ws: aiohttp.ClientWebSocketResponse) -> dict[str, Any]:
103    """Fetch areas from websocket."""
104    await ws.send_json({"id": next_id.increment(), "type": GET_CONFIG})
105    data = await ws.receive_json()
106    _LOGGER.info(data["result"])
107    assert data["result"]
108    return data["result"]  # type: ignore[no-any-return]
109
110
111async def fetch_areas(ws: aiohttp.ClientWebSocketResponse) -> dict[str, inventory.Area]:
112    """Fetch areas from websocket."""
113    await ws.send_json({"id": next_id.increment(), "type": AREA_REGISTRY_LIST})
114    data = await ws.receive_json()
115    areas = {
116        area["area_id"]: inventory.Area(
117            id=slugify.slugify(area["name"], separator="_"),
118            name=area["name"],
119            floor=area["floor_id"],
120        )
121        for area in data["result"]
122    }
123    return areas
124
125
126async def fetch_devices(
127    ws: aiohttp.ClientWebSocketResponse, areas: dict[str, inventory.Area]
128) -> dict[str, inventory.Device]:
129    """Fetch areas from websocket."""
130
131    await ws.send_json({"id": next_id.increment(), "type": DEVICE_REGISTRY_LIST})
132    data = await ws.receive_json()
133
134    devices = {}
135    for device in data["result"]:
136        if device.get("disabled_by") is not None:
137            continue
138        inv_device = inventory.Device(
139            name=device["name"],
140            id=slugify.slugify(device["name"], separator="_"),
141        )
142        if any(device.get(n) for n in ("model", "manufacturer", "sw_version")):
143            device_info = common.DeviceInfo()
144            if model := device.get("model"):
145                device_info.model = model
146            if manufacturer := device.get("manufacturer"):
147                device_info.manufacturer = manufacturer
148            if sw_version := device.get("sw_version"):
149                device_info.model = sw_version
150            inv_device.info = device_info
151        if area_id := device.get("area_id"):
152            inv_device.area = areas[area_id].id
153        devices[device["id"]] = inv_device
154    return devices
155
156
157async def fetch_entities(
158    ws: aiohttp.ClientWebSocketResponse,
159    areas: dict[str, inventory.Area],
160    devices: dict[str, inventory.Device],
161) -> dict[str, inventory.Entity]:
162    """Fetch devices from websocket."""
163    await ws.send_json({"id": next_id.increment(), "type": ENTITY_REGISTRY_LIST})
164    data = await ws.receive_json()
165
166    entities = {}
167    for entity in data["result"]:
168        if entity.get("disabled_by") is not None:
169            continue
170        if entity.get("hidden_by") is not None:
171            continue
172        if entity.get("entity_category") in ("diagnostic", "config"):
173            continue
174        entity_id = entity["entity_id"]
175        domain = entity_id.split(".", maxsplit=1)[0]
176        if domain not in DOMAINS:
177            _LOGGER.debug(
178                "Skipping entity with unsupported domain %s (%s)",
179                entity_id,
180                domain,
181            )
182            continue
183        inv_entity = inventory.Entity(
184            name=entity["name"],
185            id=entity_id,
186        )
187        if area_id := entity.get("area_id"):
188            inv_entity.area = areas[area_id].id
189        if device_id := entity.get("device_id"):
190            inv_entity.device = devices[device_id].id
191        entities[entity_id] = inv_entity
192    return entities
193
194
195async def build_states(
196    ws: aiohttp.ClientWebSocketResponse,
197    entities: dict[str, inventory.Entity],
198    unit_system: dict[str, str],
199) -> list[inventory.Entity]:
200    """Fetch states from websocket and update inventory."""
201
202    await ws.send_json({"id": next_id.increment(), "type": GET_STATES})
203    data = await ws.receive_json()
204
205    temperature_unit = unit_system["temperature"]
206    results = []
207    for state in data["result"]:
208        entity_id = state["entity_id"]
209        if (inv_entity := entities.get(entity_id)) is None:
210            continue
211        entity_state = state["state"]
212        if entity_state in ("unavailable", "unknown"):
213            continue
214        inv_entity.state = entity_state
215        if (attributes := state.get("attributes")) is not None:
216            if friendly_name := attributes.get("friendly_name"):
217                inv_entity.name = friendly_name.strip()
218            inv_attributes = {
219                k: v
220                for k, v in attributes.items()
221                if (v is not None) and (k not in STRIP_ATTRIBUTES)
222            }
223
224            if entity_id.startswith("climate."):
225                inv_attributes["unit_of_measurement"] = temperature_unit
226            if inv_attributes:
227                inv_entity.attributes = inv_attributes
228        results.append(inv_entity)
229    return results
230
231
232async def run(args: argparse.Namespace) -> int:
233    url = args.homeassistant_url
234    auth_token = args.auth_token
235
236    async with aiohttp.ClientSession() as session:
237        area_url = f"{url}/api/websocket"
238        _LOGGER.info("Fetching areas from %s", url)
239        async with session.ws_connect(area_url) as ws:
240            await auth_login(ws, auth_token)
241
242            config = await fetch_config(ws)
243            unit_system = config["unit_system"]
244            areas = await fetch_areas(ws)
245            devices = await fetch_devices(ws, areas)
246            entities = await fetch_entities(ws, areas, devices)
247
248            required_device_ids = set({entity.device for entity in entities.values()})
249
250            # Only include required devices for entities
251            inv = inventory.Inventory()
252
253            # Fetch state for all relevant entities
254            inv.entities = await build_states(ws, entities, unit_system)
255            inv.areas = list(areas.values())
256            inv.devices = [
257                device
258                for device in devices.values()
259                if device.id in required_device_ids
260            ]
261
262            print(inv.yaml())
263
264    return 0