Add CameraCapabilities (#128455)

homeassistant/components/camera/__init__.py

@@ -6,7 +6,7 @@ import asyncio
 import collections
 from collections.abc import Awaitable, Callable, Coroutine
 from contextlib import suppress
-from dataclasses import asdict
+from dataclasses import asdict, dataclass
 from datetime import datetime, timedelta
 from enum import IntFlag
 from functools import partial
@@ -18,7 +18,7 @@ from typing import Any, Final, final
 
 from aiohttp import hdrs, web
 import attr
-from propcache import cached_property
+from propcache import cached_property, under_cached_property
 import voluptuous as vol
 from webrtc_models import RTCIceServer
 
@@ -177,6 +177,13 @@ class Image:
     content: bytes = attr.ib()
 
 
+@dataclass(frozen=True)
+class CameraCapabilities:
+    """Camera capabilities."""
+
+    frontend_stream_types: set[StreamType]
+
+
 @bind_hass
 async def async_request_stream(hass: HomeAssistant, entity_id: str, fmt: str) -> str:
     """Request a stream for a camera entity."""
@@ -352,6 +359,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     websocket_api.async_register_command(hass, ws_camera_stream)
     websocket_api.async_register_command(hass, websocket_get_prefs)
     websocket_api.async_register_command(hass, websocket_update_prefs)
+    websocket_api.async_register_command(hass, ws_camera_capabilities)
     async_register_ws(hass)
 
     await component.async_setup(config)
@@ -463,6 +471,7 @@ class Camera(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
 
     def __init__(self) -> None:
         """Initialize a camera."""
+        self._cache: dict[str, Any] = {}
         self.stream: Stream | None = None
         self.stream_options: dict[str, str | bool | float] = {}
         self.content_type: str = DEFAULT_CONTENT_TYPE
@@ -791,6 +800,7 @@ class Camera(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
         if old_provider != new_provider or old_legacy_provider != new_legacy_provider:
             self._webrtc_provider = new_provider
             self._legacy_webrtc_provider = new_legacy_provider
+            self._invalidate_camera_capabilities_cache()
             if write_state:
                 self.async_write_ha_state()
 
@@ -840,6 +850,31 @@ class Camera(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
         if self._webrtc_provider:
             self._webrtc_provider.async_close_session(session_id)
 
+    @callback
+    def _invalidate_camera_capabilities_cache(self) -> None:
+        """Invalidate the camera capabilities cache."""
+        self._cache.pop("camera_capabilities", None)
+
+    @final
+    @under_cached_property
+    def camera_capabilities(self) -> CameraCapabilities:
+        """Return the camera capabilities."""
+        frontend_stream_types = set()
+        if CameraEntityFeature.STREAM in self.supported_features_compat:
+            if (
+                type(self).async_handle_web_rtc_offer
+                != Camera.async_handle_web_rtc_offer
+            ):
+                # The camera has a native WebRTC implementation
+                frontend_stream_types.add(StreamType.WEB_RTC)
+            else:
+                frontend_stream_types.add(StreamType.HLS)
+
+                if self._webrtc_provider:
+                    frontend_stream_types.add(StreamType.WEB_RTC)
+
+        return CameraCapabilities(frontend_stream_types)
+
 
 class CameraView(HomeAssistantView):
     """Base CameraView."""
@@ -930,6 +965,24 @@ class CameraMjpegStream(CameraView):
             raise web.HTTPBadRequest from err
 
 
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "camera/capabilities",
+        vol.Required("entity_id"): cv.entity_id,
+    }
+)
+@websocket_api.async_response
+async def ws_camera_capabilities(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Handle get camera capabilities websocket command.
+
+    Async friendly.
+    """
+    camera = get_camera_from_entity_id(hass, msg["entity_id"])
+    connection.send_result(msg["id"], asdict(camera.camera_capabilities))
+
+
 @websocket_api.websocket_command(
     {
         vol.Required("type"): "camera/stream",

tests/components/camera/common.py

@@ -6,6 +6,13 @@ components. Instead call the service directly.
 
 from unittest.mock import Mock
 
+from homeassistant.components.camera import Camera
+from homeassistant.components.camera.webrtc import (
+    CameraWebRTCProvider,
+    async_register_webrtc_provider,
+)
+from homeassistant.core import HomeAssistant
+
 EMPTY_8_6_JPEG = b"empty_8_6"
 WEBRTC_ANSWER = "a=sendonly"
 STREAM_SOURCE = "rtsp://127.0.0.1/stream"
@@ -23,3 +30,25 @@ def mock_turbo_jpeg(
     mocked_turbo_jpeg.scale_with_quality.return_value = EMPTY_8_6_JPEG
     mocked_turbo_jpeg.encode.return_value = EMPTY_8_6_JPEG
     return mocked_turbo_jpeg
+
+
+async def add_webrtc_provider(hass: HomeAssistant) -> CameraWebRTCProvider:
+    """Add test WebRTC provider."""
+
+    class SomeTestProvider(CameraWebRTCProvider):
+        """Test provider."""
+
+        async def async_is_supported(self, stream_source: str) -> bool:
+            """Determine if the provider supports the stream source."""
+            return True
+
+        async def async_handle_web_rtc_offer(
+            self, camera: Camera, offer_sdp: str
+        ) -> str | None:
+            """Handle the WebRTC offer and return an answer."""
+            return "answer"
+
+    provider = SomeTestProvider()
+    async_register_webrtc_provider(hass, provider)
+    await hass.async_block_till_done()
+    return provider

tests/components/camera/test_init.py

@@ -13,8 +13,11 @@ from homeassistant.components.camera.const import (
     DOMAIN,
     PREF_ORIENTATION,
     PREF_PRELOAD_STREAM,
+    StreamType,
 )
+from homeassistant.components.camera.helper import get_camera_from_entity_id
 from homeassistant.components.websocket_api import TYPE_RESULT
+from homeassistant.config_entries import ConfigEntry, ConfigFlow
 from homeassistant.const import (
     ATTR_ENTITY_ID,
     EVENT_HOMEASSISTANT_STARTED,
@@ -27,12 +30,24 @@ from homeassistant.helpers import entity_registry as er, issue_registry as ir
 from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util
 
-from .common import EMPTY_8_6_JPEG, STREAM_SOURCE, mock_turbo_jpeg
+from .common import (
+    EMPTY_8_6_JPEG,
+    STREAM_SOURCE,
+    WEBRTC_ANSWER,
+    add_webrtc_provider,
+    mock_turbo_jpeg,
+)
 
 from tests.common import (
+    MockConfigEntry,
+    MockModule,
     async_fire_time_changed,
     help_test_all,
     import_and_test_deprecated_constant_enum,
+    mock_config_flow,
+    mock_integration,
+    mock_platform,
+    setup_test_component_platform,
 )
 from tests.typing import ClientSessionGenerator, WebSocketGenerator
 
@@ -885,3 +900,114 @@ async def test_entity_picture_url_changes_on_token_update(hass: HomeAssistant) -
     new_entity_picture = camera_state.attributes["entity_picture"]
     assert new_entity_picture != original_picture
     assert "token=" in new_entity_picture
+
+
+async def _test_capabilities(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    entity_id: str,
+    expected_stream_types: set[StreamType],
+    expected_stream_types_with_webrtc_provider: set[StreamType],
+) -> None:
+    """Test camera capabilities."""
+    await async_setup_component(hass, "camera", {})
+    await hass.async_block_till_done()
+
+    async def test(expected_types: set[StreamType]) -> None:
+        camera_obj = get_camera_from_entity_id(hass, entity_id)
+        capabilities = camera_obj.camera_capabilities
+        assert capabilities == camera.CameraCapabilities(expected_types)
+
+        # Request capabilities through WebSocket
+        client = await hass_ws_client(hass)
+        await client.send_json_auto_id(
+            {"type": "camera/capabilities", "entity_id": entity_id}
+        )
+        msg = await client.receive_json()
+
+        # Assert WebSocket response
+        assert msg["type"] == TYPE_RESULT
+        assert msg["success"]
+        assert msg["result"] == {"frontend_stream_types": list(expected_types)}
+
+    await test(expected_stream_types)
+
+    # Test with WebRTC provider
+    await add_webrtc_provider(hass)
+    await test(expected_stream_types_with_webrtc_provider)
+
+
+@pytest.mark.usefixtures("mock_camera", "mock_stream_source")
+async def test_camera_capabilities_hls(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test HLS camera capabilities."""
+    await _test_capabilities(
+        hass,
+        hass_ws_client,
+        "camera.demo_camera",
+        {StreamType.HLS},
+        {StreamType.HLS, StreamType.WEB_RTC},
+    )
+
+
+async def test_camera_capabilities_webrtc(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test WebRTC camera capabilities."""
+
+    # Cannot use the fixture mock_camera_web_rtc as it's mocking Camera.async_handle_web_rtc_offer
+    # Camera capabilities are determined by by checking if the function was overwritten(implemented) or not
+    class MockCamera(camera.Camera):
+        """Mock Camera Entity."""
+
+        _attr_name = "Test"
+        _attr_supported_features: camera.CameraEntityFeature = (
+            camera.CameraEntityFeature.STREAM
+        )
+
+        async def stream_source(self) -> str | None:
+            return STREAM_SOURCE
+
+        async def async_handle_web_rtc_offer(self, offer_sdp: str) -> str | None:
+            return WEBRTC_ANSWER
+
+    domain = "test"
+
+    entry = MockConfigEntry(domain=domain)
+    entry.add_to_hass(hass)
+
+    async def async_setup_entry_init(
+        hass: HomeAssistant, config_entry: ConfigEntry
+    ) -> bool:
+        """Set up test config entry."""
+        await hass.config_entries.async_forward_entry_setups(config_entry, [DOMAIN])
+        return True
+
+    async def async_unload_entry_init(
+        hass: HomeAssistant, config_entry: ConfigEntry
+    ) -> bool:
+        """Unload test config entry."""
+        await hass.config_entries.async_forward_entry_unload(config_entry, DOMAIN)
+        return True
+
+    mock_integration(
+        hass,
+        MockModule(
+            domain,
+            async_setup_entry=async_setup_entry_init,
+            async_unload_entry=async_unload_entry_init,
+        ),
+    )
+    setup_test_component_platform(hass, DOMAIN, [MockCamera()], from_config_entry=True)
+    mock_platform(hass, f"{domain}.config_flow", Mock())
+
+    with mock_config_flow(domain, ConfigFlow):
+        assert await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    await _test_capabilities(
+        hass, hass_ws_client, "camera.test", {StreamType.WEB_RTC}, {StreamType.WEB_RTC}
+    )