google_nest_sdm.device

  1"""A device from the Smart Device Management API."""
  2
  3from __future__ import annotations
  4
  5import datetime
  6import logging
  7from typing import Any, Awaitable, Callable
  8from dataclasses import dataclass, field, fields, asdict
  9
 10from mashumaro import field_options, DataClassDictMixin
 11from mashumaro.config import BaseConfig
 12from mashumaro.types import SerializationStrategy
 13
 14from . import camera_traits, device_traits, doorbell_traits, thermostat_traits
 15from .auth import AbstractAuth
 16from .doorbell_traits import DoorbellChimeTrait
 17from .diagnostics import Diagnostics, redact_data
 18from .event import EventMessage, EventProcessingError
 19from .event_media import EventMediaManager
 20from .traits import Command
 21from .model import TraitDataClass, SDM_PREFIX, TRAITS
 22
 23_LOGGER = logging.getLogger(__name__)
 24
 25
 26@dataclass
 27class ParentRelation(DataClassDictMixin):
 28    """Represents the parent structure/room of the current resource."""
 29
 30    parent: str
 31    display_name: str = field(metadata=field_options(alias="displayName"))
 32
 33    class Config(BaseConfig):
 34        serialize_by_alias = True
 35
 36
 37@dataclass
 38class TraitTypes(TraitDataClass):
 39    """Data model for parsing traits in the Google Nest SDM API."""
 40
 41    # Device Traits
 42    connectivity: device_traits.ConnectivityTrait | None = field(
 43        metadata=field_options(
 44            alias="sdm.devices.traits.Connectivity",
 45        ),
 46        default=None,
 47    )
 48    fan: device_traits.FanTrait | None = field(
 49        metadata=field_options(
 50            alias="sdm.devices.traits.Fan",
 51        ),
 52        default=None,
 53    )
 54    info: device_traits.InfoTrait | None = field(
 55        metadata=field_options(
 56            alias="sdm.devices.traits.Info",
 57        ),
 58        default=None,
 59    )
 60    humidity: device_traits.HumidityTrait | None = field(
 61        metadata=field_options(
 62            alias="sdm.devices.traits.Humidity",
 63        ),
 64        default=None,
 65    )
 66    temperature: device_traits.TemperatureTrait | None = field(
 67        metadata=field_options(
 68            alias="sdm.devices.traits.Temperature",
 69        ),
 70        default=None,
 71    )
 72
 73    # Thermostat Traits
 74    thermostat_eco: thermostat_traits.ThermostatEcoTrait | None = field(
 75        metadata=field_options(
 76            alias="sdm.devices.traits.ThermostatEco",
 77        ),
 78        default=None,
 79    )
 80    thermostat_hvac: thermostat_traits.ThermostatHvacTrait | None = field(
 81        metadata=field_options(
 82            alias="sdm.devices.traits.ThermostatHvac",
 83        ),
 84        default=None,
 85    )
 86    thermostat_mode: thermostat_traits.ThermostatModeTrait | None = field(
 87        metadata=field_options(
 88            alias="sdm.devices.traits.ThermostatMode",
 89        ),
 90        default=None,
 91    )
 92    thermostat_temperature_setpoint: (
 93        thermostat_traits.ThermostatTemperatureSetpointTrait | None
 94    ) = field(
 95        metadata=field_options(
 96            alias="sdm.devices.traits.ThermostatTemperatureSetpoint",
 97        ),
 98        default=None,
 99    )
100
101    # # Camera Traits
102    camera_image: camera_traits.CameraImageTrait | None = field(
103        metadata=field_options(
104            alias="sdm.devices.traits.CameraImage",
105        ),
106        default=None,
107    )
108    camera_live_stream: camera_traits.CameraLiveStreamTrait | None = field(
109        metadata=field_options(alias="sdm.devices.traits.CameraLiveStream"),
110        default=None,
111    )
112    camera_event_image: camera_traits.CameraEventImageTrait | None = field(
113        metadata=field_options(
114            alias="sdm.devices.traits.CameraEventImage",
115        ),
116        default=None,
117    )
118    camera_motion: camera_traits.CameraMotionTrait | None = field(
119        metadata=field_options(
120            alias="sdm.devices.traits.CameraMotion",
121        ),
122        default=None,
123    )
124    camera_person: camera_traits.CameraPersonTrait | None = field(
125        metadata=field_options(
126            alias="sdm.devices.traits.CameraPerson",
127        ),
128        default=None,
129    )
130    camera_sound: camera_traits.CameraSoundTrait | None = field(
131        metadata=field_options(
132            alias="sdm.devices.traits.CameraSound",
133        ),
134        default=None,
135    )
136    camera_clip_preview: camera_traits.CameraClipPreviewTrait | None = field(
137        metadata=field_options(
138            alias="sdm.devices.traits.CameraClipPreview",
139        ),
140        default=None,
141    )
142
143    # # Doorbell Traits
144    doorbell_chime: doorbell_traits.DoorbellChimeTrait | None = field(
145        metadata=field_options(
146            alias="sdm.devices.traits.DoorbellChime",
147        ),
148        default=None,
149    )
150
151
152class ParentRelationsSerializationStrategy(SerializationStrategy, use_annotations=True):
153    """Parser to ignore invalid parent relations."""
154
155    def serialize(self, value: list[ParentRelation]) -> list[dict[str, Any]]:
156        return [x.to_dict() for x in value]
157
158    def deserialize(self, value: list[dict[str, Any]]) -> list[ParentRelation]:
159        return [
160            ParentRelation.from_dict(relation)
161            for relation in value
162            if "parent" in relation and "displayName" in relation
163        ]
164
165
166def _name_required() -> str:
167    """Raise an error if the name field is not provided.
168
169    This is a workaround for the fact that dataclasses children can't have
170    default fields out of order from the subclass.
171    """
172    raise ValueError("Field 'name' is required")
173
174
175@dataclass
176class Device(TraitTypes):
177    """Class that represents a device object in the Google Nest SDM API."""
178
179    name: str = field(default_factory=_name_required)
180    """Resource name of the device such as 'enterprises/XYZ/devices/123'."""
181
182    type: str | None = None
183    """Type of device for display purposes.
184
185    The device type should not be used to deduce or infer functionality of
186    the actual device it is assigned to. Instead, use the returned traits for
187    the device.
188    """
189
190    relations: list[ParentRelation] = field(
191        metadata=field_options(alias="parentRelations"), default_factory=list
192    )
193    """Represents the parent structure or room of the device."""
194
195    _auth: AbstractAuth = field(init=False, metadata={"serialize": "omit"})
196    _diagnostics: Diagnostics = field(init=False, metadata={"serialize": "omit"})
197    _event_media_manager: EventMediaManager = field(
198        init=False, metadata={"serialize": "omit"}
199    )
200    _callbacks: list[Callable[[EventMessage], Awaitable[None]]] = field(
201        init=False, metadata={"serialize": "omit"}, default_factory=list
202    )
203    _trait_event_ts: dict[str, datetime.datetime] = field(
204        init=False, metadata={"serialize": "omit"}, default_factory=dict
205    )
206
207    @staticmethod
208    def MakeDevice(raw_data: dict[str, Any], auth: AbstractAuth) -> Device:
209        """Create a device with the appropriate traits."""
210
211        # Hack for incorrect nest API response values
212        if (type := raw_data.get("type")) and type == "sdm.devices.types.DOORBELL":
213            if TRAITS not in raw_data:
214                raw_data[TRAITS] = {}
215            raw_data[TRAITS][DoorbellChimeTrait.NAME] = {}
216
217        device: Device = Device.parse_trait_object(raw_data)
218        device._auth = auth
219        device._diagnostics = Diagnostics()
220        cmd = Command(raw_data["name"], auth, device._diagnostics.subkey("command"))
221        for trait in device.traits.values():
222            if hasattr(trait, "_cmd"):
223                trait._cmd = cmd
224
225        event_traits = {
226            trait.EVENT_NAME
227            for trait in device.traits.values()
228            if hasattr(trait, "EVENT_NAME")
229        }
230        device._event_media_manager = EventMediaManager(
231            device.name or "",
232            device.traits,
233            event_traits,
234            diagnostics=device._diagnostics.subkey("event_media"),
235        )
236        return device
237
238    def add_update_listener(self, target: Callable[[], None]) -> Callable[[], None]:
239        """Register a simple event listener notified on updates.
240
241        This will not block on media being fetched. To wait for media, use
242        the callback form the `EventMediaManager`.
243
244        The return value is a callable that will unregister the callback.
245        """
246
247        async def handle_event(event_message: EventMessage) -> None:
248            target()
249
250        return self.add_event_callback(handle_event)
251
252    def add_event_callback(
253        self, target: Callable[[EventMessage], Awaitable[None]]
254    ) -> Callable[[], None]:
255        """Register an event callback for updates to this device.
256
257        This will not block on media being fetched. To wait for media, use
258        the callback form the `EventMediaManager`.
259
260        The return value is a callable that will unregister the callback.
261        """
262        self._callbacks.append(target)
263
264        def remove_callback() -> None:
265            """Remove the event_callback."""
266            self._callbacks.remove(target)
267
268        return remove_callback
269
270    async def async_handle_event(self, event_message: EventMessage) -> None:
271        """Process an event from the pubsub subscriber.
272
273        This will invoke any directly registered callbacks (before fetching media)
274        as well as any callbacks registered with the event media manager that
275        fire post-media.
276        """
277        _LOGGER.debug(
278            "Processing update %s @ %s", event_message.event_id, event_message.timestamp
279        )
280        if not event_message.resource_update_name:
281            raise EventProcessingError("Event was not resource update event")
282        if self.name != event_message.resource_update_name:
283            raise EventProcessingError(
284                f"Mismatch {self.name} != {event_message.resource_update_name}"
285            )
286        self._async_handle_traits(event_message)
287        for callback in self._callbacks:
288            await callback(event_message)
289        await self._event_media_manager.async_handle_events(event_message)
290
291    def _async_handle_traits(self, event_message: EventMessage) -> None:
292        traits = event_message.resource_update_traits
293        if not traits:
294            return
295        # Parse the traits using a separate object, then overwrite
296        # each present field with an updated copy of the original trait with
297        # the new fields merged in.
298        _LOGGER.debug("Trait update %s", traits)
299        parsed_traits = TraitTypes.parse_trait_object({TRAITS: traits})
300        self._async_update_traits(parsed_traits, event_message.timestamp)
301
302    def _async_update_traits(
303        self, parsed_traits: TraitTypes, timestamp: datetime.datetime
304    ) -> None:
305        if timestamp.tzinfo is None:
306            timestamp = timestamp.replace(tzinfo=datetime.UTC)
307        for trait_field in fields(parsed_traits):
308            if (
309                (alias := trait_field.metadata.get("alias")) is None
310                or not alias.startswith(SDM_PREFIX)
311                or not (new := getattr(parsed_traits, trait_field.name))
312            ):
313                continue
314            # Discard updates to traits that are newer than the update
315            if (ts := self._trait_timestamp(trait_field.name)) and ts > timestamp:
316                _LOGGER.debug("Discarding stale update (%s)", timestamp)
317                continue
318
319            # Only merge updates into existing models, updating the existing
320            # fields present in the update trait
321            if not (existing := getattr(self, trait_field.name)):
322                continue
323            for k, v in asdict(new).items():
324                if v is not None:
325                    setattr(existing, k, v)
326            self._trait_event_ts[trait_field.name] = timestamp
327
328    def merge_from_update(self, new_device: Device) -> None:
329        """Merge fields from an updated device object.
330
331        This is used when refreshing the device list from the API.
332        """
333        self._async_update_traits(new_device, datetime.datetime.now(datetime.UTC))
334
335    def _trait_timestamp(self, trait_field_name: str) -> datetime.datetime | None:
336        """Get the last update timestamp for a given trait field."""
337        if (ts := self._trait_event_ts.get(trait_field_name)) is None:
338            return None
339        if ts.tzinfo is None:
340            return ts.replace(tzinfo=datetime.UTC)
341        return ts
342
343    @property
344    def event_media_manager(self) -> EventMediaManager:
345        return self._event_media_manager
346
347    @property
348    def parent_relations(self) -> dict:
349        """Room or structure for the device."""
350        return {relation.parent: relation.display_name for relation in self.relations}
351
352    def delete_relation(self, parent: str) -> None:
353        """Remove a device relationship with the parent."""
354        self.relations = [
355            relation for relation in self.relations if relation.parent != parent
356        ]
357
358    def create_relation(self, relation: ParentRelation) -> None:
359        """Add a new device relation."""
360        self.relations.append(relation)
361
362    def get_diagnostics(self) -> dict[str, Any]:
363        return {
364            "data": redact_data(self.raw_data),
365            **self._diagnostics.as_dict(),
366        }
367
368    class Config(TraitTypes.Config):
369        serialization_strategy = {
370            list[ParentRelation]: ParentRelationsSerializationStrategy(),
371        }