Stream Record Service (#22456)

* Initial commit of record service for live streams

* fix lint

* update service descriptions

* add tests

* fix lint

homeassistant/components/camera/__init__.py

@@ -20,7 +20,7 @@
 
 from homeassistant.core import callback
 from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, \
-    SERVICE_TURN_ON, EVENT_HOMEASSISTANT_START
+    SERVICE_TURN_ON, EVENT_HOMEASSISTANT_START, CONF_FILENAME
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.loader import bind_hass
 from homeassistant.helpers.entity import Entity
@@ -33,7 +33,8 @@
     SERVICE_PLAY_MEDIA, DOMAIN as DOMAIN_MP)
 from homeassistant.components.stream import request_stream
 from homeassistant.components.stream.const import (
-    OUTPUT_FORMATS, FORMAT_CONTENT_TYPE)
+    OUTPUT_FORMATS, FORMAT_CONTENT_TYPE, CONF_STREAM_SOURCE, CONF_LOOKBACK,
+    CONF_DURATION, SERVICE_RECORD, DOMAIN as DOMAIN_STREAM)
 from homeassistant.components import websocket_api
 import homeassistant.helpers.config_validation as cv
 
@@ -85,6 +86,12 @@
     vol.Optional(ATTR_FORMAT, default='hls'): vol.In(OUTPUT_FORMATS),
 })
 
+CAMERA_SERVICE_RECORD = CAMERA_SERVICE_SCHEMA.extend({
+    vol.Required(CONF_FILENAME): cv.template,
+    vol.Optional(CONF_DURATION, default=30): int,
+    vol.Optional(CONF_LOOKBACK, default=0): int,
+})
+
 WS_TYPE_CAMERA_THUMBNAIL = 'camera_thumbnail'
 SCHEMA_WS_CAMERA_THUMBNAIL = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
     vol.Required('type'): WS_TYPE_CAMERA_THUMBNAIL,
@@ -260,6 +267,10 @@ def update_tokens(time):
         SERVICE_PLAY_STREAM, CAMERA_SERVICE_PLAY_STREAM,
         async_handle_play_stream_service
     )
+    component.async_register_entity_service(
+        SERVICE_RECORD, CAMERA_SERVICE_RECORD,
+        async_handle_record_service
+    )
 
     return True
 
@@ -640,3 +651,27 @@ async def async_handle_play_stream_service(camera, service_call):
     await hass.services.async_call(
         DOMAIN_MP, SERVICE_PLAY_MEDIA, data,
         blocking=True, context=service_call.context)
+
+
+async def async_handle_record_service(camera, call):
+    """Handle stream recording service calls."""
+    if not camera.stream_source:
+        raise HomeAssistantError("{} does not support record service"
+                                 .format(camera.entity_id))
+
+    hass = camera.hass
+    filename = call.data[CONF_FILENAME]
+    filename.hass = hass
+    video_path = filename.async_render(
+        variables={ATTR_ENTITY_ID: camera})
+
+    data = {
+        CONF_STREAM_SOURCE: camera.stream_source,
+        CONF_FILENAME: video_path,
+        CONF_DURATION: call.data[CONF_DURATION],
+        CONF_LOOKBACK: call.data[CONF_LOOKBACK],
+    }
+
+    await hass.services.async_call(
+        DOMAIN_STREAM, SERVICE_RECORD, data,
+        blocking=True, context=call.context)

homeassistant/components/camera/services.yaml

@@ -51,6 +51,22 @@ play_stream:
       description: (Optional) Stream format supported by media player.
       example: 'hls'
 
+record:
+  description: Record live camera feed.
+  fields:
+    entity_id:
+      description: Name of entities to record.
+      example: 'camera.living_room_camera'
+    filename:
+      description: Template of a Filename. Variable is entity_id.  Must be mp4.
+      example: '/tmp/snapshot_{{ entity_id }}.mp4'
+    duration:
+      description: (Optional) Target recording length (in seconds). Default: 30
+      example: 30
+    lookback:
+      description: (Optional) Target lookback period (in seconds) to include in addition to duration.  Only available if there is currently an active HLS stream.
+      example: 4
+
 local_file_update_file_path:
   description: Update the file_path for a local_file camera.
   fields:

homeassistant/components/stream/__init__.py

@@ -10,15 +10,19 @@
 import voluptuous as vol
 
 from homeassistant.auth.util import generate_secret
-from homeassistant.const import EVENT_HOMEASSISTANT_STOP
+import homeassistant.helpers.config_validation as cv
+from homeassistant.const import EVENT_HOMEASSISTANT_STOP, CONF_FILENAME
 from homeassistant.core import callback
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.loader import bind_hass
 
-from .const import DOMAIN, ATTR_STREAMS, ATTR_ENDPOINTS
+from .const import (
+    DOMAIN, ATTR_STREAMS, ATTR_ENDPOINTS, CONF_STREAM_SOURCE,
+    CONF_DURATION, CONF_LOOKBACK, SERVICE_RECORD)
 from .core import PROVIDERS
 from .worker import stream_worker
 from .hls import async_setup_hls
+from .recorder import async_setup_recorder
 
 REQUIREMENTS = ['av==6.1.2']
 
@@ -30,6 +34,16 @@
     DOMAIN: vol.Schema({}),
 }, extra=vol.ALLOW_EXTRA)
 
+STREAM_SERVICE_SCHEMA = vol.Schema({
+    vol.Required(CONF_STREAM_SOURCE): cv.string,
+})
+
+SERVICE_RECORD_SCHEMA = STREAM_SERVICE_SCHEMA.extend({
+    vol.Required(CONF_FILENAME): cv.string,
+    vol.Optional(CONF_DURATION, default=30): int,
+    vol.Optional(CONF_LOOKBACK, default=0): int,
+})
+
 # Set log level to error for libav
 logging.getLogger('libav').setLevel(logging.ERROR)
 
@@ -82,6 +96,9 @@ async def async_setup(hass, config):
     hls_endpoint = async_setup_hls(hass)
     hass.data[DOMAIN][ATTR_ENDPOINTS]['hls'] = hls_endpoint
 
+    # Setup Recorder
+    async_setup_recorder(hass)
+
     @callback
     def shutdown(event):
         """Stop all stream workers."""
@@ -92,6 +109,13 @@ def shutdown(event):
 
     hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, shutdown)
 
+    async def async_record(call):
+        """Call record stream service handler."""
+        await async_handle_record_service(hass, call)
+
+    hass.services.async_register(DOMAIN, SERVICE_RECORD,
+                                 async_record, schema=SERVICE_RECORD_SCHEMA)
+
     return True
 
 
@@ -119,15 +143,15 @@ def outputs(self):
 
     def add_provider(self, fmt):
         """Add provider output stream."""
-        provider = PROVIDERS[fmt](self)
-        if not self._outputs.get(provider.format):
-            self._outputs[provider.format] = provider
-        return self._outputs[provider.format]
+        if not self._outputs.get(fmt):
+            provider = PROVIDERS[fmt](self)
+            self._outputs[fmt] = provider
+        return self._outputs[fmt]
 
     def remove_provider(self, provider):
         """Remove provider output stream."""
-        if provider.format in self._outputs:
-            del self._outputs[provider.format]
+        if provider.name in self._outputs:
+            del self._outputs[provider.name]
             self.check_idle()
 
         if not self._outputs:
@@ -165,3 +189,44 @@ def _stop(self):
             self._thread.join()
             self._thread = None
             _LOGGER.info("Stopped stream: %s", self.source)
+
+
+async def async_handle_record_service(hass, call):
+    """Handle save video service calls."""
+    stream_source = call.data[CONF_STREAM_SOURCE]
+    video_path = call.data[CONF_FILENAME]
+    duration = call.data[CONF_DURATION]
+    lookback = call.data[CONF_LOOKBACK]
+
+    # Check for file access
+    if not hass.config.is_allowed_path(video_path):
+        raise HomeAssistantError("Can't write {}, no access to path!"
+                                 .format(video_path))
+
+    # Check for active stream
+    streams = hass.data[DOMAIN][ATTR_STREAMS]
+    stream = streams.get(stream_source)
+    if not stream:
+        stream = Stream(hass, stream_source)
+        streams[stream_source] = stream
+
+    # Add recorder
+    recorder = stream.outputs.get('recorder')
+    if recorder:
+        raise HomeAssistantError("Stream already recording to {}!"
+                                 .format(recorder.video_path))
+
+    recorder = stream.add_provider('recorder')
+    recorder.video_path = video_path
+    recorder.timeout = duration
+
+    stream.start()
+
+    # Take advantage of lookback
+    hls = stream.outputs.get('hls')
+    if lookback > 0 and hls:
+        num_segments = min(int(lookback // hls.target_duration),
+                           hls.num_segments)
+        # Wait for latest segment, then add the lookback
+        await hls.recv()
+        recorder.prepend(list(hls.get_segment())[-num_segments:])

homeassistant/components/stream/const.py

@@ -1,10 +1,16 @@
 """Constants for Stream component."""
 DOMAIN = 'stream'
 
+CONF_STREAM_SOURCE = 'stream_source'
+CONF_LOOKBACK = 'lookback'
+CONF_DURATION = 'duration'
+
 ATTR_ENDPOINTS = 'endpoints'
 ATTR_STREAMS = 'streams'
 ATTR_KEEPALIVE = 'keepalive'
 
+SERVICE_RECORD = 'record'
+
 OUTPUT_FORMATS = ['hls']
 
 FORMAT_CONTENT_TYPE = {

homeassistant/components/stream/core.py

@@ -41,15 +41,21 @@ class StreamOutput:
 
     num_segments = 3
 
-    def __init__(self, stream) -> None:
+    def __init__(self, stream, timeout: int = 300) -> None:
         """Initialize a stream output."""
         self.idle = False
+        self.timeout = timeout
         self._stream = stream
         self._cursor = None
         self._event = asyncio.Event()
         self._segments = deque(maxlen=self.num_segments)
         self._unsub = None
 
+    @property
+    def name(self) -> str:
+        """Return provider name."""
+        return None
+
     @property
     def format(self) -> str:
         """Return container format."""
@@ -82,7 +88,8 @@ def get_segment(self, sequence: int = None) -> Any:
         # Reset idle timeout
         if self._unsub is not None:
             self._unsub()
-        self._unsub = async_call_later(self._stream.hass, 300, self._timeout)
+        self._unsub = async_call_later(
+            self._stream.hass, self.timeout, self._timeout)
 
         if not sequence:
             return self._segments
@@ -111,14 +118,14 @@ def put(self, segment: Segment) -> None:
         # Start idle timeout when we start recieving data
         if self._unsub is None:
             self._unsub = async_call_later(
-                self._stream.hass, 300, self._timeout)
+                self._stream.hass, self.timeout, self._timeout)
 
         if segment is None:
             self._event.set()
             # Cleanup provider
             if self._unsub is not None:
                 self._unsub()
-            self._cleanup()
+            self.cleanup()
             return
 
         self._segments.append(segment)
@@ -133,11 +140,11 @@ def _timeout(self, _now=None):
             self.idle = True
             self._stream.check_idle()
         else:
-            self._cleanup()
+            self.cleanup()
 
-    def _cleanup(self):
-        """Remove provider."""
-        self._segments = []
+    def cleanup(self):
+        """Handle cleanup."""
+        self._segments = deque(maxlen=self.num_segments)
         self._stream.remove_provider(self)
 
 

homeassistant/components/stream/hls.py

@@ -110,6 +110,11 @@ def render(self, track, start_time):
 class HlsStreamOutput(StreamOutput):
     """Represents HLS Output formats."""
 
+    @property
+    def name(self) -> str:
+        """Return provider name."""
+        return 'hls'
+
     @property
     def format(self) -> str:
         """Return container format."""