-#!/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')
-
¶