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

Merge pull request #160 from fronzbot/api-change

Browse Source 
Api change
This commit is contained in:
Kevin Fronczak
2019-03-01 14:55:38 -05:00
committed by GitHub
parent aff95ba9e9 2ca6b9edda
commit 9dda477e3d
12 changed files with 104 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+1 -1
View File
@@ -2,7 +2,7 @@ sudo: required
matrix:
fast_finish: true
include:
- python: "3.6"
- python: "3.5.3"
env: TOXENV=lint
- python: "3.5.3"
env: TOXENV=py35
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 = {}
for camera in response['cameras']:
camera_network = str(camera['network_id'])
camera_name = camera['name']
camera_id = camera['id']
camera_info = {'name': camera_name, 'id': camera_id}
if camera_network not in all_cameras:
all_cameras[camera_network] = []
all_cameras[camera_network].append(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
+3 -6
View File
@@ -94,9 +94,9 @@ 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.camera_id = str(config['id'])
self.network_id = str(config['network_id'])
self.serial = config['serial']
self.motion_enabled = config['enabled']
@@ -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/helpers/util.py
-2
View File
@@ -107,8 +107,6 @@ class BlinkException(Exception):
class BlinkAuthenticationException(BlinkException):
"""Class to throw authentication exception."""
pass
class BlinkURLHandler():
"""Class that handles Blink URLS."""
blinkpy/sync_module.py
+28 -24
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.
@@ -30,12 +30,12 @@ class BlinkSyncModule():
self.sync_id = None
self.host = None
self.summary = None
self.homescreen = None
self.network_info = None
self.events = []
self.cameras = CaseInsensitiveDict({})
self.motion = {}
self.last_record = {}
self.camera_list = camera_list
@property
def attributes(self):
@@ -96,20 +96,26 @@ 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:
name = camera_config['name']
self.cameras[name] = BlinkCamera(self)
self.motion[name] = False
self.cameras[name].update(camera_config,
force_cache=True,
force=True)
try:
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
camera_info = self.get_camera_info(camera_config['id'])
self.cameras[name].update(camera_info,
force_cache=True,
force=True)
except KeyError:
_LOGGER.error("Could not create cameras instances for %s",
self.name,
exc_info=True)
return False
return True
@@ -127,14 +133,13 @@ class BlinkSyncModule():
exc_info=True)
return False
def get_camera_info(self, **kwargs):
def get_camera_info(self, camera_id):
"""Retrieve camera information."""
force = kwargs.pop('force', False)
response = api.request_cameras(self.blink,
self.network_id,
force=force)
response = api.request_camera_info(self.blink,
self.network_id,
camera_id)
try:
return response['devicestatus']
return response['camera'][0]
except (TypeError, KeyError):
_LOGGER.error("Could not extract camera info: %s",
response,
@@ -143,15 +148,14 @@ 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)
camera_info = self.get_camera_info()
self.check_new_videos()
for camera_config in camera_info:
name = camera_config['name']
self.cameras[name].update(camera_config, force_cache=force_cache)
for camera_name in self.cameras.keys():
camera_id = self.cameras[camera_name].camera_id
camera_info = self.get_camera_info(camera_id)
self.cameras[camera_name].update(camera_info,
force_cache=force_cache)
def check_new_videos(self):
"""Check if new videos since last refresh."""
requirements_test.txt
+1 -1
View File
@@ -1,6 +1,6 @@
flake8==3.5.0
flake8-docstrings==1.3.0
pylint==2.1.1
pylint==2.3.0
pydocstyle==2.1.1
pytest==3.7.1
pytest-cov>=2.3.1
tests/mock_responses.py
-2
View File
@@ -67,5 +67,3 @@ def mocked_session_send(*args, **kwargs):
class MockURLHandler(BlinkURLHandler):
"""Mocks URL Handler in blinkpy module."""
pass
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
+24 -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()
@@ -160,3 +163,23 @@ class TestBlinkSetup(unittest.TestCase):
"""Check that we appropriately handle unexpected login info."""
mock_login.return_value = None
self.assertFalse(self.blink.get_auth_token())
@mock.patch('blinkpy.api.request_homescreen')
def test_get_cameras(self, mock_home, mock_sess):
"""Check retrieval of camera information."""
mock_home.return_value = {
'cameras': [{'name': 'foo', 'network_id': 1234, 'id': 5678},
{'name': 'bar', 'network_id': 1234, 'id': 5679},
{'name': 'test', 'network_id': 4321, 'id': 0000}]
}
result = self.blink.get_cameras()
self.assertEqual(result, {'1234': [{'name': 'foo', 'id': 5678},
{'name': 'bar', 'id': 5679}],
'4321': [{'name': 'test', 'id': 0000}]})
@mock.patch('blinkpy.api.request_homescreen')
def test_get_cameras_failure(self, mock_home, mock_sess):
"""Check that on failure we initialize empty info and move on."""
mock_home.return_value = {}
result = self.blink.get_cameras()
self.assertEqual(result, {})
tests/test_cameras.py
+5 -5
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
@@ -59,7 +59,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
"""Test that we can properly update camera properties."""
config = {
'name': 'new',
'camera_id': 1234,
'id': 1234,
'network_id': 5678,
'serial': '12345678',
'enabled': False,
@@ -115,7 +115,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
}
config = {
'name': 'new',
'camera_id': 1234,
'id': 1234,
'network_id': 5678,
'serial': '12345678',
'enabled': False,
@@ -144,7 +144,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
self.camera.last_record = ['1']
config = {
'name': 'new',
'camera_id': 1234,
'id': 1234,
'network_id': 5678,
'serial': '12345678',
'enabled': False,
@@ -176,7 +176,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
mock_sess.return_value = 'foobar'
config = {
'name': 'new',
'camera_id': 1234,
'id': 1234,
'network_id': 5678,
'serial': '12345678',
'enabled': False,
tests/test_sync_module.py
+7 -17
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': {
@@ -52,8 +55,9 @@ class TestBlinkSyncModule(unittest.TestCase):
def test_get_camera_info(self, mock_resp):
"""Test get camera info function."""
mock_resp.return_value = {'devicestatus': True}
self.assertEqual(self.blink.sync['test'].get_camera_info(), True)
mock_resp.return_value = {'camera': ['foobar']}
self.assertEqual(self.blink.sync['test'].get_camera_info('1234'),
'foobar')
def test_check_new_videos(self, mock_resp):
"""Test recent video response."""
@@ -126,20 +130,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