google_nest_sdm.event

  1"""Events from pubsub subscriber."""
  2
  3from __future__ import annotations
  4
  5from abc import ABC, abstractmethod
  6import base64
  7import binascii
  8from dataclasses import dataclass, field
  9import datetime
 10from enum import StrEnum
 11import hashlib
 12import json
 13import logging
 14import traceback
 15from typing import Any, Iterable, Mapping, ClassVar
 16
 17from mashumaro import DataClassDictMixin, field_options
 18from mashumaro.config import (
 19    BaseConfig,
 20)
 21from mashumaro.types import SerializationStrategy
 22
 23from .auth import AbstractAuth
 24from .exceptions import DecodeException
 25from .registry import Registry
 26
 27__all__ = [
 28    "EventMessage",
 29    "CameraMotionEvent",
 30    "CameraPersonEvent",
 31    "CameraSoundEvent",
 32    "DoorbellChimeEvent",
 33    "CameraClipPreviewEvent",
 34    "EventImageType",
 35    "EventProcessingError",
 36]
 37
 38EVENT_ID = "eventId"
 39EVENT_SESSION_ID = "eventSessionId"
 40TIMESTAMP = "timestamp"
 41RESOURCE_UPDATE = "resourceUpdate"
 42NAME = "name"
 43TRAITS = "traits"
 44EVENTS = "events"
 45PREVIEW_URL = "previewUrl"
 46ZONES = "zones"
 47EVENT_THREAD_STATE_ENDED = "ENDED"
 48
 49# Event images expire 30 seconds after the event is published
 50EVENT_IMAGE_EXPIRE_SECS = 30
 51
 52# Camera clip previews don't list an expiration in the API. Lets say 15 minutes
 53# as an arbitrary number for now.
 54CAMERA_CLIP_PREVIEW_EXPIRE_SECS = 15 * 60
 55
 56EVENT_MAP = Registry()
 57
 58_LOGGER = logging.getLogger(__name__)
 59
 60
 61class EventType(StrEnum):
 62    """Types of events."""
 63
 64    CAMERA_MOTION = "sdm.devices.events.CameraMotion.Motion"
 65    CAMERA_PERSON = "sdm.devices.events.CameraPerson.Person"
 66    CAMERA_SOUND = "sdm.devices.events.CameraSound.Sound"
 67    DOORBELL_CHIME = "sdm.devices.events.DoorbellChime.Chime"
 68    CAMERA_CLIP_PREVIEW = "sdm.devices.events.CameraClipPreview.ClipPreview"
 69
 70
 71class EventProcessingError(Exception):
 72    """Raised when there was an error handling an event."""
 73
 74
 75@dataclass(frozen=True)
 76class EventImageContentType(DataClassDictMixin):
 77    """Event image content type."""
 78
 79    content_type: str
 80
 81    def __str__(self) -> str:
 82        """Return a string representation of the event image type."""
 83        return self.content_type
 84
 85
 86class EventImageType(ABC):
 87    IMAGE = EventImageContentType("image/jpeg")
 88    CLIP_PREVIEW = EventImageContentType("video/mp4")
 89    IMAGE_PREVIEW = EventImageContentType("image/gif")
 90
 91    @staticmethod
 92    def from_string(content_type: str) -> EventImageContentType:
 93        """Parse an EventImageType from a string representation."""
 94        if content_type == EventImageType.CLIP_PREVIEW.content_type:
 95            return EventImageType.CLIP_PREVIEW
 96        elif content_type == EventImageType.IMAGE.content_type:
 97            return EventImageType.IMAGE
 98        elif content_type == EventImageType.IMAGE_PREVIEW.content_type:
 99            return EventImageType.IMAGE_PREVIEW
100        else:
101            return EventImageContentType(content_type)
102
103
104@dataclass
105class EventToken:
106    """Identifier for a unique event."""
107
108    event_session_id: str = field(metadata=field_options(alias="eventSessionId"))
109    event_id: str = field(metadata=field_options(alias="eventId"))
110
111    def encode(self) -> str:
112        """Encode the event token as a serialized string."""
113        data = [self.event_session_id, self.event_id]
114        b = json.dumps(data).encode("utf-8")
115        return base64.b64encode(b).decode("utf-8")
116
117    @staticmethod
118    def decode(content: str) -> EventToken:
119        """Decode an event token into a class."""
120        try:
121            s = base64.b64decode(content).decode("utf-8")
122        except binascii.Error as err:
123            raise DecodeException from err
124        data = json.loads(s)
125        if not isinstance(data, list) or len(data) != 2:
126            raise DecodeException("Unexpected data type: %s", data)
127        return EventToken(data[0], data[1])
128
129    def __repr__(self) -> str:
130        if not self.event_id:
131            return "<EventToken event_session_id" + self.event_session_id + ">"
132        return (
133            "<EventToken event_session_id"
134            + self.event_session_id
135            + " event_id="
136            + self.event_id
137            + ">"
138        )
139
140
141class EventImageTypeSerializationStrategy(SerializationStrategy):
142    def serialize(self, value: EventImageContentType) -> str:
143        return value.content_type
144
145    def deserialize(self, value: str) -> EventImageContentType:
146        return EventImageType.from_string(value)
147
148
149class UtcDateTimeSerializationStrategy(SerializationStrategy):
150    """Parser to ensure all datetimes have timezones."""
151
152    def serialize(self, value: datetime.datetime) -> str:
153        return value.isoformat()
154
155    def deserialize(self, value: str) -> datetime.datetime:
156        dt = datetime.datetime.fromisoformat(value)
157        if dt.tzinfo is None:
158            return dt.replace(tzinfo=datetime.timezone.utc)
159        return dt
160
161
162@dataclass
163class ImageEventBase(DataClassDictMixin, ABC):
164    """Base class for all image related event types."""
165
166    event_session_id: str = field(metadata=field_options(alias="eventSessionId"))
167    """ID used to associate separate messages with a single event."""
168
169    timestamp: datetime.datetime
170    """Timestamp when the event occurred."""
171
172    event_id: str = field(metadata=field_options(alias="eventId"), default="")
173    """ID used to associate separate messages with a single event."""
174
175    event_image_type: EventImageContentType = field(default=EventImageType.IMAGE)
176    """Type of the event."""
177
178    zones: list[str] = field(default_factory=list)
179    """List of zones for the event."""
180
181    @property
182    def event_token(self) -> str:
183        """An identifier of this session / event combination."""
184        token = EventToken(self.event_session_id, self.event_id)
185        return token.encode()
186
187    @property
188    @abstractmethod
189    def event_type(self) -> EventType:
190        """The type of event."""
191
192    @property
193    def expires_at(self) -> datetime.datetime:
194        """Timestamp when the message expires."""
195        return self.timestamp + datetime.timedelta(seconds=EVENT_IMAGE_EXPIRE_SECS)
196
197    @property
198    def is_expired(self) -> bool:
199        """Return true if the event expiration has passed."""
200        now = datetime.datetime.now(tz=datetime.timezone.utc)
201        return self.expires_at < now
202
203    def as_dict(self) -> dict[str, Any]:
204        """Return as a dict form that can be serialized for persistence."""
205        return {
206            "event_type": self.event_type,
207            "event_data": self.to_dict(),
208            "timestamp": self.timestamp.isoformat(),
209            "event_image_type": str(self.event_image_type),
210        }
211
212    @staticmethod
213    def parse_event_dict(data: dict[str, Any]) -> ImageEventBase | None:
214        """Parse from a persisted serialized dictionary."""
215        event_type = data["event_type"]
216        event_data = data["event_data"]
217        event = _BuildEvent(event_type, event_data)
218        if event and "event_image_type" in data:
219            event.event_image_type = EventImageType.from_string(
220                data["event_image_type"]
221            )
222        return event
223
224    class Config(BaseConfig):
225        serialization_strategy = {
226            EventImageContentType: EventImageTypeSerializationStrategy(),
227            datetime.datetime: UtcDateTimeSerializationStrategy(),
228        }
229        code_generation_options = [
230            "TO_DICT_ADD_BY_ALIAS_FLAG",
231            "TO_DICT_ADD_OMIT_NONE_FLAG",
232        ]
233        allow_deserialization_not_by_alias = True
234
235
236@EVENT_MAP.register()
237@dataclass
238class CameraMotionEvent(ImageEventBase):
239    """Motion has been detected by the camera."""
240
241    NAME: ClassVar[EventType] = EventType.CAMERA_MOTION
242    event_image_type: EventImageContentType = field(default=EventImageType.IMAGE)
243
244    @property
245    def event_type(self) -> EventType:
246        """The type of event."""
247        return EventType.CAMERA_MOTION
248
249
250@EVENT_MAP.register()
251@dataclass
252class CameraPersonEvent(ImageEventBase):
253    """A person has been detected by the camera."""
254
255    NAME: ClassVar[EventType] = EventType.CAMERA_PERSON
256    event_image_type: EventImageContentType = field(default=EventImageType.IMAGE)
257
258    @property
259    def event_type(self) -> EventType:
260        """The type of event."""
261        return EventType.CAMERA_PERSON
262
263
264@EVENT_MAP.register()
265@dataclass
266class CameraSoundEvent(ImageEventBase):
267    """Sound has been detected by the camera."""
268
269    NAME: ClassVar[EventType] = EventType.CAMERA_SOUND
270    event_image_type: EventImageContentType = field(default=EventImageType.IMAGE)
271
272    @property
273    def event_type(self) -> EventType:
274        """The type of event."""
275        return EventType.CAMERA_SOUND
276
277
278@EVENT_MAP.register()
279@dataclass
280class DoorbellChimeEvent(ImageEventBase):
281    """The doorbell has been pressed."""
282
283    NAME: ClassVar[EventType] = EventType.DOORBELL_CHIME
284    event_image_type: EventImageContentType = field(default=EventImageType.IMAGE)
285
286    @property
287    def event_type(self) -> EventType:
288        """The type of event."""
289        return EventType.DOORBELL_CHIME
290
291
292@EVENT_MAP.register()
293@dataclass
294class CameraClipPreviewEvent(ImageEventBase):
295    """A video clip is available for preview, without extra download."""
296
297    NAME: ClassVar[EventType] = EventType.CAMERA_CLIP_PREVIEW
298    event_image_type: EventImageContentType = field(default=EventImageType.CLIP_PREVIEW)
299
300    preview_url: str = field(metadata=field_options(alias="previewUrl"), default="")
301    """A url 10 second frame video file in mp4 format."""
302
303    @property
304    def event_type(self) -> EventType:
305        """The type of event."""
306        return EventType.CAMERA_CLIP_PREVIEW
307
308    @classmethod
309    def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]:
310        """Validate the event id to use a URL hash as the event id.
311
312        Since clip preview events already have a url associated with them,
313        we don't have an event id for downloading the image.
314        """
315        if not (preview_url := d.get("previewUrl", d.get("preview_url"))):
316            raise ValueError("missing required field previewUrl")
317        d["eventId"] = hashlib.blake2b(preview_url.encode()).hexdigest()
318        return d
319
320    @property
321    def expires_at(self) -> datetime.datetime:
322        """Event ids do not expire."""
323        return self.timestamp + datetime.timedelta(
324            seconds=CAMERA_CLIP_PREVIEW_EXPIRE_SECS
325        )
326
327
328@dataclass
329class RelationUpdate(DataClassDictMixin):
330    """Represents a relational update for a resource."""
331
332    type: str
333    """Type of relation event 'CREATED', 'UPDATED', 'DELETED'."""
334
335    subject: str
336    """Resource that the object is now in relation with."""
337
338    object: str
339    """Resource that triggered the event."""
340
341
342def _BuildEvent(
343    event_type: str, event_data: Mapping[str, Any]
344) -> ImageEventBase | None:
345    if event_type not in EVENT_MAP:
346        _LOGGER.debug("Event type %s not found (%s)", event_type, EVENT_MAP.keys())
347        return None
348    cls = EVENT_MAP[event_type]
349    try:
350        return cls.from_dict(event_data)  # type: ignore
351    except Exception as err:
352        traceback.print_exc()
353        _LOGGER.debug("Failed to parse event: %s (event_data=%s)", err, event_data)
354        raise err
355
356
357def session_event_image_type(events: Iterable[ImageEventBase]) -> EventImageContentType:
358    """Determine the event type to use based on the events in the session."""
359    for event in events:
360        if event.event_image_type != EventImageType.IMAGE:
361            return event.event_image_type
362    return EventImageType.IMAGE
363
364
365class UpdateEventsSerializationStrategy(SerializationStrategy, use_annotations=True):
366    """Parser to ignore invalid parent relations."""
367
368    def serialize(self, value: dict[str, ImageEventBase]) -> dict[str, Any]:
369        return {k: v.to_dict(by_alias=True) for k, v in value.items()}
370
371    def deserialize(self, value: dict[str, Any]) -> dict[str, ImageEventBase]:
372        result = {}
373        for event_type, event_data in value.items():
374            image_event = _BuildEvent(event_type, event_data)
375            if not image_event:
376                continue
377            result[event_type] = image_event
378        return result
379
380
381@dataclass
382class EventMessage(DataClassDictMixin):
383    """Event for a change in trait value or device action."""
384
385    timestamp: datetime.datetime
386    event_id: str = field(metadata=field_options(alias="eventId"))
387    resource_update_name: str | None = field(default=None)
388    resource_update_events: dict[str, ImageEventBase] | None = field(default=None)
389    resource_update_traits: dict[str, Any] | None = field(default=None)
390    event_thread_state: str | None = field(
391        metadata=field_options(alias="eventThreadState"), default=None
392    )
393    relation_update: RelationUpdate | None = field(
394        metadata=field_options(alias="relationUpdate"), default=None
395    )
396
397    _auth: AbstractAuth = field(init=False, metadata={"serialize": "omit"})
398
399    @classmethod
400    def create_event(
401        cls, raw_data: dict[str, Any], auth: AbstractAuth
402    ) -> "EventMessage":
403        """Initialize an EventMessage."""
404        event_data = {**raw_data}
405        _LOGGER.debug("EventMessage raw_data=%s", event_data)
406        if update := event_data.get(RESOURCE_UPDATE):
407            if name := update.get(NAME):
408                event_data["resource_update_name"] = name
409            if events := update.get(EVENTS):
410                timestamp = event_data.get(TIMESTAMP)
411                for event_updates in events.values():
412                    event_updates[TIMESTAMP] = timestamp
413                event_data["resource_update_events"] = events
414            if traits := update.get(TRAITS):
415                event_data["resource_update_traits"] = traits
416                event_data["resource_update_traits"][NAME] = update.get(NAME)
417
418        event = cls.from_dict(event_data)
419        event._auth = auth
420        return event
421
422    @property
423    def event_sessions(self) -> dict[str, dict[str, ImageEventBase]] | None:
424        events = self.resource_update_events
425        if not events:
426            return None
427        event_sessions: dict[str, dict[str, ImageEventBase]] = {}
428        for event_name, event in events.items():
429            d = event_sessions.get(event.event_session_id, {})
430            d[event_name] = event
431            event_sessions[event.event_session_id] = d
432        # Build associations between all events
433        for event_session_id, event_dict in event_sessions.items():
434            event_image_type = session_event_image_type(events.values())
435            for event_type, event in event_dict.items():
436                event.event_image_type = event_image_type
437        return event_sessions
438
439    @property
440    def raw_data(self) -> dict[str, Any]:
441        """Return raw data for the event."""
442        return self.to_dict(by_alias=True)
443
444    def with_events(
445        self,
446        event_keys: Iterable[str],
447        merge_data: dict[str, ImageEventBase] | None = None,
448    ) -> EventMessage:
449        """Create a new EventMessage minus some existing events by key."""
450        new_message = EventMessage.create_event(self.to_dict(by_alias=True), self._auth)
451        if not merge_data:
452            merge_data = {}
453        new_events = {}
454        for key in event_keys:
455            if (
456                new_message.resource_update_events
457                and key in new_message.resource_update_events
458            ):
459                new_events[key] = new_message.resource_update_events[key]
460            elif merge_data and key in merge_data:
461                new_events[key] = merge_data[key]
462        new_message.resource_update_events = new_events
463        return new_message
464
465    @property
466    def is_thread_ended(self) -> bool:
467        """Return true if the message indicates the thread is ended."""
468        return self.event_thread_state == EVENT_THREAD_STATE_ENDED
469
470    def __repr__(self) -> str:
471        """Debug information."""
472        return f"EventMessage{self.to_dict()}"
473
474    class Config(BaseConfig):
475        serialization_strategy = {
476            dict[str, ImageEventBase]: UpdateEventsSerializationStrategy(),
477            datetime.datetime: UtcDateTimeSerializationStrategy(),
478        }
479        code_generation_options = [
480            "TO_DICT_ADD_BY_ALIAS_FLAG",
481            "TO_DICT_ADD_OMIT_NONE_FLAG",
482        ]