diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ca0291e --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.cache/* +.tox/* +__pycache__/* +htmlcov/* +.coverage +*.pyc +*.egg*/* +dist/* +MANIFEST +README diff --git a/README.md b/README.md deleted file mode 100644 index 23dd94f..0000000 --- a/README.md +++ /dev/null @@ -1,2 +0,0 @@ -# blinkpy -A Python library for the Blink Camera system diff --git a/README.rst b/README.rst new file mode 100644 index 0000000..dd06ca7 --- /dev/null +++ b/README.rst @@ -0,0 +1,126 @@ +**blinkpy** +============ +A Python library for the Blink Camera system + +**Disclaimers** +=============== +Published under the MIT license - See LICENSE file for more details. + +"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark owned by Immedia Inc., see www.blinkforhome.com for more information. +I am in no way affiliated with Blink, nor Immedia Inc. + +Original protocol hacking by MattTW : https://github.com/MattTW/BlinkMonitorProtocol + +**Installation** +================ +``pip3 install blinkpy`` + +**Purpose** +=========== +This library was built with the intention of allowing easy communication with Blink camera systems, specifically so I can add a module into homeassistant https://home-assistant.io + +**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') +`` + +If you leave out either of those parameters, you need to call the login function which will prompt for your username and password +`` +blink.login() +`` + +Once the login information is entered, you can run the `setup_system()` function which will attempt to authenticate with Blink servers using your username and password, obtain network ids, and create a list of cameras. +The cameras are of a BlinkCamera class, of which the following parameters can be used (the code below creates a Blink object and iterates through each camera found) +`` +blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD') +blink.setup_system() + +for camera in blink.cameras: + print(camera.name) # Name of the camera + print(camera.id) # Integer id of the camera (assigned by Blink) + print(camera.armed) # Whether the device is armed/disarmed (ie. detecting motion) + print(camera.clip) # Link to last motion clip captured + print(camera.thumbnail) # Link to current camera thumbnail + print(camera.temperature) # Current camera temperature (not super accurate, but might be useful for someone) + print(camera.battery) # Current battery level... I think the value ranges from 0-3, but not quite sure yet. + print(camera.notifications) # Number of unread notifications (ie. motion alerts that haven't been viewed) + print(camera.motion) # Dictionary containing values for keys ['video', 'image', 'time'] + # which correspond to last motion recorded, thumbnail of last motion, and timestamp of last motion +`` + +**class Blink** +--------------- +The following properties/methods are availiable to the Blink class + +**Blink.cameras** +Returns a list of BlinkCamera objects found by the system + +**Blink.network_id** +Returns the current network id + +**Blink.account_id** +Returns the account id + +**Blink.events** +Returns a list of events recorded by blink. This information will contain links to any motion caught by an armed camera. + +**Blink.online** +Returns online status of sync module (True = online, False = offline) + +**Blink.last_motion()** +Finds last motion information for each camera and stores it in the BlinkCamera.motion field + +**Blink.arm** +Set to True to arm, False to disarm. Can be used to see the status of the system as well + +**Blink.refresh()** +Forces a refresh of all camera information + +**Blink.get_summary()** +Returns json formatted summary of the system + +**Blink.get_cameras()** +Finds all cameras in the system and creates them + +**Blink.set_links()** +Gives each BlinkCamera object the links needed to find recent images and videos + +**Blink.login()** +Prompts user for login information + +**Blink.get_auth_token()** +Uses login information to retrieve authorization token from Blink for further communication + +**Blink.get_ids()** +Retrieves the network_id and account_id from Blink in order to access video and image pages on their server + +**Blink.setup_system()** +A wrapper script that calls: +`` +self.get_auth_token() +self.get_ids() +self.get_camers() +self.set_links() +`` + +**class BlinkCamera** +--------------------- +This class is just a wrapper for each individual camera in order to make communication with individual cameras less clunky. The following properties/methods are availiable (in addition to the ones mentioned earlier) + +**BlinkCamera.snap_picture()** +Takes an image with the camera and saves it as the new thumbnail. The Blink.refresh() method should be called after this if you want to store the new thumbnail link + +**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 + + + + + + + + diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..550c8cd --- /dev/null +++ b/__init__.py @@ -0,0 +1 @@ +from blinkpy import Blink diff --git a/blinkpy.py b/blinkpy.py new file mode 100644 index 0000000..575990c --- /dev/null +++ b/blinkpy.py @@ -0,0 +1,344 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +''' +blinkpy by Kevin Fronczak - A Blink camera Python library +https://github.com/fronzbot/blinkpy +Original protocol hacking by MattTW : https://github.com/MattTW/BlinkMonitorProtocol + +Published under the MIT license - See LICENSE file for more details. + +"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark owned by Immedia Inc., see www.blinkforhome.com for more information. +I am in no way affiliated with Blink, nor Immedia Inc. +''' + +import logging +import requests +import getpass +import json + +__version__ = '0.2.0' + +LOGIN_URL = 'https://prod.immedia-semi.com/login' +BASE_URL = 'https://prod.immedia-semi.com' +HOST_URL = 'prod.immedia-semi.com' + +logger = logging.getLogger('blinkpy') + + +def _request(url, data=None, headers=None, type='get'): + """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() + else: + raise ValueError("Cannot perform requests of type " + type) + + if 'message' in response.keys(): + raise BlinkAuthenticationException(response['code'], response['message']) + + return response + + +class BlinkException(Exception): + def __init__(self, id, message): + self.id = id + self.message = message + + +class BlinkAuthenticationException(BlinkException): + pass + + +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 + + @property + def id(self): + return self._ID + + @property + def name(self): + return self._NAME + + @name.setter + def name(self, value): + self._NAME = value + + @property + def armed(self): + return self._STATUS + + @property + def clip(self): + return self._CLIP + + @clip.setter + def clip(self, value): + self._CLIP = value + + @property + def thumbnail(self): + return self._THUMB + + @thumbnail.setter + def thumbnail(self, value): + self._THUMB = value + + @property + def temperature(self): + return self._TEMPERATURE + + @temperature.setter + def temperature(self, value): + self._TEMPERATURE = value + + @property + def battery(self): + return self._BATTERY + + @battery.setter + def battery(self, value): + self._BATTERY = value + + @property + def notifications(self): + return self._NOTIFICATIONS + + @notifications.setter + def notifications(self, value): + self._NOTIFICATIONS = value + + @property + def image_link(self): + return self._IMAGE_LINK + + @image_link.setter + def image_link(self, value): + self._IMAGE_LINK = value + + @property + def arm_link(self): + return self._ARM_LINK + + @arm_link.setter + def arm_link(self, value): + self._ARM_LINK = value + + @property + def header(self): + return self._HEADER + + @header.setter + def header(self, value): + self._HEADER = value + + @property + def motion(self): + return self._MOTION + + @motion.setter + def motion(self, value): + """Sets link to last motion and timestamp""" + self._MOTION = value + + def snap_picture(self): + """Takes a picture with camera to create a new thumbnail""" + _request(self._IMAGE_LINK, headers=self._HEADER, type='post') + + def set_motion_detect(self, enable): + """Sets motion detection""" + url = self._ARM_LINK + if enable: + _request(url + 'enable', headers=self._HEADER, type='post') + else: + _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._NOTIFICATIONS = values['notifications'] + + +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 = [] + + @property + def cameras(self): + return self._CAMERAS + + @property + def network_id(self): + return self._NETWORKID + + @property + def account_id(self): + return self._ACCOUNTID + + @property + def events(self): + """Gets all events on server""" + url = BASE_URL + '/events/network/' + self._NETWORKID + headers = self._AUTH_HEADER + self._EVENTS = _request(url, headers=headers, type='get')['events'] + return self._EVENTS + + @property + def online(self): + """Returns True or False depending on if sync module is online/offline""" + url = BASE_URL + 'network/' + self._NETWORKID + '/syncmodules' + headers = self._AUTH_HEADER + online_dict = {'online': True, 'offline': False} + return online_dict[_request(url, headers=headers, type='get')['syncmodule']['status']] + + def last_motion(self): + """Finds last motion of each camera""" + recent = self.events + for element in recent: + try: + camera_id = element['camera_id'] + 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']} + except KeyError: + pass + + @property + def arm(self): + """Returns status of sync module: armed/disarmed""" + return self.get_summary()['network']['armed'] + + @arm.setter + def arm(self, value): + """Arms or disarms system. Arms/disarms all if camera not named""" + if value: + value_to_append = 'arm' + else: + value_to_append = 'disarm' + url = BASE_URL + '/network/' + self._NETWORKID + '/' + value_to_append + _request(url, headers=self._AUTH_HEADER, type='post') + + def refresh(self): + """Gets all blink cameras and pulls their most recent status""" + response = self.get_summary()['devices'] + + for camera in self._CAMERAS: + for element in response: + try: + if element['id'] == camera.id: + camera.update(element) + except KeyError: + pass + + def get_summary(self): + """Gets a full summary of device information""" + url = BASE_URL + '/homescreen' + headers = self._AUTH_HEADER + + if self._AUTH_HEADER is None: + raise BlinkException(0, "Authentication header incorrect. Are you sure you logged in and received your token?") + + return _request(url, headers=headers, type='get') + + def get_cameras(self): + """Finds and creates cameras""" + response = self.get_summary()['devices'] + 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'] + ) + ) + + def set_links(self): + """Sets access links and required headers for each camera in system""" + for camera in self._CAMERAS: + image_url = BASE_URL + '/network/' + self._NETWORKID + '/camera/' + camera.id + '/thumbnail' + 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 + + def setup_system(self): + """Method logs in and sets auth token and network ids for future requests""" + if self._username is None or self._password is None: + raise BlinkAuthenticationException(3, "Cannot authenticate since either password or username has not been set") + + self.get_auth_token() + self.get_ids() + self.get_cameras() + self.set_links() + + def login(self): + """Prompts user for username and password""" + self._username = input("Username:") + self._password = getpass.getpass("Password:") + + def get_auth_token(self): + """Retrieves the authentication token from Blink""" + if not isinstance(self._username, str): + raise BlinkAuthenticationException(0, "Username must be a string") + if not isinstance(self._password, str): + raise BlinkAuthenticationException(0, "Password must be a string") + + headers = {'Host': HOST_URL, + 'Content-Type': 'application/json' + } + 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, + 'TOKEN_AUTH': self._TOKEN + } + + def get_ids(self): + """Sets the network ID and Account ID""" + url = BASE_URL + '/networks' + headers = self._AUTH_HEADER + + if self._AUTH_HEADER is None: + raise BlinkException(0, "Authentication header incorrect. Are you sure you logged in and received your token?") + + response = _request(url, headers=headers, type='get') + self._NETWORKID = str(response['networks'][0]['id']) + self._ACCOUNTID = str(response['networks'][0]['account_id']) diff --git a/examples/arm_system_camera.py b/examples/arm_system_camera.py new file mode 100644 index 0000000..7d0c240 --- /dev/null +++ b/examples/arm_system_camera.py @@ -0,0 +1,22 @@ +#!/usr/bin/python +import blinkpy +import time + +blink = blinkpy.Blink() +blink.login() +blink.setup_system() + +for camera in blink.cameras: + print('Arming ' + camera.name) + camera.set_motion_detect(True) + time.sleep(5) + blink.refresh() + +print('Arming Blink') +blink.arm = True +time.sleep(5) +print('Blink armed? ' + str(blink.arm)) +print('Disarming Blink') +time.sleep(5) +blink.arm = False +print('Blink armed? ' + str(blink.arm)) diff --git a/examples/list_all_cameras.py b/examples/list_all_cameras.py new file mode 100644 index 0000000..c5bf179 --- /dev/null +++ b/examples/list_all_cameras.py @@ -0,0 +1,18 @@ +#!/usr/bin/python +import blinkpy + +blink = blinkpy.Blink() +blink.login() +blink.setup_system() + +for camera in blink.cameras: + print(camera.name) # Name of the camera + print(camera.id) # Integer id of the camera (assigned by Blink) + print(camera.armed) # Whether the device is armed/disarmed (ie. detecting motion) + print(camera.clip) # Link to last motion clip captured + print(camera.thumbnail) # Link to current camera thumbnail + print(camera.temperature) # Current camera temperature (not super accurate, but might be useful for someone) + print(camera.battery) # Current battery level... I think the value ranges from 0-3, but not quite sure yet. + print(camera.notifications) # Number of unread notifications (ie. motion alerts that haven't been viewed) + print(camera.motion) # Dictionary containing values for keys ['video', 'image', 'time'] + # which correspond to last motion recorded, thumbnail of last motion, and timestamp of last motion \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ac3953b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +pytest>=2.9.2 +pytest-cov>=2.3.1 +pytest-timeout>=1.0.0 +requests>=2.12.4 +testtools \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..f62073e --- /dev/null +++ b/setup.py @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- + +from blinkpy import __version__ +from setuptools import setup + +setup( + name = 'blinkpy', + version = __version__, + description = 'A Blink camera Python library', + long_description='A library that communicates with Blink cameras', + author = 'Kevin Fronczak', + author_email = "kfronczak@gmail.com", + license='MIT', + url = 'https://github.com/fronzbot/blinkpy', + py_modules=['blinkpy'], + install_requires=['requests'], + classifiers=[ + 'Development Status :: 4 - Beta', + 'Intended Audience :: Developers', + 'License :: OSI Approved :: MIT License', + 'Programming Language :: Python :: 3.4', + 'Programming Language :: Python :: 3.5', + 'Programming Language :: Python :: 3.6', + 'Environment :: Plugins', + 'Environment :: Web Environment' + ] +) \ No newline at end of file diff --git a/tests/test_blink_requests.py b/tests/test_blink_requests.py new file mode 100644 index 0000000..bfa9247 --- /dev/null +++ b/tests/test_blink_requests.py @@ -0,0 +1,82 @@ +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 + \ No newline at end of file diff --git a/tests/test_classes.py b/tests/test_classes.py new file mode 100644 index 0000000..deec86c --- /dev/null +++ b/tests/test_classes.py @@ -0,0 +1,86 @@ +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 + \ No newline at end of file diff --git a/tests/test_import.py b/tests/test_import.py new file mode 100644 index 0000000..41efe52 --- /dev/null +++ b/tests/test_import.py @@ -0,0 +1,6 @@ +import testtools +import blinkpy + +class TestImport(testtools.TestCase): + def test_import(self): + pass \ No newline at end of file diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..5ade3d4 --- /dev/null +++ b/tox.ini @@ -0,0 +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 +exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build \ No newline at end of file