synthetic_home.synthetic_home

  1"""Data model for home assistant synthetic home."""
  2
  3from dataclasses import dataclass, field
  4import pathlib
  5import logging
  6import slugify
  7from typing import Any
  8
  9from mashumaro.codecs.yaml import yaml_decode
 10
 11from synthetic_home.exceptions import SyntheticHomeError
 12from synthetic_home.device_types import (
 13    DeviceTypeRegistry,
 14    DeviceStateStrategy,
 15    DeviceState,
 16    EntityEntry,
 17    merge_entity_state_attributes,
 18)
 19from . import common, inventory
 20from .inventory import DEFAULT_SEPARATOR
 21from .device_types import load_device_type_registry
 22
 23__all__ = [
 24    "SyntheticHome",
 25    "Device",
 26    "build_device_state",
 27    "load_synthetic_home",
 28    "read_config_content",
 29]
 30
 31
 32_LOGGER = logging.getLogger(__name__)
 33
 34
 35@dataclass
 36class Device:
 37    """A synthetic device."""
 38
 39    name: str
 40    """A human readable name for the device."""
 41
 42    device_type: str | None = None
 43    """The type of the device in the device registry that determines how it maps to entities."""
 44
 45    device_info: common.DeviceInfo | None = None
 46    """Device make and model information."""
 47
 48    device_state: str | dict | DeviceState | None = None
 49    """A list of pre-canned RestorableStateAttributes specified by the key.
 50
 51    These are used for restoring a device into a specific state supported by the
 52    device type. This is used to use a label rather than specifying low level
 53    entity details. This is an alternative to specifying low level attributes above.
 54
 55    Restorable attributes overwrite normal attributes since they can be reloaded
 56    at runtime.
 57    """
 58
 59    entity_entries: dict[str, list[EntityEntry]] = field(default_factory=dict)
 60    """The validated set of entity entries"""
 61
 62    def merge(
 63        self,
 64        device_state: DeviceState | None = None,
 65        entity_entries: dict[str, list[EntityEntry]] | None = None,
 66    ) -> "Device":
 67        """Merge the existing device with a new device state."""
 68        return Device(
 69            name=self.name,
 70            device_type=self.device_type,
 71            device_info=self.device_info,
 72            device_state=device_state or self.device_state,
 73            entity_entries=entity_entries or self.entity_entries,
 74        )
 75
 76
 77def build_device_state(device: Device, registry: DeviceTypeRegistry) -> Device:
 78    """Validate the device and return a new instance."""
 79    if (device_type := registry.device_types.get(device.device_type or "")) is None:
 80        raise SyntheticHomeError(
 81            f"Device {device} has device_type {device.device_type} not found in: {registry.device_types}"
 82        )
 83
 84    if device.device_state is not None:
 85        # Lookup a device state and merge it into the current state
 86        if (
 87            isinstance(device.device_state, str)
 88            and device.device_state not in device_type.device_states_dict
 89        ):
 90            raise SyntheticHomeError(
 91                f"Device {device}\nhas state '{device.device_state}'\n not in: {device_type.device_states_dict}"
 92            )
 93        if isinstance(device.device_state, dict):
 94            _LOGGER.debug(
 95                "Parsing device state from dictionary: %s", device.device_state
 96            )
 97            strategy = DeviceStateStrategy()
 98            device = device.merge(
 99                device_state=strategy.deserialize(("custom", device.device_state))
100            )
101            _LOGGER.debug("Parsed=%s", device)
102
103    device_state: DeviceState | None = None
104    _LOGGER.debug("Checking device state: %s", device.device_state)
105    if (
106        device.device_state is None or isinstance(device.device_state, DeviceState)
107    ) and device_type.device_states:
108        # Pick the first device state as the default
109        device_state = device_type.device_states[0]
110        if isinstance(device.device_state, DeviceState):
111            device_state = device_state.merge(device.device_state)
112    elif device.device_state is not None and isinstance(device.device_state, str):
113        device_state = device_type.device_states_dict[device.device_state]
114    else:
115        raise SyntheticHomeError(f"Device did not declare a device state: {device}")
116
117    _LOGGER.debug("Merging entity attributes for device state: %s", device_state)
118    entity_entries = {
119        platform: [
120            merge_entity_state_attributes(
121                platform, entity_entry, device_state.entity_states
122            )
123            for entity_entry in entity_entries
124        ]
125        for platform, entity_entries in device_type.entities.items()
126    }
127    return device.merge(device_state=device_state, entity_entries=entity_entries)
128
129
130@dataclass
131class SyntheticHome:
132    """Data about a synthetic home."""
133
134    name: str
135    """A human readable name for the home."""
136
137    # Devices by area
138    devices: dict[str, list[Device]] = field(default_factory=dict)
139
140    # Services for the home not for a specific area
141    services: list[Device] = field(default_factory=list)
142
143    # Device types supported by the home.
144    device_type_registry: DeviceTypeRegistry | None = None
145
146    def __post_init__(self) -> None:
147        """Build the complete device state."""
148        if self.device_type_registry is None:
149            self.device_type_registry = load_device_type_registry()
150        self.devices = {
151            key: [
152                build_device_state(device, self.device_type_registry)
153                for device in devices
154            ]
155            for key, devices in self.devices.items()
156        }
157        self.services = [
158            build_device_state(device, self.device_type_registry)
159            for device in self.services
160        ]
161
162
163def read_config_content(config_file: pathlib.Path) -> str:
164    """Create configuration file content, exposed for patching."""
165    with config_file.open("r") as f:
166        return f.read()
167
168
169def load_synthetic_home(config_file: pathlib.Path) -> SyntheticHome:
170    """Load synthetic home configuration from disk."""
171    try:
172        content = read_config_content(config_file)
173    except FileNotFoundError:
174        raise SyntheticHomeError(f"Configuration file '{config_file}' does not exist")
175    try:
176        return yaml_decode(content, SyntheticHome)
177    except ValueError as err:
178        raise SyntheticHomeError(f"Could not parse config file '{config_file}': {err}")
179
180
181def yaml_state_value(v: Any) -> Any:
182    """Convert a entity state value to yaml."""
183    if isinstance(v, bool) or isinstance(v, float) or isinstance(v, list):
184        return v
185    return str(v)
186
187
188def build_entities(area_id: str | None, device_entry: Device) -> list[inventory.Entity]:
189    """Build the set of entities for the device entry."""
190    entities = []
191    device_name = device_entry.name.replace("_", " ").title()
192    device_id = slugify.slugify(device_entry.name, separator=DEFAULT_SEPARATOR)
193
194    for platform, entity_entries in device_entry.entity_entries.items():
195        for entity_entry in entity_entries:
196            # Each entity in this platform needs a unique name, but
197            # if the key is in the name it's the primary to avoid "Motion motion"
198            entity_name = device_name
199            if platform == "sensor" or (
200                platform == "binary_sensor"
201                and entity_entry.key not in device_entry.name.lower()
202            ):
203                entity_name = f"{device_name} {entity_entry.key.capitalize()}"
204            entity_id = f"{platform}.{slugify.slugify(entity_name, separator=DEFAULT_SEPARATOR)}"
205            entity = inventory.Entity(
206                name=entity_name,
207                id=entity_id,
208                device=device_id,
209            )
210            if area_id:
211                entity.area = area_id
212            attributes: dict[str, str | list[str] | int | float] = {}
213            if entity_entry.attributes:
214                attributes.update(entity_entry.attributes)
215            state = attributes.pop("state", None)
216            if state is not None:
217                entity.state = yaml_state_value(state)
218            if attributes:
219                entity.attributes = attributes
220            entities.append(entity)
221    return entities
222
223
224def build_inventory(home: SyntheticHome) -> inventory.Inventory:
225    """Build a home inventory from the synthetic home definition.
226
227    This is a flattened set of areas, entities, and devices.
228    """
229
230    inv = inventory.Inventory()
231    pairs: list[tuple[str | None, list[Device]]] = [*home.devices.items()]
232    if home.services:
233        pairs.append((None, home.services))
234
235    device_ids: set[str] = set()
236    entities = []
237    for area_name, devices in pairs:
238        if area_name:
239            area_id = slugify.slugify(area_name, separator=DEFAULT_SEPARATOR)
240            inv.areas.append(inventory.Area(name=area_name, id=area_id))
241        else:
242            area_id = None
243
244        for device_entry in devices:
245            # Make computer generated device names more friendly
246            device_name = device_entry.name.replace("_", " ").title()
247            device_id = slugify.slugify(device_entry.name, separator=DEFAULT_SEPARATOR)
248            if device_id in device_ids:
249                device_entry.name = f"{area_name}_{device_entry.name}"
250                device_name = device_entry.name.replace("_", " ").title()
251                device_id = slugify.slugify(device_name, separator=DEFAULT_SEPARATOR)
252            device_ids.add(device_id)
253            device = inventory.Device(
254                name=device_name,
255                id=device_id,
256                info=device_entry.device_info,
257            )
258            if area_id:
259                device.area = area_id
260            inv.devices.append(device)
261            entities.extend(build_entities(area_id, device_entry))
262    if entities:
263        inv.entities = entities
264    return inv