Added cameras dict to blink to keep track of ALL cameras

This commit is contained in:
Kevin Fronczak
2018-11-23 17:47:44 -05:00
parent 84f987d29c
commit 5a8af75a5b
4 changed files with 36 additions and 7 deletions
+5 -6
View File
@@ -62,16 +62,15 @@ The below code will display cameras and their available attributes:
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD')
blink.start()
for sync_name, sync in blink.sync.items():
for name, camera in blink.sync[sync_name].cameras.items():
print(name) # Name of the camera
print(camera.attributes) # Print available attributes of camera
for name, camera in blink.cameras.items():
print(name) # Name of the camera
print(camera.attributes) # Print available attributes of camera
The most recent images and videos can be accessed as a bytes-object via internal variables. These can be updated with calls to ``Blink.refresh()`` but will only make a request if motion has been detected or other changes have been found. This can be overridden with the ``force_cache`` flag, but this should be used for debugging only since it overrides the internal request throttling.
.. code:: python
camera = blink.sync['SYNC NAME'].camera['SOME CAMERA NAME']
camera = blink.cameras['SOME CAMERA NAME']
blink.refresh(force_cache=True) # force a cache update USE WITH CAUTION
camera.image_from_cache.raw # bytes-like image object (jpg)
camera.video_from_cache.raw # bytes-like video object (mp4)
@@ -80,7 +79,7 @@ The ``blinkpy`` api also allows for saving images and videos to a file and snapp
.. code:: python
camera = blink.sync['SYNC NAME'].camera['SOME CAMERA NAME']
camera = blink.cameras['SOME CAMERA NAME']
camera.snap_picture() # Take a new picture with the camera
blink.refresh() # Get new information from server
camera.image_to_file('/local/path/for/image.jpg')
+10 -1
View File
@@ -19,7 +19,7 @@ import blinkpy.helpers.errors as ERROR
from blinkpy import api
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers.util import (
create_session, BlinkURLHandler,
create_session, merge_dicts, BlinkURLHandler,
BlinkAuthenticationException)
from blinkpy.helpers.constants import (
BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL)
@@ -57,6 +57,7 @@ class Blink():
self.refresh_rate = refresh_rate
self.session = None
self.networks = []
self.cameras = CaseInsensitiveDict({})
self._login_url = LOGIN_URL
@property
@@ -81,6 +82,7 @@ class Blink():
sync_module = BlinkSyncModule(self, network_name, network_id)
sync_module.start()
self.sync[network_name] = sync_module
self.cameras = self.merge_cameras()
def login(self):
"""Prompt user for username and password."""
@@ -175,3 +177,10 @@ class Blink():
self.last_refresh = current_time
return True
return False
def merge_cameras(self):
"""Merge all sync camera dicts into one."""
combined = CaseInsensitiveDict({})
for sync_name, sync in self.sync.items():
combined = merge_dicts(combined, sync.cameras)
return combined
+9
View File
@@ -9,6 +9,15 @@ import blinkpy.helpers.errors as ERROR
_LOGGER = logging.getLogger(__name__)
def merge_dicts(dict_a, dict_b):
"""Merge two dictionaries into one."""
duplicates = [val for val in dict_a if val in dict_b]
if duplicates:
_LOGGER.warning(("Duplicates found during merge: %s. "
"Renaming is recommended."), duplicates)
return {**dict_a, **dict_b}
def create_session():
"""Create a session for blink communication."""
sess = Session()
+12
View File
@@ -70,3 +70,15 @@ class TestBlinkFunctions(unittest.TestCase):
self.assertEqual(self.blink.region, 'UNKNOWN')
# pylint: disable=protected-access
self.assertEqual(self.blink._token, 'foobar123')
def test_merge_cameras(self, mock_sess):
"""Test merge camera functionality."""
first_dict = {'foo': 'bar', 'test': 123}
next_dict = {'foobar': 456, 'bar': 'foo'}
self.blink.sync['foo'] = BlinkSyncModule(self.blink, 'foo', 1)
self.blink.sync['bar'] = BlinkSyncModule(self.blink, 'bar', 2)
self.blink.sync['foo'].cameras = first_dict
self.blink.sync['bar'].cameras = next_dict
result = self.blink.merge_cameras()
expected = {'foo': 'bar', 'test': 123, 'foobar': 456, 'bar': 'foo'}
self.assertEqual(expected, result)