User camera unique id in go2rtc if available

homeassistant/components/go2rtc/__init__.py

@@ -67,7 +67,7 @@
     RECOMMENDED_VERSION,
 )
 from .server import Server
-from .util import get_go2rtc_unix_socket_path
+from .util import get_camera_identifier, get_go2rtc_unix_socket_path
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -308,7 +308,7 @@ async def async_handle_async_webrtc_offer(
             return
 
         self._sessions[session_id] = ws_client = Go2RtcWsClient(
-            self._session, self._url, source=camera.entity_id
+            self._session, self._url, source=get_camera_identifier(camera)
         )
 
         @callback
@@ -354,7 +354,7 @@ async def async_get_image(
         """Get an image from the camera."""
         await self._update_stream_source(camera)
         return await self._rest_client.get_jpeg_snapshot(
-            camera.entity_id, width, height
+            get_camera_identifier(camera), width, height
         )
 
     async def _update_stream_source(self, camera: Camera) -> None:
@@ -399,18 +399,19 @@ async def _update_stream_source(self, camera: Camera) -> None:
                     stream_source += "#rotate=90"
 
         streams = await self._rest_client.streams.list()
+        identifier = get_camera_identifier(camera)
 
-        if (stream := streams.get(camera.entity_id)) is None or not any(
+        if (stream := streams.get(identifier)) is None or not any(
             stream_source == producer.url for producer in stream.producers
         ):
             await self._rest_client.streams.add(
-                camera.entity_id,
+                identifier,
                 [
                     stream_source,
                     # We are setting any ffmpeg rtsp related logs to debug
                     # Connection problems to the camera will be logged by the first stream
                     # Therefore setting it to debug will not hide any important logs
-                    f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
+                    f"ffmpeg:{identifier}#audio=opus#query=log_level=debug",
                 ],
             )
 

homeassistant/components/go2rtc/util.py

@@ -1,12 +1,27 @@
 """Go2rtc utility functions."""
 
 from pathlib import Path
+import string
+from urllib.parse import quote
+
+from homeassistant.components.camera import Camera
 
 _HA_MANAGED_UNIX_SOCKET_FILE = "go2rtc.sock"
+# Go2rtc is not validating the camera identifier, but some characters (e.g. : or #)
+# have special meaning in URLs and could cause issues.
+_SAFE_CHARS = string.ascii_letters + string.digits + "._-"
 
 
 def get_go2rtc_unix_socket_path(path: str | Path) -> str:
     """Get the Go2rtc unix socket path."""
     if not isinstance(path, Path):
         path = Path(path)
     return str(path / _HA_MANAGED_UNIX_SOCKET_FILE)
+
+
+def get_camera_identifier(camera: Camera) -> str:
+    """Get the Go2rtc camera identifier."""
+    attr = camera.entity_id
+    if camera.unique_id is not None:
+        attr = f"{camera.platform.platform_name}_{camera.unique_id}"
+    return quote(attr, safe=_SAFE_CHARS)

tests/components/go2rtc/__init__.py

@@ -9,10 +9,11 @@ class MockCamera(Camera):
     _attr_name = "Test"
     _attr_supported_features: CameraEntityFeature = CameraEntityFeature.STREAM
 
-    def __init__(self) -> None:
+    def __init__(self, unique_id: str | None) -> None:
         """Initialize the mock entity."""
         super().__init__()
         self._stream_source: str | None = "rtsp://stream"
+        self._attr_unique_id = unique_id
 
     def set_stream_source(self, stream_source: str | None) -> None:
         """Set the stream source."""

tests/components/go2rtc/conftest.py

@@ -185,10 +185,17 @@ def integration_config_entry(hass: HomeAssistant) -> ConfigEntry:
     return entry
 
 
+@pytest.fixture
+def camera_unique_id() -> str | None:
+    """Camera unique ID."""
+    return "camera_unique_id"
+
+
 @pytest.fixture
 async def init_test_integration(
     hass: HomeAssistant,
     integration_config_entry: ConfigEntry,
+    camera_unique_id: str | None,
 ) -> MockCamera:
     """Initialize components."""
 
@@ -218,7 +225,7 @@ async def async_unload_entry_init(
             async_unload_entry=async_unload_entry_init,
         ),
     )
-    test_camera = MockCamera()
+    test_camera = MockCamera(camera_unique_id)
     setup_test_component_platform(
         hass, CAMERA_DOMAIN, [test_camera], from_config_entry=True
     )

tests/components/go2rtc/test_init.py

@@ -42,7 +42,10 @@
     DOMAIN,
     RECOMMENDED_VERSION,
 )
-from homeassistant.components.go2rtc.util import get_go2rtc_unix_socket_path
+from homeassistant.components.go2rtc.util import (
+    get_camera_identifier,
+    get_go2rtc_unix_socket_path,
+)
 from homeassistant.components.stream import Orientation
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import CONF_PASSWORD, CONF_URL, CONF_USERNAME
@@ -87,7 +90,7 @@ async def _test_setup_and_signaling(
     camera: MockCamera,
 ) -> None:
     """Test the go2rtc config entry."""
-    entity_id = camera.entity_id
+    identifier = get_camera_identifier(camera)
     assert camera.camera_capabilities.frontend_stream_types == {StreamType.HLS}
 
     assert await async_setup_component(hass, DOMAIN, config)
@@ -124,35 +127,35 @@ async def test(session: str) -> None:
     await test("sesion_1")
 
     rest_client.streams.add.assert_called_once_with(
-        entity_id,
+        identifier,
         [
             "rtsp://stream",
-            f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
+            f"ffmpeg:{identifier}#audio=opus#query=log_level=debug",
         ],
     )
 
     # Stream exists but the source is different
     rest_client.streams.add.reset_mock()
     rest_client.streams.list.return_value = {
-        entity_id: Stream([Producer("rtsp://different")])
+        identifier: Stream([Producer("rtsp://different")])
     }
 
     receive_message_callback.reset_mock()
     ws_client.reset_mock()
     await test("session_2")
 
     rest_client.streams.add.assert_called_once_with(
-        entity_id,
+        identifier,
         [
             "rtsp://stream",
-            f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
+            f"ffmpeg:{identifier}#audio=opus#query=log_level=debug",
         ],
     )
 
     # If the stream is already added, the stream should not be added again.
     rest_client.streams.add.reset_mock()
     rest_client.streams.list.return_value = {
-        entity_id: Stream([Producer("rtsp://stream")])
+        identifier: Stream([Producer("rtsp://stream")])
     }
 
     receive_message_callback.reset_mock()
@@ -192,6 +195,14 @@ async def test(session: str) -> None:
     "mock_is_docker_env",
     "mock_go2rtc_entry",
 )
+@pytest.mark.parametrize(
+    "camera_unique_id",
+    [
+        "camera_unique_id",
+        None,
+    ],
+    indirect=True,
+)
 @pytest.mark.parametrize(
     ("config", "ui_enabled", "expected_username", "expected_password"),
     [
@@ -283,6 +294,14 @@ def after_setup() -> None:
 
 
 @pytest.mark.usefixtures("mock_go2rtc_entry")
+@pytest.mark.parametrize(
+    "camera_unique_id",
+    [
+        "camera_unique_id",
+        None,
+    ],
+    indirect=True,
+)
 @pytest.mark.parametrize(
     ("go2rtc_binary", "is_docker_env"),
     [
@@ -816,13 +835,14 @@ async def test_generic_workaround(
         image = await async_get_image(hass, camera.entity_id)
         assert image.content == image_bytes
 
-    rest_client.streams.add.assert_called_once_with(
-        camera.entity_id,
-        [
-            "ffmpeg:https://my_stream_url.m3u8",
-            f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
-        ],
-    )
+        identifier = get_camera_identifier(camera)
+        rest_client.streams.add.assert_called_once_with(
+            identifier,
+            [
+                "ffmpeg:https://my_stream_url.m3u8",
+                f"ffmpeg:{identifier}#audio=opus#query=log_level=debug",
+            ],
+        )
 
 
 async def _test_camera_orientation(
@@ -849,11 +869,12 @@ async def _test_camera_orientation(
     await camera_fn(hass, camera)
 
     # Verify the stream was configured correctly
+    identifier = get_camera_identifier(camera)
     rest_client.streams.add.assert_called_once_with(
-        camera.entity_id,
+        identifier,
         [
             expected_stream_source,
-            f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
+            f"ffmpeg:{identifier}#audio=opus#query=log_level=debug",
         ],
     )
 

tests/components/go2rtc/test_util.py

@@ -0,0 +1,44 @@
+"""Go2rtc utility function tests."""
+
+from unittest.mock import Mock
+
+import pytest
+
+from homeassistant.components.camera import Camera
+from homeassistant.components.go2rtc.util import get_camera_identifier
+
+
+@pytest.mark.parametrize(
+    ("unique_id", "entity_id", "expected"),
+    [
+        # Prefer unique_id over entity_id
+        ("unique123", "camera.test", "test_unique123"),
+        # Fall back to entity_id when unique_id is None
+        (None, "camera.test", "camera.test"),
+        # Safe characters pass through
+        ("abc-def_ghi.123", "camera.test", "test_abc-def_ghi.123"),
+        # Special characters are percent-encoded
+        ("cam#1", "camera.test", "test_cam%231"),
+        ("cam:1", "camera.test", "test_cam%3A1"),
+        ("cam/1", "camera.test", "test_cam%2F1"),
+        ("cam?1", "camera.test", "test_cam%3F1"),
+        ("cam&1", "camera.test", "test_cam%261"),
+        ("cam=1", "camera.test", "test_cam%3D1"),
+        ("cam%1", "camera.test", "test_cam%251"),
+        ("cam 1", "camera.test", "test_cam%201"),
+        ("cam@1", "camera.test", "test_cam%401"),
+        ("cam_1", "camera.test", "test_cam_1"),
+        ("cam%231", "camera.test", "test_cam%25231"),
+        # Non-ASCII: UTF-8 byte-wise encoding (€ = E2 82 AC)
+        ("cam€1", "camera.test", "test_cam%E2%82%AC1"),
+    ],
+)
+def test_get_camera_identifier(
+    unique_id: str | None, entity_id: str, expected: str
+) -> None:
+    """Test get_camera_identifier sanitizes and prefers unique_id."""
+    camera = Mock(spec_set=Camera)
+    camera.platform.platform_name = "test"
+    camera.unique_id = unique_id
+    camera.entity_id = entity_id
+    assert get_camera_identifier(camera) == expected