Started adding more camera tests, reorganized package

This commit is contained in:
Kevin Fronczak
2017-03-11 21:53:58 -05:00
parent 94dab01e6b
commit d54a6cbb71
15 changed files with 376 additions and 303 deletions
+4
View File
@@ -0,0 +1,4 @@
include README.rst
include LICENSE
include tests/*.py
include helpers/*.py
+76 -84
View File
@@ -1,9 +1,9 @@
**blinkpy** |Build Status| |Coverage Status| blinkpy |Build Status| |Coverage Status|
============ ============
A Python library for the Blink Camera system A Python library for the Blink Camera system
**Disclaimers** Disclaimer:
=============== ~~~~~~~~~~~~~~~
Published under the MIT license - See LICENSE file for more details. 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.
@@ -11,113 +11,105 @@ I am in no way affiliated with Blink, nor Immedia Inc.
Original protocol hacking by MattTW : https://github.com/MattTW/BlinkMonitorProtocol Original protocol hacking by MattTW : https://github.com/MattTW/BlinkMonitorProtocol
**Installation** API calls faster than 60 seconds is not recommended as it can overwhelm Blink's servers. Please use this module responsibly.
Installation
================ ================
``pip3 install blinkpy`` ``pip3 install blinkpy``
**Purpose** 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 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** Usage
========= =========
In terms of usage, you just need to instantiate the module with a username and password In terms of usage, you just need to instantiate the module with a username and password
::
import blinkpy .. code:: python
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR 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 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() .. code:: python
import blinkpy
blink = blinkpy.Blink()
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. 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) 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: .. code:: python
print(camera.name) # Name of the camera
print(camera.id) # Integer id of the camera (assigned by Blink) import blinkpy
print(camera.armed) # Whether the device is armed/disarmed (ie. detecting motion)
print(camera.clip) # Link to last motion clip captured blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD')
print(camera.thumbnail) # Link to current camera thumbnail blink.setup_system()
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. for camera in blink.cameras:
print(camera.notifications) # Number of unread notifications (ie. motion alerts that haven't been viewed) print(camera.name) # Name of the camera
print(camera.motion) # Dictionary containing values for keys ['video', 'image', 'time'] print(camera.id) # Integer id of the camera (assigned by Blink)
# which correspond to last motion recorded, thumbnail of last motion, and timestamp of last motion 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** Class Descriptions
--------------- ===================
The following properties/methods are availiable to the Blink class .. code:: python
**Blink.cameras** class Blink()
Returns a list of BlinkCamera objects found by the system
**Blink.network_id** * ``Blink.cameras`` Returns a dictionary of ``BlinkCamera`` objects where the key corresponds to the camera name and the value is the actual BlinkCamera object.
Returns the current network id * ``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 ``BlinkCamera`` objects to represent 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:
.. code:: python
**Blink.account_id** Blink.get_auth_token()
Returns the account id Blink.get_ids()
Blink.get_camers()
Blink.set_links()
**Blink.events** .. code:: python
Returns a list of events recorded by blink. This information will contain links to any motion caught by an armed camera.
**Blink.online** class BlinkCamera(config, urls)
Returns online status of sync module (True = online, False = offline)
The ``BlinkCamera`` class expects to receive:
* A dictionary ``config`` that contains the camera name, device id, armed status, thumbnail url, camera temperature, camery battery level, number of notifications, and region id
* A ``BlinkURLHandler`` object that contains all the links necessary for communication.
**Blink.last_motion()** Ultimately, 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):
Finds last motion information for each camera and stores it in the BlinkCamera.motion field * ``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.
* ``BlinkCamera.image_to_file(path)`` This will write the current thumbnail to the location indicated in 'path'
* ``BlinkCamera.image_refresh()`` Refreshes the current thumbnail.
**Blink.arm** .. code:: python
Set to True to arm, False to disarm. Can be used to see the status of the system as well
**Blink.refresh()** class BlinkURLHandler(region_id)
Forces a refresh of all camera information
The ``BlinkURLHandler`` class expects to be initialized with the region id found in the ``Blink.get_auth_token()`` function. The class will then create the necessary links required for various communication.
**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
**BlinkCamera.image_to_file(path)**
This will write the current thumbnail to the location indicated in 'path'
.. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=master .. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=master
:target: https://travis-ci.org/fronzbot/blinkpy :target: https://travis-ci.org/fronzbot/blinkpy
+5 -1
View File
@@ -1 +1,5 @@
from blinkpy import Blink """Init file for blinkpy."""
from blinkpy.blinkpy import Blink
from blinkpy.blinkpy import BlinkAuthenticationException
from blinkpy.blinkpy import BlinkException
from blinkpy.blinkpy import BlinkURLHandler
+69 -74
View File
@@ -1,8 +1,8 @@
#!/usr/bin/python #!/usr/bin/python
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""
blinkpy by Kevin Fronczak - A Blink camera Python library.
'''
blinkpy by Kevin Fronczak - A Blink camera Python library
https://github.com/fronzbot/blinkpy https://github.com/fronzbot/blinkpy
Original protocol hacking by MattTW : Original protocol hacking by MattTW :
https://github.com/MattTW/BlinkMonitorProtocol https://github.com/MattTW/BlinkMonitorProtocol
@@ -10,19 +10,19 @@ Published under the MIT license - See LICENSE file for more details.
"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark "Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark
owned by Immedia Inc., see www.blinkforhome.com for more information. owned by Immedia Inc., see www.blinkforhome.com for more information.
I am in no way affiliated with Blink, nor Immedia Inc. I am in no way affiliated with Blink, nor Immedia Inc.
''' """
import json import json
import getpass import getpass
import requests import requests
import errors as ERROR import helpers.errors as ERROR
from constants import (BLINK_URL, LOGIN_URL, from helpers.constants import (BLINK_URL, LOGIN_URL,
DEFAULT_URL, ONLINE) DEFAULT_URL, ONLINE)
def _request(url, data=None, headers=None, reqtype='get', def _request(url, data=None, headers=None, reqtype='get',
stream=False, json_resp=True): stream=False, json_resp=True):
"""Wrapper function for request""" """Wrapper function for request."""
if reqtype is 'post': if reqtype is 'post':
response = requests.post(url, headers=headers, response = requests.post(url, headers=headers,
data=data).json() data=data).json()
@@ -44,23 +44,23 @@ def _request(url, data=None, headers=None, reqtype='get',
# pylint: disable=super-init-not-called # pylint: disable=super-init-not-called
class BlinkException(Exception): class BlinkException(Exception):
""" """Class to throw general blink exception."""
Class to throw general blink exception.
"""
def __init__(self, errcode): def __init__(self, errcode):
"""Initialize BlinkException."""
self.errid = errcode[0] self.errid = errcode[0]
self.message = errcode[1] self.message = errcode[1]
class BlinkAuthenticationException(BlinkException): class BlinkAuthenticationException(BlinkException):
""" """Class to throw authentication exception."""
Class to throw authentication exception.
"""
pass pass
class BlinkURLHandler(object): class BlinkURLHandler(object):
"""Class that handles Blink URLS""" """Class that handles Blink URLS."""
def __init__(self, region_id): def __init__(self, region_id):
"""Initialize the urls.""" """Initialize the urls."""
self.base_url = 'https://' + region_id + '.' + BLINK_URL self.base_url = 'https://' + region_id + '.' + BLINK_URL
@@ -72,7 +72,9 @@ class BlinkURLHandler(object):
class BlinkCamera(object): class BlinkCamera(object):
"""Class to initialize individual camera.""" """Class to initialize individual camera."""
def __init__(self, config, urls): def __init__(self, config, urls):
"""Initiailize BlinkCamera."""
self.urls = urls self.urls = urls
self._id = str(config['device_id']) self._id = str(config['device_id'])
self._name = config['name'] self._name = config['name']
@@ -91,125 +93,125 @@ class BlinkCamera(object):
@property @property
# pylint: disable=invalid-name # pylint: disable=invalid-name
def id(self): def id(self):
"""Returns camera id""" """Return camera id."""
return self._id return self._id
@property @property
def name(self): def name(self):
"""Returns camera name""" """Return camera name."""
return self._name return self._name
@name.setter @name.setter
def name(self, value): def name(self, value):
"""Sets camera name""" """Set camera name."""
self._name = value self._name = value
@property @property
def region_id(self): def region_id(self):
"""Returns region id""" """Return region id."""
return self._region_id return self._region_id
@property @property
def armed(self): def armed(self):
"""Returns camera arm status""" """Return camera arm status."""
return self._status return self._status
@property @property
def clip(self): def clip(self):
"""Returns current clip""" """Return current clip."""
return self._clip return self._clip
@clip.setter @clip.setter
def clip(self, value): def clip(self, value):
"""Sets current clip""" """Set current clip."""
self._clip = value self._clip = value
@property @property
def thumbnail(self): def thumbnail(self):
"""Returns current thumbnail""" """Return current thumbnail."""
return self._thumb return self._thumb
@thumbnail.setter @thumbnail.setter
def thumbnail(self, value): def thumbnail(self, value):
"""Sets current thumbnail""" """Set current thumbnail."""
self._thumb = value self._thumb = value
@property @property
def temperature(self): def temperature(self):
"""Returns camera temperature""" """Return camera temperature."""
return self._temperature return self._temperature
@temperature.setter @temperature.setter
def temperature(self, value): def temperature(self, value):
"""Sets camera temperature""" """Set camera temperature."""
self._temperature = value self._temperature = value
@property @property
def battery(self): def battery(self):
"""Returns battery level""" """Return battery level."""
return self._battery return self._battery
@battery.setter @battery.setter
def battery(self, value): def battery(self, value):
"""Sets battery level""" """Set battery level."""
self._battery = value self._battery = value
@property @property
def notifications(self): def notifications(self):
"""Returns number of notifications""" """Return number of notifications."""
return self._notifications return self._notifications
@notifications.setter @notifications.setter
def notifications(self, value): def notifications(self, value):
"""Sets number of notifications""" """Set number of notifications."""
self._notifications = value self._notifications = value
@property @property
def image_link(self): def image_link(self):
"""Returns image link""" """Return image link."""
return self._image_link return self._image_link
@image_link.setter @image_link.setter
def image_link(self, value): def image_link(self, value):
"""Sets image link""" """Set image link."""
self._image_link = value self._image_link = value
@property @property
def arm_link(self): def arm_link(self):
"""Returns link to arm camera""" """Return link to arm camera."""
return self._arm_link return self._arm_link
@arm_link.setter @arm_link.setter
def arm_link(self, value): def arm_link(self, value):
"""Sets link to arm camera""" """Set link to arm camera."""
self._arm_link = value self._arm_link = value
@property @property
def header(self): def header(self):
"""Returns request header""" """Return request header."""
return self._header return self._header
@header.setter @header.setter
def header(self, value): def header(self, value):
"""Sets request header""" """Set request header."""
self._header = value self._header = value
@property @property
def motion(self): def motion(self):
"""Returns last motion event detail""" """Return last motion event detail."""
return self._motion return self._motion
@motion.setter @motion.setter
def motion(self, value): def motion(self, value):
"""Sets link to last motion and timestamp""" """Set link to last motion and timestamp."""
self._motion = value self._motion = value
def snap_picture(self): def snap_picture(self):
"""Takes a picture with camera to create a new thumbnail""" """Take a picture with camera to create a new thumbnail."""
_request(self._image_link, headers=self._header, reqtype='post') _request(self._image_link, headers=self._header, reqtype='post')
def set_motion_detect(self, enable): def set_motion_detect(self, enable):
"""Sets motion detection""" """Set motion detection."""
url = self._arm_link url = self._arm_link
if enable: if enable:
_request(url + 'enable', headers=self._header, reqtype='post') _request(url + 'enable', headers=self._header, reqtype='post')
@@ -217,7 +219,7 @@ class BlinkCamera(object):
_request(url + 'disable', headers=self._header, reqtype='post') _request(url + 'disable', headers=self._header, reqtype='post')
def update(self, values): def update(self, values):
"""Updates camera information""" """Update camera information."""
self._name = values['name'] self._name = values['name']
self._status = values['armed'] self._status = values['armed']
self._thumb = self.urls.base_url + values['thumbnail'] + '.jpg' self._thumb = self.urls.base_url + values['thumbnail'] + '.jpg'
@@ -227,7 +229,7 @@ class BlinkCamera(object):
self._notifications = values['notifications'] self._notifications = values['notifications']
def image_refresh(self): def image_refresh(self):
"""Refreshs current thumbnail""" """Refresh current thumbnail."""
url = self.urls.home_url url = self.urls.home_url
response = _request(url, headers=self._header, response = _request(url, headers=self._header,
reqtype='get')['devices'] reqtype='get')['devices']
@@ -242,7 +244,7 @@ class BlinkCamera(object):
return None return None
def image_to_file(self, path): def image_to_file(self, path):
"""Writes image to file""" """Write image to file."""
thumb = self.image_refresh() thumb = self.image_refresh()
response = _request(thumb, headers=self._header, response = _request(thumb, headers=self._header,
stream=True, json_resp=False) stream=True, json_resp=False)
@@ -253,9 +255,10 @@ class BlinkCamera(object):
class Blink(object): class Blink(object):
"""Class to initialize communication and sync module""" """Class to initialize communication and sync module."""
def __init__(self, username=None, password=None): def __init__(self, username=None, password=None):
"""Constructor for class""" """Initialize Blink system."""
self._username = username self._username = username
self._password = password self._password = password
self._token = None self._token = None
@@ -272,12 +275,12 @@ class Blink(object):
@property @property
def cameras(self): def cameras(self):
"""Returns camera/id pairs""" """Return camera/id pairs."""
return self._cameras return self._cameras
@property @property
def camera_thumbs(self): def camera_thumbs(self):
"""Returns camera thumbnails""" """Return camera thumbnails."""
self.refresh() self.refresh()
data = {} data = {}
for name, camera in self._cameras.items(): for name, camera in self._cameras.items():
@@ -287,32 +290,32 @@ class Blink(object):
@property @property
def id_table(self): def id_table(self):
"""Returns id/camera pairs""" """Return id/camera pairs."""
return self._idlookup return self._idlookup
@property @property
def network_id(self): def network_id(self):
"""Returns network id""" """Return network id."""
return self._network_id return self._network_id
@property @property
def account_id(self): def account_id(self):
"""Returns account id""" """Return account id."""
return self._account_id return self._account_id
@property @property
def region(self): def region(self):
"""Returns current region""" """Return current region."""
return self._region return self._region
@property @property
def region_id(self): def region_id(self):
"""Returns region id""" """Return region id."""
return self._region_id return self._region_id
@property @property
def events(self): def events(self):
"""Gets all events on server""" """Get all events on server."""
url = self.urls.event_url + self._network_id url = self.urls.event_url + self._network_id
headers = self._auth_header headers = self._auth_header
self._events = _request(url, headers=headers, self._events = _request(url, headers=headers,
@@ -321,17 +324,14 @@ class Blink(object):
@property @property
def online(self): def online(self):
""" """Return boolean system online status."""
Returns True or False depending on if
sync module is online/offline
"""
url = self.urls.network_url + self._network_id + '/syncmodules' url = self.urls.network_url + self._network_id + '/syncmodules'
headers = self._auth_header headers = self._auth_header
return ONLINE[_request(url, headers=headers, return ONLINE[_request(url, headers=headers,
reqtype='get')['syncmodule']['status']] reqtype='get')['syncmodule']['status']]
def last_motion(self): def last_motion(self):
"""Finds last motion of each camera""" """Find last motion of each camera."""
recent = self.events recent = self.events
for element in recent: for element in recent:
try: try:
@@ -348,15 +348,12 @@ class Blink(object):
@property @property
def arm(self): def arm(self):
"""Returns status of sync module: armed/disarmed""" """Return status of sync module: armed/disarmed."""
return self.get_summary()['network']['armed'] return self.get_summary()['network']['armed']
@arm.setter @arm.setter
def arm(self, value): def arm(self, value):
""" """Arm or disarm system."""
Arms or disarms system.
Arms/disarms all if camera not named.
"""
if value: if value:
value_to_append = 'arm' value_to_append = 'arm'
else: else:
@@ -365,7 +362,7 @@ class Blink(object):
_request(url, headers=self._auth_header, reqtype='post') _request(url, headers=self._auth_header, reqtype='post')
def refresh(self): def refresh(self):
"""Gets all blink cameras and pulls their most recent status""" """Get all blink cameras and pulls their most recent status."""
response = self.get_summary()['devices'] response = self.get_summary()['devices']
for name in self._cameras: for name in self._cameras:
@@ -378,7 +375,7 @@ class Blink(object):
pass pass
def get_summary(self): def get_summary(self):
"""Gets a full summary of device information""" """Get a full summary of device information."""
url = self.urls.home_url url = self.urls.home_url
headers = self._auth_header headers = self._auth_header
@@ -388,7 +385,7 @@ class Blink(object):
return _request(url, headers=headers, reqtype='get') return _request(url, headers=headers, reqtype='get')
def get_cameras(self): def get_cameras(self):
"""Finds and creates cameras""" """Find and creates cameras."""
response = self.get_summary()['devices'] response = self.get_summary()['devices']
for element in response: for element in response:
if ('device_type' in element and if ('device_type' in element and
@@ -400,10 +397,7 @@ class Blink(object):
self._idlookup[device.id] = device.name self._idlookup[device.id] = device.name
def set_links(self): def set_links(self):
""" """Set access links and required headers for each camera in system."""
Sets access links and required headers
for each camera in system
"""
for name in self._cameras: for name in self._cameras:
camera = self._cameras[name] camera = self._cameras[name]
network_id_url = self.urls.network_url + self._network_id network_id_url = self.urls.network_url + self._network_id
@@ -415,8 +409,9 @@ class Blink(object):
def setup_system(self): def setup_system(self):
""" """
Method logs in and sets auth token and Wrapper for various setup functions.
network ids for future requests.
Method logs in and sets auth token, urls, and ids for future requests.
""" """
if self._username is None or self._password is None: if self._username is None or self._password is None:
raise BlinkAuthenticationException(ERROR.AUTHENTICATE) raise BlinkAuthenticationException(ERROR.AUTHENTICATE)
@@ -427,12 +422,12 @@ class Blink(object):
self.set_links() self.set_links()
def login(self): def login(self):
"""Prompts user for username and password""" """Prompt user for username and password."""
self._username = input("Username:") self._username = input("Username:")
self._password = getpass.getpass("Password:") self._password = getpass.getpass("Password:")
def get_auth_token(self): def get_auth_token(self):
"""Retrieves the authentication token from Blink""" """Retrieve the authentication token from Blink."""
if not isinstance(self._username, str): if not isinstance(self._username, str):
raise BlinkAuthenticationException(ERROR.USERNAME) raise BlinkAuthenticationException(ERROR.USERNAME)
if not isinstance(self._password, str): if not isinstance(self._password, str):
@@ -456,7 +451,7 @@ class Blink(object):
self.urls = BlinkURLHandler(self._region_id) self.urls = BlinkURLHandler(self._region_id)
def get_ids(self): def get_ids(self):
"""Sets the network ID and Account ID""" """Set the network ID and Account ID."""
url = self.urls.networks_url url = self.urls.networks_url
headers = self._auth_header headers = self._auth_header
-22
View File
@@ -1,22 +0,0 @@
#!/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
@@ -1,18 +0,0 @@
#!/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
+3
View File
@@ -0,0 +1,3 @@
"""Init file for blinkpy helper functions."""
from helpers import constants
from helpers import errors
View File
+5 -6
View File
@@ -1,11 +1,10 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from setuptools import setup from setuptools import setup
from constants import (__version__, PROJECT_PACKAGE_NAME, from helpers.constants import (__version__, PROJECT_PACKAGE_NAME,
PROJECT_LICENSE, PROJECT_URL, PROJECT_LICENSE, PROJECT_URL,
PROJECT_EMAIL, PROJECT_DESCRIPTION, PROJECT_EMAIL, PROJECT_DESCRIPTION,
PROJECT_CLASSIFIERS, PROJECT_AUTHOR, PROJECT_CLASSIFIERS, PROJECT_AUTHOR,
PROJECT_LONG_DESCRIPTION) PROJECT_LONG_DESCRIPTION)
setup( setup(
name = PROJECT_PACKAGE_NAME, name = PROJECT_PACKAGE_NAME,
+1
View File
@@ -0,0 +1 @@
"""Init file for tests directory."""
+83 -62
View File
@@ -1,18 +1,26 @@
import constants as const """
Mock responses that mimic actual responses from Blink servers.
This file should be updated any time the Blink server responses
change so we can make sure blinkpy can still communicate.
"""
import helpers.constants as const
"""Fake device attributes."""
NETWORKS_RESPONSE = {} NETWORKS_RESPONSE = {}
NETWORKS_RESPONSE['summary'] = {'onboarded': True, 'name': 'Nilfgaard'} NETWORKS_RESPONSE['summary'] = {'onboarded': True, 'name': 'Nilfgaard'}
NETWORKS_RESPONSE['networks'] = [{'network_key': None, NETWORKS_RESPONSE['networks'] = [{
'name': NETWORKS_RESPONSE['summary']['name'], 'network_key': None,
'account_id': 1989, 'name': NETWORKS_RESPONSE['summary']['name'],
'id': 9898, 'account_id': 1989,
'encryption_key': None, 'id': 9898,
'armed': True, 'encryption_key': None,
'ping_interval': 60, 'armed': True,
'video_destination': 'server', 'ping_interval': 60,
'arm_string': 'Armed', 'video_destination': 'server',
'feature_plan_id': None}] 'arm_string': 'Armed',
'feature_plan_id': None
}]
FIRST_CAMERA = {'device_type': 'camera', FIRST_CAMERA = {'device_type': 'camera',
'notifications': 1, 'notifications': 1,
@@ -50,50 +58,51 @@ SYNC_MODULE = {'updated_at': '1970-01-01T01:00:00+00:00',
'device_id': 7990, 'device_id': 7990,
'status': 'online'} 'status': 'online'}
FIRST_EVENT = {'camera_name': FIRST_CAMERA['name'], FIRST_EVENT = {'camera_name': FIRST_CAMERA['name'],
'updated_at': '2017-10-10T03:37:37+00:00', 'updated_at': '2017-10-10T03:37:37+00:00',
'sync_module_id': None, 'sync_module_id': None,
'camera': FIRST_CAMERA['device_id'], 'camera': FIRST_CAMERA['device_id'],
'type': 'motion', 'type': 'motion',
'duration': None, 'duration': None,
'status': None, 'status': None,
'created_at': '2017-01-28T19:51:52+00:00', 'created_at': '2017-01-28T19:51:52+00:00',
'camera_id': FIRST_CAMERA['device_id'], 'camera_id': FIRST_CAMERA['device_id'],
'id': 666777888, 'id': 666777888,
'siren_id': None, 'siren_id': None,
'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'], 'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'],
'notified': True, 'notified': True,
'siren': None, 'siren': None,
'syncmodule': None, 'syncmodule': None,
'video_url': FIRST_CAMERA['thumbnail'] + '.mp4', 'video_url': FIRST_CAMERA['thumbnail'] + '.mp4',
'command_id': None, 'command_id': None,
'network_id': None, 'network_id': None,
'account': NETWORKS_RESPONSE['networks'][0]['account_id'], 'account': NETWORKS_RESPONSE['networks'][0]['account_id'],
'video_id': 123000321} 'video_id': 123000321}
SECOND_EVENT = {'updated_at': '2017-01-28T19:51:29+00:00', SECOND_EVENT = {'updated_at': '2017-01-28T19:51:29+00:00',
'sync_module_id': SYNC_MODULE['device_id'], 'sync_module_id': SYNC_MODULE['device_id'],
'camera': None, 'camera': None,
'type': 'armed', 'type': 'armed',
'duration': None, 'duration': None,
'status': None, 'status': None,
'created_at': '2017-01-28T19:51:29+00:00', 'created_at': '2017-01-28T19:51:29+00:00',
'camera_id': None, 'camera_id': None,
'id': 333777555, 'id': 333777555,
'siren_id': None, 'siren_id': None,
'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'], 'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'],
'notified': False, 'notified': False,
'siren': None, 'siren': None,
'syncmodule': SYNC_MODULE['device_id'], 'syncmodule': SYNC_MODULE['device_id'],
'command_id': None, 'command_id': None,
'network_id': None, 'network_id': None,
'account': NETWORKS_RESPONSE['networks'][0]['account_id']} 'account': NETWORKS_RESPONSE['networks'][0]['account_id']}
"""Fake response content.""" """Fake response content."""
LOGIN_RESPONSE = {} LOGIN_RESPONSE = {}
LOGIN_RESPONSE['region'] = {'ciri': 'Cintra'} LOGIN_RESPONSE['region'] = {'ciri': 'Cintra'}
LOGIN_RESPONSE['networks'] = {NETWORKS_RESPONSE['networks'][0]['id'] : NETWORKS_RESPONSE['summary']} LOGIN_RESPONSE['networks'] = {
NETWORKS_RESPONSE['networks'][0]['id']: NETWORKS_RESPONSE['summary']
}
LOGIN_RESPONSE['authtoken'] = {'authtoken': 'foobar7117', 'message': 'auth'} LOGIN_RESPONSE['authtoken'] = {'authtoken': 'foobar7117', 'message': 'auth'}
RESPONSE = {} RESPONSE = {}
@@ -111,42 +120,54 @@ RESPONSE['syncmodule'] = {'name': 'Vengerberg', 'status': 'online'}
def mocked_requests_post(*args, **kwargs): def mocked_requests_post(*args, **kwargs):
"""Mock post request.""" """Mock post request."""
class MockPostResponse: class MockPostResponse:
def __init__(self,json_data,status_code): """Class for mock post response."""
def __init__(self, json_data, status_code):
"""Initialze mock post response."""
self.json_data = json_data self.json_data = json_data
self.status_code = status_code self.status_code = status_code
def json(self): def json(self):
"""Return json data from post request."""
return self.json_data return self.json_data
url_tail = args[0].split("/")[-1] url_tail = args[0].split("/")[-1]
if args[0] == const.LOGIN_URL: if args[0] == const.LOGIN_URL:
return MockPostResponse(LOGIN_RESPONSE, 200) return MockPostResponse(LOGIN_RESPONSE, 200)
elif url_tail == 'arm' or url_tail == 'disarm': elif url_tail == 'arm' or url_tail == 'disarm':
# pylint: disable=global-variable-not-assigned
global NETWORKS_RESPONSE global NETWORKS_RESPONSE
# pylint: disable=global-variable-not-assigned
global RESPONSE global RESPONSE
NETWORKS_RESPONSE['networks'][0]['armed'] = url_tail == 'arm' NETWORKS_RESPONSE['networks'][0]['armed'] = url_tail == 'arm'
RESPONSE['network']['armed'] = NETWORKS_RESPONSE['networks'][0]['armed'] RESPONSE['network']['armed'] = url_tail == 'arm'
return MockPostResponse({}, 200) return MockPostResponse({}, 200)
return MockPostResponse({'message':'ERROR','code':404}, 404) return MockPostResponse({'message': 'ERROR', 'code': 404}, 404)
def mocked_requests_get(*args, **kwargs): def mocked_requests_get(*args, **kwargs):
"""Mock get request.""" """Mock get request."""
class MockGetResponse: class MockGetResponse:
def __init__(self,json_data,status_code): """Class for mock get response."""
def __init__(self, json_data, status_code):
"""Initialze mock get response."""
self.json_data = json_data self.json_data = json_data
self.status_code = status_code self.status_code = status_code
def json(self):
return self.json_data
def json(self):
"""Return json data from post request."""
return self.json_data
# pylint: disable=unused-variable
(region_id, region), = LOGIN_RESPONSE['region'].items() (region_id, region), = LOGIN_RESPONSE['region'].items()
set_region_id = args[0].split('/')[2].split('.')[0] set_region_id = args[0].split('/')[2].split('.')[0]
NETURL = 'https://' + set_region_id + '.' + const.BLINK_URL + '/networks' neturl = 'https://' + set_region_id + '.' + const.BLINK_URL + '/networks'
if args[0] == NETURL: if args[0] == neturl:
return MockGetResponse(NETWORKS_RESPONSE, 200) return MockGetResponse(NETWORKS_RESPONSE, 200)
elif set_region_id != region_id: elif set_region_id != region_id:
raise ConnectionError('Received url ' + args[0]) raise ConnectionError('Received url ' + args[0])
else: else:
return MockGetResponse(RESPONSE, 200) return MockGetResponse(RESPONSE, 200)
return MockGetResponse({'message':'ERROR','code':404}, 404) return MockGetResponse({'message': 'ERROR', 'code': 404}, 404)
+80 -1
View File
@@ -1 +1,80 @@
# TODO """
Tests the camera initialization and attributes of
individual BlinkCamera instantiations.
"""
import unittest
from unittest import mock
import blinkpy
import tests.mock_responses as mresp
USERNAME = 'foobar'
PASSWORD = 'deadbeef'
class TestBlinkCameraSetup(unittest.TestCase):
"""Test the Blink class in blinkpy."""
def setUp(self):
"""Set up Blink module."""
self.blink = blinkpy.Blink(username=USERNAME,
password=PASSWORD)
def tearDown(self):
"""Clean up after test."""
self.blink = None
@mock.patch('blinkpy.blinkpy.requests.post',
side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.blinkpy.requests.get',
side_effect=mresp.mocked_requests_get)
def test_camera_values_from_setup(self, mock_get, mock_post):
"""Tests all property values after camera setup."""
self.blink.setup_system()
# Get expected test values
test_network_id = str(mresp.NETWORKS_RESPONSE['networks'][0]['id'])
# pylint: disable=unused-variable
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
# pylint: disable=protected-access
expected_header = self.blink._auth_header
test_urls = blinkpy.BlinkURLHandler(region_id)
test_cameras = dict()
for element in mresp.RESPONSE['devices']:
if ('device_type' in element and
element['device_type'] == 'camera'):
test_cameras[element['name']] = {
'device_id': str(element['device_id']),
'armed': element['armed'],
'thumbnail': (test_urls.base_url +
element['thumbnail'] + '.jpg'),
'temperature': element['temp'],
'battery': element['battery'],
'notifications': element['notifications']
}
test_net_id_url = test_urls.network_url + test_network_id
for name in self.blink.cameras:
camera = self.blink.cameras[name]
self.assertEqual(name, camera.name)
if name in test_cameras:
self.assertEqual(camera.id,
test_cameras[name]['device_id'])
self.assertEqual(camera.armed,
test_cameras[name]['armed'])
self.assertEqual(camera.thumbnail,
test_cameras[name]['thumbnail'])
self.assertEqual(camera.temperature,
test_cameras[name]['temperature'])
self.assertEqual(camera.battery,
test_cameras[name]['battery'])
self.assertEqual(camera.notifications,
test_cameras[name]['notifications'])
else:
self.fail("Camera wasn't initialized: " + name)
expected_arm_link = test_net_id_url + '/camera/' + camera.id + '/'
expected_image_link = expected_arm_link + 'thumbnail'
self.assertEqual(camera.image_link, expected_image_link)
self.assertEqual(camera.arm_link, expected_arm_link)
self.assertEqual(camera.header, expected_header)
+45 -24
View File
@@ -1,15 +1,22 @@
import requests """
Tests the system initialization and attributes of
the main Blink system. Tests if we properly catch
any communication related errors at startup.
"""
import unittest import unittest
from unittest import mock from unittest import mock
import blinkpy import blinkpy
import tests.mock_responses as mresp import tests.mock_responses as mresp
import constants as const import helpers.constants as const
USERNAME = 'foobar' USERNAME = 'foobar'
PASSWORD = 'deadbeef' PASSWORD = 'deadbeef'
class TestBlinkSetup(unittest.TestCase): class TestBlinkSetup(unittest.TestCase):
"""Test the Blink class in blinkpy.""" """Test the Blink class in blinkpy."""
def setUp(self): def setUp(self):
"""Set up Blink module.""" """Set up Blink module."""
self.blink_no_cred = blinkpy.Blink() self.blink_no_cred = blinkpy.Blink()
@@ -23,7 +30,9 @@ class TestBlinkSetup(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Verify we can initialize blink.""" """Verify we can initialize blink."""
# pylint: disable=protected-access
self.assertEqual(self.blink._username, USERNAME) self.assertEqual(self.blink._username, USERNAME)
# pylint: disable=protected-access
self.assertEqual(self.blink._password, PASSWORD) self.assertEqual(self.blink._password, PASSWORD)
def test_no_credentials(self): def test_no_credentials(self):
@@ -35,62 +44,72 @@ class TestBlinkSetup(unittest.TestCase):
def test_no_auth_header(self): def test_no_auth_header(self):
"""Check that we throw an excpetion when no auth header given.""" """Check that we throw an excpetion when no auth header given."""
# pylint: disable=unused-variable
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items() (region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
self.blink.urls = blinkpy.BlinkURLHandler(region_id) self.blink.urls = blinkpy.BlinkURLHandler(region_id)
with self.assertRaises(blinkpy.BlinkException): with self.assertRaises(blinkpy.BlinkException):
self.blink.get_ids() self.blink.get_ids()
@mock.patch('blinkpy.getpass.getpass') @mock.patch('blinkpy.blinkpy.getpass.getpass')
def test_manual_login(self, getpwd): def test_manual_login(self, getpwd):
"""Check that we can manually use the login() function.""" """Check that we can manually use the login() function."""
getpwd.return_value = PASSWORD getpwd.return_value = PASSWORD
with mock.patch('builtins.input', return_value=USERNAME): with mock.patch('builtins.input', return_value=USERNAME):
self.blink_no_cred.login() self.blink_no_cred.login()
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._username, USERNAME) self.assertEqual(self.blink_no_cred._username, USERNAME)
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._password, PASSWORD) self.assertEqual(self.blink_no_cred._password, PASSWORD)
@mock.patch('blinkpy.requests.post', side_effect=mresp.mocked_requests_post) @mock.patch('blinkpy.blinkpy.requests.post',
@mock.patch('blinkpy.requests.get', side_effect=mresp.mocked_requests_get) side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.blinkpy.requests.get',
side_effect=mresp.mocked_requests_get)
def test_full_setup(self, mock_get, mock_post): def test_full_setup(self, mock_get, mock_post):
"""Check that we can set Blink up properly.""" """Check that we can set Blink up properly."""
self.blink.setup_system() self.blink.setup_system()
# Get all test values # Get all test values
authtoken = mresp.LOGIN_RESPONSE['authtoken']['authtoken'] authtoken = mresp.LOGIN_RESPONSE['authtoken']['authtoken']
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items() (region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
host = region_id + '.' + const.BLINK_URL host = region_id + '.' + const.BLINK_URL
network_id = mresp.NETWORKS_RESPONSE['networks'][0]['id'] network_id = mresp.NETWORKS_RESPONSE['networks'][0]['id']
account_id = mresp.NETWORKS_RESPONSE['networks'][0]['account_id'] account_id = mresp.NETWORKS_RESPONSE['networks'][0]['account_id']
TestURLs = blinkpy.BlinkURLHandler(region_id) test_urls = blinkpy.BlinkURLHandler(region_id)
TestCameras = list() test_cameras = list()
TestCameraID = dict() test_camera_id = dict()
for element in mresp.RESPONSE['devices']: for element in mresp.RESPONSE['devices']:
if 'device_type' in element and element['device_type'] == 'camera': if ('device_type' in element and
TestCameras.append(element['name']) element['device_type'] == 'camera'):
TestCameraID[str(element['device_id'])] = element['name'] test_cameras.append(element['name'])
test_camera_id[str(element['device_id'])] = element['name']
# Check that all links have been set properly # Check that all links have been set properly
self.assertEqual(self.blink.region_id, region_id) self.assertEqual(self.blink.region_id, region_id)
self.assertEqual(self.blink.urls.base_url, TestURLs.base_url) self.assertEqual(self.blink.urls.base_url, test_urls.base_url)
self.assertEqual(self.blink.urls.home_url, TestURLs.home_url) self.assertEqual(self.blink.urls.home_url, test_urls.home_url)
self.assertEqual(self.blink.urls.event_url, TestURLs.event_url) self.assertEqual(self.blink.urls.event_url, test_urls.event_url)
self.assertEqual(self.blink.urls.network_url, TestURLs.network_url) self.assertEqual(self.blink.urls.network_url, test_urls.network_url)
self.assertEqual(self.blink.urls.networks_url, TestURLs.networks_url) self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url)
# Check that all properties have been set after startup # Check that all properties have been set after startup
# pylint: disable=protected-access
self.assertEqual(self.blink._token, authtoken) self.assertEqual(self.blink._token, authtoken)
# pylint: disable=protected-access
self.assertEqual(self.blink._host, host) self.assertEqual(self.blink._host, host)
self.assertEqual(self.blink.network_id, str(network_id)) self.assertEqual(self.blink.network_id, str(network_id))
self.assertEqual(self.blink.account_id, str(account_id)) self.assertEqual(self.blink.account_id, str(account_id))
self.assertEqual(self.blink.region, region) self.assertEqual(self.blink.region, region)
# Verify we have initialized all the cameras # Verify we have initialized all the cameras
self.assertEqual(self.blink.id_table, TestCameraID) self.assertEqual(self.blink.id_table, test_camera_id)
for camera in TestCameras: for camera in test_cameras:
self.assertTrue(camera in self.blink.cameras) self.assertTrue(camera in self.blink.cameras)
@mock.patch('blinkpy.requests.post', side_effect=mresp.mocked_requests_post) @mock.patch('blinkpy.blinkpy.requests.post',
@mock.patch('blinkpy.requests.get', side_effect=mresp.mocked_requests_get) side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.blinkpy.requests.get',
side_effect=mresp.mocked_requests_get)
def test_arm_disarm_system(self, mock_get, mock_post): def test_arm_disarm_system(self, mock_get, mock_post):
"""Check that we can arm/disarm the system""" """Check that we can arm/disarm the system"""
self.blink.setup_system() self.blink.setup_system()
@@ -99,8 +118,10 @@ class TestBlinkSetup(unittest.TestCase):
self.blink.arm = True self.blink.arm = True
self.assertIs(self.blink.arm, True) self.assertIs(self.blink.arm, True)
@mock.patch('blinkpy.requests.post', side_effect=mresp.mocked_requests_post) @mock.patch('blinkpy.blinkpy.requests.post',
@mock.patch('blinkpy.requests.get', side_effect=mresp.mocked_requests_get) side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.blinkpy.requests.get',
side_effect=mresp.mocked_requests_get)
def test_check_online_status(self, mock_get, mock_post): def test_check_online_status(self, mock_get, mock_post):
"""Check that we can get our online status.""" """Check that we can get our online status."""
self.blink.setup_system() self.blink.setup_system()
+5 -11
View File
@@ -4,10 +4,10 @@ skip_missing_interpreters = True
[testenv] [testenv]
setenv = setenv =
LAND=en_US.UTF-8 LANG=en_US.UTF-8
PYTHONPATH = {toxinidir} PYTHONPATH = {toxinidir}
commands = commands =
py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report html {posargs} py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report term {posargs}
deps = deps =
-r{toxinidir}/requirements.txt -r{toxinidir}/requirements.txt
-r{toxinidir}/requirements_test.txt -r{toxinidir}/requirements_test.txt
@@ -19,12 +19,6 @@ deps =
basepython = python3 basepython = python3
ignore_errors = True ignore_errors = True
commands = commands =
;flake8 blinkpy.py tests pylint --rcfile={toxinidir}/pylintrc --load-plugins=pylint.extensions.mccabe blinkpy.py tests
flake8 blinkpy.py flake8 blinkpy.py tests
;pylint blinkpy.py tests pydocstyle blinkpy.py tests
pylint blinkpy.py
;pydocstyle blinkpy.py tests
;[flake8]
;ignore = E501
;exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build