add proxy view for unifiprotect to grab snapshot at specific time (#133546)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/unifiprotect/__init__.py

@@ -45,7 +45,12 @@ from .utils import (
     async_create_api_client,
     async_get_devices,
 )
-from .views import ThumbnailProxyView, VideoEventProxyView, VideoProxyView
+from .views import (
+    SnapshotProxyView,
+    ThumbnailProxyView,
+    VideoEventProxyView,
+    VideoProxyView,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -173,6 +178,7 @@ async def _async_setup_entry(
     data_service.async_setup()
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     hass.http.register_view(ThumbnailProxyView(hass))
+    hass.http.register_view(SnapshotProxyView(hass))
     hass.http.register_view(VideoProxyView(hass))
     hass.http.register_view(VideoEventProxyView(hass))
 

homeassistant/components/unifiprotect/views.py

@@ -44,6 +44,34 @@ def async_generate_thumbnail_url(
     return f"{url}?{urlencode(params)}"
 
 
+@callback
+def async_generate_snapshot_url(
+    nvr_id: str,
+    camera_id: str,
+    timestamp: datetime,
+    width: int | None = None,
+    height: int | None = None,
+) -> str:
+    """Generate URL for event thumbnail."""
+
+    url_format = SnapshotProxyView.url
+    if TYPE_CHECKING:
+        assert url_format is not None
+    url = url_format.format(
+        nvr_id=nvr_id,
+        camera_id=camera_id,
+        timestamp=timestamp.replace(microsecond=0).isoformat(),
+    )
+
+    params = {}
+    if width is not None:
+        params["width"] = str(width)
+    if height is not None:
+        params["height"] = str(height)
+
+    return f"{url}?{urlencode(params)}"
+
+
 @callback
 def async_generate_event_video_url(event: Event) -> str:
     """Generate URL for event video."""
@@ -188,6 +216,59 @@ class ThumbnailProxyView(ProtectProxyView):
         return web.Response(body=thumbnail, content_type="image/jpeg")
 
 
+class SnapshotProxyView(ProtectProxyView):
+    """View to proxy snapshots at specified time from UniFi Protect."""
+
+    url = "/api/unifiprotect/snapshot/{nvr_id}/{camera_id}/{timestamp}"
+    name = "api:unifiprotect_snapshot"
+
+    async def get(
+        self, request: web.Request, nvr_id: str, camera_id: str, timestamp: str
+    ) -> web.Response:
+        """Get snapshot."""
+
+        data = self._get_data_or_404(nvr_id)
+        if isinstance(data, web.Response):
+            return data
+
+        camera = self._async_get_camera(data, camera_id)
+        if camera is None:
+            return _404(f"Invalid camera ID: {camera_id}")
+        if not camera.can_read_media(data.api.bootstrap.auth_user):
+            return _403(f"User cannot read media from camera: {camera.id}")
+
+        width: int | str | None = request.query.get("width")
+        height: int | str | None = request.query.get("height")
+
+        if width is not None:
+            try:
+                width = int(width)
+            except ValueError:
+                return _400("Invalid width param")
+        if height is not None:
+            try:
+                height = int(height)
+            except ValueError:
+                return _400("Invalid height param")
+
+        try:
+            timestamp_dt = datetime.fromisoformat(timestamp)
+        except ValueError:
+            return _400("Invalid timestamp")
+
+        try:
+            snapshot = await camera.get_snapshot(
+                width=width, height=height, dt=timestamp_dt
+            )
+        except ClientError as err:
+            return _404(err)
+
+        if snapshot is None:
+            return _404("snapshot not found")
+
+        return web.Response(body=snapshot, content_type="image/jpeg")
+
+
 class VideoProxyView(ProtectProxyView):
     """View to proxy video clips from UniFi Protect."""
 

tests/components/unifiprotect/test_views.py

@@ -12,6 +12,7 @@ from uiprotect.exceptions import ClientError
 from homeassistant.components.unifiprotect.views import (
     async_generate_event_video_url,
     async_generate_proxy_event_video_url,
+    async_generate_snapshot_url,
     async_generate_thumbnail_url,
 )
 from homeassistant.core import HomeAssistant
@@ -169,6 +170,231 @@ async def test_thumbnail_invalid_entry_entry_id(
     assert response.status == 404
 
 
+async def test_snapshot_bad_nvr_id(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot URL with bad NVR id."""
+
+    ufp.api.request = AsyncMock()
+    await init_entry(hass, ufp, [camera])
+
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, fixed_now)
+    url = url.replace(ufp.api.bootstrap.nvr.id, "bad_id")
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 404
+    ufp.api.request.assert_not_called()
+
+
+async def test_snapshot_bad_camera_id(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot URL with bad camera id."""
+
+    ufp.api.request = AsyncMock()
+    await init_entry(hass, ufp, [camera])
+
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, fixed_now)
+    url = url.replace(camera.id, "bad_id")
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 404
+    ufp.api.request.assert_not_called()
+
+
+async def test_snapshot_bad_camera_perms(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot URL with bad camera perms."""
+
+    ufp.api.request = AsyncMock()
+    await init_entry(hass, ufp, [camera])
+
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, fixed_now)
+
+    ufp.api.bootstrap.auth_user.all_permissions = []
+    ufp.api.bootstrap.auth_user._perm_cache = {}
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 403
+    ufp.api.request.assert_not_called()
+
+
+async def test_snapshot_bad_timestamp(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot URL with bad timestamp params."""
+
+    ufp.api.request = AsyncMock()
+    await init_entry(hass, ufp, [camera])
+
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, fixed_now)
+    url = url.replace(fixed_now.replace(microsecond=0).isoformat(), "bad_time")
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 400
+    ufp.api.request.assert_not_called()
+
+
+async def test_snapshot_client_error(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot triggers client error at API."""
+
+    ufp.api.get_camera_snapshot = AsyncMock(side_effect=ClientError())
+
+    tomorrow = fixed_now + timedelta(days=1)
+
+    await init_entry(hass, ufp, [camera])
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, tomorrow)
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 404
+    ufp.api.get_camera_snapshot.assert_called_once()
+
+
+async def test_snapshot_notfound(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot not found."""
+
+    ufp.api.get_camera_snapshot = AsyncMock(return_value=None)
+
+    tomorrow = fixed_now + timedelta(days=1)
+
+    await init_entry(hass, ufp, [camera])
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, tomorrow)
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 404
+    ufp.api.get_camera_snapshot.assert_called_once()
+
+
+@pytest.mark.parametrize(("width", "height"), [("test", None), (None, "test")])
+async def test_snapshot_bad_params(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+    width: Any,
+    height: Any,
+) -> None:
+    """Test invalid bad query parameters."""
+
+    ufp.api.request = AsyncMock()
+    await init_entry(hass, ufp, [camera])
+
+    url = async_generate_snapshot_url(
+        ufp.api.bootstrap.nvr.id, camera.id, fixed_now, width=width, height=height
+    )
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    assert response.status == 400
+    ufp.api.request.assert_not_called()
+
+
+async def test_snapshot(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+) -> None:
+    """Test snapshot at timestamp in URL."""
+
+    ufp.api.get_camera_snapshot = AsyncMock(return_value=b"testtest")
+    await init_entry(hass, ufp, [camera])
+
+    # replace microseconds to match behavior of underlying library
+    fixed_now = fixed_now.replace(microsecond=0)
+    url = async_generate_snapshot_url(ufp.api.bootstrap.nvr.id, camera.id, fixed_now)
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    # verify when height is None that it is called with camera high channel height
+    height = camera.high_camera_channel.height
+
+    assert response.status == 200
+    assert response.content_type == "image/jpeg"
+    assert await response.content.read() == b"testtest"
+    ufp.api.get_camera_snapshot.assert_called_once_with(
+        camera.id, None, height, dt=fixed_now
+    )
+
+
+@pytest.mark.parametrize(("width", "height"), [(123, None), (None, 456), (123, 456)])
+async def test_snapshot_with_dimensions(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    ufp: MockUFPFixture,
+    camera: Camera,
+    fixed_now: datetime,
+    width: Any,
+    height: Any,
+) -> None:
+    """Test snapshot at timestamp in URL with specified width and height."""
+
+    ufp.api.get_camera_snapshot = AsyncMock(return_value=b"testtest")
+    await init_entry(hass, ufp, [camera])
+
+    # Replace microseconds to match behavior of underlying library
+    fixed_now = fixed_now.replace(microsecond=0)
+    url = async_generate_snapshot_url(
+        ufp.api.bootstrap.nvr.id, camera.id, fixed_now, width=width, height=height
+    )
+
+    http_client = await hass_client()
+    response = cast(ClientResponse, await http_client.get(url))
+
+    # Assertions
+    assert response.status == 200
+    assert response.content_type == "image/jpeg"
+    assert await response.content.read() == b"testtest"
+    ufp.api.get_camera_snapshot.assert_called_once_with(
+        camera.id, width, height, dt=fixed_now
+    )
+
+
 async def test_video_bad_event(
     hass: HomeAssistant,
     ufp: MockUFPFixture,