synthetic_home.device_types

  1"""Data model for device type definitions.
  2
  3These device devices and how they map to a set of entities. States can be set
  4on a device then translated into how that state affects entities.
  5"""
  6
  7from functools import cache
  8from collections.abc import Generator
  9from dataclasses import dataclass, field
 10from importlib import resources
 11from importlib.resources.abc import Traversable
 12import logging
 13from typing import Any
 14import yaml
 15
 16from mashumaro.codecs.yaml import yaml_decode
 17from mashumaro.exceptions import MissingField
 18from mashumaro import field_options, DataClassDictMixin
 19from mashumaro.types import SerializationStrategy
 20
 21from .exceptions import SyntheticHomeError
 22
 23__all__ = [
 24    "DeviceTypeRegistry",
 25    "DeviceType",
 26    "DeviceState",
 27    "EntityState",
 28    "EntityEntry",
 29    "load_device_type_registry",
 30]
 31
 32
 33_LOGGER = logging.getLogger(__name__)
 34
 35
 36DEVICE_TYPES_RESOURCE_PATH = resources.files().joinpath("registry")
 37
 38
 39class KeyedObjectListStrategy(SerializationStrategy):
 40    """A predefined entity state parser."""
 41
 42    def __init__(self, object_strategy: SerializationStrategy) -> None:
 43        """Initialize KeyedObjectStrategy."""
 44        self._object_strategy = object_strategy
 45
 46    def deserialize(self, value: Any) -> list[Any]:
 47        """Deserialize the object."""
 48        if not value:
 49            return []
 50        if isinstance(value, dict):
 51            return [
 52                self._object_strategy.deserialize((key, state))
 53                for key, state in value.items()
 54            ]
 55        raise ValueError(f"Expected 'dict' representing the object list, got: {value}")
 56
 57
 58@dataclass
 59class EntityEntry(DataClassDictMixin):
 60    """Defines an entity type.
 61
 62    An entity is the lowest level object that maps to a behavior or trait of a device.
 63    """
 64
 65    key: str
 66    """The entity description key"""
 67
 68    attributes: dict[str, str | list[str]] = field(default_factory=dict)
 69    """Attributes supported by this entity."""
 70
 71
 72class EntityDictStrategy(SerializationStrategy):
 73    """A parser of the entity entry dict."""
 74
 75    def deserialize(
 76        self, value: dict[str, dict[str, dict[str, str | list[str]]]]
 77    ) -> dict[str, list[EntityEntry]]:
 78        """Deserialize the object."""
 79        return {
 80            domain: [
 81                EntityEntry(key=key, attributes=attributes)
 82                for key, attributes in values.items()
 83            ]
 84            for domain, values in value.items()
 85        }
 86
 87
 88@dataclass
 89class EntityState(DataClassDictMixin):
 90    """Represents a pre-defined entity state."""
 91
 92    domain: str
 93    """The domain of the entity."""
 94
 95    key: str
 96    """The entity key identifying the entity."""
 97
 98    state: str | dict[str, Any]
 99    """The values that make up the entity state."""
100
101    @property
102    def domain_key(self) -> str:
103        """Return the full domain.key."""
104        return f"{self.domain}.{self.key}"
105
106    def merge(self, new: "EntityState") -> "EntityState":
107        """Merge with an additional value."""
108        merged_state: str | bool | dict[str, Any]
109        if isinstance(new.state, dict):
110            if isinstance(self.state, dict):
111                merged_state = {
112                    **self.state,
113                    **new.state,
114                }
115            else:
116                merged_state = {"state": self.state, **new.state}
117        else:
118            if isinstance(self.state, dict):
119                merged_state = {
120                    **self.state,
121                    "state": new.state,
122                }
123            else:
124                merged_state = new.state
125        return EntityState(
126            domain=self.domain,
127            key=self.key,
128            state=merged_state,
129        )
130
131
132def merge_entity_state_attributes(
133    platform: str, entry: EntityEntry, entity_states: list[EntityState]
134) -> EntityEntry:
135    """Merge state values from an EntityEntry into a new EntityEntry."""
136    for state in entity_states:
137        if state.domain == platform and state.key == entry.key:
138            if isinstance(state.state, dict):
139                extras = state.state
140            else:
141                extras = {"state": state.state}
142            return EntityEntry(
143                key=entry.key,
144                attributes={
145                    **entry.attributes,
146                    **extras,
147                },
148            )
149    _LOGGER.debug("No state to merge: %s", entry)
150    return entry
151
152
153class EntityStateStrategy(SerializationStrategy):
154    """A predefined entity state parser."""
155
156    def deserialize(self, value: tuple[str, Any]) -> EntityState:
157        """Deserialize the object."""
158        key = value[0]
159        state = value[1]
160        parts = key.split(".")
161        if len(parts) != 2:
162            raise ValueError(f"Expected '<domain>.<entitiy-key>', got {key}")
163        return EntityState(domain=parts[0], key=parts[1], state=state)
164
165
166@dataclass
167class DeviceState(DataClassDictMixin):
168    """Represents a pre-defined device state.
169
170    This is used to map a pre-defined devie state to the values that should be
171    set on individual entities. These are typically "interesting" states of the
172    device that can be enumerated during evaluation. For example, instead of
173    explicitly specifying specific temperature values, a predefined state could
174    implement "warm" and "cool".
175    """
176
177    name: str
178    """An identifier that names this state."""
179
180    entity_states: list[EntityState]
181    """An identifier for this set of attributes used for labeling"""
182
183    def merge(self, new_state: "DeviceState") -> "DeviceState":
184        """Return a new DeviceState with merged entity states."""
185        states = {entity.domain_key: entity for entity in self.entity_states}
186        for entity in new_state.entity_states:
187            if entity.domain_key in states:
188                states[entity.domain_key] = states[entity.domain_key].merge(entity)
189            else:
190                states[entity.domain_key] = entity
191        return DeviceState(name=self.name, entity_states=list(states.values()))
192
193
194class DeviceStateStrategy(SerializationStrategy):
195    """A predefined device state parser."""
196
197    _child_strategy = KeyedObjectListStrategy(EntityStateStrategy())
198
199    def deserialize(self, value: tuple[str, Any]) -> DeviceState:
200        """Deserialize the object."""
201        if not isinstance(value, tuple):
202            raise ValueError(
203                f"Expected 'tuple' representing the DeviceState object, got: {value}"
204            )
205        return DeviceState(
206            name=value[0], entity_states=self._child_strategy.deserialize(value[1])
207        )
208
209
210@dataclass(frozen=True)
211class DeviceType(DataClassDictMixin):
212    """Defines of device type."""
213
214    device_type: str
215    """The identifier for the device e.g. 'smart-lock'"""
216
217    desc: str
218    """The human readable description of the device."""
219
220    device_states: list[DeviceState] = field(
221        metadata=field_options(
222            serialization_strategy=KeyedObjectListStrategy(DeviceStateStrategy())
223        ),
224    )
225    """A series of different attribute values that are most interesting to use during evaluation."""
226
227    entities: dict[str, list[EntityEntry]] = field(
228        metadata=field_options(serialization_strategy=EntityDictStrategy()),
229        default_factory=dict,
230    )
231    """Entity platforms and their entity description keys"""
232
233    @property
234    def device_states_dict(self) -> dict[str, DeviceState]:
235        """Get the predefined state by the specified key."""
236        return {state.name: state for state in self.device_states}
237
238    @property
239    def entity_dict(self) -> dict[str, EntityEntry]:
240        """Get a flat map of all entity entries."""
241        return {
242            f"{platform}.{entry.key}": entry
243            for platform, entries in self.entities.items()
244            for entry in entries
245        }
246
247    def __post_init__(self) -> None:
248        """Validate the DeviceType."""
249        entity_dict = self.entity_dict
250        for device_state in self.device_states:
251            for entity_state in device_state.entity_states:
252                if entity_state.domain_key not in entity_dict:
253                    raise ValueError(
254                        f"Device '{self.device_type}' state '{device_state.name}' references "
255                        f"invalid entity '{entity_state.domain_key}' not in {list(entity_dict.keys())}"
256                    )
257
258
259@dataclass
260class DeviceTypeRegistry:
261    """The registry of all DeviceType objects."""
262
263    device_types: dict[str, DeviceType] = field(default_factory=dict)
264
265
266def _read_device_types(
267    device_types_path: Traversable,
268) -> Generator[DeviceType, None, None]:
269    """Read device types from the device type directory."""
270    _LOGGER.debug("Loading device type registry from %s", device_types_path.absolute())  # type: ignore[attr-defined]
271
272    for device_type_file in device_types_path.iterdir():
273        if not device_type_file.name.endswith(".yaml"):
274            continue
275        try:
276            with device_type_file.open("r") as f:
277                content = f.read()
278        except FileNotFoundError:
279            raise SyntheticHomeError(
280                f"Configuration file '{device_type_file}' does not exist"
281            )
282
283        try:
284            device_type: DeviceType = yaml_decode(content, DeviceType)
285        except MissingField as err:
286            raise SyntheticHomeError(f"Unable to decode file {device_type_file}: {err}")
287        except yaml.YAMLError as err:
288            raise SyntheticHomeError(f"Unable to decode file {device_type_file}: {err}")
289        if device_type_file.name != f"{device_type.device_type}.yaml":
290            raise SyntheticHomeError(
291                f"Device type '{device_type.device_type}' name does not match filename '{device_type_file.name}'"
292            )
293
294        yield device_type
295
296
297@cache
298def load_device_type_registry() -> DeviceTypeRegistry:
299    """Load device types from the yaml configuration files."""
300    device_types = {}
301    for device_type in _read_device_types(DEVICE_TYPES_RESOURCE_PATH):
302        if device_type.device_type in device_types:
303            raise SyntheticHomeError(
304                f"Device registry contains duplicate device type '{device_type.device_type}"
305            )
306        device_types[device_type.device_type] = device_type
307    return DeviceTypeRegistry(device_types=device_types)