caelebfi/blinkpy
Archived
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #149 from fronzbot/add-download-videos

Browse Source 
Add download videos routine
This commit is contained in:
Kevin Fronczak
2019-01-31 10:56:08 -08:00
committed by GitHub
parent 663a128469 02937c0ef5
commit d5f866bd34
4 changed files with 197 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.rst
+28
View File
@@ -51,6 +51,23 @@ The simplest way to use this package from a terminal is to call ``Blink.start()`
If you would like to log in without setting up the cameras or system, you can simply call the ``Blink.login()`` function which will prompt for a username and password and then authenticate with the server. This is useful if you want to avoid use of the ``start()`` function which simply acts as a wrapper for more targeted API methods.
At initialization, you may also set the logging level of the ``blinkpy`` library like so (default is ``INFO``:
.. code:: python
import logging
from blinkpy import blinkpy
blink = blinkpy.Blink(..., loglevel=logging.<LEVEL>)
blink.start()
You can also disable logging of duplicate entries via the ``allow_duplicate_logs`` flag (default is ``True``):
.. code:: python
from blinkpy import blinkpy
blink = blinkpy.Blink(..., allow_duplicate_logs=False)
blink.start()
Cameras are instantiated as individual ``BlinkCamera`` classes within a ``BlinkSyncModule`` instance. All of your sync modules are stored within the ``Blink.sync`` dictionary and can be accessed using the name of the sync module as the key (this is the name of your sync module in the Blink App).
The below code will display cameras and their available attributes:
@@ -84,6 +101,17 @@ The ``blinkpy`` api also allows for saving images and videos to a file and snapp
blink.refresh() # Get new information from server
camera.image_to_file('/local/path/for/image.jpg')
camera.video_to_file('/local/path/for/video.mp4')
You can also use this library to download all videos from the server. In order to do this, you must specify a ``path``. You may also specifiy a how far back in time to go to retrieve videos via the ``since=`` variable (a simple string such as ``"2017/09/21"`` is sufficient), as well as how many pages to traverse via the ``page=`` variable. Note that by default, the library will search the first ten pages which is sufficient in most use cases. Additionally, you can specidy one or more cameras via the ``camera=`` property. This can be a single string indicating the name of the camera, or a list of camera names. By default, it is set to the string ``'all'`` to grab videos from all cameras.
Example usage, which downloads all videos recorded since July 4th, 2018 at 9:34am to the ``/home/blink`` directory:
.. code:: python
blink = blinkpy.Blink(username="YOUR USER NAME", password="YOUR PASSWORD")
blink.start()
blink.download_videos('/home/blink', since='2018/07/04 09:34')
.. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=dev
:target: https://travis-ci.org/fronzbot/blinkpy
blinkpy/blinkpy.py
+104 -5
View File
@@ -13,15 +13,20 @@ owned by Immedia Inc., see www.blinkforhome.com for more information.
blinkpy is in no way affiliated with Blink, nor Immedia Inc.
"""
import os.path
import time
import getpass
import logging
from shutil import copyfileobj
from requests.structures import CaseInsensitiveDict
import blinkpy.helpers.errors as ERROR
from dateutil.parser import parse
from blinkpy import api
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers import errors as ERROR
from blinkpy.helpers.util import (
create_session, merge_dicts, BlinkURLHandler,
create_session, merge_dicts, get_time, BlinkURLHandler,
BlinkAuthenticationException, RepeatLogHandler)
from blinkpy.helpers.constants import (
BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL)
@@ -36,7 +41,8 @@ class Blink():
"""Class to initialize communication."""
def __init__(self, username=None, password=None,
refresh_rate=REFRESH_RATE):
refresh_rate=REFRESH_RATE, loglevel=logging.INFO,
allow_duplicate_logs=True):
"""
Initialize Blink system.
@@ -44,6 +50,9 @@ class Blink():
:param password: Blink password
:param refresh_rate: Refresh rate of blink information.
Defaults to 15 (seconds)
:param loglevel: Sets the log level for the logger.
:param allow_duplicate_logs: Set to 'False' to only allow a log
message to be logged once.
"""
self._username = username
self._password = password
@@ -61,9 +70,13 @@ class Blink():
self.session = create_session()
self.networks = []
self.cameras = CaseInsensitiveDict({})
self.video_list = CaseInsensitiveDict({})
self._login_url = LOGIN_URL
self.version = __version__
self.allow_duplicate_logs = allow_duplicate_logs
_LOGGER.addHandler(RepeatLogHandler())
_LOGGER.setLevel(loglevel)
@property
def auth_header(self):
@@ -77,8 +90,8 @@ class Blink():
Method logs in and sets auth token, urls, and ids for future requests.
Essentially this is just a wrapper function for ease of use.
"""
_LOGGER.handlers.pop()
_LOGGER.addHandler(RepeatLogHandler())
if not self.allow_duplicate_logs:
self._reset_logger()
if self._username is None or self._password is None:
if not self.login():
return
@@ -197,3 +210,89 @@ class Blink():
for sync in self.sync:
combined = merge_dicts(combined, self.sync[sync].cameras)
return combined
def download_videos(self, path, since=None, camera='all', stop=10):
"""
Download all videos from server since specified time.
:param path: Path to write files. /path/<cameraname>_<recorddate>.mp4
:param since: Date and time to get videos from.
Ex: "2018/07/28 12:33:00" to retrieve videos since
July 28th 2018 at 12:33:00
:param camera: Camera name to retrieve. Defaults to "all".
Use a list for multiple cameras.
:param stop: Page to stop on (~25 items per page. Default page 10).
"""
# Reset the handler so we don't filter out messages during this method.
if not self.allow_duplicate_logs:
self._reset_logger()
if since is None:
since_epochs = self.last_refresh
else:
parsed_datetime = parse(since, fuzzy=True)
since_epochs = parsed_datetime.timestamp()
formatted_date = get_time(time_to_convert=since_epochs)
_LOGGER.info("Retrieving videos since %s", formatted_date)
if not isinstance(camera, list):
camera = [camera]
for page in range(1, stop):
response = api.request_videos(self, time=since_epochs, page=page)
_LOGGER.debug("Processing page %s", page)
try:
result = response['videos']
if not result:
raise IndexError
except (KeyError, IndexError):
_LOGGER.info("No videos found on page %s. Exiting.", page)
break
self._parse_downloaded_items(result, camera, path)
def _parse_downloaded_items(self, result, camera, path):
"""Parse downloaded videos."""
for item in result:
try:
created_at = item['created_at']
camera_name = item['camera_name']
is_deleted = item['deleted']
address = item['address']
except KeyError:
_LOGGER.info("Missing clip information, skipping...")
continue
if camera_name not in camera and 'all' not in camera:
_LOGGER.debug("Skipping videos for %s.", camera_name)
continue
if is_deleted:
_LOGGER.debug("%s: %s is marked as deleted.",
camera_name,
address)
continue
clip_address = "{}{}".format(self.urls.base_url, address)
filename = "{}_{}.mp4".format(camera_name, created_at)
filename = os.path.join(path, filename)
if os.path.isfile(filename):
_LOGGER.info("%s already exists, skipping...", filename)
continue
response = api.http_get(self, url=clip_address,
stream=True, json=False)
with open(filename, 'wb') as vidfile:
copyfileobj(response.raw, vidfile)
_LOGGER.info("Downloaded video to %s", filename)
# pylint: disable=no-self-use
def _reset_logger(self):
"""Reset the log handler."""
for handler in _LOGGER.handlers:
handler.close()
_LOGGER.removeHandler(handler)
_LOGGER.addHandler(RepeatLogHandler())
requirements.txt
+1
View File
@@ -1,2 +1,3 @@
python-dateutil==2.7.5
requests>=2.20.0
testtools==2.3.0
tests/test_blink_functions.py
+64 -2
View File
@@ -1,12 +1,12 @@
"""Tests camera and system functions."""
import unittest
from unittest import mock
import logging
from requests import Request
from blinkpy import blinkpy
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers.util import create_session
from blinkpy.helpers.util import create_session, get_time
import tests.mock_responses as mresp
USERNAME = 'foobar'
@@ -86,3 +86,65 @@ class TestBlinkFunctions(unittest.TestCase):
result = self.blink.merge_cameras()
expected = {'foo': 'bar', 'test': 123, 'foobar': 456, 'bar': 'foo'}
self.assertEqual(expected, result)
@mock.patch('blinkpy.blinkpy.api.request_videos')
def test_download_video_exit(self, mock_req, mock_sess):
"""Test we exit method when provided bad response."""
blink = blinkpy.Blink(loglevel=logging.DEBUG)
blink.last_refresh = 0
mock_req.return_value = {}
formatted_date = get_time(blink.last_refresh)
expected_log = [
"INFO:blinkpy:Retrieving videos since {}".format(formatted_date),
"DEBUG:blinkpy:Processing page 1",
"INFO:blinkpy:No videos found on page 1. Exiting."
]
with self.assertLogs() as dl_log:
blink.download_videos('/tmp')
self.assertEqual(dl_log.output, expected_log)
@mock.patch('blinkpy.blinkpy.api.request_videos')
def test_parse_downloaded_items(self, mock_req, mock_sess):
"""Test ability to parse downloaded items list."""
blink = blinkpy.Blink(loglevel=logging.DEBUG)
generic_entry = {
'created_at': '1970',
'camera_name': 'foo',
'deleted': True,
'address': '/bar.mp4'
}
result = [generic_entry]
mock_req.return_value = {'videos': result}
blink.last_refresh = 0
formatted_date = get_time(blink.last_refresh)
expected_log = [
"INFO:blinkpy:Retrieving videos since {}".format(formatted_date),
"DEBUG:blinkpy:Processing page 1",
"DEBUG:blinkpy:foo: /bar.mp4 is marked as deleted."
]
with self.assertLogs() as dl_log:
blink.download_videos('/tmp', stop=2)
self.assertEqual(dl_log.output, expected_log)
@mock.patch('blinkpy.blinkpy.api.request_videos')
def test_parse_camera_not_in_list(self, mock_req, mock_sess):
"""Test ability to parse downloaded items list."""
blink = blinkpy.Blink(loglevel=logging.DEBUG)
generic_entry = {
'created_at': '1970',
'camera_name': 'foo',
'deleted': True,
'address': '/bar.mp4'
}
result = [generic_entry]
mock_req.return_value = {'videos': result}
blink.last_refresh = 0
formatted_date = get_time(blink.last_refresh)
expected_log = [
"INFO:blinkpy:Retrieving videos since {}".format(formatted_date),
"DEBUG:blinkpy:Processing page 1",
"DEBUG:blinkpy:Skipping videos for foo."
]
with self.assertLogs() as dl_log:
blink.download_videos('/tmp', camera='bar', stop=2)
self.assertEqual(dl_log.output, expected_log)