Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* initial commit of streams * refactor stream component * refactor so stream formats are not considered a platform * initial test and minor refactor * fix linting * update requirements * need av in tests as well * fix import in class def vs method * fix travis and docker builds * address code review comments * fix logger, add stream start/stop logs, listen to HASS stop * address additional code review comments * beef up tests * fix tests * fix lint * add stream_source to onvif camera * address pr comments * add keepalive to camera play_stream service * remove keepalive and move import * implement registry and have output provider remove itself from stream after idle, set libav log level to error
- Loading branch information
Showing
18 changed files
with
993 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| """ | ||
| Provide functionality to stream video source. | ||
| For more details about this component, please refer to the documentation at | ||
| https://home-assistant.io/components/stream/ | ||
| """ | ||
| import logging | ||
| import threading | ||
|
|
||
| import voluptuous as vol | ||
|
|
||
| from homeassistant.auth.util import generate_secret | ||
| from homeassistant.const import EVENT_HOMEASSISTANT_STOP | ||
| 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 .core import PROVIDERS | ||
| from .worker import stream_worker | ||
| from .hls import async_setup_hls | ||
|
|
||
| REQUIREMENTS = ['av==6.1.2'] | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
| DEPENDENCIES = ['http'] | ||
|
|
||
| CONFIG_SCHEMA = vol.Schema({ | ||
| DOMAIN: vol.Schema({}), | ||
| }, extra=vol.ALLOW_EXTRA) | ||
|
|
||
| # Set log level to error for libav | ||
| logging.getLogger('libav').setLevel(logging.ERROR) | ||
|
|
||
|
|
||
| @bind_hass | ||
| def request_stream(hass, stream_source, *, fmt='hls', | ||
| keepalive=False, options=None): | ||
| """Set up stream with token.""" | ||
| if DOMAIN not in hass.config.components: | ||
| raise HomeAssistantError("Stream component is not set up.") | ||
|
|
||
| if options is None: | ||
| options = {} | ||
|
|
||
| try: | ||
| streams = hass.data[DOMAIN][ATTR_STREAMS] | ||
| stream = streams.get(stream_source) | ||
| if not stream: | ||
| stream = Stream(hass, stream_source, | ||
| options=options, keepalive=keepalive) | ||
| streams[stream_source] = stream | ||
|
|
||
| # Add provider | ||
| stream.add_provider(fmt) | ||
|
|
||
| if not stream.access_token: | ||
| stream.access_token = generate_secret() | ||
| stream.start() | ||
| return hass.data[DOMAIN][ATTR_ENDPOINTS][fmt].format( | ||
| hass.config.api.base_url, stream.access_token) | ||
| except Exception: | ||
| raise HomeAssistantError('Unable to get stream') | ||
|
|
||
|
|
||
| async def async_setup(hass, config): | ||
| """Set up stream.""" | ||
| hass.data[DOMAIN] = {} | ||
| hass.data[DOMAIN][ATTR_ENDPOINTS] = {} | ||
| hass.data[DOMAIN][ATTR_STREAMS] = {} | ||
|
|
||
| # Setup HLS | ||
| hls_endpoint = async_setup_hls(hass) | ||
| hass.data[DOMAIN][ATTR_ENDPOINTS]['hls'] = hls_endpoint | ||
|
|
||
| @callback | ||
| def shutdown(event): | ||
| """Stop all stream workers.""" | ||
| for stream in hass.data[DOMAIN][ATTR_STREAMS].values(): | ||
| stream.keepalive = False | ||
| stream.stop() | ||
| _LOGGER.info("Stopped stream workers.") | ||
|
|
||
| hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, shutdown) | ||
|
|
||
| return True | ||
|
|
||
|
|
||
| class Stream: | ||
| """Represents a single stream.""" | ||
|
|
||
| def __init__(self, hass, source, options=None, keepalive=False): | ||
| """Initialize a stream.""" | ||
| self.hass = hass | ||
| self.source = source | ||
| self.options = options | ||
| self.keepalive = keepalive | ||
| self.access_token = None | ||
| self._thread = None | ||
| self._thread_quit = None | ||
| self._outputs = {} | ||
|
|
||
| if self.options is None: | ||
| self.options = {} | ||
|
|
||
| @property | ||
| def outputs(self): | ||
| """Return stream outputs.""" | ||
| return self._outputs | ||
|
|
||
| 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] | ||
|
|
||
| def remove_provider(self, provider): | ||
| """Remove provider output stream.""" | ||
| if provider.format in self._outputs: | ||
| del self._outputs[provider.format] | ||
|
|
||
| if not self._outputs: | ||
| self.stop() | ||
|
|
||
| def start(self): | ||
| """Start a stream.""" | ||
| if self._thread is None or not self._thread.isAlive(): | ||
| self._thread_quit = threading.Event() | ||
| self._thread = threading.Thread( | ||
| name='stream_worker', | ||
| target=stream_worker, | ||
| args=( | ||
| self.hass, self, self._thread_quit)) | ||
| self._thread.start() | ||
| _LOGGER.info("Started stream: %s", self.source) | ||
|
|
||
| def stop(self): | ||
| """Remove outputs and access token.""" | ||
| self._outputs = {} | ||
| self.access_token = None | ||
|
|
||
| if not self.keepalive: | ||
| self._stop() | ||
|
|
||
| def _stop(self): | ||
| """Stop worker thread.""" | ||
| if self._thread is not None: | ||
| self._thread_quit.set() | ||
| self._thread.join() | ||
| self._thread = None | ||
| _LOGGER.info("Stopped stream: %s", self.source) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| """Constants for Stream component.""" | ||
| DOMAIN = 'stream' | ||
|
|
||
| ATTR_ENDPOINTS = 'endpoints' | ||
| ATTR_STREAMS = 'streams' | ||
| ATTR_KEEPALIVE = 'keepalive' | ||
|
|
||
| OUTPUT_FORMATS = ['hls'] | ||
|
|
||
| FORMAT_CONTENT_TYPE = { | ||
| 'hls': 'application/vnd.apple.mpegurl' | ||
| } | ||
|
|
||
| AUDIO_SAMPLE_RATE = 44100 |
Oops, something went wrong.