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

Changed camera to dict, added id->camera lookup, added ability to write image to file

Browse Source
This commit is contained in:
Kevin Fronczak
2017-01-25 23:34:11 -05:00
parent 666efc7ebe
commit e5c63917e9
6 changed files with 274 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -8,3 +8,4 @@ htmlcov/*
dist/*
MANIFEST
README
.sh
README.rst
+3 -1
View File
@@ -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'
blinkpy.py
+90 -60
View File
@@ -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:
tests/test_blink_requests.py
+84 -81
View File
@@ -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
tests/test_classes.py
+74 -85
View File
@@ -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)
tox.ini
+22 -22
View File
@@ -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