From e7fef30f9337274c598ecb3b8df5c67ee8f41661 Mon Sep 17 00:00:00 2001 From: Maciej Bieniek Date: Tue, 15 Sep 2026 18:13:58 +0200 Subject: [PATCH] Use the Shelly Camera `snapshot` endpoint to get a still image from the camera (#182139) --- homeassistant/components/shelly/camera.py | 38 +++++++-- homeassistant/components/shelly/strings.json | 3 + tests/components/shelly/test_camera.py | 90 +++++++++++++++++++- 3 files changed, 122 insertions(+), 9 deletions(-) diff --git a/homeassistant/components/shelly/camera.py b/homeassistant/components/shelly/camera.py index 8e8662d90692..051561b57c19 100644 --- a/homeassistant/components/shelly/camera.py +++ b/homeassistant/components/shelly/camera.py @@ -1,9 +1,11 @@ """Support for Shelly cameras.""" from dataclasses import dataclass -from typing import Final, override +from typing import TYPE_CHECKING, Final, override from urllib.parse import quote +from aioshelly.exceptions import DeviceConnectionError, HttpCallError, InvalidAuthError + from homeassistant.components.camera import ( Camera, CameraEntityDescription, @@ -11,8 +13,10 @@ from homeassistant.components.camera import ( ) from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from .const import DOMAIN from .coordinator import ShellyConfigEntry, ShellyRpcCoordinator from .entity import ( RpcEntityDescription, @@ -134,7 +138,31 @@ class ShellyCameraEntity(ShellyRpcAttributeEntity, Camera): return f"rtsp://{host}/stream/{self.entity_description.stream}" @override - @property - def use_stream_for_stills(self) -> bool: - """Use the RTSP stream to generate still images.""" - return True + async def async_camera_image( + self, width: int | None = None, height: int | None = None + ) -> bytes | None: + """Return a still image from the camera snapshot endpoint.""" + if TYPE_CHECKING: + assert self._id is not None + + try: + return await self.coordinator.device.camera_get_image(self._id) + except DeviceConnectionError as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="device_communication_error", + translation_placeholders={ + "device": self.coordinator.name, + }, + ) from err + except HttpCallError as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="http_call_error", + translation_placeholders={ + "device": self.coordinator.name, + }, + ) from err + except InvalidAuthError: + await self.coordinator.async_shutdown_device_and_start_reauth() + return None diff --git a/homeassistant/components/shelly/strings.json b/homeassistant/components/shelly/strings.json index 17f4be9a709b..8bff186c92f4 100644 --- a/homeassistant/components/shelly/strings.json +++ b/homeassistant/components/shelly/strings.json @@ -658,6 +658,9 @@ "firmware_unsupported": { "message": "{device} is running an unsupported firmware, please update the firmware" }, + "http_call_error": { + "message": "HTTP call error occurred for {device}" + }, "invalid_trigger": { "message": "Invalid device automation trigger (type, subtype): {trigger}" }, diff --git a/tests/components/shelly/test_camera.py b/tests/components/shelly/test_camera.py index fd49c646c309..bfd839f0e149 100644 --- a/tests/components/shelly/test_camera.py +++ b/tests/components/shelly/test_camera.py @@ -2,9 +2,15 @@ from collections.abc import Generator from copy import deepcopy -from unittest.mock import Mock, patch +from unittest.mock import AsyncMock, Mock, patch from aioshelly.const import MODEL_CAMERA +from aioshelly.exceptions import ( + DeviceConnectionError, + DeviceConnectionTimeoutError, + HttpCallError, + InvalidAuthError, +) import pytest from syrupy.assertion import SnapshotAssertion @@ -14,7 +20,8 @@ from homeassistant.components.camera import ( CameraState, get_camera_from_entity_id, ) -from homeassistant.components.shelly.const import CONF_SLEEP_PERIOD +from homeassistant.components.shelly.const import CONF_SLEEP_PERIOD, DOMAIN +from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState from homeassistant.const import ( CONF_HOST, CONF_MODEL, @@ -23,6 +30,7 @@ from homeassistant.const import ( Platform, ) from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_registry import EntityRegistry from . import MOCK_MAC, init_integration, patch_platforms, register_entity @@ -97,11 +105,85 @@ async def test_camera_use_stream_for_stills( hass: HomeAssistant, mock_camera_rpc_device: Mock, ) -> None: - """Test use_stream_for_stills returns True (still images from the RTSP stream).""" + """Test use_stream_for_stills returns False (still images from HTTP snapshot).""" await init_integration(hass, 3, model=MODEL_CAMERA) camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID) - assert camera.use_stream_for_stills is True + assert camera.use_stream_for_stills is False + + +async def test_camera_image( + hass: HomeAssistant, + mock_camera_rpc_device: Mock, +) -> None: + """Test async_camera_image returns snapshot from the device.""" + await init_integration(hass, 3, model=MODEL_CAMERA) + + camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID) + mock_camera_rpc_device.camera_get_image = AsyncMock(return_value=b"fake-image") + + assert await camera.async_camera_image() == b"fake-image" + assert mock_camera_rpc_device.camera_get_image.call_count == 1 + assert mock_camera_rpc_device.camera_get_image.call_args[0] == (0,) + + +@pytest.mark.parametrize( + ("exception", "error"), + [ + ( + DeviceConnectionTimeoutError, + "Device communication error occurred for Test name", + ), + ( + DeviceConnectionError, + "Device communication error occurred for Test name", + ), + ( + HttpCallError(500, "Server error"), + "HTTP call error occurred for Test name", + ), + ], +) +async def test_camera_image_exc( + hass: HomeAssistant, + mock_camera_rpc_device: Mock, + exception: Exception, + error: str, +) -> None: + """Test camera snapshot with exception.""" + await init_integration(hass, 3, model=MODEL_CAMERA) + + camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID) + mock_camera_rpc_device.camera_get_image = AsyncMock(side_effect=exception) + + with pytest.raises(HomeAssistantError, match=error): + await camera.async_camera_image() + + +async def test_camera_image_reauth_error( + hass: HomeAssistant, mock_camera_rpc_device: Mock +) -> None: + """Test camera snapshot with authentication error starts reauth.""" + entry = await init_integration(hass, 3, model=MODEL_CAMERA) + + camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID) + mock_camera_rpc_device.camera_get_image = AsyncMock( + side_effect=InvalidAuthError(401) + ) + + assert await camera.async_camera_image() is None + assert entry.state is ConfigEntryState.LOADED + + flows = hass.config_entries.flow.async_progress() + assert len(flows) == 1 + + flow = flows[0] + assert flow.get("step_id") == "reauth_confirm" + assert flow.get("handler") == DOMAIN + + assert "context" in flow + assert flow["context"].get("source") == SOURCE_REAUTH + assert flow["context"].get("entry_id") == entry.entry_id async def test_camera_stream_source(