+#!/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 json
+import getpass
+from shutil import copyfileobj
+import requests
+from requests.structures import CaseInsensitiveDict
+import blinkpy.helpers.errors as ERROR
+from blinkpy.helpers.constants import (
+ BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL,
+ DEFAULT_URL, ONLINE
+)
+
+
+[docs]def _attempt_reauthorization(blink):
+
"""Attempt to refresh auth token and links."""
+
headers = blink.get_auth_token()
+
blink.set_links()
+
return headers
+
+
+[docs]def _request(blink, url='http://google.com', data=None, headers=None,
+
reqtype='get', stream=False, json_resp=True, is_retry=False):
+
"""Perform server requests and check if reauthorization neccessary."""
+
if reqtype == 'post':
+
response = requests.post(url, headers=headers,
+
data=data)
+
elif reqtype == 'get':
+
response = requests.get(url, headers=headers,
+
stream=stream)
+
else:
+
raise BlinkException(ERROR.REQUEST)
+
+
if json_resp and 'code' in response.json():
+
if is_retry:
+
raise BlinkAuthenticationException(
+
(response.json()['code'], response.json()['message']))
+
else:
+
headers = _attempt_reauthorization(blink)
+
return _request(blink, url=url, data=data, headers=headers,
+
reqtype=reqtype, stream=stream,
+
json_resp=json_resp, is_retry=True)
+
# pylint: disable=no-else-return
+
if json_resp:
+
return response.json()
+
else:
+
return response
+
+
+# pylint: disable=super-init-not-called
+class BlinkException(Exception):
+ """Class to throw general blink exception."""
+
+ def __init__(self, errcode):
+ """Initialize BlinkException."""
+ self.errid = errcode[0]
+ self.message = errcode[1]
+
+
+class BlinkAuthenticationException(BlinkException):
+ """Class to throw authentication exception."""
+
+ pass
+
+
+[docs]class BlinkURLHandler(object):
+
"""Class that handles Blink URLS."""
+
+
def __init__(self, region_id):
+
"""Initialize the urls."""
+
self.base_url = "https://rest.{}.{}".format(region_id, BLINK_URL)
+
self.home_url = "{}/homescreen".format(self.base_url)
+
self.event_url = "{}/events/network".format(self.base_url)
+
self.network_url = "{}/network".format(self.base_url)
+
self.networks_url = "{}/networks".format(self.base_url)
+
self.video_url = "{}/api/v2/videos".format(self.base_url)
+
+
+[docs]class BlinkCamera(object):
+
"""Class to initialize individual camera."""
+
+
def __init__(self, config, blink):
+
"""Initiailize BlinkCamera."""
+
self.blink = blink
+
self.urls = self.blink.urls
+
self.id = str(config['device_id']) # pylint: disable=invalid-name
+
self.name = config['name']
+
self._status = config['armed']
+
self.thumbnail = "{}{}.jpg".format(self.urls.base_url,
+
config['thumbnail'])
+
self.clip = "{}{}".format(self.urls.base_url, config['video'])
+
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 armed(self):
+
"""Return camera arm status."""
+
return self._status
+
+
@property
+
def battery_string(self):
+
"""Return string indicating battery status."""
+
status = "Unknown"
+
if self.battery > 1 and self.battery <= 3:
+
status = "OK"
+
elif self.battery >= 0:
+
status = "Low"
+
return status
+
+
[docs] def snap_picture(self):
+
"""Take a picture with camera to create a new thumbnail."""
+
_request(self.blink, url=self.image_link,
+
headers=self.header, reqtype='post')
+
+
[docs] def set_motion_detect(self, enable):
+
"""Set motion detection."""
+
url = self.arm_link
+
if enable:
+
_request(self.blink, url=url + 'enable',
+
headers=self.header, reqtype='post')
+
else:
+
_request(self.blink, url=url + 'disable',
+
headers=self.header, reqtype='post')
+
+
[docs] def update(self, values):
+
"""Update camera information."""
+
self.name = values['name']
+
self._status = values['armed']
+
self.thumbnail = "{}{}.jpg".format(
+
self.urls.base_url, values['thumbnail'])
+
self.clip = "{}{}".format(
+
self.urls.base_url, values['video'])
+
self.temperature = values['temp']
+
self.battery = values['battery']
+
self.notifications = values['notifications']
+
+
[docs] def image_refresh(self):
+
"""Refresh current thumbnail."""
+
url = self.urls.home_url
+
response = _request(self.blink, url=url, headers=self.header,
+
reqtype='get')['devices']
+
for element in response:
+
try:
+
if str(element['device_id']) == self.id:
+
self.thumbnail = (
+
"{}{}.jpg".format(
+
self.urls.base_url, element['thumbnail'])
+
)
+
return self.thumbnail
+
except KeyError:
+
pass
+
return None
+
+
[docs] def image_to_file(self, path):
+
"""Write image to file."""
+
thumb = self.image_refresh()
+
response = _request(self.blink, url=thumb, headers=self.header,
+
reqtype='get', stream=True, json_resp=False)
+
if response.status_code == 200:
+
with open(path, 'wb') as imgfile:
+
copyfileobj(response.raw, imgfile)
+
+
[docs] def video_to_file(self, path):
+
"""Write video to file."""
+
response = _request(self.blink, url=self.clip, headers=self.header,
+
reqtype='get', stream=True, json_resp=False)
+
with open(path, 'wb') as vidfile:
+
copyfileobj(response.raw, vidfile)
+
+
+[docs]class Blink(object):
+
"""Class to initialize communication and sync module."""
+
+
def __init__(self, username=None, password=None):
+
"""Initialize Blink system."""
+
self._username = username
+
self._password = password
+
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 = CaseInsensitiveDict({})
+
self._idlookup = {}
+
self.urls = None
+
self._video_count = 0
+
self._all_videos = {}
+
self._summary = None
+
+
@property
+
def camera_thumbs(self):
+
"""Return camera thumbnails."""
+
self.refresh()
+
data = {}
+
for name, camera in self.cameras.items():
+
data[name] = camera.thumbnail
+
+
return data
+
+
@property
+
def id_table(self):
+
"""Return id/camera pairs."""
+
return self._idlookup
+
+
@property
+
def video_count(self):
+
"""Return number of videos on server."""
+
url = "{}/count".format(self.urls.video_url)
+
headers = self._auth_header
+
self._video_count = _request(self, url=url, headers=headers,
+
reqtype='get')['count']
+
return self._video_count
+
+
@property
+
def events(self):
+
"""Get all events on server."""
+
return self._events
+
+
@property
+
def online(self):
+
"""Return boolean system online status."""
+
return ONLINE[self._status_request()['syncmodule']['status']]
+
+
@property
+
def videos(self):
+
"""Return video list."""
+
return self._all_videos
+
+
@property
+
def summary(self):
+
"""Get a full summary of device information."""
+
return self._summary
+
+
@property
+
def arm(self):
+
"""Return status of sync module: armed/disarmed."""
+
return self.summary['network']['armed']
+
+
@arm.setter
+
def arm(self, value):
+
"""Arm or disarm system."""
+
if value:
+
value_to_append = 'arm'
+
else:
+
value_to_append = 'disarm'
+
url = "{}/{}/{}".format(self.urls.network_url,
+
self.network_id,
+
value_to_append)
+
_request(self, url=url, headers=self._auth_header, reqtype='post')
+
+
[docs] def refresh(self):
+
"""Get all blink cameras and pulls their most recent status."""
+
self._summary = self._summary_request()
+
self._events = self._events_request()
+
response = self.summary['devices']
+
self.get_videos()
+
for name in self.cameras:
+
camera = self.cameras[name]
+
for element in response:
+
try:
+
if str(element['device_id']) == camera.id:
+
element['video'] = self.videos[name][0]['clip']
+
element['thumbnail'] = self.videos[name][0]['thumb']
+
camera.update(element)
+
except KeyError:
+
pass
+
return None
+
+
[docs] def get_videos(self, start_page=0, end_page=1):
+
"""Retrieve last recorded videos per camera."""
+
videos = list()
+
for page_num in range(start_page, end_page + 1):
+
this_page = self._video_request(page_num)
+
if not this_page:
+
break
+
videos.append(this_page)
+
+
for page in videos:
+
for entry in page:
+
camera_name = entry['camera_name']
+
clip_addr = entry['address']
+
thumb_addr = entry['thumbnail']
+
try:
+
self._all_videos[camera_name].append(
+
{
+
'clip': clip_addr,
+
'thumb': thumb_addr,
+
}
+
)
+
except KeyError:
+
self._all_videos[camera_name] = [
+
{
+
'clip': clip_addr,
+
'thumb': thumb_addr,
+
}
+
]
+
+
[docs] def get_cameras(self):
+
"""Find and creates cameras."""
+
self._summary = self._summary_request()
+
response = self.summary['devices']
+
for element in response:
+
if ('device_type' in element and
+
element['device_type'] == 'camera'):
+
# Add region to config
+
element['region_id'] = self.region_id
+
try:
+
name = element['name']
+
element['video'] = self.videos[name][0]['clip']
+
element['thumbnail'] = self.videos[name][0]['thumb']
+
except KeyError:
+
element['video'] = None
+
element['thumbnail'] = None
+
device = BlinkCamera(element, self)
+
self.cameras[device.name] = device
+
self._idlookup[device.id] = device.name
+
+
[docs] def set_links(self):
+
"""Set access links and required headers for each camera in system."""
+
for name in self.cameras:
+
camera = self.cameras[name]
+
network_id_url = "{}/{}".format(self.urls.network_url,
+
self.network_id)
+
image_url = "{}/camera/{}/thumbnail".format(network_id_url,
+
camera.id)
+
arm_url = "{}/camera/{}/".format(network_id_url,
+
camera.id)
+
camera.image_link = image_url
+
camera.arm_link = arm_url
+
camera.header = self._auth_header
+
+
[docs] def setup_system(self):
+
"""
+
Perform full system setup.
+
+
Method logs in and sets auth token, urls, and ids for future requests.
+
Essentially this is just a wrapper function for ease of use.
+
"""
+
if self._username is None or self._password is None:
+
raise BlinkAuthenticationException(ERROR.AUTHENTICATE)
+
+
self.get_auth_token()
+
self.get_ids()
+
self.get_videos()
+
if self.video_count > 0:
+
self.get_cameras()
+
self.set_links()
+
self._events = self._events_request()
+
+
[docs] def login(self):
+
"""Prompt user for username and password."""
+
self._username = input("Username:")
+
self._password = getpass.getpass("Password:")
+
+
[docs] def get_auth_token(self):
+
"""Retrieve the authentication token from Blink."""
+
if not isinstance(self._username, str):
+
raise BlinkAuthenticationException(ERROR.USERNAME)
+
if not isinstance(self._password, str):
+
raise BlinkAuthenticationException(ERROR.PASSWORD)
+
+
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"
+
})
+
response = _request(self, url=LOGIN_URL, headers=headers,
+
data=data, json_resp=False, reqtype='post')
+
if response.status_code == 200:
+
response = response.json()
+
(self.region_id, self.region), = response['region'].items()
+
else:
+
response = _request(self, url=LOGIN_BACKUP_URL, headers=headers,
+
data=data, reqtype='post')
+
self.region_id = 'piri'
+
self.region = "UNKNOWN"
+
+
self._host = "{}.{}".format(self.region_id, BLINK_URL)
+
self._token = response['authtoken']['authtoken']
+
+
self._auth_header = {'Host': self._host,
+
'TOKEN_AUTH': self._token}
+
+
self.urls = BlinkURLHandler(self.region_id)
+
+
return self._auth_header
+
+
[docs] def get_ids(self):
+
"""Set the network ID and Account ID."""
+
response = self._network_request()
+
self.network_id = str(response['networks'][0]['id'])
+
self.account_id = str(response['networks'][0]['account_id'])
+
+
def _video_request(self, page=0):
+
"""Perform a request for videos."""
+
url = "{}/page/{}".format(self.urls.video_url, page)
+
headers = self._auth_header
+
return _request(self, url=url, headers=headers, reqtype='get')
+
+
def _summary_request(self):
+
"""Get blink summary."""
+
url = self.urls.home_url
+
headers = self._auth_header
+
if headers is None:
+
raise BlinkException(ERROR.AUTH_TOKEN)
+
return _request(self, url=url, headers=headers, reqtype='get')
+
+
def _network_request(self):
+
"""Get network and account information."""
+
url = self.urls.networks_url
+
headers = self._auth_header
+
if headers is None:
+
raise BlinkException(ERROR.AUTH_TOKEN)
+
return _request(self, url=url, headers=headers, reqtype='get')
+
+
def _events_request(self):
+
"""Get events on server."""
+
url = "{}/{}".format(self.urls.event_url, self.network_id)
+
headers = self._auth_header
+
return _request(self, url=url, headers=headers, reqtype='get')
+
+
def _status_request(self):
+
"""Get syncmodule status."""
+
url = "{}/{}/syncmodules".format(self.urls.network_url,
+
self.network_id)
+
headers = self._auth_header
+
return _request(self, url=url, headers=headers, reqtype='get')
+
¶