Add image support to nest SDM api (#42556)

homeassistant/components/ffmpeg/__init__.py

@@ -1,7 +1,9 @@
 """Support for FFmpeg."""
+import asyncio
 import re
+from typing import Optional
 
-from haffmpeg.tools import FFVersion
+from haffmpeg.tools import IMAGE_JPEG, FFVersion, ImageFrame
 import voluptuous as vol
 
 from homeassistant.const import (
@@ -17,6 +19,7 @@ from homeassistant.helpers.dispatcher import (
     async_dispatcher_send,
 )
 from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.typing import HomeAssistantType
 
 DOMAIN = "ffmpeg"
 
@@ -86,6 +89,21 @@ async def async_setup(hass, config):
     return True
 
 
+async def async_get_image(
+    hass: HomeAssistantType,
+    input_source: str,
+    output_format: str = IMAGE_JPEG,
+    extra_cmd: Optional[str] = None,
+):
+    """Get an image from a frame of an RTSP stream."""
+    manager = hass.data[DATA_FFMPEG]
+    ffmpeg = ImageFrame(manager.binary, loop=hass.loop)
+    image = await asyncio.shield(
+        ffmpeg.get_image(input_source, output_format=output_format, extra_cmd=extra_cmd)
+    )
+    return image
+
+
 class FFmpegManager:
     """Helper for ha-ffmpeg."""
 

homeassistant/components/ffmpeg/camera.py

@@ -1,8 +1,7 @@
 """Support for Cameras with FFmpeg as decoder."""
-import asyncio
 
 from haffmpeg.camera import CameraMjpeg
-from haffmpeg.tools import IMAGE_JPEG, ImageFrame
+from haffmpeg.tools import IMAGE_JPEG
 import voluptuous as vol
 
 from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_STREAM, Camera
@@ -10,7 +9,7 @@ from homeassistant.const import CONF_NAME
 from homeassistant.helpers.aiohttp_client import async_aiohttp_proxy_stream
 import homeassistant.helpers.config_validation as cv
 
-from . import CONF_EXTRA_ARGUMENTS, CONF_INPUT, DATA_FFMPEG
+from . import CONF_EXTRA_ARGUMENTS, CONF_INPUT, DATA_FFMPEG, async_get_image
 
 DEFAULT_NAME = "FFmpeg"
 DEFAULT_ARGUMENTS = "-pred 1"
@@ -52,15 +51,12 @@ class FFmpegCamera(Camera):
 
     async def async_camera_image(self):
         """Return a still image response from the camera."""
-
+        return await async_get_image(
-        ffmpeg = ImageFrame(self._manager.binary, loop=self.hass.loop)
+            self.hass,
-
+            self._input,
-        image = await asyncio.shield(
+            output_format=IMAGE_JPEG,
-            ffmpeg.get_image(
+            extra_cmd=self._extra_arguments,
-                self._input, output_format=IMAGE_JPEG, extra_cmd=self._extra_arguments
         )
-        )
-        return image
 
     async def handle_async_mjpeg_stream(self, request):
         """Generate an HTTP MJPEG stream from the camera."""

homeassistant/components/nest/camera_sdm.py

@@ -5,11 +5,14 @@ from typing import Optional
 
 from google_nest_sdm.camera_traits import CameraImageTrait, CameraLiveStreamTrait
 from google_nest_sdm.device import Device
+from haffmpeg.tools import IMAGE_JPEG
 
 from homeassistant.components.camera import SUPPORT_STREAM, Camera
+from homeassistant.components.ffmpeg import async_get_image
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.typing import HomeAssistantType
+from homeassistant.util.dt import utcnow
 
 from .const import DOMAIN, SIGNAL_NEST_UPDATE
 from .device_info import DeviceInfo
@@ -45,6 +48,7 @@ class NestCamera(Camera):
         super().__init__()
         self._device = device
         self._device_info = DeviceInfo(device)
+        self._stream = None
 
     @property
     def should_poll(self) -> bool:
@@ -90,11 +94,21 @@ class NestCamera(Camera):
         if CameraLiveStreamTrait.NAME not in self._device.traits:
             return None
         trait = self._device.traits[CameraLiveStreamTrait.NAME]
-        rtsp_stream = await trait.generate_rtsp_stream()
+        now = utcnow()
-        # Note: This is only valid for a few minutes, and probably needs
+        if not self._stream:
-        # to be improved with an occasional call to .extend_rtsp_stream() which
+            logging.debug("Fetching stream url")
-        # returns a new rtsp_stream object.
+            self._stream = await trait.generate_rtsp_stream()
-        return rtsp_stream.rtsp_stream_url
+        elif self._stream.expires_at < now:
+            logging.debug("Stream expired, extending stream")
+            new_stream = await self._stream.extend_rtsp_stream()
+            self._stream = new_stream
+        return self._stream.rtsp_stream_url
+
+    async def async_will_remove_from_hass(self):
+        """Invalidates the RTSP token when unloaded."""
+        if self._stream:
+            logging.debug("Invalidating stream")
+            await self._stream.stop_rtsp_stream()
 
     async def async_added_to_hass(self):
         """Run when entity is added to register update signal handler."""
@@ -109,7 +123,7 @@ class NestCamera(Camera):
 
     async def async_camera_image(self):
         """Return bytes of camera image."""
-        # No support for still images yet.  Still images are only available
+        stream_url = await self.stream_source()
-        # in response to an event on the feed.  For now, suppress a
+        if not stream_url:
-        # NotImplementedError in the parent class.
             return None
+        return await async_get_image(self.hass, stream_url, output_format=IMAGE_JPEG)

homeassistant/components/nest/manifest.json

@@ -2,7 +2,7 @@
   "domain": "nest",
   "name": "Nest",
   "config_flow": true,
-  "dependencies": ["http"],
+  "dependencies": ["ffmpeg", "http"],
   "documentation": "https://www.home-assistant.io/integrations/nest",
   "requirements": [
       "python-nest==4.1.0",

tests/components/nest/camera_sdm_test.py

@@ -5,17 +5,38 @@ These tests fake out the subscriber/devicemanager, and are not using a real
 pubsub subscriber.
 """
 
+import datetime
+from typing import List
+
 from google_nest_sdm.auth import AbstractAuth
 from google_nest_sdm.device import Device
 
 from homeassistant.components import camera
 from homeassistant.components.camera import STATE_IDLE
+from homeassistant.util.dt import utcnow
 
 from .common import async_setup_sdm_platform
 
+from tests.async_mock import patch
+
 PLATFORM = "camera"
 CAMERA_DEVICE_TYPE = "sdm.devices.types.CAMERA"
 DEVICE_ID = "some-device-id"
+DEVICE_TRAITS = {
+    "sdm.devices.traits.Info": {
+        "customName": "My Camera",
+    },
+    "sdm.devices.traits.CameraLiveStream": {
+        "maxVideoResolution": {
+            "width": 640,
+            "height": 480,
+        },
+        "videoCodecs": ["H264"],
+        "audioCodecs": ["AAC"],
+    },
+}
+DATETIME_FORMAT = "YY-MM-DDTHH:MM:SS"
+DOMAIN = "nest"
 
 
 class FakeResponse:
@@ -37,10 +58,10 @@ class FakeResponse:
 class FakeAuth(AbstractAuth):
     """Fake authentication object that returns fake responses."""
 
-    def __init__(self, response: FakeResponse):
+    def __init__(self, responses: List[FakeResponse]):
         """Initialize the FakeAuth."""
         super().__init__(None, "")
-        self._response = response
+        self._responses = responses
 
     async def async_get_access_token(self):
         """Return a fake access token."""
@@ -52,7 +73,7 @@ class FakeAuth(AbstractAuth):
 
     async def request(self, method: str, url: str, **kwargs):
         """Pass through the FakeResponse."""
-        return self._response
+        return self._responses.pop(0)
 
 
 async def async_setup_camera(hass, traits={}, auth=None):
@@ -91,22 +112,7 @@ async def test_ineligible_device(hass):
 
 async def test_camera_device(hass):
     """Test a basic camera with a live stream."""
-    await async_setup_camera(
+    await async_setup_camera(hass, DEVICE_TRAITS)
-        hass,
-        {
-            "sdm.devices.traits.Info": {
-                "customName": "My Camera",
-            },
-            "sdm.devices.traits.CameraLiveStream": {
-                "maxVideoResolution": {
-                    "width": 640,
-                    "height": 480,
-                },
-                "videoCodecs": ["H264"],
-                "audioCodecs": ["AAC"],
-            },
-        },
-    )
 
     assert len(hass.states.async_all()) == 1
     camera = hass.states.get("camera.my_camera")
@@ -126,34 +132,75 @@ async def test_camera_device(hass):
     assert device.identifiers == {("nest", DEVICE_ID)}
 
 
-async def test_camera_stream(hass):
+async def test_camera_stream(hass, aiohttp_client):
     """Test a basic camera and fetch its live stream."""
+    now = utcnow()
+    expiration = now + datetime.timedelta(seconds=100)
     response = FakeResponse(
         {
             "results": {
                 "streamUrls": {"rtspUrl": "rtsp://some/url?auth=g.0.streamingToken"},
                 "streamExtensionToken": "g.1.extensionToken",
                 "streamToken": "g.0.streamingToken",
-                "expiresAt": "2018-01-04T18:30:00.000Z",
+                "expiresAt": expiration.isoformat(timespec="seconds"),
             },
         }
     )
+    await async_setup_camera(hass, DEVICE_TRAITS, auth=FakeAuth([response]))
+
+    assert len(hass.states.async_all()) == 1
+    cam = hass.states.get("camera.my_camera")
+    assert cam is not None
+    assert cam.state == STATE_IDLE
+
+    stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
+    assert stream_source == "rtsp://some/url?auth=g.0.streamingToken"
+
+    with patch(
+        "homeassistant.components.ffmpeg.ImageFrame.get_image",
+        autopatch=True,
+        return_value=b"image bytes",
+    ):
+        image = await camera.async_get_image(hass, "camera.my_camera")
+
+    assert image.content == b"image bytes"
+
+
+async def test_refresh_expired_stream_token(hass, aiohttp_client):
+    """Test a camera stream expiration and refresh."""
+    now = utcnow()
+    past = now - datetime.timedelta(seconds=100)
+    future = now + datetime.timedelta(seconds=100)
+    responses = [
+        FakeResponse(
+            {
+                "results": {
+                    "streamUrls": {
+                        "rtspUrl": "rtsp://some/url?auth=g.0.streamingToken"
+                    },
+                    "streamExtensionToken": "g.1.extensionToken",
+                    "streamToken": "g.0.streamingToken",
+                    "expiresAt": past.isoformat(timespec="seconds"),
+                },
+            }
+        ),
+        FakeResponse(
+            {
+                "results": {
+                    "streamUrls": {
+                        "rtspUrl": "rtsp://some/url?auth=g.2.streamingToken"
+                    },
+                    "streamExtensionToken": "g.3.extensionToken",
+                    "streamToken": "g.2.streamingToken",
+                    "expiresAt": future.isoformat(timespec="seconds"),
+                },
+            }
+        ),
+    ]
     await async_setup_camera(
         hass,
-        {
+        DEVICE_TRAITS,
-            "sdm.devices.traits.Info": {
+        auth=FakeAuth(responses),
-                "customName": "My Camera",
-            },
-            "sdm.devices.traits.CameraLiveStream": {
-                "maxVideoResolution": {
-                    "width": 640,
-                    "height": 480,
-                },
-                "videoCodecs": ["H264"],
-                "audioCodecs": ["AAC"],
-            },
-        },
-        auth=FakeAuth(response),
     )
 
     assert len(hass.states.async_all()) == 1
@@ -163,3 +210,49 @@ async def test_camera_stream(hass):
 
     stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
     assert stream_source == "rtsp://some/url?auth=g.0.streamingToken"
+
+    # On second fetch, notice the stream is expired and fetch again
+    stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
+    assert stream_source == "rtsp://some/url?auth=g.2.streamingToken"
+
+    # Stream is not expired; Same url returned
+    stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
+    assert stream_source == "rtsp://some/url?auth=g.2.streamingToken"
+
+
+async def test_camera_removed(hass, aiohttp_client):
+    """Test case where entities are removed and stream tokens expired."""
+    now = utcnow()
+    expiration = now + datetime.timedelta(seconds=100)
+    responses = [
+        FakeResponse(
+            {
+                "results": {
+                    "streamUrls": {
+                        "rtspUrl": "rtsp://some/url?auth=g.0.streamingToken"
+                    },
+                    "streamExtensionToken": "g.1.extensionToken",
+                    "streamToken": "g.0.streamingToken",
+                    "expiresAt": expiration.isoformat(timespec="seconds"),
+                },
+            }
+        ),
+        FakeResponse({"results": {}}),
+    ]
+    await async_setup_camera(
+        hass,
+        DEVICE_TRAITS,
+        auth=FakeAuth(responses),
+    )
+
+    assert len(hass.states.async_all()) == 1
+    cam = hass.states.get("camera.my_camera")
+    assert cam is not None
+    assert cam.state == STATE_IDLE
+
+    stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
+    assert stream_source == "rtsp://some/url?auth=g.0.streamingToken"
+
+    for config_entry in hass.config_entries.async_entries(DOMAIN):
+        await hass.config_entries.async_remove(config_entry.entry_id)
+    assert len(hass.states.async_all()) == 0