Add `rtsptowebrtc` integration (#59660)

* Add initial version of the webrtc integration

Add the webrtc integration. This integration proxies the
signal 'offer' from the client to a RTSPtoWebRTCP server that
returns an 'answer'.

The RTSPtoWebRTC server is a go binary based on pion, and this is
what is currently used by the WebRTC custom_component:
https://github.com/AlexxIT/WebRTC
https://github.com/deepch/RTSPtoWebRTC

* Readability improvements for webrtc

* Reach 100% test coverage

* Use rtsp-to-webrtc client library package

* Rename webrtc to rtstptowebrtc

This is to reflect naming as one type of approach to webrtc since other webrtc integrations would look very different.

* Remove internal quality scale

* Bump rtsptowebrtc to support heartbeats

* Shorten server url variable and remove const.py

* Add config flow validation for RTSPtoWebRTC server

* Add RTSPtoWebRTC server health checks

* Accept translation suggestion

* Apply suggestions from code review

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

* Update rtsptowebrtc to use new camera registry API

Update rtsptowebrtc to use new API added in #62962

* Remove unused variable

* Fix lint and typing errors for python 3.8

* Rename to rtsp_to_webrtc to follow standards

* Use async_on_unload for unsubscribing camera webrtc provider

* Remove unnecessary translations in config flow

* Remove unnecessary configuration setup

* Cleanup test setup and typing

* Patch integration setup to avoid starting the whole integration

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

.strict-typing

@@ -117,6 +117,7 @@ homeassistant.components.renault.*
 homeassistant.components.ridwell.*
 homeassistant.components.rituals_perfume_genie.*
 homeassistant.components.rpi_power.*
+homeassistant.components.rtsp_to_webrtc.*
 homeassistant.components.samsungtv.*
 homeassistant.components.scene.*
 homeassistant.components.select.*

CODEOWNERS

@@ -776,6 +776,8 @@ tests/components/roon/* @pavoni
 homeassistant/components/rpi_gpio_pwm/* @soldag
 homeassistant/components/rpi_power/* @shenxn @swetoast
 tests/components/rpi_power/* @shenxn @swetoast
+homeassistant/components/rtsp_to_webrtc/* @allenporter
+tests/components/rtsp_to_webrtc/* @allenporter
 homeassistant/components/ruckus_unleashed/* @gabe565
 tests/components/ruckus_unleashed/* @gabe565
 homeassistant/components/safe_mode/* @home-assistant/core

homeassistant/components/rtsp_to_webrtc/__init__.py

@@ -0,0 +1,83 @@
+"""RTSPtoWebRTC integration with an external RTSPToWebRTC Server.
+
+WebRTC uses a direct communication from the client (e.g. a web browser) to a
+camera device. Home Assistant acts as the signal path for initial set up,
+passing through the client offer and returning a camera answer, then the client
+and camera communicate directly.
+
+However, not all cameras natively support WebRTC. This integration is a shim
+for camera devices that support RTSP streams only, relying on an external
+server RTSPToWebRTC that is a proxy. Home Assistant does not participate in
+the offer/answer SDP protocol, other than as a signal path pass through.
+
+Other integrations may use this integration with these steps:
+- Check if this integration is loaded
+- Call is_suported_stream_source for compatibility
+- Call async_offer_for_stream_source to get back an answer for a client offer
+"""
+
+from __future__ import annotations
+
+import logging
+
+import async_timeout
+from rtsp_to_webrtc.client import Client
+from rtsp_to_webrtc.exceptions import ClientError, ResponseError
+
+from homeassistant.components import camera
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+_LOGGER = logging.getLogger(__name__)
+
+DOMAIN = "rtsp_to_webrtc"
+DATA_SERVER_URL = "server_url"
+DATA_UNSUB = "unsub"
+TIMEOUT = 10
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up RTSPtoWebRTC from a config entry."""
+    hass.data.setdefault(DOMAIN, {})
+
+    client = Client(async_get_clientsession(hass), entry.data[DATA_SERVER_URL])
+    try:
+        async with async_timeout.timeout(TIMEOUT):
+            await client.heartbeat()
+    except ResponseError as err:
+        raise ConfigEntryNotReady from err
+    except (TimeoutError, ClientError) as err:
+        raise ConfigEntryNotReady from err
+
+    async def async_offer_for_stream_source(
+        stream_source: str,
+        offer_sdp: str,
+    ) -> str:
+        """Handle the signal path for a WebRTC stream.
+
+        This signal path is used to route the offer created by the client to the
+        proxy server that translates a stream to WebRTC. The communication for
+        the stream itself happens directly between the client and proxy.
+        """
+        try:
+            async with async_timeout.timeout(TIMEOUT):
+                return await client.offer(offer_sdp, stream_source)
+        except TimeoutError as err:
+            raise HomeAssistantError("Timeout talking to RTSPtoWebRTC server") from err
+        except ClientError as err:
+            raise HomeAssistantError(str(err)) from err
+
+    entry.async_on_unload(
+        camera.async_register_rtsp_to_web_rtc_provider(
+            hass, DOMAIN, async_offer_for_stream_source
+        )
+    )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    return True

homeassistant/components/rtsp_to_webrtc/config_flow.py

@@ -0,0 +1,65 @@
+"""Config flow for RTSPtoWebRTC."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+from urllib.parse import urlparse
+
+import rtsp_to_webrtc
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from . import DATA_SERVER_URL, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+DATA_SCHEMA = vol.Schema({vol.Required(DATA_SERVER_URL): str})
+
+
+class RTSPToWebRTCConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """RTSPtoWebRTC config flow."""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Configure the RTSPtoWebRTC server url."""
+        if self._async_current_entries():
+            return self.async_abort(reason="single_instance_allowed")
+
+        if user_input is None:
+            return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
+
+        url = user_input[DATA_SERVER_URL]
+        result = urlparse(url)
+        if not all([result.scheme, result.netloc]):
+            return self.async_show_form(
+                step_id="user",
+                data_schema=DATA_SCHEMA,
+                errors={DATA_SERVER_URL: "invalid_url"},
+            )
+
+        errors = {}
+        client = rtsp_to_webrtc.client.Client(async_get_clientsession(self.hass), url)
+        try:
+            await client.heartbeat()
+        except rtsp_to_webrtc.exceptions.ResponseError as err:
+            _LOGGER.error("RTSPtoWebRTC server failure: %s", str(err))
+            errors["base"] = "server_failure"
+        except rtsp_to_webrtc.exceptions.ClientError as err:
+            _LOGGER.error("RTSPtoWebRTC communication failure: %s", str(err))
+            errors["base"] = "server_unreachable"
+        if errors:
+            return self.async_show_form(
+                step_id="user",
+                data_schema=DATA_SCHEMA,
+                errors=errors,
+            )
+
+        await self.async_set_unique_id(DOMAIN)
+        return self.async_create_entry(
+            title=url,
+            data={DATA_SERVER_URL: url},
+        )

homeassistant/components/rtsp_to_webrtc/manifest.json

@@ -0,0 +1,12 @@
+{
+  "domain": "rtsp_to_webrtc",
+  "name": "RTSPtoWebRTC",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/rtsp_to_webrtc",
+  "requirements": ["rtsp-to-webrtc==0.2.7"],
+  "dependencies": ["camera"],
+  "codeowners": [
+    "@allenporter"
+  ],
+  "iot_class": "local_push"
+}

homeassistant/components/rtsp_to_webrtc/strings.json

@@ -0,0 +1,21 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "title": "Configure RTSPtoWebRTC",
+        "description": "The RTSPtoWebRTC integration requires a server to translate RTSP streams into WebRTC. Enter the URL to the RTSPtoWebRTC server.",
+        "data": {
+          "server_url": "RTSPtoWebRTC server URL e.g. https://example.com"
+        }
+      }
+    },
+    "error": {
+      "invalid_url": "Must be a valid RTSPtoWebRTC server URL e.g. https://example.com",
+      "server_failure": "RTSPtoWebRTC server returned an error. Check logs for more information.",
+      "server_unreachable": "Unable to communicate with RTSPtoWebRTC server. Check logs for more information."
+    },
+    "abort": {
+      "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
+    }
+  }
+}

homeassistant/components/rtsp_to_webrtc/translations/en.json

@@ -0,0 +1,25 @@
+{
+    "config": {
+        "abort": {
+            "no_devices_found": "No devices found on the network",
+            "single_instance_allowed": "Already configured. Only a single configuration possible."
+        },
+        "error": {
+            "invalid_url": "Must be a valid RTSPtoWebRTC server URL e.g. https://example.com",
+            "server_failure": "RTSPtoWebRTC server returned an error. Check logs for more information.",
+            "server_unreachable": "Unable to communicate with RTSPtoWebRTC server. Check logs for more information."
+        },
+        "step": {
+            "confirm": {
+                "description": "Do you want to start set up?"
+            },
+            "user": {
+                "data": {
+                    "server_url": "RTSPtoWebRTC server URL e.g. https://example.com"
+                },
+                "description": "The RTSPtoWebRTC integration requires a server to translate RTSP streams into WebRTC. Enter the URL to the RTSPtoWebRTC server.\n",
+                "title": "Configure RTSPtoWebRTC"
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -262,6 +262,7 @@ FLOWS = [
     "roomba",
     "roon",
     "rpi_power",
+    "rtsp_to_webrtc",
     "ruckus_unleashed",
     "samsungtv",
     "screenlogic",

mypy.ini

@@ -1298,6 +1298,17 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.rtsp_to_webrtc.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.samsungtv.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -2110,6 +2110,9 @@ rpi-bad-power==0.1.0
 # homeassistant.components.rpi_rf
 # rpi-rf==0.9.7
 
+# homeassistant.components.rtsp_to_webrtc
+rtsp-to-webrtc==0.2.7
+
 # homeassistant.components.russound_rnet
 russound==0.1.9
 

requirements_test_all.txt

@@ -1275,6 +1275,9 @@ roonapi==0.0.38
 # homeassistant.components.rpi_power
 rpi-bad-power==0.1.0
 
+# homeassistant.components.rtsp_to_webrtc
+rtsp-to-webrtc==0.2.7
+
 # homeassistant.components.yamaha
 rxv==0.7.0
 

tests/components/rtsp_to_webrtc/__init__.py

@@ -0,0 +1 @@
+"""Tests for the RTSPtoWebRTC integration."""

tests/components/rtsp_to_webrtc/test_config_flow.py

@@ -0,0 +1,110 @@
+"""Test the RTSPtoWebRTC config flow."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import rtsp_to_webrtc
+
+from homeassistant import config_entries
+from homeassistant.components.rtsp_to_webrtc import DOMAIN
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def test_web_full_flow(hass: HomeAssistant) -> None:
+    """Check full flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result.get("type") == "form"
+    assert result.get("step_id") == "user"
+    assert result.get("data_schema").schema.get("server_url") == str
+    assert not result.get("errors")
+    assert "flow_id" in result
+    with patch("rtsp_to_webrtc.client.Client.heartbeat"), patch(
+        "homeassistant.components.rtsp_to_webrtc.async_setup_entry",
+        return_value=True,
+    ) as mock_setup:
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"server_url": "https://example.com"}
+        )
+        assert result.get("type") == "create_entry"
+        assert "result" in result
+        assert result["result"].data == {"server_url": "https://example.com"}
+
+        assert len(mock_setup.mock_calls) == 1
+
+
+async def test_single_config_entry(hass: HomeAssistant) -> None:
+    """Test that only a single config entry is allowed."""
+    old_entry = MockConfigEntry(domain=DOMAIN, data={"example": True})
+    old_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result.get("type") == "abort"
+    assert result.get("reason") == "single_instance_allowed"
+
+
+async def test_invalid_url(hass: HomeAssistant) -> None:
+    """Check full flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result.get("type") == "form"
+    assert result.get("step_id") == "user"
+    assert result.get("data_schema").schema.get("server_url") == str
+    assert not result.get("errors")
+    assert "flow_id" in result
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], {"server_url": "not-a-url"}
+    )
+
+    assert result.get("type") == "form"
+    assert result.get("step_id") == "user"
+    assert result.get("errors") == {"server_url": "invalid_url"}
+
+
+async def test_server_unreachable(hass: HomeAssistant) -> None:
+    """Exercise case where the server is unreachable."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result.get("type") == "form"
+    assert result.get("step_id") == "user"
+    assert not result.get("errors")
+    assert "flow_id" in result
+    with patch(
+        "rtsp_to_webrtc.client.Client.heartbeat",
+        side_effect=rtsp_to_webrtc.exceptions.ClientError(),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"server_url": "https://example.com"}
+        )
+        assert result.get("type") == "form"
+        assert result.get("step_id") == "user"
+        assert result.get("errors") == {"base": "server_unreachable"}
+
+
+async def test_server_failure(hass: HomeAssistant) -> None:
+    """Exercise case where server returns a failure."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result.get("type") == "form"
+    assert result.get("step_id") == "user"
+    assert not result.get("errors")
+    assert "flow_id" in result
+    with patch(
+        "rtsp_to_webrtc.client.Client.heartbeat",
+        side_effect=rtsp_to_webrtc.exceptions.ResponseError(),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"server_url": "https://example.com"}
+        )
+        assert result.get("type") == "form"
+        assert result.get("step_id") == "user"
+        assert result.get("errors") == {"base": "server_failure"}

tests/components/rtsp_to_webrtc/test_init.py

@@ -0,0 +1,204 @@
+"""Tests for RTSPtoWebRTC inititalization."""
+
+from __future__ import annotations
+
+import base64
+from typing import Any, AsyncGenerator, Awaitable, Callable
+from unittest.mock import patch
+
+import aiohttp
+import pytest
+import rtsp_to_webrtc
+
+from homeassistant.components import camera
+from homeassistant.components.rtsp_to_webrtc import DOMAIN
+from homeassistant.components.websocket_api.const import TYPE_RESULT
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+STREAM_SOURCE = "rtsp://example.com"
+# The webrtc component does not inspect the details of the offer and answer,
+# and is only a pass through.
+OFFER_SDP = "v=0\r\no=carol 28908764872 28908764872 IN IP4 100.3.6.6\r\n..."
+ANSWER_SDP = "v=0\r\no=bob 2890844730 2890844730 IN IP4 host.example.com\r\n..."
+
+SERVER_URL = "http://127.0.0.1:8083"
+
+CONFIG_ENTRY_DATA = {"server_url": SERVER_URL}
+
+
+@pytest.fixture
+async def mock_camera(hass) -> AsyncGenerator[None, None]:
+    """Initialize a demo camera platform."""
+    assert await async_setup_component(
+        hass, "camera", {camera.DOMAIN: {"platform": "demo"}}
+    )
+    await hass.async_block_till_done()
+    with patch(
+        "homeassistant.components.demo.camera.Path.read_bytes",
+        return_value=b"Test",
+    ), patch(
+        "homeassistant.components.camera.Camera.stream_source",
+        return_value=STREAM_SOURCE,
+    ), patch(
+        "homeassistant.components.camera.Camera.supported_features",
+        return_value=camera.SUPPORT_STREAM,
+    ):
+        yield
+
+
+async def async_setup_rtsp_to_webrtc(hass: HomeAssistant) -> None:
+    """Set up the component."""
+    return await async_setup_component(hass, DOMAIN, {})
+
+
+async def test_setup_success(hass: HomeAssistant) -> None:
+    """Test successful setup and unload."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA)
+    config_entry.add_to_hass(hass)
+
+    with patch("rtsp_to_webrtc.client.Client.heartbeat"):
+        assert await async_setup_rtsp_to_webrtc(hass)
+        await hass.async_block_till_done()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.LOADED
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert not hass.data.get(DOMAIN)
+    assert config_entry.state is ConfigEntryState.NOT_LOADED
+
+
+async def test_invalid_config_entry(hass: HomeAssistant) -> None:
+    """Test a config entry with missing required fields."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data={})
+    config_entry.add_to_hass(hass)
+
+    assert await async_setup_rtsp_to_webrtc(hass)
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.SETUP_ERROR
+
+
+async def test_setup_server_failure(hass: HomeAssistant) -> None:
+    """Test server responds with a failure on startup."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA)
+    config_entry.add_to_hass(hass)
+
+    with patch(
+        "rtsp_to_webrtc.client.Client.heartbeat",
+        side_effect=rtsp_to_webrtc.exceptions.ResponseError(),
+    ):
+        assert await async_setup_rtsp_to_webrtc(hass)
+        await hass.async_block_till_done()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.SETUP_RETRY
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+
+async def test_setup_communication_failure(hass: HomeAssistant) -> None:
+    """Test unable to talk to server on startup."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA)
+    config_entry.add_to_hass(hass)
+
+    with patch(
+        "rtsp_to_webrtc.client.Client.heartbeat",
+        side_effect=rtsp_to_webrtc.exceptions.ClientError(),
+    ):
+        assert await async_setup_rtsp_to_webrtc(hass)
+        await hass.async_block_till_done()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.SETUP_RETRY
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+
+async def test_offer_for_stream_source(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: Callable[[...], Awaitable[aiohttp.ClientWebSocketResponse]],
+    mock_camera: Any,
+) -> None:
+    """Test successful response from RTSPtoWebRTC server."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA)
+    config_entry.add_to_hass(hass)
+
+    with patch("rtsp_to_webrtc.client.Client.heartbeat"):
+        assert await async_setup_rtsp_to_webrtc(hass)
+        await hass.async_block_till_done()
+
+    aioclient_mock.post(
+        f"{SERVER_URL}/stream",
+        json={"sdp64": base64.b64encode(ANSWER_SDP.encode("utf-8")).decode("utf-8")},
+    )
+
+    client = await hass_ws_client(hass)
+    await client.send_json(
+        {
+            "id": 1,
+            "type": "camera/web_rtc_offer",
+            "entity_id": "camera.demo_camera",
+            "offer": OFFER_SDP,
+        }
+    )
+    response = await client.receive_json()
+    assert response.get("id") == 1
+    assert response.get("type") == TYPE_RESULT
+    assert response.get("success")
+    assert "result" in response
+    assert response["result"].get("answer") == ANSWER_SDP
+    assert "error" not in response
+
+
+async def test_offer_failure(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: Callable[[...], Awaitable[aiohttp.ClientWebSocketResponse]],
+    mock_camera: Any,
+) -> None:
+    """Test a transient failure talking to RTSPtoWebRTC server."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA)
+    config_entry.add_to_hass(hass)
+
+    with patch("rtsp_to_webrtc.client.Client.heartbeat"):
+        assert await async_setup_rtsp_to_webrtc(hass)
+        await hass.async_block_till_done()
+
+    aioclient_mock.post(
+        f"{SERVER_URL}/stream",
+        exc=aiohttp.ClientError,
+    )
+
+    client = await hass_ws_client(hass)
+    await client.send_json(
+        {
+            "id": 2,
+            "type": "camera/web_rtc_offer",
+            "entity_id": "camera.demo_camera",
+            "offer": OFFER_SDP,
+        }
+    )
+    response = await client.receive_json()
+    assert response.get("id") == 2
+    assert response.get("type") == TYPE_RESULT
+    assert "success" in response
+    assert not response.get("success")
+    assert "error" in response
+    assert response["error"].get("code") == "web_rtc_offer_failed"
+    assert "message" in response["error"]
+    assert "RTSPtoWebRTC server communication failure" in response["error"]["message"]