Add homekit camera support (#32527)

* Add homekit camera support

* Cleanup pyhapcamera inheritance

* Add camera to homekit manifest

* Use upstream pyhap server handler in homekit

* Remove unused homekit constants

* Fix lint errors in homekit camera

* Update homekit camera log messages

* Black after conflict fixes

* More conflict fixes

* missing srtp

* Allow streaming retry when ffmpeg fails to connect

* Fix inherit of camera config, force kill ffmpeg on failure

* Fix audio (Home Assistant only comes with OPUS)

* Fix audio (Home Assistant only comes with OPUS)

* Add camera to the list of supported domains.

* add a test for camera creation

* Add a basic test (still needs more as its only at 44% cover)

* let super handle reconfigure_stream

* Remove scaling as it does not appear to be needed and causes artifacts

* Some more basic tests

* make sure no exceptions when finding the source from the entity

* make sure the bridge forwards get_snapshot

* restore full coverage to accessories.py

* revert usage of super for start/stop stream

* one more test

* more mocking

* Remove -tune zerolatency, disable reconfigure_stream

* Restore -tune zerolatency

Co-authored-by: John Carr <john.carr@unrouted.co.uk>
Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/homekit/__init__.py

@@ -527,6 +527,7 @@ class HomeKit:
 
     def _start(self, bridged_states):
         from . import (  # noqa: F401 pylint: disable=unused-import, import-outside-toplevel
+            type_cameras,
             type_covers,
             type_fans,
             type_lights,

homeassistant/components/homekit/accessories.py

@@ -208,6 +208,9 @@ def get_accessory(hass, driver, state, aid, config):
     elif state.domain == "water_heater":
         a_type = "WaterHeater"
 
+    elif state.domain == "camera":
+        a_type = "Camera"
+
     if a_type is None:
         return None
 
@@ -219,10 +222,19 @@ class HomeAccessory(Accessory):
     """Adapter class for Accessory."""
 
     def __init__(
-        self, hass, driver, name, entity_id, aid, config, category=CATEGORY_OTHER
+        self,
+        hass,
+        driver,
+        name,
+        entity_id,
+        aid,
+        config,
+        *args,
+        category=CATEGORY_OTHER,
+        **kwargs,
     ):
         """Initialize a Accessory object."""
-        super().__init__(driver, name, aid=aid)
+        super().__init__(driver=driver, display_name=name, aid=aid, *args, **kwargs)
         model = split_entity_id(entity_id)[0].replace("_", " ").title()
         self.set_info_service(
             firmware_revision=__version__,
@@ -459,6 +471,18 @@ class HomeBridge(Bridge):
     def setup_message(self):
         """Prevent print of pyhap setup message to terminal."""
 
+    def get_snapshot(self, info):
+        """Get snapshot from accessory if supported."""
+        acc = self.accessories.get(info["aid"])
+        if acc is None:
+            raise ValueError("Requested snapshot for missing accessory")
+        if not hasattr(acc, "get_snapshot"):
+            raise ValueError(
+                "Got a request for snapshot, but the Accessory "
+                'does not define a "get_snapshot" method'
+            )
+        return acc.get_snapshot(info)
+
 
 class HomeDriver(AccessoryDriver):
     """Adapter class for AccessoryDriver."""

homeassistant/components/homekit/config_flow.py

@@ -38,6 +38,7 @@ SUPPORTED_DOMAINS = [
     "alarm_control_panel",
     "automation",
     "binary_sensor",
+    "camera",
     "climate",
     "cover",
     "demo",

homeassistant/components/homekit/const.py

@@ -19,6 +19,8 @@ ATTR_VALUE = "value"
 
 # #### Config ####
 CONF_ADVERTISE_IP = "advertise_ip"
+CONF_AUDIO_MAP = "audio_map"
+CONF_AUDIO_PACKET_SIZE = "audio_packet_size"
 CONF_AUTO_START = "auto_start"
 CONF_ENTITY_CONFIG = "entity_config"
 CONF_FEATURE = "feature"
@@ -27,16 +29,31 @@ CONF_FILTER = "filter"
 CONF_LINKED_BATTERY_SENSOR = "linked_battery_sensor"
 CONF_LINKED_BATTERY_CHARGING_SENSOR = "linked_battery_charging_sensor"
 CONF_LOW_BATTERY_THRESHOLD = "low_battery_threshold"
+CONF_MAX_FPS = "max_fps"
+CONF_MAX_HEIGHT = "max_height"
+CONF_MAX_WIDTH = "max_width"
 CONF_SAFE_MODE = "safe_mode"
 CONF_ZEROCONF_DEFAULT_INTERFACE = "zeroconf_default_interface"
+CONF_STREAM_ADDRESS = "stream_address"
+CONF_STREAM_SOURCE = "stream_source"
+CONF_SUPPORT_AUDIO = "support_audio"
+CONF_VIDEO_MAP = "video_map"
+CONF_VIDEO_PACKET_SIZE = "video_packet_size"
 
 # #### Config Defaults ####
+DEFAULT_AUDIO_MAP = "0:a:0"
+DEFAULT_AUDIO_PACKET_SIZE = 188
 DEFAULT_AUTO_START = True
 DEFAULT_LOW_BATTERY_THRESHOLD = 20
+DEFAULT_MAX_FPS = 30
+DEFAULT_MAX_HEIGHT = 1080
+DEFAULT_MAX_WIDTH = 1920
 DEFAULT_PORT = 51827
 DEFAULT_CONFIG_FLOW_PORT = 51828
 DEFAULT_SAFE_MODE = False
 DEFAULT_ZEROCONF_DEFAULT_INTERFACE = False
+DEFAULT_VIDEO_MAP = "0:v:0"
+DEFAULT_VIDEO_PACKET_SIZE = 1316
 
 # #### Features ####
 FEATURE_ON_OFF = "on_off"

homeassistant/components/homekit/manifest.json

@@ -3,7 +3,7 @@
   "name": "HomeKit",
   "documentation": "https://www.home-assistant.io/integrations/homekit",
   "requirements": ["HAP-python==2.8.3","fnvhash==0.1.0","PyQRCode==1.2.1","base36==0.1.1"],
-  "dependencies": ["http"],
+  "dependencies": ["http", "camera", "ffmpeg"],
   "after_dependencies": ["logbook"],
   "codeowners": ["@bdraco"],
   "config_flow": true

homeassistant/components/homekit/type_cameras.py

@@ -0,0 +1,241 @@
+"""Class to hold all camera accessories."""
+import asyncio
+import logging
+
+from haffmpeg.core import HAFFmpeg
+from pyhap.camera import (
+    VIDEO_CODEC_PARAM_LEVEL_TYPES,
+    VIDEO_CODEC_PARAM_PROFILE_ID_TYPES,
+    Camera as PyhapCamera,
+)
+from pyhap.const import CATEGORY_CAMERA
+
+from homeassistant.components.camera.const import DOMAIN as DOMAIN_CAMERA
+from homeassistant.components.ffmpeg import DATA_FFMPEG
+from homeassistant.util import get_local_ip
+
+from .accessories import TYPES, HomeAccessory
+from .const import (
+    CONF_AUDIO_MAP,
+    CONF_AUDIO_PACKET_SIZE,
+    CONF_MAX_FPS,
+    CONF_MAX_HEIGHT,
+    CONF_MAX_WIDTH,
+    CONF_STREAM_ADDRESS,
+    CONF_STREAM_SOURCE,
+    CONF_SUPPORT_AUDIO,
+    CONF_VIDEO_MAP,
+    CONF_VIDEO_PACKET_SIZE,
+)
+from .util import CAMERA_SCHEMA
+
+_LOGGER = logging.getLogger(__name__)
+
+VIDEO_OUTPUT = (
+    "-map {v_map} -an "
+    "-c:v libx264 -profile:v {v_profile} -tune zerolatency -pix_fmt yuv420p "
+    "-r {fps} "
+    "-b:v {v_max_bitrate}k -bufsize {v_bufsize}k -maxrate {v_max_bitrate}k "
+    "-payload_type 99 "
+    "-ssrc {v_ssrc} -f rtp "
+    "-srtp_out_suite AES_CM_128_HMAC_SHA1_80 -srtp_out_params {v_srtp_key} "
+    "srtp://{address}:{v_port}?rtcpport={v_port}&"
+    "localrtcpport={v_port}&pkt_size={v_pkt_size}"
+)
+
+AUDIO_ENCODER_OPUS = "libopus -application lowdelay"
+
+AUDIO_OUTPUT = (
+    "-map {a_map} -vn "
+    "-c:a {a_encoder} "
+    "-ac 1 -ar {a_sample_rate}k "
+    "-b:a {a_max_bitrate}k -bufsize {a_bufsize}k "
+    "-payload_type 110 "
+    "-ssrc {a_ssrc} -f rtp "
+    "-srtp_out_suite AES_CM_128_HMAC_SHA1_80 -srtp_out_params {a_srtp_key} "
+    "srtp://{address}:{a_port}?rtcpport={a_port}&"
+    "localrtcpport={a_port}&pkt_size={a_pkt_size}"
+)
+
+SLOW_RESOLUTIONS = [
+    (320, 180, 15),
+    (320, 240, 15),
+]
+
+RESOLUTIONS = [
+    (320, 180),
+    (320, 240),
+    (480, 270),
+    (480, 360),
+    (640, 360),
+    (640, 480),
+    (1024, 576),
+    (1024, 768),
+    (1280, 720),
+    (1280, 960),
+    (1920, 1080),
+]
+
+VIDEO_PROFILE_NAMES = ["baseline", "main", "high"]
+
+
+@TYPES.register("Camera")
+class Camera(HomeAccessory, PyhapCamera):
+    """Generate a Camera accessory."""
+
+    def __init__(self, hass, driver, name, entity_id, aid, config):
+        """Initialize a Camera accessory object."""
+        self._ffmpeg = hass.data[DATA_FFMPEG]
+        self._camera = hass.data[DOMAIN_CAMERA]
+        config_w_defaults = CAMERA_SCHEMA(config)
+
+        max_fps = config_w_defaults[CONF_MAX_FPS]
+        max_width = config_w_defaults[CONF_MAX_WIDTH]
+        max_height = config_w_defaults[CONF_MAX_HEIGHT]
+        resolutions = [
+            (w, h, fps)
+            for w, h, fps in SLOW_RESOLUTIONS
+            if w <= max_width and h <= max_height and fps < max_fps
+        ] + [
+            (w, h, max_fps)
+            for w, h in RESOLUTIONS
+            if w <= max_width and h <= max_height
+        ]
+
+        video_options = {
+            "codec": {
+                "profiles": [
+                    VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["BASELINE"],
+                    VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["MAIN"],
+                    VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["HIGH"],
+                ],
+                "levels": [
+                    VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE3_1"],
+                    VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE3_2"],
+                    VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE4_0"],
+                ],
+            },
+            "resolutions": resolutions,
+        }
+        audio_options = {"codecs": [{"type": "OPUS", "samplerate": 24}]}
+
+        stream_address = config_w_defaults.get(CONF_STREAM_ADDRESS, get_local_ip())
+
+        options = {
+            "video": video_options,
+            "audio": audio_options,
+            "address": stream_address,
+            "srtp": True,
+        }
+
+        super().__init__(
+            hass,
+            driver,
+            name,
+            entity_id,
+            aid,
+            config_w_defaults,
+            category=CATEGORY_CAMERA,
+            options=options,
+        )
+
+    def update_state(self, new_state):
+        """Handle state change to update HomeKit value."""
+        pass  # pylint: disable=unnecessary-pass
+
+    async def _async_get_stream_source(self):
+        """Find the camera stream source url."""
+        camera = self._camera.get_entity(self.entity_id)
+        if not camera or not camera.is_on:
+            return None
+        stream_source = self.config.get(CONF_STREAM_SOURCE)
+        if stream_source:
+            return stream_source
+        try:
+            return await camera.stream_source()
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception(
+                "Failed to get stream source - this could be a transient error or your camera might not be compatible with HomeKit yet"
+            )
+
+    async def start_stream(self, session_info, stream_config):
+        """Start a new stream with the given configuration."""
+        _LOGGER.debug(
+            "[%s] Starting stream with the following parameters: %s",
+            session_info["id"],
+            stream_config,
+        )
+        input_source = await self._async_get_stream_source()
+        if not input_source:
+            _LOGGER.error("Camera has no stream source")
+            return False
+        if "-i " not in input_source:
+            input_source = "-i " + input_source
+        output_vars = stream_config.copy()
+        output_vars.update(
+            {
+                "v_profile": VIDEO_PROFILE_NAMES[
+                    int.from_bytes(stream_config["v_profile_id"], byteorder="big")
+                ],
+                "v_bufsize": stream_config["v_max_bitrate"] * 2,
+                "v_map": self.config[CONF_VIDEO_MAP],
+                "v_pkt_size": self.config[CONF_VIDEO_PACKET_SIZE],
+                "a_bufsize": stream_config["a_max_bitrate"] * 2,
+                "a_map": self.config[CONF_AUDIO_MAP],
+                "a_pkt_size": self.config[CONF_AUDIO_PACKET_SIZE],
+                "a_encoder": AUDIO_ENCODER_OPUS,
+            }
+        )
+        output = VIDEO_OUTPUT.format(**output_vars)
+        if self.config[CONF_SUPPORT_AUDIO]:
+            output = output + " " + AUDIO_OUTPUT.format(**output_vars)
+        _LOGGER.debug("FFmpeg output settings: %s", output)
+        stream = HAFFmpeg(self._ffmpeg.binary, loop=self.driver.loop)
+        opened = await stream.open(
+            cmd=[], input_source=input_source, output=output, stdout_pipe=False
+        )
+        if not opened:
+            _LOGGER.error("Failed to open ffmpeg stream")
+            return False
+        session_info["stream"] = stream
+        _LOGGER.info(
+            "[%s] Started stream process - PID %d",
+            session_info["id"],
+            stream.process.pid,
+        )
+        return True
+
+    async def stop_stream(self, session_info):
+        """Stop the stream for the given ``session_id``."""
+        session_id = session_info["id"]
+        stream = session_info.get("stream")
+        if not stream:
+            _LOGGER.debug("No stream for session ID %s", session_id)
+        _LOGGER.info("[%s] Stopping stream.", session_id)
+
+        try:
+            await stream.close()
+            return
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Failed to gracefully close stream.")
+
+        try:
+            await stream.kill()
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Failed to forcefully close stream.")
+        _LOGGER.debug("Stream process stopped forcefully.")
+
+    async def reconfigure_stream(self, session_info, stream_config):
+        """Reconfigure the stream so that it uses the given ``stream_config``."""
+        return True
+
+    def get_snapshot(self, image_size):
+        """Return a jpeg of a snapshot from the camera."""
+        return (
+            asyncio.run_coroutine_threadsafe(
+                self.hass.components.camera.async_get_image(self.entity_id),
+                self.hass.loop,
+            )
+            .result()
+            .content
+        )

homeassistant/components/homekit/util.py

@@ -1,6 +1,7 @@
 """Collection of useful functions for the HomeKit component."""
 from collections import OrderedDict, namedtuple
 import io
+import ipaddress
 import logging
 import os
 import secrets
@@ -23,11 +24,28 @@ from homeassistant.helpers.storage import STORAGE_DIR
 import homeassistant.util.temperature as temp_util
 
 from .const import (
+    CONF_AUDIO_MAP,
+    CONF_AUDIO_PACKET_SIZE,
     CONF_FEATURE,
     CONF_FEATURE_LIST,
     CONF_LINKED_BATTERY_SENSOR,
     CONF_LOW_BATTERY_THRESHOLD,
+    CONF_MAX_FPS,
+    CONF_MAX_HEIGHT,
+    CONF_MAX_WIDTH,
+    CONF_STREAM_ADDRESS,
+    CONF_STREAM_SOURCE,
+    CONF_SUPPORT_AUDIO,
+    CONF_VIDEO_MAP,
+    CONF_VIDEO_PACKET_SIZE,
+    DEFAULT_AUDIO_MAP,
+    DEFAULT_AUDIO_PACKET_SIZE,
     DEFAULT_LOW_BATTERY_THRESHOLD,
+    DEFAULT_MAX_FPS,
+    DEFAULT_MAX_HEIGHT,
+    DEFAULT_MAX_WIDTH,
+    DEFAULT_VIDEO_MAP,
+    DEFAULT_VIDEO_PACKET_SIZE,
     DOMAIN,
     FEATURE_ON_OFF,
     FEATURE_PLAY_PAUSE,
@@ -62,6 +80,25 @@ FEATURE_SCHEMA = BASIC_INFO_SCHEMA.extend(
     {vol.Optional(CONF_FEATURE_LIST, default=None): cv.ensure_list}
 )
 
+CAMERA_SCHEMA = BASIC_INFO_SCHEMA.extend(
+    {
+        vol.Optional(CONF_STREAM_ADDRESS): vol.All(ipaddress.ip_address, cv.string),
+        vol.Optional(CONF_STREAM_SOURCE): cv.string,
+        vol.Optional(CONF_SUPPORT_AUDIO, default=False): cv.boolean,
+        vol.Optional(CONF_MAX_WIDTH, default=DEFAULT_MAX_WIDTH): cv.positive_int,
+        vol.Optional(CONF_MAX_HEIGHT, default=DEFAULT_MAX_HEIGHT): cv.positive_int,
+        vol.Optional(CONF_MAX_FPS, default=DEFAULT_MAX_FPS): cv.positive_int,
+        vol.Optional(CONF_AUDIO_MAP, default=DEFAULT_AUDIO_MAP): cv.string,
+        vol.Optional(CONF_VIDEO_MAP, default=DEFAULT_VIDEO_MAP): cv.string,
+        vol.Optional(
+            CONF_AUDIO_PACKET_SIZE, default=DEFAULT_AUDIO_PACKET_SIZE
+        ): cv.positive_int,
+        vol.Optional(
+            CONF_VIDEO_PACKET_SIZE, default=DEFAULT_VIDEO_PACKET_SIZE
+        ): cv.positive_int,
+    }
+)
+
 CODE_SCHEMA = BASIC_INFO_SCHEMA.extend(
     {vol.Optional(ATTR_CODE, default=None): vol.Any(None, cv.string)}
 )
@@ -162,6 +199,9 @@ def validate_entity_config(values):
                 feature_list[key] = params
             config[CONF_FEATURE_LIST] = feature_list
 
+        elif domain == "camera":
+            config = CAMERA_SCHEMA(config)
+
         elif domain == "switch":
             config = SWITCH_TYPE_SCHEMA(config)
 

tests/components/homekit/test_get_accessories.py

@@ -264,3 +264,15 @@ def test_type_vacuum(type_name, entity_id, state, attrs):
         entity_state = State(entity_id, state, attrs)
         get_accessory(None, None, entity_state, 2, {})
     assert mock_type.called
+
+
+@pytest.mark.parametrize(
+    "type_name, entity_id, state, attrs", [("Camera", "camera.basic", "on", {})],
+)
+def test_type_camera(type_name, entity_id, state, attrs):
+    """Test if camera types are associated correctly."""
+    mock_type = Mock()
+    with patch.dict(TYPES, {type_name: mock_type}):
+        entity_state = State(entity_id, state, attrs)
+        get_accessory(None, None, entity_state, 2, {})
+    assert mock_type.called

tests/components/homekit/test_type_cameras.py

@@ -0,0 +1,352 @@
+"""Test different accessory types: Camera."""
+
+from uuid import UUID
+
+from pyhap.accessory_driver import AccessoryDriver
+import pytest
+
+from homeassistant.components import camera, ffmpeg
+from homeassistant.components.homekit.accessories import HomeBridge
+from homeassistant.components.homekit.const import (
+    CONF_STREAM_SOURCE,
+    CONF_SUPPORT_AUDIO,
+)
+from homeassistant.components.homekit.type_cameras import Camera
+from homeassistant.components.homekit.type_switches import Switch
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.setup import async_setup_component
+
+from tests.async_mock import AsyncMock, MagicMock, patch
+
+
+@pytest.fixture()
+def run_driver(hass):
+    """Return a custom AccessoryDriver instance for HomeKit accessory init."""
+    with patch("pyhap.accessory_driver.Zeroconf"), patch(
+        "pyhap.accessory_driver.AccessoryEncoder"
+    ), patch("pyhap.accessory_driver.HAPServer"), patch(
+        "pyhap.accessory_driver.AccessoryDriver.publish"
+    ), patch(
+        "pyhap.accessory_driver.AccessoryDriver.persist"
+    ):
+        yield AccessoryDriver(
+            pincode=b"123-45-678", address="127.0.0.1", loop=hass.loop
+        )
+
+
+def _get_working_mock_ffmpeg():
+    """Return a working ffmpeg."""
+    ffmpeg = MagicMock()
+    ffmpeg.open = AsyncMock(return_value=True)
+    ffmpeg.close = AsyncMock(return_value=True)
+    ffmpeg.kill = AsyncMock(return_value=True)
+    return ffmpeg
+
+
+def _get_failing_mock_ffmpeg():
+    """Return an ffmpeg that fails to shutdown."""
+    ffmpeg = MagicMock()
+    ffmpeg.open = AsyncMock(return_value=False)
+    ffmpeg.close = AsyncMock(side_effect=OSError)
+    ffmpeg.kill = AsyncMock(side_effect=OSError)
+    return ffmpeg
+
+
+async def test_camera_stream_source_configured(hass, run_driver, events):
+    """Test a camera that can stream with a configured source."""
+    await async_setup_component(hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
+    await async_setup_component(
+        hass, camera.DOMAIN, {camera.DOMAIN: {"platform": "demo"}}
+    )
+
+    entity_id = "camera.demo_camera"
+
+    hass.states.async_set(entity_id, None)
+    await hass.async_block_till_done()
+    acc = Camera(
+        hass,
+        run_driver,
+        "Camera",
+        entity_id,
+        2,
+        {CONF_STREAM_SOURCE: "/dev/null", CONF_SUPPORT_AUDIO: True},
+    )
+    not_camera_acc = Switch(hass, run_driver, "Switch", entity_id, 4, {},)
+    bridge = HomeBridge("hass", run_driver, "Test Bridge")
+    bridge.add_accessory(acc)
+    bridge.add_accessory(not_camera_acc)
+
+    await acc.run_handler()
+
+    assert acc.aid == 2
+    assert acc.category == 17  # Camera
+
+    stream_service = acc.get_service("CameraRTPStreamManagement")
+    endpoints_config_char = stream_service.get_characteristic("SetupEndpoints")
+    assert endpoints_config_char.setter_callback
+    stream_config_char = stream_service.get_characteristic(
+        "SelectedRTPStreamConfiguration"
+    )
+    assert stream_config_char.setter_callback
+    acc.set_endpoints(
+        "ARAzA9UDF8xGmrZykkNqcaL2AgEAAxoBAQACDTE5Mi4xNjguMjA4LjUDAi7IBAKkxwQlAQEAAhDN0+Y0tZ4jzoO0ske9UsjpAw6D76oVXnoi7DbawIG4CwUlAQEAAhCyGcROB8P7vFRDzNF2xrK1Aw6NdcLugju9yCfkWVSaVAYEDoAsAAcEpxV8AA=="
+    )
+
+    with patch(
+        "homeassistant.components.demo.camera.DemoCamera.stream_source",
+        return_value=None,
+    ), patch(
+        "homeassistant.components.homekit.type_cameras.HAFFmpeg",
+        return_value=_get_working_mock_ffmpeg(),
+    ):
+        acc.set_selected_stream_configuration(
+            "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+        )
+        await hass.async_block_till_done()
+
+    session_info = {
+        "id": "mock",
+        "v_srtp_key": "key",
+        "a_srtp_key": "key",
+        "v_port": "0",
+        "a_port": "0",
+        "address": "0.0.0.0",
+    }
+    acc.sessions[UUID("3303d503-17cc-469a-b672-92436a71a2f6")] = session_info
+
+    with patch(
+        "homeassistant.components.demo.camera.DemoCamera.stream_source",
+        return_value="rtsp://example.local",
+    ), patch(
+        "homeassistant.components.homekit.type_cameras.HAFFmpeg",
+        return_value=_get_working_mock_ffmpeg(),
+    ):
+        acc.set_selected_stream_configuration(
+            "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+        )
+        await acc.stop_stream(session_info)
+        # Calling a second time should not throw
+        await acc.stop_stream(session_info)
+        await hass.async_block_till_done()
+
+    assert await hass.async_add_executor_job(acc.get_snapshot, 1024)
+
+    # Verify the bridge only forwards get_snapshot for
+    # cameras and valid accessory ids
+    assert await hass.async_add_executor_job(bridge.get_snapshot, {"aid": 2})
+    with pytest.raises(ValueError):
+        assert await hass.async_add_executor_job(bridge.get_snapshot, {"aid": 3})
+    with pytest.raises(ValueError):
+        assert await hass.async_add_executor_job(bridge.get_snapshot, {"aid": 4})
+
+
+async def test_camera_stream_source_configured_with_failing_ffmpeg(
+    hass, run_driver, events
+):
+    """Test a camera that can stream with a configured source with ffmpeg failing."""
+    await async_setup_component(hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
+    await async_setup_component(
+        hass, camera.DOMAIN, {camera.DOMAIN: {"platform": "demo"}}
+    )
+
+    entity_id = "camera.demo_camera"
+
+    hass.states.async_set(entity_id, None)
+    await hass.async_block_till_done()
+    acc = Camera(
+        hass,
+        run_driver,
+        "Camera",
+        entity_id,
+        2,
+        {CONF_STREAM_SOURCE: "/dev/null", CONF_SUPPORT_AUDIO: True},
+    )
+    not_camera_acc = Switch(hass, run_driver, "Switch", entity_id, 4, {},)
+    bridge = HomeBridge("hass", run_driver, "Test Bridge")
+    bridge.add_accessory(acc)
+    bridge.add_accessory(not_camera_acc)
+
+    await acc.run_handler()
+
+    assert acc.aid == 2
+    assert acc.category == 17  # Camera
+
+    stream_service = acc.get_service("CameraRTPStreamManagement")
+    endpoints_config_char = stream_service.get_characteristic("SetupEndpoints")
+    assert endpoints_config_char.setter_callback
+    stream_config_char = stream_service.get_characteristic(
+        "SelectedRTPStreamConfiguration"
+    )
+    assert stream_config_char.setter_callback
+    acc.set_endpoints(
+        "ARAzA9UDF8xGmrZykkNqcaL2AgEAAxoBAQACDTE5Mi4xNjguMjA4LjUDAi7IBAKkxwQlAQEAAhDN0+Y0tZ4jzoO0ske9UsjpAw6D76oVXnoi7DbawIG4CwUlAQEAAhCyGcROB8P7vFRDzNF2xrK1Aw6NdcLugju9yCfkWVSaVAYEDoAsAAcEpxV8AA=="
+    )
+
+    session_info = {
+        "id": "mock",
+        "v_srtp_key": "key",
+        "a_srtp_key": "key",
+        "v_port": "0",
+        "a_port": "0",
+        "address": "0.0.0.0",
+    }
+    acc.sessions[UUID("3303d503-17cc-469a-b672-92436a71a2f6")] = session_info
+
+    with patch(
+        "homeassistant.components.demo.camera.DemoCamera.stream_source",
+        return_value="rtsp://example.local",
+    ), patch(
+        "homeassistant.components.homekit.type_cameras.HAFFmpeg",
+        return_value=_get_failing_mock_ffmpeg(),
+    ):
+        acc.set_selected_stream_configuration(
+            "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+        )
+        await acc.stop_stream(session_info)
+        # Calling a second time should not throw
+        await acc.stop_stream(session_info)
+        await hass.async_block_till_done()
+
+
+async def test_camera_stream_source_found(hass, run_driver, events):
+    """Test a camera that can stream and we get the source from the entity."""
+    await async_setup_component(hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
+    await async_setup_component(
+        hass, camera.DOMAIN, {camera.DOMAIN: {"platform": "demo"}}
+    )
+
+    entity_id = "camera.demo_camera"
+
+    hass.states.async_set(entity_id, None)
+    await hass.async_block_till_done()
+    acc = Camera(hass, run_driver, "Camera", entity_id, 2, {},)
+    await acc.run_handler()
+
+    assert acc.aid == 2
+    assert acc.category == 17  # Camera
+
+    stream_service = acc.get_service("CameraRTPStreamManagement")
+    endpoints_config_char = stream_service.get_characteristic("SetupEndpoints")
+    assert endpoints_config_char.setter_callback
+    stream_config_char = stream_service.get_characteristic(
+        "SelectedRTPStreamConfiguration"
+    )
+    assert stream_config_char.setter_callback
+    acc.set_endpoints(
+        "ARAzA9UDF8xGmrZykkNqcaL2AgEAAxoBAQACDTE5Mi4xNjguMjA4LjUDAi7IBAKkxwQlAQEAAhDN0+Y0tZ4jzoO0ske9UsjpAw6D76oVXnoi7DbawIG4CwUlAQEAAhCyGcROB8P7vFRDzNF2xrK1Aw6NdcLugju9yCfkWVSaVAYEDoAsAAcEpxV8AA=="
+    )
+
+    session_info = {
+        "id": "mock",
+        "v_srtp_key": "key",
+        "a_srtp_key": "key",
+        "v_port": "0",
+        "a_port": "0",
+        "address": "0.0.0.0",
+    }
+    acc.sessions[UUID("3303d503-17cc-469a-b672-92436a71a2f6")] = session_info
+
+    with patch(
+        "homeassistant.components.demo.camera.DemoCamera.stream_source",
+        return_value="rtsp://example.local",
+    ), patch(
+        "homeassistant.components.homekit.type_cameras.HAFFmpeg",
+        return_value=_get_working_mock_ffmpeg(),
+    ):
+        acc.set_selected_stream_configuration(
+            "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+        )
+        await acc.stop_stream(session_info)
+        await hass.async_block_till_done()
+
+    with patch(
+        "homeassistant.components.demo.camera.DemoCamera.stream_source",
+        return_value="rtsp://example.local",
+    ), patch(
+        "homeassistant.components.homekit.type_cameras.HAFFmpeg",
+        return_value=_get_working_mock_ffmpeg(),
+    ):
+        acc.set_selected_stream_configuration(
+            "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+        )
+        await acc.stop_stream(session_info)
+        await hass.async_block_till_done()
+
+
+async def test_camera_stream_source_fails(hass, run_driver, events):
+    """Test a camera that can stream and we cannot get the source from the entity."""
+    await async_setup_component(hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
+    await async_setup_component(
+        hass, camera.DOMAIN, {camera.DOMAIN: {"platform": "demo"}}
+    )
+
+    entity_id = "camera.demo_camera"
+
+    hass.states.async_set(entity_id, None)
+    await hass.async_block_till_done()
+    acc = Camera(hass, run_driver, "Camera", entity_id, 2, {},)
+    await acc.run_handler()
+
+    assert acc.aid == 2
+    assert acc.category == 17  # Camera
+
+    stream_service = acc.get_service("CameraRTPStreamManagement")
+    endpoints_config_char = stream_service.get_characteristic("SetupEndpoints")
+    assert endpoints_config_char.setter_callback
+    stream_config_char = stream_service.get_characteristic(
+        "SelectedRTPStreamConfiguration"
+    )
+    assert stream_config_char.setter_callback
+    acc.set_endpoints(
+        "ARAzA9UDF8xGmrZykkNqcaL2AgEAAxoBAQACDTE5Mi4xNjguMjA4LjUDAi7IBAKkxwQlAQEAAhDN0+Y0tZ4jzoO0ske9UsjpAw6D76oVXnoi7DbawIG4CwUlAQEAAhCyGcROB8P7vFRDzNF2xrK1Aw6NdcLugju9yCfkWVSaVAYEDoAsAAcEpxV8AA=="
+    )
+
+    session_info = {
+        "id": "mock",
+        "v_srtp_key": "key",
+        "a_srtp_key": "key",
+        "v_port": "0",
+        "a_port": "0",
+        "address": "0.0.0.0",
+    }
+    acc.sessions[UUID("3303d503-17cc-469a-b672-92436a71a2f6")] = session_info
+
+    with patch(
+        "homeassistant.components.demo.camera.DemoCamera.stream_source",
+        side_effect=OSError,
+    ), patch(
+        "homeassistant.components.homekit.type_cameras.HAFFmpeg",
+        return_value=_get_working_mock_ffmpeg(),
+    ):
+        acc.set_selected_stream_configuration(
+            "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+        )
+        await acc.stop_stream(session_info)
+        await hass.async_block_till_done()
+
+
+async def test_camera_with_no_stream(hass, run_driver, events):
+    """Test a camera that cannot stream."""
+    await async_setup_component(hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
+    await async_setup_component(hass, camera.DOMAIN, {camera.DOMAIN: {}})
+
+    entity_id = "camera.demo_camera"
+
+    hass.states.async_set(entity_id, None)
+    await hass.async_block_till_done()
+    acc = Camera(hass, run_driver, "Camera", entity_id, 2, {},)
+    await acc.run_handler()
+
+    assert acc.aid == 2
+    assert acc.category == 17  # Camera
+
+    acc.set_endpoints(
+        "ARAzA9UDF8xGmrZykkNqcaL2AgEAAxoBAQACDTE5Mi4xNjguMjA4LjUDAi7IBAKkxwQlAQEAAhDN0+Y0tZ4jzoO0ske9UsjpAw6D76oVXnoi7DbawIG4CwUlAQEAAhCyGcROB8P7vFRDzNF2xrK1Aw6NdcLugju9yCfkWVSaVAYEDoAsAAcEpxV8AA=="
+    )
+    acc.set_selected_stream_configuration(
+        "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
+    )
+    await hass.async_block_till_done()
+
+    with pytest.raises(HomeAssistantError):
+        await hass.async_add_executor_job(acc.get_snapshot, 1024)