diff --git a/.gitignore b/.gitignore index ca0291e..561086d 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,4 @@ htmlcov/* dist/* MANIFEST README +.sh diff --git a/README.rst b/README.rst index 85590ee..6ce352d 100644 --- a/README.rst +++ b/README.rst @@ -22,7 +22,6 @@ This library was built with the intention of allowing easy communication with Bl **Usage** ========= In terms of usage, you just need to instantiate the module with a username and password - :: import blinkpy blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD') @@ -117,6 +116,9 @@ Takes an image with the camera and saves it as the new thumbnail. The Blink.ref **BlinkCamera.set_motion_detect(enable=True/False)** Sending True to this function will enable motion detection for the camera. Setting to False will disable motion detection +**BlinkCamera.image_to_file(path)** +This will write the current thumbnail to the location indicated in 'path' + diff --git a/blinkpy.py b/blinkpy.py index 575990c..d77b5fb 100644 --- a/blinkpy.py +++ b/blinkpy.py @@ -17,25 +17,28 @@ import requests import getpass import json -__version__ = '0.2.0' +__version__ = '0.3.0' -LOGIN_URL = 'https://prod.immedia-semi.com/login' -BASE_URL = 'https://prod.immedia-semi.com' -HOST_URL = 'prod.immedia-semi.com' +BLINK_URL = 'immedia-semi.com' +LOGIN_URL = 'https://prod.' + BLINK_URL + '/login' +BASE_URL = 'https://prod.' + BLINK_URL +DEFAULT_URL = 'prod.' + BLINK_URL logger = logging.getLogger('blinkpy') -def _request(url, data=None, headers=None, type='get'): +def _request(url, data=None, headers=None, type='get', stream=False, json=True): """Wrapper function for request""" if type is 'post': response = requests.post(url, headers=headers, data=data).json() - elif type is 'get': - response = requests.get(url, headers=headers).json() + elif type is 'get' and json: + response = requests.get(url, headers=headers, stream=stream).json() + elif type is 'get' and not json: + response = requests.get(url, headers=headers, stream=stream) else: raise ValueError("Cannot perform requests of type " + type) - if 'message' in response.keys(): + if json and 'message' in response.keys(): raise BlinkAuthenticationException(response['code'], response['message']) return response @@ -53,19 +56,20 @@ class BlinkAuthenticationException(BlinkException): class BlinkCamera(object): """Class to initialize individual camera""" - def __init__(self, name, id, status, thumb, clip, temperature, battery, notifications): - self._ID = id - self._NAME = name - self._STATUS = status - self._THUMB = thumb - self._CLIP = clip - self._TEMPERATURE = temperature - self._BATTERY = battery - self._NOTIFICATIONS = notifications - self._MOTION = {} - self._HEADER = None - self._IMAGE_LINK = None - self._ARM_LINK = None + def __init__(self, config): + self._ID = str(config['device_id']) + self._NAME = config['name'] + self._STATUS = config['armed'] + self._THUMB = BASE_URL + config['thumbnail'] + '.jpg' + self._CLIP = BASE_URL + config['thumbnail'] + '.mp4' + self._TEMPERATURE = config['temp'] + self._BATTERY = config['battery'] + self._NOTIFICATIONS = config['notifications'] + self._MOTION = {} + self._HEADER = None + self._IMAGE_LINK = None + self._ARM_LINK = None + self._REGION_ID = config['region_id'] @property def id(self): @@ -79,6 +83,10 @@ class BlinkCamera(object): def name(self, value): self._NAME = value + @property + def region_id(self): + return self._REGION_ID + @property def armed(self): return self._STATUS @@ -93,6 +101,7 @@ class BlinkCamera(object): @property def thumbnail(self): + # RUN THUMB ACQ HERE return self._THUMB @thumbnail.setter @@ -169,32 +178,47 @@ class BlinkCamera(object): _request(url + 'disable', headers=self._HEADER, type='post') def update(self, values): - self._NAME = values['name'] - self._STATUS = values['armed'] - self._THUMB = BASE_URL + values['thumbnail'] + '.jpg' - self._CLIP = BASE_URL + values['thumbnail'] + '.mp4' - self._TEMPERATURE = values['temp'] - self._BATTERY = values['battery'] + self._NAME = values['name'] + self._STATUS = values['armed'] + self._THUMB = BASE_URL + values['thumbnail'] + '.jpg' + self._CLIP = BASE_URL + values['thumbnail'] + '.mp4' + self._TEMPERATURE = values['temp'] + self._BATTERY = values['battery'] self._NOTIFICATIONS = values['notifications'] + def image_to_file(self, path): + response = _request(self._THUMB, headers=self._HEADER, stream=True, json=False) + if response.status_code == 200: + with open(path, 'wb') as f: + for chunk in response: + f.write(chunk) + class Blink(object): """Class to initialize communication and sync module""" def __init__(self, username=None, password=None): """Constructor for class""" - self._username = username - self._password = password - self._TOKEN = None - self._AUTH_HEADER = None - self._NETWORKID = None - self._ACCOUNTID = None - self._EVENTS = [] - self._CAMERAS = [] + self._username = username + self._password = password + self._TOKEN = None + self._AUTH_HEADER = None + self._NETWORKID = None + self._ACCOUNTID = None + self._REGION = None + self._REGION_ID = None + self._HOST = None + self._EVENTS = [] + self._CAMERAS = {} + self._IDLOOKUP = {} @property def cameras(self): return self._CAMERAS + @property + def id_table(self): + return self._IDLOOKUP + @property def network_id(self): return self._NETWORKID @@ -203,6 +227,14 @@ class Blink(object): def account_id(self): return self._ACCOUNTID + @property + def region(self): + return self._REGION + + @property + def region_id(self): + return self._REGION_ID + @property def events(self): """Gets all events on server""" @@ -224,10 +256,12 @@ class Blink(object): recent = self.events for element in recent: try: - camera_id = element['camera_id'] + camera_id = str(element['camera_id']) + camera_name = self._IDLOOKUP[camera_id] + camera = self._CAMERAS[camera_name] if element['type'] == 'motion': url = BASE_URL + element['video_url'] - self._CAMERAS[str(camera_id)].motion = {'video': url, 'image': url[:-3] + 'jpg', 'time': element['created_at']} + camera.motion = {'video': url, 'image': url[:-3] + 'jpg', 'time': element['created_at']} except KeyError: pass @@ -250,7 +284,7 @@ class Blink(object): """Gets all blink cameras and pulls their most recent status""" response = self.get_summary()['devices'] - for camera in self._CAMERAS: + for name, camera in self._CAMERAS.items(): for element in response: try: if element['id'] == camera.id: @@ -260,7 +294,7 @@ class Blink(object): def get_summary(self): """Gets a full summary of device information""" - url = BASE_URL + '/homescreen' + url = BASE_URL + '/homescreen' headers = self._AUTH_HEADER if self._AUTH_HEADER is None: @@ -274,24 +308,17 @@ class Blink(object): for element in response: if 'device_type' in element.keys(): if element['device_type'] == 'camera': - self._CAMERAS.append( - BlinkCamera( - element['name'], - str(element['device_id']), - element['armed'], - BASE_URL + element['thumbnail'] + '.jpg', - BASE_URL + element['thumbnail'] + '.mp4', - element['temp'], - element['battery'], - element['notifications'] - ) - ) + # Add region to config + element['region_id'] = self._REGION_ID + device = BlinkCamera(element) + self._CAMERAS[device.name] = device + self._IDLOOKUP[device.id] = device.name def set_links(self): """Sets access links and required headers for each camera in system""" - for camera in self._CAMERAS: + for name, camera in self._CAMERAS.items(): image_url = BASE_URL + '/network/' + self._NETWORKID + '/camera/' + camera.id + '/thumbnail' - arm_url = BASE_URL + '/network/' + self._NETWORKID + '/camera/' + camera.id + '/' + arm_url = BASE_URL + '/network/' + self._NETWORKID + '/camera/' + camera.id + '/' camera.image_link = image_url camera.arm_link = arm_url camera.header = self._AUTH_HEADER @@ -318,22 +345,25 @@ class Blink(object): if not isinstance(self._password, str): raise BlinkAuthenticationException(0, "Password must be a string") - headers = {'Host': HOST_URL, + headers = {'Host': DEFAULT_URL, 'Content-Type': 'application/json' } - data = json.dumps({"email": self._username, - "password": self._password, - "client_specifier": "iPhone 9.2 | 2.2 | 222" - }) + data = json.dumps({ + "email": self._username, + "password": self._password, + "client_specifier": "iPhone 9.2 | 2.2 | 222" + }) response = _request(LOGIN_URL, headers=headers, data=data, type='post') self._TOKEN = response['authtoken']['authtoken'] - self._AUTH_HEADER = {'Host': HOST_URL, + (self._REGION_ID, self._REGION), = response['region'].items() + self._HOST = self._REGION + '.' + BLINK_URL + self._AUTH_HEADER = {'Host': self._HOST, 'TOKEN_AUTH': self._TOKEN } def get_ids(self): """Sets the network ID and Account ID""" - url = BASE_URL + '/networks' + url = BASE_URL + '/networks' headers = self._AUTH_HEADER if self._AUTH_HEADER is None: diff --git a/tests/test_blink_requests.py b/tests/test_blink_requests.py index bfa9247..806e0b6 100644 --- a/tests/test_blink_requests.py +++ b/tests/test_blink_requests.py @@ -1,82 +1,85 @@ -import blinkpy -import requests -import unittest -from unittest import mock -from blinkpy import LOGIN_URL -from blinkpy import BASE_URL -from blinkpy import HOST_URL - -def mocked_requests_post(*args, **kwargs): - class MockPostResponse: - def __init__(self,json_data,status_code): - self.json_data = json_data - self.status_code = status_code - def json(self): - return self.json_data - - if args[0] == LOGIN_URL: - return MockPostResponse({"authtoken":{"authtoken":"abcd1234"}}, 200) - elif args[0].split("/")[-1] == 'arm': - return MockPostResponse({"armed":True}, 200) - elif args[0].split("/")[-1] == 'disarm': - return MockPostResponse({"armed":False}, 200) - else: - return MockPostResponse({}, 200) - - return MockPostResponse({'message':'ERROR','code':404}, 404) - -def mocked_requests_get(*args, **kwargs): - class MockGetResponse: - def __init__(self,json_data,status_code): - self.json_data = json_data - self.status_code = status_code - def json(self): - return self.json_data - - if args[0] == BASE_URL + '/networks': - return MockGetResponse({'networks':[{"id":7777,"account_id":3333},{"nothing":"nothing"}]}, 200) - else: - return MockGetResponse({'devices':[{'device_type':'camera','name':'test','device_id':123,'armed':False,'thumbnail':'/some/url','temp':65,'battery':3,'notifications':1}, - {'device_type':'camera','name':'test2','device_id':321,'armed':True,'thumbnail':'/some/new/url','temp':56,'battery':5,'notifications':0}, - {'device_type':'None'} - ], - 'events':[{'camera_id':123, 'type':'motion', 'video_url':'/some/dumb/location.mp4'}, - {'camera_id':321, 'type':'None'} - ], - 'syncmodule':{'name':'SyncName', 'status':'online'}, - 'network':{'name':'Sync','armed':True, 'notifications':4} - }, - 200 - ) - - - return MockGetResponse({'message':'ERROR','code':404}, 404) - -class TestBlinkRequests(unittest.TestCase): - @mock.patch('blinkpy.requests.post', side_effect=mocked_requests_post) - @mock.patch('blinkpy.requests.get', side_effect=mocked_requests_get) - def test_blink_setup(self, mock_get, mock_post): - blink = blinkpy.Blink(username='user',password='password') - blink.setup_system() - - self.assertEqual(blink.network_id, '7777') - self.assertEqual(blink.account_id, '3333') - self.assertEqual(blink.online, True) - self.assertEqual(blink.arm, True) - - @mock.patch('blinkpy.requests.post', side_effect=mocked_requests_post) - @mock.patch('blinkpy.requests.get', side_effect=mocked_requests_get) - def test_blink_camera_setup_and_motion(self, mock_get, mock_post): - blink = blinkpy.Blink(username='user',password='password') - blink.setup_system() - blink.last_motion() - for camera in blink.cameras: - if camera.id == 123: - assert camera.name is 'test' - assert camera.armed is False - assert camera.motion['video'] is BASE_URL+'/some/dumb/location.mp4' - elif camera.id == 321: - assert camera.name is 'test2' - assert camera.armed is True - assert len(camera.motion.keys()) is 0 +import blinkpy +import requests +import unittest +from unittest import mock +from blinkpy import LOGIN_URL +from blinkpy import BASE_URL + +def mocked_requests_post(*args, **kwargs): + class MockPostResponse: + def __init__(self,json_data,status_code): + self.json_data = json_data + self.status_code = status_code + def json(self): + return self.json_data + + if args[0] == LOGIN_URL: + return MockPostResponse({"region":{"test": "Notacountry"}, "authtoken":{"authtoken":"abcd1234"}}, 200) + elif args[0].split("/")[-1] == 'arm': + return MockPostResponse({"armed":True}, 200) + elif args[0].split("/")[-1] == 'disarm': + return MockPostResponse({"armed":False}, 200) + else: + return MockPostResponse({}, 200) + + return MockPostResponse({'message':'ERROR','code':404}, 404) + +def mocked_requests_get(*args, **kwargs): + class MockGetResponse: + def __init__(self,json_data,status_code): + self.json_data = json_data + self.status_code = status_code + def json(self): + return self.json_data + + if args[0] == BASE_URL + '/networks': + return MockGetResponse({'networks':[{"id":7777,"account_id":3333},{"nothing":"nothing"}]}, 200) + else: + return MockGetResponse({'devices':[{'device_type':'camera','name':'test','device_id':123,'armed':False,'thumbnail':'/some/url','temp':65,'battery':3,'notifications':1}, + {'device_type':'camera','name':'test2','device_id':321,'armed':True,'thumbnail':'/some/new/url','temp':56,'battery':5,'notifications':0}, + {'device_type':'None'} + ], + 'events':[{'camera_id':123, 'type':'motion', 'video_url':'/some/dumb/location.mp4', 'created_at':'2017-01-01'}, + {'camera_id':321, 'type':'None'} + ], + 'syncmodule':{'name':'SyncName', 'status':'online'}, + 'network':{'name':'Sync','armed':True, 'notifications':4} + }, + 200 + ) + + + return MockGetResponse({'message':'ERROR','code':404}, 404) + +class TestBlinkRequests(unittest.TestCase): + @mock.patch('blinkpy.requests.post', side_effect=mocked_requests_post) + @mock.patch('blinkpy.requests.get', side_effect=mocked_requests_get) + def test_blink_setup(self, mock_get, mock_post): + blink = blinkpy.Blink(username='user',password='password') + blink.setup_system() + + self.assertEqual(blink.network_id, '7777') + self.assertEqual(blink.account_id, '3333') + self.assertEqual(blink.region, 'Notacountry') + self.assertEqual(blink.region_id, 'test') + self.assertEqual(blink.online, True) + self.assertEqual(blink.arm, True) + + @mock.patch('blinkpy.requests.post', side_effect=mocked_requests_post) + @mock.patch('blinkpy.requests.get', side_effect=mocked_requests_get) + def test_blink_camera_setup_and_motion(self, mock_get, mock_post): + blink = blinkpy.Blink(username='user',password='password') + blink.setup_system() + blink.last_motion() + for name, camera in blink.cameras.items(): + if camera.id == '123': + self.assertEqual(name, 'test') + self.assertEqual(camera.armed, False) + self.assertEqual(camera.motion['video'], BASE_URL+'/some/dumb/location.mp4') + elif camera.id == '321': + self.assertEqual(name, 'test2') + self.assertEqual(camera.armed, True) + self.assertEqual(len(camera.motion.keys()), 0) + else: + assert False is True \ No newline at end of file diff --git a/tests/test_classes.py b/tests/test_classes.py index deec86c..6a3b50d 100644 --- a/tests/test_classes.py +++ b/tests/test_classes.py @@ -1,86 +1,75 @@ -import testtools -import blinkpy -from blinkpy import BASE_URL - -class TestBlink(testtools.TestCase): - def test_blink_instance(self): - blink = blinkpy.Blink() - - def test_blink_camera_attributes(self): - test_name = 'name' - test_id = 1248 - test_status = True - test_url = 'https://www.abcdef.com/abc/123/abc456/_-()98~/X.jpg' - test_temp = 70 - test_battery = 3 - test_notif = 0 - camera = blinkpy.BlinkCamera(test_name, test_id, test_status, test_url, test_url, test_temp, test_battery, test_notif) - - # Check that values are properly stored and can be recalled - assert camera.name is test_name - assert camera.id is test_id - assert camera.armed is test_status - assert camera.clip is test_url - assert camera.thumbnail is test_url - assert camera.temperature is test_temp - assert camera.battery is test_battery - assert camera.notifications is test_notif - - # Check that values can be changed with individual methods - test_name = test_name +'new' - test_url = test_url + '/NEWAPPEND' - test_temp = test_temp - 10 - test_battery = test_battery - 1 - test_notif = test_notif + 3 - test_header = {'Header':'test','Test':'1234'} - test_img = 'http://image-link.com' - test_arm = 'http://arm-link.com' - test_motion = {'video':'url','image':'url','time':'timestamp'} - - camera.name = test_name - camera.clip = test_url - camera.thumbnail = test_url - camera.temperature = test_temp - camera.battery = test_battery - camera.notifications = test_notif - camera.image_link = test_img - camera.arm_link = test_arm - camera.header = test_header - camera.motion = test_motion - - # Check that values are properly stored and can be recalled - assert camera.name is test_name - assert camera.id is test_id - assert camera.armed is test_status - assert camera.clip is test_url - assert camera.thumbnail is test_url - assert camera.temperature is test_temp - assert camera.battery is test_battery - assert camera.notifications is test_notif - assert camera.header is test_header - assert camera.image_link is test_img - assert camera.arm_link is test_arm - assert camera.motion is test_motion - - # Verify bulk update function - test_name = test_name +' last' - test_status = True - test_url = '-this-is-a-test/for_realz' - test_temp = test_temp + 7 - test_battery = test_battery - 1 - test_notif = test_notif - 1 - values = {'name':test_name, 'armed':test_status, 'thumbnail':test_url, 'temp':test_temp, 'battery':test_battery, 'notifications':test_notif} - - camera.update(values) - # Check that values are properly stored and can be recalled - jpg_url = BASE_URL + test_url +'.jpg' - mp4_url = BASE_URL + test_url +'.mp4' - assert camera.name is test_name - assert camera.id is test_id - assert camera.armed is test_status - assert camera.clip == mp4_url - assert camera.thumbnail == jpg_url - assert camera.temperature is test_temp - assert camera.battery is test_battery - assert camera.notifications is test_notif +import testtools +import unittest +import blinkpy +from blinkpy import BASE_URL + +class TestBlink(unittest.TestCase): + def test_blink_instance(self): + blink = blinkpy.Blink() + + def test_blink_camera_attributes(self): + config = {'name':'name', 'device_id': 1248, 'armed':True, 'thumbnail':'/test/url/image', 'temp':70, 'battery':3, 'notifications':0, 'region_id':'rid'} + + camera = blinkpy.BlinkCamera(config) + + # Check that values are properly stored and can be recalled + self.assertEqual(camera.name, config['name']) + self.assertEqual(camera.id, str(config['device_id'])) + self.assertEqual(camera.armed, config['armed']) + self.assertEqual(camera.clip, BASE_URL + config['thumbnail'] + '.mp4') + self.assertEqual(camera.thumbnail, BASE_URL + config['thumbnail'] + '.jpg') + self.assertEqual(camera.temperature, config['temp']) + self.assertEqual(camera.battery, config['battery']) + self.assertEqual(camera.notifications, config['notifications']) + self.assertEqual(camera.region_id, config['region_id']) + + # Check that values can be changed with individual methods + test_header = {'Header':'test','Test':'1234'} + test_img = 'http://image-link.com' + test_arm = 'http://arm-link.com' + test_motion = {'video':'url','image':'url','time':'timestamp'} + + camera.name = config['name']+'_new' + camera.clip = BASE_URL + config['thumbnail'] + '.mp4_new' + camera.thumbnail = BASE_URL + config['thumbnail'] + '.jpg_new' + camera.temperature = config['temp'] + 10 + camera.battery = config['battery'] + 1 + camera.notifications = config['notifications'] + 1 + camera.image_link = test_img + camera.arm_link = test_arm + camera.header = test_header + camera.motion = test_motion + + # Check that values are properly stored and can be recalled + self.assertEqual(camera.name , config['name']+'_new') + self.assertEqual(camera.clip , BASE_URL + config['thumbnail'] + '.mp4_new') + self.assertEqual(camera.thumbnail, BASE_URL + config['thumbnail'] + '.jpg_new') + self.assertEqual(camera.temperature, config['temp'] + 10) + self.assertEqual(camera.battery, config['battery'] + 1) + self.assertEqual(camera.notifications, config['notifications'] + 1) + self.assertEqual(camera.image_link, test_img) + self.assertEqual(camera.arm_link, test_arm) + self.assertEqual(camera.header, test_header) + self.assertEqual(camera.motion, test_motion) + + # Verify bulk update function + test_name = camera.name +' last' + test_status = True + test_url = '-this-is-a-test/for_realz' + test_temp = camera.temperature + 7 + test_battery = camera.battery - 1 + test_notif = camera.notifications - 1 + values = {'name':test_name, 'armed':test_status, 'thumbnail':test_url, 'temp':test_temp, 'battery':test_battery, 'notifications':test_notif} + + camera.update(values) + # Check that values are properly stored and can be recalled + jpg_url = BASE_URL + test_url +'.jpg' + mp4_url = BASE_URL + test_url +'.mp4' + self.assertEqual(camera.name, test_name) + self.assertEqual(camera.armed, test_status) + self.assertEqual(camera.clip, mp4_url) + self.assertEqual(camera.thumbnail, jpg_url) + self.assertEqual(camera.temperature, test_temp) + self.assertEqual(camera.battery, test_battery) + self.assertEqual(camera.notifications, test_notif) \ No newline at end of file diff --git a/tox.ini b/tox.ini index 5ade3d4..660f208 100644 --- a/tox.ini +++ b/tox.ini @@ -1,23 +1,23 @@ -[tox] -envlist = py34, pep8 - -[testenv] -setenv = - LAND=en_US.UTF-8 - PYTHONPATH = {toxinidir} -commands = - py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report html {posargs} -deps = - -r{toxinidir}/requirements.txt - -[testenv:pep8] -deps = - -r{toxinidir}/requirements.txt - flake8 -basepython = python3 -commands = - flake8 blinkpy.py - -[flake8] -ignore = E501, E221 +[tox] +envlist = py34, pep8 + +[testenv] +setenv = + LAND=en_US.UTF-8 + PYTHONPATH = {toxinidir} +commands = + py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report html {posargs} +deps = + -r{toxinidir}/requirements.txt + +[testenv:pep8] +deps = + -r{toxinidir}/requirements.txt + flake8 +basepython = python3 +commands = + flake8 blinkpy.py + +[flake8] +ignore = E501 exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build \ No newline at end of file