google_nest_sdm.camera_traits

  1"""Traits belonging to camera devices."""
  2
  3from __future__ import annotations
  4
  5from abc import ABC, abstractmethod
  6from dataclasses import dataclass, field
  7import datetime
  8from enum import Enum
  9import logging
 10from typing import ClassVar
 11import urllib.parse as urlparse
 12
 13from mashumaro import DataClassDictMixin, field_options
 14from mashumaro.config import BaseConfig
 15from mashumaro.types import SerializationStrategy
 16
 17from .event import (
 18    CameraClipPreviewEvent,
 19    CameraMotionEvent,
 20    CameraPersonEvent,
 21    CameraSoundEvent,
 22    EventImageContentType,
 23    EventImageType,
 24    EventType,
 25)
 26from .traits import CommandDataClass, TraitType
 27from .webrtc_util import fix_sdp_answer
 28
 29__all__ = [
 30    "CameraImageTrait",
 31    "CameraLiveStreamTrait",
 32    "CameraEventImageTrait",
 33    "CameraMotionTrait",
 34    "CameraPersonTrait",
 35    "CameraSoundTrait",
 36    "CameraClipPreviewTrait",
 37    "Resolution",
 38    "Stream",
 39    "StreamUrls",
 40    "RtspStream",
 41    "WebRtcStream",
 42    "StreamingProtocol",
 43    "EventImage",
 44]
 45
 46_LOGGER = logging.getLogger(__name__)
 47
 48MAX_IMAGE_RESOLUTION = "maxImageResolution"
 49MAX_VIDEO_RESOLUTION = "maxVideoResolution"
 50WIDTH = "width"
 51HEIGHT = "height"
 52VIDEO_CODECS = "videoCodecs"
 53AUDIO_CODECS = "audioCodecs"
 54SUPPORTED_PROTOCOLS = "supportedProtocols"
 55STREAM_URLS = "streamUrls"
 56RESULTS = "results"
 57RTSP_URL = "rtspUrl"
 58STREAM_EXTENSION_TOKEN = "streamExtensionToken"
 59STREAM_TOKEN = "streamToken"
 60URL = "url"
 61TOKEN = "token"
 62ANSWER_SDP = "answerSdp"
 63MEDIA_SESSION_ID = "mediaSessionId"
 64
 65EVENT_IMAGE_CLIP_PREVIEW = "clip_preview"
 66
 67
 68@dataclass
 69class Resolution:
 70    """Maximum Resolution of an image or stream."""
 71
 72    width: int | None = None
 73    height: int | None = None
 74
 75
 76@dataclass
 77class CameraImageTrait(DataClassDictMixin):
 78    """This trait belongs to any device that supports taking images."""
 79
 80    NAME: ClassVar[TraitType] = TraitType.CAMERA_IMAGE
 81
 82    max_image_resolution: Resolution | None = field(
 83        metadata=field_options(alias="maxImageResolution"), default=None
 84    )
 85    """Maximum resolution of the camera image."""
 86
 87
 88@dataclass
 89class Stream(DataClassDictMixin, CommandDataClass, ABC):
 90    """Base class for streams."""
 91
 92    expires_at: datetime.datetime = field(metadata=field_options(alias="expiresAt"))
 93    """Time at which both streamExtensionToken and streamToken expire."""
 94
 95    @abstractmethod
 96    async def extend_stream(self) -> Stream:
 97        """Extend the lifetime of the stream."""
 98
 99    @abstractmethod
100    async def stop_stream(self) -> None:
101        """Invalidate the stream."""
102
103
104@dataclass
105class StreamUrls:
106    """Response object for stream urls"""
107
108    rtsp_url: str = field(metadata=field_options(alias="rtspUrl"))
109    """RTSP live stream URL."""
110
111
112@dataclass
113class RtspStream(Stream):
114    """Provides access an RTSP live stream URL."""
115
116    stream_urls: StreamUrls = field(metadata=field_options(alias="streamUrls"))
117    """Stream urls to access the live stream."""
118
119    stream_token: str = field(metadata=field_options(alias="streamToken"))
120    """Token to use to access an RTSP live stream."""
121
122    stream_extension_token: str = field(
123        metadata=field_options(alias="streamExtensionToken")
124    )
125    """Token to use to extend access to an RTSP live stream."""
126
127    @property
128    def rtsp_stream_url(self) -> str:
129        """RTSP live stream URL."""
130        return self.stream_urls.rtsp_url
131
132    async def extend_stream(self) -> Stream | RtspStream:
133        """Extend the lifetime of the stream."""
134        return await self.extend_rtsp_stream()
135
136    async def extend_rtsp_stream(self) -> RtspStream:
137        """Request a new RTSP live stream URL access token."""
138        data = {
139            "command": "sdm.devices.commands.CameraLiveStream.ExtendRtspStream",
140            "params": {"streamExtensionToken": self.stream_extension_token},
141        }
142        response_data = await self.cmd.execute_json(data)
143        results = response_data[RESULTS]
144        # Update the stream url with the new token
145        stream_token = results[STREAM_TOKEN]
146        parsed = urlparse.urlparse(self.rtsp_stream_url)
147        parsed = parsed._replace(query=f"auth={stream_token}")
148        url = urlparse.urlunparse(parsed)
149        results[STREAM_URLS] = {}
150        results[STREAM_URLS][RTSP_URL] = url
151        obj = RtspStream.from_dict(results)
152        obj._cmd = self.cmd
153        return obj
154
155    async def stop_stream(self) -> None:
156        """Invalidate the stream."""
157        return await self.stop_rtsp_stream()
158
159    async def stop_rtsp_stream(self) -> None:
160        """Invalidates a valid RTSP access token and stops the RTSP live stream."""
161        data = {
162            "command": "sdm.devices.commands.CameraLiveStream.StopRtspStream",
163            "params": {"streamExtensionToken": self.stream_extension_token},
164        }
165        await self.cmd.execute(data)
166
167
168@dataclass
169class WebRtcStream(Stream):
170    """Provides access an RTSP live stream URL."""
171
172    answer_sdp: str = field(metadata=field_options(alias="answerSdp"))
173    """An SDP answer to use with the local device displaying the stream."""
174
175    media_session_id: str = field(metadata=field_options(alias="mediaSessionId"))
176    """Media Session ID of the live stream."""
177
178    async def extend_stream(self) -> WebRtcStream:
179        """Request a new RTSP live stream URL access token."""
180        data = {
181            "command": "sdm.devices.commands.CameraLiveStream.ExtendWebRtcStream",
182            "params": {MEDIA_SESSION_ID: self.media_session_id},
183        }
184        response_data = await self.cmd.execute_json(data)
185        # Preserve original answerSdp, and merge with response that contains
186        # the other fields (expiresAt, and mediaSessionId.
187        results = response_data[RESULTS]
188        results[ANSWER_SDP] = self.answer_sdp
189        obj = WebRtcStream.from_dict(results)
190        obj._cmd = self.cmd
191        return obj
192
193    async def stop_stream(self) -> None:
194        """Invalidates a valid RTSP access token and stops the RTSP live stream."""
195        data = {
196            "command": "sdm.devices.commands.CameraLiveStream.StopWebRtcStream",
197            "params": {MEDIA_SESSION_ID: self.media_session_id},
198        }
199        await self.cmd.execute(data)
200
201
202class StreamingProtocol(str, Enum):
203    """Streaming protocols supported by the device."""
204
205    RTSP = "RTSP"
206    WEB_RTC = "WEB_RTC"
207
208
209def _default_streaming_protocol() -> list[StreamingProtocol]:
210    return [
211        StreamingProtocol.RTSP,
212    ]
213
214
215class StreamingProtocolSerializationStrategy(
216    SerializationStrategy, use_annotations=True
217):
218    """Parser for streaming protocols that ignores invalid values."""
219
220    def serialize(self, value: list[StreamingProtocol]) -> list[str]:
221        return [str(x.name) for x in value]
222
223    def deserialize(self, value: list[str]) -> list[StreamingProtocol]:
224        return [
225            StreamingProtocol[x] for x in value if x in StreamingProtocol.__members__
226        ] or _default_streaming_protocol()
227
228
229@dataclass
230class CameraLiveStreamTrait(DataClassDictMixin, CommandDataClass):
231    """This trait belongs to any device that supports live streaming."""
232
233    NAME: ClassVar[TraitType] = TraitType.CAMERA_LIVE_STREAM
234
235    max_video_resolution: Resolution = field(
236        metadata=field_options(alias="maxVideoResolution"), default_factory=Resolution
237    )
238    """Maximum resolution of the video live stream."""
239
240    video_codecs: list[str] = field(
241        metadata=field_options(alias="videoCodecs"), default_factory=list
242    )
243    """Video codecs supported for the live stream."""
244
245    audio_codecs: list[str] = field(
246        metadata=field_options(alias="audioCodecs"), default_factory=list
247    )
248    """Audio codecs supported for the live stream."""
249
250    supported_protocols: list[StreamingProtocol] = field(
251        metadata=field_options(alias="supportedProtocols"),
252        default_factory=_default_streaming_protocol,
253    )
254    """Streaming protocols supported for the live stream."""
255
256    async def generate_rtsp_stream(self) -> RtspStream:
257        """Request a token to access an RTSP live stream URL."""
258        if StreamingProtocol.RTSP not in self.supported_protocols:
259            raise ValueError("Device does not support RTSP stream")
260        data = {
261            "command": "sdm.devices.commands.CameraLiveStream.GenerateRtspStream",
262            "params": {},
263        }
264        response_data = await self.cmd.execute_json(data)
265        results = response_data[RESULTS]
266        obj = RtspStream.from_dict(results)
267        obj._cmd = self.cmd
268        return obj
269
270    async def generate_web_rtc_stream(self, offer_sdp: str) -> WebRtcStream:
271        """Request a token to access a Web RTC live stream URL."""
272        if StreamingProtocol.WEB_RTC not in self.supported_protocols:
273            raise ValueError("Device does not support WEB_RTC stream")
274        data = {
275            "command": "sdm.devices.commands.CameraLiveStream.GenerateWebRtcStream",
276            "params": {"offerSdp": offer_sdp},
277        }
278        response_data = await self.cmd.execute_json(data)
279        results = response_data[RESULTS]
280        obj = WebRtcStream.from_dict(results)
281        obj._cmd = self.cmd
282        _LOGGER.debug("Received answer_sdp: %s", obj.answer_sdp)
283        obj.answer_sdp = fix_sdp_answer(offer_sdp, obj.answer_sdp)
284        _LOGGER.debug("Return answer_sdp: %s", obj.answer_sdp)
285        return obj
286
287    class Config(BaseConfig):
288        serialization_strategy = {
289            list[StreamingProtocol]: StreamingProtocolSerializationStrategy(),
290        }
291        serialize_by_alias = True
292
293
294@dataclass
295class EventImage(DataClassDictMixin, CommandDataClass):
296    """Provides access to an image in response to an event.
297
298    Use a ?width or ?height query parameters to customize the resolution
299    of the downloaded image. Only one of these parameters need to specified.
300    The other parameter is scaled automatically according to the camera's
301    aspect ratio.
302
303    The token should be added as an HTTP header:
304    Authorization: Basic <token>
305    """
306
307    event_image_type: EventImageContentType
308    """Return the type of event image."""
309
310    url: str | None = field(default=None)
311    """URL to download the camera image from."""
312
313    token: str | None = field(default=None)
314    """Token to use in the HTTP Authorization header when downloading."""
315
316    async def contents(
317        self,
318        width: int | None = None,
319        height: int | None = None,
320    ) -> bytes:
321        """Download the image bytes."""
322        if width:
323            fetch_url = f"{self.url}?width={width}"
324        elif height:
325            fetch_url = f"{self.url}?width={height}"
326        else:
327            assert self.url
328            fetch_url = self.url
329        return await self.cmd.fetch_image(fetch_url, basic_auth=self.token)
330
331
332@dataclass
333class CameraEventImageTrait(DataClassDictMixin, CommandDataClass):
334    """This trait belongs to any device that generates images from events."""
335
336    NAME: ClassVar[TraitType] = TraitType.CAMERA_EVENT_IMAGE
337
338    async def generate_image(self, event_id: str) -> EventImage:
339        """Provide a URL to download a camera image."""
340        data = {
341            "command": "sdm.devices.commands.CameraEventImage.GenerateImage",
342            "params": {
343                "eventId": event_id,
344            },
345        }
346        response_data = await self.cmd.execute_json(data)
347        results = response_data[RESULTS]
348        img = EventImage(**results, event_image_type=EventImageType.IMAGE)
349        img._cmd = self.cmd
350        return img
351
352
353@dataclass
354class CameraMotionTrait:
355    """For any device that supports motion detection events."""
356
357    NAME: ClassVar[TraitType] = TraitType.CAMERA_MOTION
358    EVENT_NAME: ClassVar[EventType] = CameraMotionEvent.NAME
359
360
361@dataclass
362class CameraPersonTrait:
363    """For any device that supports person detection events."""
364
365    NAME: ClassVar[TraitType] = TraitType.CAMERA_PERSON
366    EVENT_NAME: ClassVar[EventType] = CameraPersonEvent.NAME
367
368
369@dataclass
370class CameraSoundTrait:
371    """For any device that supports sound detection events."""
372
373    NAME: ClassVar[TraitType] = TraitType.CAMERA_SOUND
374    EVENT_NAME: ClassVar[EventType] = CameraSoundEvent.NAME
375
376
377@dataclass
378class CameraClipPreviewTrait(DataClassDictMixin, CommandDataClass):
379    """For any device that supports a clip preview."""
380
381    NAME: ClassVar[TraitType] = TraitType.CAMERA_CLIP_PREVIEW
382    EVENT_NAME: ClassVar[EventType] = CameraClipPreviewEvent.NAME
383
384    async def generate_event_image(self, preview_url: str) -> EventImage | None:
385        """Provide a URL to download a camera image from the active event."""
386        img = EventImage(url=preview_url, event_image_type=EventImageType.CLIP_PREVIEW)
387        img._cmd = self.cmd
388        return img