From 7b07146afa24c91e9476c77e8a3ab87c30d38d49 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Wed, 8 Mar 2017 23:11:56 -0500 Subject: [PATCH] Constants and better style testing --- blinkpy.py | 355 +++++++++++++++++++++++++----------------- constants.py | 34 ++++ errors.py | 2 + pylintrc | 38 +++++ setup.py | 38 +++-- tests/test_servers.py | 8 + tox.ini | 18 ++- 7 files changed, 322 insertions(+), 171 deletions(-) create mode 100644 pylintrc create mode 100644 tests/test_servers.py diff --git a/blinkpy.py b/blinkpy.py index f5b33b7..7f58a4c 100644 --- a/blinkpy.py +++ b/blinkpy.py @@ -4,203 +4,242 @@ ''' blinkpy by Kevin Fronczak - A Blink camera Python library https://github.com/fronzbot/blinkpy -Original protocol hacking by MattTW : https://github.com/MattTW/BlinkMonitorProtocol +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. +"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 - -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') +import getpass +import requests +import errors as ERROR +from constants import (BLINK_URL, LOGIN_URL, + BASE_URL, DEFAULT_URL, + HOME_URL, EVENT_URL, + NETWORK_URL, NETWORKS_URL, + ONLINE) -def _request(url, data=None, headers=None, type='get', stream=False, json=True): +def _request(url, data=None, headers=None, reqtype='get', + stream=False, json_resp=True): """Wrapper function for request""" - if type is 'post': - response = requests.post(url, headers=headers, data=data).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) + if reqtype is 'post': + response = requests.post(url, headers=headers, + data=data).json() + elif reqtype is 'get' and json_resp: + response = requests.get(url, headers=headers, + stream=stream).json() + elif reqtype is 'get' and not json_resp: + response = requests.get(url, headers=headers, + stream=stream) else: - raise ValueError("Cannot perform requests of type " + type) + raise BlinkException(ERROR.REQUEST) if json and 'message' in response.keys(): - raise BlinkAuthenticationException(response['code'], response['message']) + raise BlinkAuthenticationException( + (response['code'], response['message'])) return response +# pylint: disable=super-init-not-called class BlinkException(Exception): - def __init__(self, id, message): - self.id = id - self.message = message + """ + Class to throw general blink exception. + """ + def __init__(self, errcode): + self.errid = errcode[0] + self.message = errcode[1] class BlinkAuthenticationException(BlinkException): + """ + Class to throw authentication exception. + """ pass class BlinkCamera(object): - """Class to initialize individual camera""" + """Class to initialize individual camera.""" 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'] + 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 + # pylint: disable=invalid-name def id(self): - return self._ID + """Returns camera id""" + return self._id @property def name(self): - return self._NAME + """Returns camera name""" + return self._name @name.setter def name(self, value): - self._NAME = value + """Sets camera name""" + self._name = value @property def region_id(self): - return self._REGION_ID + """Returns region id""" + return self._region_id @property def armed(self): - return self._STATUS + """Returns camera arm status""" + return self._status @property def clip(self): - return self._CLIP + """Returns current clip""" + return self._clip @clip.setter def clip(self, value): - self._CLIP = value + """Sets current clip""" + self._clip = value @property def thumbnail(self): - # RUN THUMB ACQ HERE - return self._THUMB + """Returns current thumbnail""" + return self._thumb @thumbnail.setter def thumbnail(self, value): - self._THUMB = value + """Sets current thumbnail""" + self._thumb = value @property def temperature(self): - return self._TEMPERATURE + """Returns camera temperature""" + return self._temperature @temperature.setter def temperature(self, value): - self._TEMPERATURE = value + """Sets camera temperature""" + self._temperature = value @property def battery(self): - return self._BATTERY + """Returns battery level""" + return self._battery @battery.setter def battery(self, value): - self._BATTERY = value + """Sets battery level""" + self._battery = value @property def notifications(self): - return self._NOTIFICATIONS + """Returns number of notifications""" + return self._notifications @notifications.setter def notifications(self, value): - self._NOTIFICATIONS = value + """Sets number of notifications""" + self._notifications = value @property def image_link(self): - return self._IMAGE_LINK + """Returns image link""" + return self._image_link @image_link.setter def image_link(self, value): - self._IMAGE_LINK = value + """Sets image link""" + self._image_link = value @property def arm_link(self): - return self._ARM_LINK + """Returns link to arm camera""" + return self._arm_link @arm_link.setter def arm_link(self, value): - self._ARM_LINK = value + """Sets link to arm camera""" + self._arm_link = value @property def header(self): - return self._HEADER + """Returns request header""" + return self._header @header.setter def header(self, value): - self._HEADER = value + """Sets request header""" + self._header = value @property def motion(self): - return self._MOTION + """Returns last motion event detail""" + return self._motion @motion.setter def motion(self, value): """Sets link to last motion and timestamp""" - self._MOTION = value + 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') + _request(self._image_link, headers=self._header, reqtype='post') def set_motion_detect(self, enable): """Sets motion detection""" - url = self._ARM_LINK + url = self._arm_link if enable: - _request(url + 'enable', headers=self._HEADER, type='post') + _request(url + 'enable', headers=self._header, reqtype='post') else: - _request(url + 'disable', headers=self._HEADER, type='post') + _request(url + 'disable', headers=self._header, reqtype='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'] + """Updates camera information""" + 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_refresh(self): - url = BASE_URL + '/homescreen' - response = _request(url, headers=self._HEADER, type='get')['devices'] + """Refreshs current thumbnail""" + url = HOME_URL + response = _request(url, headers=self._header, + reqtype='get')['devices'] for element in response: try: - if str(element['device_id']) == self._ID: - self._THUMB = BASE_URL + element['thumbnail'] + '.jpg' - return self._THUMB + if str(element['device_id']) == self._id: + self._thumb = BASE_URL + element['thumbnail'] + '.jpg' + return self._thumb except KeyError: pass return None def image_to_file(self, path): + """Writes image to file""" thumb = self.image_refresh() - response = _request(thumb, headers=self._HEADER, stream=True, json=False) + response = _request(thumb, headers=self._header, + stream=True, json_resp=False) if response.status_code == 200: - with open(path, 'wb') as f: + with open(path, 'wb') as imgfile: for chunk in response: - f.write(chunk) + imgfile.write(chunk) class Blink(object): @@ -209,65 +248,76 @@ class Blink(object): """Constructor for class""" 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 = {} + self._token = None + self._auth_header = None + self._network_id = None + self._account_id = None + self._region = None + self._region_id = None + self._host = None + self._events = [] + self._cameras = {} + self._idlookup = {} @property def cameras(self): - return self._CAMERAS + """Returns camera/id pairs""" + return self._cameras @property def camera_thumbs(self): + """Returns camera thumbnails""" self.refresh() data = {} - for name, camera in self._CAMERAS.items(): + for name, camera in self._cameras.items(): data[name] = camera.thumbnail return data @property def id_table(self): - return self._IDLOOKUP + """Returns id/camera pairs""" + return self._idlookup @property def network_id(self): - return self._NETWORKID + """Returns network id""" + return self._network_id @property def account_id(self): - return self._ACCOUNTID + """Returns account id""" + return self._account_id @property def region(self): - return self._REGION + """Returns current region""" + return self._region @property def region_id(self): - return self._REGION_ID + """Returns region id""" + return self._region_id @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')['event'] - return self._EVENTS + url = EVENT_URL + self._network_id + headers = self._auth_header + self._events = _request(url, headers=headers, + reqtype='get')['event'] + 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']] + """ + Returns True or False depending on if + sync module is online/offline + """ + url = NETWORK_URL + self._network_id + '/syncmodules' + headers = self._auth_header + return ONLINE[_request(url, headers=headers, + reqtype='get')['syncmodule']['status']] def last_motion(self): """Finds last motion of each camera""" @@ -275,11 +325,13 @@ class Blink(object): for element in recent: try: camera_id = str(element['camera_id']) - camera_name = self._IDLOOKUP[camera_id] - camera = self._CAMERAS[camera_name] + camera_name = self._idlookup[camera_id] + camera = self._cameras[camera_name] if element['type'] == 'motion': url = BASE_URL + element['video_url'] - camera.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 @@ -290,19 +342,23 @@ class Blink(object): @arm.setter def arm(self, value): - """Arms or disarms system. Arms/disarms all if camera not named""" + """ + 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') + url = NETWORK_URL + self._network_id + '/' + value_to_append + _request(url, headers=self._auth_header, reqtype='post') def refresh(self): """Gets all blink cameras and pulls their most recent status""" response = self.get_summary()['devices'] - for name, camera in self._CAMERAS.items(): + for name in self._cameras: + camera = self._cameras[name] for element in response: try: if str(element['device_id']) == camera.id: @@ -313,12 +369,12 @@ class Blink(object): def get_summary(self): """Gets a full summary of device information""" url = BASE_URL + '/homescreen' - headers = self._AUTH_HEADER + 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?") + if self._auth_header is None: + raise BlinkException(ERROR.AUTH_TOKEN) - return _request(url, headers=headers, type='get') + return _request(url, headers=headers, reqtype='get') def get_cameras(self): """Finds and creates cameras""" @@ -327,24 +383,32 @@ class Blink(object): if 'device_type' in element.keys(): if element['device_type'] == 'camera': # Add region to config - element['region_id'] = self._REGION_ID + element['region_id'] = self._region_id device = BlinkCamera(element) - self._CAMERAS[device.name] = device - self._IDLOOKUP[device.id] = device.name + 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 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 + '/' + """ + Sets access links and required headers + for each camera in system + """ + for name in self._cameras: + camera = self._cameras[name] + network_id_url = NETWORK_URL + self._network_id + image_url = network_id_url + '/camera/' + camera.id + '/thumbnail' + arm_url = network_id_url + '/camera/' + camera.id + '/' camera.image_link = image_url camera.arm_link = arm_url - camera.header = self._AUTH_HEADER + camera.header = self._auth_header def setup_system(self): - """Method logs in and sets auth token and network ids for future requests""" + """ + 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") + raise BlinkAuthenticationException(ERROR.AUTHENTICATE) self.get_auth_token() self.get_ids() @@ -359,34 +423,33 @@ class Blink(object): 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") + raise BlinkAuthenticationException(ERROR.USERNAME) if not isinstance(self._password, str): - raise BlinkAuthenticationException(0, "Password must be a string") + raise BlinkAuthenticationException(ERROR.PASSWORD) headers = {'Host': DEFAULT_URL, - 'Content-Type': 'application/json' - } + '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._REGION_ID, self._REGION), = response['region'].items() - self._HOST = self._REGION_ID + '.' + BLINK_URL - self._AUTH_HEADER = {'Host': self._HOST, - 'TOKEN_AUTH': self._TOKEN - } + response = _request(LOGIN_URL, headers=headers, + data=data, reqtype='post') + self._token = response['authtoken']['authtoken'] + (self._region_id, self._region), = response['region'].items() + self._host = self._region_id + '.' + 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' - headers = self._AUTH_HEADER + url = NETWORKS_URL + 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?") + if self._auth_header is None: + raise BlinkException(ERROR.AUTH_TOKEN) - response = _request(url, headers=headers, type='get') - self._NETWORKID = str(response['networks'][0]['id']) - self._ACCOUNTID = str(response['networks'][0]['account_id']) \ No newline at end of file + response = _request(url, headers=headers, reqtype='get') + self._network_id = str(response['networks'][0]['id']) + self._account_id = str(response['networks'][0]['account_id']) diff --git a/constants.py b/constants.py index 64e9daa..f09e4c8 100644 --- a/constants.py +++ b/constants.py @@ -2,6 +2,39 @@ constants.py Generates constants for use in blinkpy ''' +MAJOR_VERSION = 0 +MINOR_VERSION = 4 +PATCH_VERSION = 4 +__version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION) + +REQUIRED_PYTHON_VER = (3, 4, 2) + +PROJECT_NAME = 'blinkpy' +PROJECT_PACKAGE_NAME = 'blinkpy' +PROJECT_LICENSE = 'MIT' +PROJECT_AUTHOR = 'Kevin Fronczak' +PROJECT_COPYRIGHT = ' 2017, {}'.format(PROJECT_AUTHOR) +PROJECT_URL = 'https://github.com/fronzbot/blinkpy' +PROJECT_EMAIL = 'kfronczak@gmail.com' +PROJECT_DESCRIPTION = ('A Blink camera Python library ' + 'running on Python 3.') +PROJECT_LONG_DESCRIPTION = ('blinkpy is an open-source ' + 'unofficial API for the Blink Camera ' + 'system with the intention for easy ' + 'integration into various home ' + 'automation platforms.') +PROJECT_CLASSIFIERS = [ + 'Intended Audience :: Developers', + 'License :: OSI Approved :: MIT License', + 'Operating System :: OS Independent', + 'Programming Language :: Python :: 3.4', + 'Topic :: Home Automation' +] + +PROJECT_GITHUB_USERNAME = 'home-assistant' +PROJECT_GITHUB_REPOSITORY = 'home-assistant' + +PYPI_URL = 'https://pypi.python.org/pypi/{}'.format(PROJECT_PACKAGE_NAME) ''' URLS @@ -13,6 +46,7 @@ DEFAULT_URL = 'prod.' + BLINK_URL HOME_URL = BASE_URL + '/homescreen' EVENT_URL = BASE_URL + '/events/network/' NETWORK_URL = BASE_URL + '/network/' +NETWORKS_URL = BASE_URL + '/networks' ''' Dictionaries diff --git a/errors.py b/errors.py index 6d3fd36..725c683 100644 --- a/errors.py +++ b/errors.py @@ -1,4 +1,6 @@ USERNAME = (0, "Username must be a string") PASSWORD = (1, "Password must be a string") AUTHENTICATE = (2, "Cannot authenticate since either password or username has not been set") +AUTH_TOKEN = (3, "Authentication header incorrect. Are you sure you logged in and received your token?") +REQUEST = (4, "Cannot perform request (get/post type incorrect)") \ No newline at end of file diff --git a/pylintrc b/pylintrc new file mode 100644 index 0000000..1ec5ded --- /dev/null +++ b/pylintrc @@ -0,0 +1,38 @@ +[MASTER] +reports=no + +# Reasons disabled: +# locally-disabled - it spams too much +# duplicate-code - unavoidable +# cyclic-import - doesn't test if both import on load +# abstract-class-little-used - prevents from setting right foundation +# abstract-class-not-used - is flaky, should not show up but does +# unused-argument - generic callbacks and setup methods create a lot of warnings +# global-statement - used for the on-demand requirement installation +# redefined-variable-type - this is Python, we're duck typing! +# too-many-* - are not enforced for the sake of readability +# too-few-* - same as too-many-* +# abstract-method - with intro of async there are always methods missing + +disable= + locally-disabled, + duplicate-code, + cyclic-import, + abstract-class-little-used, + abstract-class-not-used, + unused-argument, + global-statement, + redefined-variable-type, + too-many-arguments, + too-many-branches, + too-many-instance-attributes, + too-many-locals, + too-many-public-methods, + too-many-return-statements, + too-many-statements, + too-many-lines, + too-few-public-methods, + abstract-method + +[EXCEPTIONS] +overgeneral-exceptions=Exception \ No newline at end of file diff --git a/setup.py b/setup.py index e05ff80..71b8d98 100644 --- a/setup.py +++ b/setup.py @@ -1,26 +1,24 @@ # -*- coding: utf-8 -*- from setuptools import setup +from constants import (__version__, PROJECT_PACKAGE_NAME, + PROJECT_LICENSE, PROJECT_URL, + PROJECT_EMAIL, PROJECT_DESCRIPTION, + PROJECT_CLASSIFIERS, PROJECT_AUTHOR, + PROJECT_LONG_DESCRIPTION) setup( - name = 'blinkpy', - version = '0.4.4', - 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>=2,<3'], - 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' - ] + name = PROJECT_PACKAGE_NAME, + version = __version__, + description = PROJECT_DESCRIPTION, + long_description = PROJECT_LONG_DESCRIPTION, + author = PROJECT_AUTHOR, + author_email = PROJECT_EMAIL, + license = PROJECT_LICENSE, + url = PROJECT_URL, + platforms = 'any', + py_modules = ['blinkpy'], + install_requires = ['requests>=2,<3'], + test_suite = 'tests', + classifiers = PROJECT_CLASSIFIERS ) \ No newline at end of file diff --git a/tests/test_servers.py b/tests/test_servers.py new file mode 100644 index 0000000..08e6631 --- /dev/null +++ b/tests/test_servers.py @@ -0,0 +1,8 @@ +import requests +import unittest +from constants import BASE_URL + +class TestRemoteServerResponse(unittest.TestCase): + def test_request_response(self): + response = requests.get(BASE_URL) + self.assertEqual(response.ok, True) \ No newline at end of file diff --git a/tox.ini b/tox.ini index 660f208..1441238 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,6 @@ [tox] -envlist = py34, pep8 +envlist = py34, py35, py36, lint +skip_missing_interpreters = True [testenv] setenv = @@ -10,14 +11,21 @@ commands = deps = -r{toxinidir}/requirements.txt -[testenv:pep8] +[testenv:lint] deps = -r{toxinidir}/requirements.txt flake8 + pydocstyle + pylint basepython = python3 +ignore_errors = True commands = + ;flake8 blinkpy.py tests flake8 blinkpy.py + ;pylint blinkpy.py tests + pylint blinkpy.py + ;pydocstyle blinkpy.py tests -[flake8] -ignore = E501 -exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build \ No newline at end of file +;[flake8] +;ignore = E501 +;exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build \ No newline at end of file