Use the Shelly Camera snapshot endpoint to get a still image from the camera (#182139)

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

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}"
     },

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(