Merge pull request #5 from evanjd/live_streaming

Added preliminary support for handling camera live streams

README.md

@@ -43,6 +43,7 @@ PyPi package soon.
   - End time (local or UTC)
   - Duration
   - Relevance level (indicating whether people/objects were detected)
+- Download real-time live stream data to disk or serve to your application as a raw bytes object
 - Query/filter the activity history by start time and/or activity properties (duration, relevance)
 - Download any activity video to disk or serve to your application as a raw bytes object
 - Download still images from camera to disk or serve to your application as a raw bytes object
@@ -51,7 +52,6 @@ PyPi package soon.
 
 ## Features planned
 
-- Live streaming support (soon)
 - Motion alerts (eventually)
 - Logi Circle CLI (eventually)
 
@@ -95,6 +95,27 @@ loop.run_until_complete(get_latest_activity())
 loop.close()
 ```
 
+#### Download live stream data to disk:
+
+##### You can use FFMPEG to concatenate the segments into an individual file.
+
+```python
+async def get_livestream():
+    camera = (await logi_api.cameras)[1]
+    live_stream = camera.live_stream
+    count = 1
+
+    while True:
+        # Keep writing until interrupted
+        filename = '%s-segment-%s.mp4' % (camera.name, count)
+        await live_stream.get_segment(filename=filename)
+        count += 1
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(get_livestream())
+loop.close()
+```
+
 #### Download last 24 hours activity for the 1st camera (limited to 100, 5 at a time):
 
 ```python
@@ -175,6 +196,8 @@ loop.close()
   - Replaced requests with aiohttp
   - Added support for turning camera on & off
   - Added update() method to Camera object to refresh data from server
+- 0.1.0
+  - Added preliminary support for live streams (to be improved)
 
 ## Meta
 
@@ -191,12 +214,12 @@ Distributed under the MIT license. See `LICENSE` for more information.
 
 They're very welcome, every little bit helps! I'm especially keen for help supporting devices that I do not own and cannot test with (eg. Circle 2 indoor & outdoor cameras).
 
-1. Fork it (<https://github.com/evanjd/python-logi-circle/fork>)
-2. Create your feature branch (`git checkout -b feature/fooBar`)
-3. Make sure there's no linting errors.
-4. Commit your changes (`git commit -am 'Add some fooBar'`)
-5. Push to the branch (`git push origin feature/fooBar`)
-6. Create a new Pull Request
+1. Fork it (<https://github.com/evanjd/python-logi-circle/fork>).
+2. Create your feature branch (`git checkout -b feature/fooBar`).
+3. Commit your changes (`git commit -am 'Add some fooBar'`).
+4. Add/update tests if needed, then run `tox` to confirm no test failures.
+5. Push to the branch (`git push origin feature/fooBar`).
+6. Create a new pull request!
 
 <!-- Markdown link & img dfn's -->
 

logi_circle/__init__.py

@@ -16,7 +16,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-class Logi(object):
+class Logi():
     """A Python abstraction object to Logi Circle cameras."""
 
     def __init__(self, username, password, reuse_session=True, cache_file=CACHE_FILE):

logi_circle/activity.py

@@ -11,7 +11,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-class Activity(object):
+class Activity():
     """Generic implementation for a Logi Circle activity."""
 
     def __init__(self, camera, activity, url, local_tz, logi):

logi_circle/camera.py

@@ -7,13 +7,14 @@
 from .const import (
     PROTOCOL, ACCESSORIES_ENDPOINT, ACTIVITIES_ENDPOINT, IMAGES_ENDPOINT, JPEG_CONTENT_TYPE)
 from .activity import Activity
+from .live_stream import LiveStream
 from .utils import _stream_to_file
 from .exception import UnexpectedContentType
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class Camera(object):
+class Camera():
     """Generic implementation for Logi Circle camera."""
 
     def __init__(self, logi, camera):
@@ -115,6 +116,11 @@ async def query_activity_history(self, property_filter=None, date_filter=None, d
 
         return activities
 
+    @property
+    def live_stream(self):
+        """Return LiveStream object."""
+        return LiveStream(self, self._logi)
+
     @property
     async def last_activity(self):
         """Returns the most recent activity as an Activity object."""

logi_circle/const.py

@@ -33,10 +33,12 @@
 AUTH_ENDPOINT = '/accounts/authorization'
 CAMERAS_ENDPOINT = '/accessories'
 IMAGES_ENDPOINT = '/image'
+LIVESTREAM_ENDPOINT = '/mpd'
 ACTIVITIES_ENDPOINT = '/activities'
 ACCESSORIES_ENDPOINT = '/accessories'
 VALIDATE_ENDPOINT = '/accounts/self'
 
 # Misc
 JPEG_CONTENT_TYPE = 'image/jpeg'
 VIDEO_CONTENT_TYPE = 'application/octet-stream'
+LIVESTREAM_XMLNS = 'urn:mpeg:dash:schema:mpd:2011'

logi_circle/live_stream.py

@@ -0,0 +1,144 @@
+"""Live Stream class"""
+# coding: utf-8
+# vim:sw=4:ts=4:et:
+import logging
+import asyncio
+from datetime import datetime, timedelta
+import xml.etree.ElementTree as ET
+from .const import (PROTOCOL, ACCESSORIES_ENDPOINT,
+                    LIVESTREAM_ENDPOINT, LIVESTREAM_XMLNS)
+from .utils import _combine_stream_and_write
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class LiveStream():
+    """Logi Circle DASH client."""
+
+    def __init__(self, camera, logi):
+        """Initialize LiveStream object."""
+        self._camera = camera
+        self._logi = logi
+        self._initialisation_file = None
+        self._index = None
+        self._initialised = False
+        self._mpd_data = {}
+        self._next_segment_time = None
+
+    def _get_mpd_url(self):
+        """Returns the URL for the MPD file"""
+        return '%s://%s/api%s/%s%s' % (PROTOCOL, self._camera.node_id,
+                                       ACCESSORIES_ENDPOINT, self._camera.id, LIVESTREAM_ENDPOINT)
+
+    async def _get_init_mp4(self, base_url, header_url):
+        """Downloads the initialisation file and returns a bytes object"""
+        url = '%s%s' % (base_url, header_url)
+
+        header = await self._logi._fetch(
+            url=url, relative_to_api_root=False, raw=True)
+
+        header_data = await header.read()
+        header.close()
+        return header_data
+
+    async def _get_mpd(self):
+        """Gets the MPD XML and extracts the data required to download segments"""
+
+        # Force an update to get the latest node ID
+        await self._camera.update()
+
+        # Get MPD XML and save to raw_xml var
+        url = self._get_mpd_url()
+
+        xml_req = await self._logi._fetch(
+            url=url, relative_to_api_root=False, raw=True)
+        raw_xml = await xml_req.read()
+        xml_req.close()
+
+        # Extract data from MPD XML
+        xml = ET.fromstring(raw_xml)
+        stream_config = {}
+        stream_config['base_url'] = xml.find(
+            './{%s}BaseURL' % (LIVESTREAM_XMLNS)).text
+        stream_config['header_url'] = xml.find('.//{%s}SegmentTemplate' %
+                                               (LIVESTREAM_XMLNS)).get('initialization')
+        stream_config['stream_filename_template'] = xml.find(
+            './/{%s}SegmentTemplate' % (LIVESTREAM_XMLNS)).get('media')
+        stream_config['start_index'] = int(xml.find('.//{%s}SegmentTemplate' %
+                                                    (LIVESTREAM_XMLNS)).get('startNumber'))
+        stream_config['segment_length'] = int(xml.find('.//{%s}SegmentTemplate' %
+                                                       (LIVESTREAM_XMLNS)).get('duration'))
+        return stream_config
+
+    def _build_mp4(self, segment_file):
+        """Concatenates the initialisation data and segment file to return a playable MP4"""
+
+        return self._initialisation_file + segment_file
+
+    def _set_next_stream_time(self):
+        duration = self._mpd_data['segment_length']
+        self._next_segment_time = datetime.now() + timedelta(milliseconds=duration)
+
+    def _get_time_before_next_stream(self):
+        """Time before the next segment is available, in seconds"""
+        delay = self._next_segment_time - datetime.now()
+        delay_in_seconds = delay.total_seconds()
+        return 0 if delay_in_seconds < 0 else delay_in_seconds
+
+    def _get_segment_url(self):
+        """Builds the URL to get the next video segment"""
+        base_url = self._mpd_data['base_url']
+        stream_filename_template = self._mpd_data['stream_filename_template']
+
+        file_name = stream_filename_template.replace(
+            '$Number$', str(self._index))
+        return '%s%s' % (base_url, file_name)
+
+    async def get_segment(self, filename=None):
+        """Returns the current segment video from the live stream"""
+        # Initialise if required
+        if self._initialised is False:
+            await self._initialise()
+
+        # Get current wait time
+        wait_time = self._get_time_before_next_stream()
+
+        _LOGGER.debug(
+            'Sleeping for %s seconds before grabbing next live stream segment.', wait_time)
+        # And sleep, if needed.
+        if wait_time > 0:
+            await asyncio.sleep(wait_time)
+
+        self._set_next_stream_time()
+
+        # Get segment data
+        url = self._get_segment_url()
+        segment = await self._logi._fetch(
+            url=url, relative_to_api_root=False, raw=True)
+
+        # Increment segment counter and timer for next download
+        self._index += 1
+
+        if filename:
+            # Stream to file
+            await _combine_stream_and_write(init_data=self._initialisation_file,
+                                            stream=segment.content,
+                                            filename=filename)
+            segment.close()
+        else:
+            # Return binary object
+            content = await segment.read()
+            segment.close()
+            return self._build_mp4(content)
+
+    async def _initialise(self):
+        """Sets up the live stream so that it's ready to output video data"""
+
+        # Get stream config and cache header
+        self._mpd_data = await self._get_mpd()
+        self._initialisation_file = await self._get_init_mp4(self._mpd_data['base_url'], self._mpd_data['header_url'])
+        self._index = self._mpd_data['start_index']
+        self._initialised = True
+
+        # Delay stream until one segment is ready
+        self._set_next_stream_time()

logi_circle/utils.py

@@ -37,6 +37,17 @@ async def _handle_response(request, raw):
         return resp
 
 
+async def _combine_stream_and_write(init_data, stream, filename):
+    """Combine DASH init data and stream and write to file"""
+    with open(filename, 'wb') as file_handle:
+        file_handle.write(init_data)
+        while True:
+            chunk = await stream.read(1024)
+            if not chunk:
+                break
+            file_handle.write(chunk)
+
+
 async def _stream_to_file(stream, filename):
     """Stream aiohttp response to file"""
     with open(filename, 'wb') as file_handle:

setup.py

@@ -10,7 +10,7 @@ def readme():
 setup(
     name='logi_circle',
     packages=['logi_circle'],
-    version='0.0.4',
+    version='0.1.0',
     description='A Python library to communicate with Logi Circle cameras',
     long_description=readme(),
     author='Evan Bruhn',
@@ -33,7 +33,6 @@ def readme():
         'Operating System :: OS Independent',
         'Framework :: AsyncIO',
         'Programming Language :: Python',
-        'Programming Language :: Python :: 3.5',
         'Programming Language :: Python :: 3.6',
         'Programming Language :: Python :: 3.7',
         'Topic :: Home Automation',