synthetic_home.synthetic_home

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