Ensure homekit camera stream can be restarted after failure

homeassistant/components/homekit/const.py

@@ -98,6 +98,7 @@
 SERV_ACCESSORY_INFO = "AccessoryInformation"
 SERV_AIR_QUALITY_SENSOR = "AirQualitySensor"
 SERV_BATTERY_SERVICE = "BatteryService"
+SERV_CAMERA_RTP_STREAM_MANAGEMENT = "CameraRTPStreamManagement"
 SERV_CARBON_DIOXIDE_SENSOR = "CarbonDioxideSensor"
 SERV_CARBON_MONOXIDE_SENSOR = "CarbonMonoxideSensor"
 SERV_CONTACT_SENSOR = "ContactSensor"
@@ -177,6 +178,7 @@
 CHAR_SLEEP_DISCOVER_MODE = "SleepDiscoveryMode"
 CHAR_SMOKE_DETECTED = "SmokeDetected"
 CHAR_STATUS_LOW_BATTERY = "StatusLowBattery"
+CHAR_STREAMING_STRATUS = "StreamingStatus"
 CHAR_SWING_MODE = "SwingMode"
 CHAR_TARGET_DOOR_STATE = "TargetDoorState"
 CHAR_TARGET_HEATING_COOLING = "TargetHeatingCoolingState"

homeassistant/components/homekit/type_cameras.py

@@ -1,9 +1,11 @@
 """Class to hold all camera accessories."""
 import asyncio
+from datetime import timedelta
 import logging
 
 from haffmpeg.core import HAFFmpeg
 from pyhap.camera import (
+    STREAMING_STATUS,
     VIDEO_CODEC_PARAM_LEVEL_TYPES,
     VIDEO_CODEC_PARAM_PROFILE_ID_TYPES,
     Camera as PyhapCamera,
@@ -13,10 +15,12 @@
 from homeassistant.components.camera.const import DOMAIN as DOMAIN_CAMERA
 from homeassistant.components.ffmpeg import DATA_FFMPEG
 from homeassistant.core import callback
+from homeassistant.helpers.event import async_track_time_interval
 from homeassistant.util import get_local_ip
 
 from .accessories import TYPES, HomeAccessory
 from .const import (
+    CHAR_STREAMING_STRATUS,
     CONF_AUDIO_CODEC,
     CONF_AUDIO_MAP,
     CONF_AUDIO_PACKET_SIZE,
@@ -29,9 +33,10 @@
     CONF_VIDEO_CODEC,
     CONF_VIDEO_MAP,
     CONF_VIDEO_PACKET_SIZE,
+    SERV_CAMERA_RTP_STREAM_MANAGEMENT,
 )
 from .img_util import scale_jpeg_camera_image
-from .util import CAMERA_SCHEMA
+from .util import CAMERA_SCHEMA, pid_is_alive
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -84,6 +89,11 @@
 
 VIDEO_PROFILE_NAMES = ["baseline", "main", "high"]
 
+FFMPEG_WATCH_INTERVAL = timedelta(seconds=5)
+FFMPEG_WATCHER = "ffmpeg_watcher"
+FFMPEG_PID = "ffmpeg_pid"
+SESSION_ID = "session_id"
+
 
 @TYPES.register("Camera")
 class Camera(HomeAccessory, PyhapCamera):
@@ -92,6 +102,7 @@ class Camera(HomeAccessory, PyhapCamera):
     def __init__(self, hass, driver, name, entity_id, aid, config):
         """Initialize a Camera accessory object."""
         self._ffmpeg = hass.data[DATA_FFMPEG]
+        self._cur_session = None
         self._camera = hass.data[DOMAIN_CAMERA]
         config_w_defaults = CAMERA_SCHEMA(config)
 
@@ -159,11 +170,14 @@ async def _async_get_stream_source(self):
         if stream_source:
             return stream_source
         try:
-            return await camera.stream_source()
+            stream_source = 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"
             )
+        if stream_source:
+            self.config[CONF_STREAM_SOURCE] = stream_source
+        return stream_source
 
     async def start_stream(self, session_info, stream_config):
         """Start a new stream with the given configuration."""
@@ -222,27 +236,69 @@ async def start_stream(self, session_info, stream_config):
             session_info["id"],
             stream.process.pid,
         )
-        return True
+
+        ffmpeg_watcher = async_track_time_interval(
+            self.hass, self._async_ffmpeg_watch, FFMPEG_WATCH_INTERVAL
+        )
+        self._cur_session = {
+            FFMPEG_WATCHER: ffmpeg_watcher,
+            FFMPEG_PID: stream.process.pid,
+            SESSION_ID: session_info["id"],
+        }
+
+        return await self._async_ffmpeg_watch(0)
+
+    async def _async_ffmpeg_watch(self, _):
+        """Check to make sure ffmpeg is still running and cleanup if not."""
+        ffmpeg_pid = self._cur_session[FFMPEG_PID]
+        session_id = self._cur_session[SESSION_ID]
+        if pid_is_alive(ffmpeg_pid):
+            return True
+
+        _LOGGER.warning("Streaming process ended unexpectedly - PID %d", ffmpeg_pid)
+        self._async_stop_ffmpeg_watch()
+        self._async_set_streaming_available(session_id)
+        return False
+
+    @callback
+    def _async_stop_ffmpeg_watch(self):
+        """Cleanup a streaming session after stopping."""
+        if not self._cur_session:
+            return
+        self._cur_session[FFMPEG_WATCHER]()
+        self._cur_session = None
+
+    @callback
+    def _async_set_streaming_available(self, session_id):
+        """Free the session so they can start another."""
+        self.streaming_status = STREAMING_STATUS["AVAILABLE"]
+        self.get_service(SERV_CAMERA_RTP_STREAM_MANAGEMENT).get_characteristic(
+            CHAR_STREAMING_STRATUS
+        ).notify()
 
     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.")
+        self._async_stop_ffmpeg_watch()
+
+        if not pid_is_alive(stream.process.pid):
+            _LOGGER.info("[%s] Stream already stopped.", session_id)
+            return True
+
+        for shutdown_method in ["close", "kill"]:
+            _LOGGER.info("[%s] %s stream.", session_id, shutdown_method)
+            try:
+                await getattr(stream, shutdown_method)()
+                return
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception(
+                    "[%s] Failed to %s stream.", session_id, shutdown_method
+                )
 
     async def reconfigure_stream(self, session_info, stream_config):
         """Reconfigure the stream so that it uses the given ``stream_config``."""

homeassistant/components/homekit/util.py

@@ -472,3 +472,13 @@ def find_next_available_port(start_port: int):
             if port == MAX_PORT:
                 raise
             continue
+
+
+def pid_is_alive(pid):
+    """Check to see if a process is alive."""
+    try:
+        os.kill(pid, 0)
+        return True
+    except OSError:
+        pass
+    return False