synthetic_home.device_types

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