Initial release of 0.2.0. Really overhauled the first cut and adding a decent amount of testing coverage

This commit is contained in:
Kevin Fronczak
2017-01-21 20:32:15 -05:00
parent 7b3a33deb8
commit bbdd6125e8
13 changed files with 750 additions and 2 deletions
+10
View File
@@ -0,0 +1,10 @@
.cache/*
.tox/*
__pycache__/*
htmlcov/*
.coverage
*.pyc
*.egg*/*
dist/*
MANIFEST
README
-2
View File
@@ -1,2 +0,0 @@
# blinkpy
A Python library for the Blink Camera system
+126
View File
@@ -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
+1
View File
@@ -0,0 +1 @@
from blinkpy import Blink
+344
View File
@@ -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'])
+22
View File
@@ -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))
+18
View File
@@ -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
+5
View File
@@ -0,0 +1,5 @@
pytest>=2.9.2
pytest-cov>=2.3.1
pytest-timeout>=1.0.0
requests>=2.12.4
testtools
+27
View File
@@ -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'
]
)
+82
View File
@@ -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
+86
View File
@@ -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
+6
View File
@@ -0,0 +1,6 @@
import testtools
import blinkpy
class TestImport(testtools.TestCase):
def test_import(self):
pass
+23
View File
@@ -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