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

New method to retrieve cameras due to api change

Browse Source
This commit is contained in:
Kevin Fronczak
2019-03-01 11:35:30 -05:00
parent aff95ba9e9
commit 602b6160ab
8 changed files with 55 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blinkpy/api.py
+6 -6
View File
@@ -102,9 +102,11 @@ def request_command_status(blink, network, command_id):
return http_get(blink, url)
@Throttle(seconds=MIN_THROTTLE_TIME)
def request_homescreen(blink):
"""Request homescreen info."""
url = "{}/homescreen".format(blink.urls.base_url)
url = "{}/api/v3/accounts/{}/homescreen".format(blink.urls.base_url,
blink.account_id)
return http_get(blink, url)
@@ -183,7 +185,6 @@ def request_cameras(blink, network):
return http_get(blink, url)
@Throttle(seconds=MIN_THROTTLE_TIME)
def request_camera_info(blink, network, camera_id):
"""
Request camera info for one camera.
@@ -192,13 +193,12 @@ def request_camera_info(blink, network, camera_id):
:param network: Sync module network id.
:param camera_id: Camera ID of camera to request info from.
"""
url = "{}/network/{}/camera/{}".format(blink.urls.base_url,
network,
camera_id)
url = "{}/network/{}/camera/{}/config".format(blink.urls.base_url,
network,
camera_id)
return http_get(blink, url)
@Throttle(seconds=MIN_THROTTLE_TIME)
def request_camera_sensors(blink, network, camera_id):
"""
Request camera sensor info for one camera.
blinkpy/blinkpy.py
+27 -1
View File
@@ -88,9 +88,16 @@ class Blink():
elif not self.get_auth_token():
return
camera_list = self.get_cameras()
networks = self.get_ids()
for network_name, network_id in networks.items():
sync_module = BlinkSyncModule(self, network_name, network_id)
if network_id not in camera_list.keys():
camera_list[network_id] = {}
_LOGGER.warning("No cameras found for %s", network_name)
sync_module = BlinkSyncModule(self,
network_name,
network_id,
camera_list[network_id])
sync_module.start()
self.sync[network_name] = sync_module
self.cameras = self.merge_cameras()
@@ -170,6 +177,25 @@ class Blink():
self.network_ids = all_networks
return network_dict
def get_cameras(self):
"""Retrieve a camera list for each onboarded network."""
response = api.request_homescreen(self)
try:
all_cameras = response['cameras']
for camera in all_cameras:
camera_network = camera['network_id']
camera_name = camera['name']
camera_id = camera['id']
camera_info = {'name': camera_name, 'id': camera_id}
if camera_network in all_cameras:
all_cameras[camera_network].append(camera_info)
else:
all_cameras[camera_network] = [camera_info]
return all_cameras
except KeyError:
_LOGGER.error("Initialization failue. Could not retrieve cameras.")
return {}
def refresh(self, force_cache=False):
"""
Perform a system refresh.
blinkpy/camera.py
+2 -5
View File
@@ -94,7 +94,7 @@ class BlinkCamera():
def update(self, config, force_cache=False, **kwargs):
"""Update camera info."""
force = kwargs.pop('force', False)
# force = kwargs.pop('force', False)
self.name = config['name']
self.camera_id = str(config['camera_id'])
self.network_id = str(config['network_id'])
@@ -108,15 +108,12 @@ class BlinkCamera():
# Retrieve calibrated temperature from special endpoint
resp = api.request_camera_sensors(self.sync.blink,
self.network_id,
self.camera_id,
force=force)
self.camera_id)
try:
self.temperature_calibrated = resp['temp']
except KeyError:
self.temperature_calibrated = self.temperature
_LOGGER.warning("Could not retrieve calibrated temperature.")
except TypeError:
_LOGGER.debug("API call temporarily throttled.")
# Check if thumbnail exists in config, if not try to
# get it from the homescreen info in teh sync module
blinkpy/sync_module.py
+9 -7
View File
@@ -13,7 +13,7 @@ _LOGGER = logging.getLogger(__name__)
class BlinkSyncModule():
"""Class to initialize sync module."""
def __init__(self, blink, network_name, network_id):
def __init__(self, blink, network_name, network_id, camera_list):
"""
Initialize Blink sync module.
@@ -36,6 +36,7 @@ class BlinkSyncModule():
self.cameras = CaseInsensitiveDict({})
self.motion = {}
self.last_record = {}
self.camera_list = camera_list
@property
def attributes(self):
@@ -96,18 +97,20 @@ class BlinkSyncModule():
response,
exc_info=True)
self.events = self.get_events(force=True)
self.homescreen = api.request_homescreen(self.blink)
self.network_info = api.request_network_status(self.blink,
self.network_id)
self.check_new_videos()
camera_info = self.get_camera_info()
for camera_config in camera_info:
for camera_config in self.camera_list:
if 'name' not in camera_config:
break
name = camera_config['name']
self.cameras[name] = BlinkCamera(self)
self.motion[name] = False
self.cameras[name].update(camera_config,
camera_info = api.request_camera_info(self.blink,
self.network_id,
camera_config['id'])
self.cameras[name].update(camera_info,
force_cache=True,
force=True)
@@ -143,7 +146,6 @@ class BlinkSyncModule():
def refresh(self, force_cache=False):
"""Get all blink cameras and pulls their most recent status."""
self.events = self.get_events()
self.homescreen = api.request_homescreen(self.blink)
self.network_info = api.request_network_status(self.blink,
self.network_id)
tests/test_blink_functions.py
+2 -2
View File
@@ -79,8 +79,8 @@ class TestBlinkFunctions(unittest.TestCase):
"""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'] = 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()
tests/test_blinkpy.py
+4 -1
View File
@@ -31,7 +31,10 @@ class TestBlinkSetup(unittest.TestCase):
self.blink_no_cred = Blink()
self.blink = Blink(username=USERNAME,
password=PASSWORD)
self.blink.sync['test'] = BlinkSyncModule(self.blink, 'test', '1234')
self.blink.sync['test'] = BlinkSyncModule(self.blink,
'test',
'1234',
[])
self.blink.urls = BlinkURLHandler('test')
self.blink.session = create_session()
tests/test_cameras.py
+1 -1
View File
@@ -46,7 +46,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
self.blink._auth_header = header
self.blink.session = create_session()
self.blink.urls = BlinkURLHandler('test')
self.blink.sync['test'] = BlinkSyncModule(self.blink, 'test', 1234)
self.blink.sync['test'] = BlinkSyncModule(self.blink, 'test', 1234, [])
self.camera = BlinkCamera(self.blink.sync['test'])
self.camera.name = 'foobar'
self.blink.sync['test'].cameras['foobar'] = self.camera
tests/test_sync_module.py
+4 -15
View File
@@ -24,7 +24,10 @@ class TestBlinkSyncModule(unittest.TestCase):
'TOKEN_AUTH': 'foobar123'
}
self.blink.urls = blinkpy.BlinkURLHandler('test')
self.blink.sync['test'] = BlinkSyncModule(self.blink, 'test', '1234')
self.blink.sync['test'] = BlinkSyncModule(self.blink,
'test',
'1234',
[])
self.camera = BlinkCamera(self.blink.sync)
self.mock_start = [
{'syncmodule': {
@@ -126,20 +129,6 @@ class TestBlinkSyncModule(unittest.TestCase):
self.blink.sync['test'].start()
self.assertEqual(self.blink.sync['test'].network_id, 8675309)
def test_unexpected_events(self, mock_resp):
"""Test unexpected events response."""
self.mock_start[1] = None
mock_resp.side_effect = self.mock_start
self.blink.sync['test'].start()
self.assertEqual(self.blink.sync['test'].events, False)
def test_missing_events(self, mock_resp):
"""Test missing events key from response."""
self.mock_start[1] = {}
mock_resp.side_effect = self.mock_start
self.blink.sync['test'].start()
self.assertEqual(self.blink.sync['test'].events, False)
def test_unexpected_camera_info(self, mock_resp):
"""Test unexpected camera info response."""
self.blink.sync['test'].cameras['foo'] = None