Source code for pyarlo.camera
# coding: utf-8
"""Generic Python Class file for Netgear Arlo camera module."""
import logging
from pyarlo.const import (
RESET_CAM_ENDPOINT, STREAM_ENDPOINT, STREAMING_BODY,
SNAPSHOTS_ENDPOINT, SNAPSHOTS_BODY, PRELOAD_DAYS)
from pyarlo.media import ArloMediaLibrary
from pyarlo.utils import http_get
from pyarlo.utils import assert_is_dict
_LOGGER = logging.getLogger(__name__)
[docs]
class ArloCamera(object):
"""Arlo Camera module implementation."""
def __init__(self, name, attrs, arlo_session,
min_days_vdo_cache=PRELOAD_DAYS):
"""Initialize Arlo camera object.
:param name: Camera name
:param attrs: Camera attributes
:param arlo_session: PyArlo shared session
:param min_days_vdo_cache: min. days to preload in video cache
"""
self.name = name
self._attrs = attrs
self._session = arlo_session
self._cached_videos = None
self._min_days_vdo_cache = min_days_vdo_cache
# make sure self._attrs is a dict
self._attrs = assert_is_dict(self._attrs)
def __repr__(self):
"""Representation string of object."""
return "<{0}: {1}>".format(self.__class__.__name__, self.name)
@property
def attrs(self):
"""Return device attributes."""
return self._attrs
@attrs.setter
def attrs(self, value):
"""Override device attributes."""
self._attrs = value
@property
def min_days_vdo_cache(self):
"""Return minimal days to lookup when building the video cache."""
return self._min_days_vdo_cache
@min_days_vdo_cache.setter
def min_days_vdo_cache(self, value):
"""Set minimal days to lookup when building the video cache."""
self._min_days_vdo_cache = value
# pylint: disable=invalid-name
@property
def device_id(self):
"""Return device_id."""
if self._attrs is not None:
return self._attrs.get('deviceId')
return None
@property
def serial_number(self):
"""Return serial_number."""
return self.device_id
@property
def device_type(self):
"""Return device_type."""
if self._attrs is not None:
return self._attrs.get('deviceType')
return None
@property
def model_id(self):
"""Return model_id."""
if self._attrs is not None:
return self._attrs.get('modelId')
return None
@property
def hw_version(self):
"""Return hardware version."""
if self._attrs is not None:
return self._attrs.get('properties').get('hwVersion')
return None
@property
def parent_id(self):
"""Return camera parentID."""
if self._attrs is not None:
return self._attrs.get('parentId')
return None
@property
def timezone(self):
"""Return timezone."""
if self._attrs is not None:
return self._attrs.get('properties').get('olsonTimeZone')
return None
@property
def unique_id(self):
"""Return unique_id."""
if self._attrs is not None:
return self._attrs.get('uniqueId')
return None
@property
def user_id(self):
"""Return userID."""
if self._attrs is not None:
return self._attrs.get('userId')
return None
@property
def unseen_videos(self):
"""Return number of unseen videos."""
if self._attrs is not None:
return self._attrs.get('mediaObjectCount')
return None
[docs]
def unseen_videos_reset(self):
"""Reset the unseen videos counter."""
url = RESET_CAM_ENDPOINT.format(self.unique_id)
ret = self._session.query(url).get('success')
return ret
@property
def user_role(self):
"""Return userRole."""
if self._attrs is not None:
return self._attrs.get('userRole')
return None
@property
def last_image(self):
"""Return last image captured by camera."""
if self._attrs is not None:
return http_get(self._attrs.get('presignedLastImageUrl'))
return None
@property
def last_image_from_cache(self):
"""
Return last thumbnail present in self._cached_images.
This is useful in Home Assistant when the ArloHub has not
updated all information, but the camera.arlo already pulled
the last image. Using this method, everything is kept synced.
"""
if self.last_video:
return http_get(self.last_video.thumbnail_url)
return None
@property
def last_video(self):
"""Return the last <ArloVideo> object from camera."""
if self._cached_videos is None:
self.make_video_cache()
if self._cached_videos:
return self._cached_videos[0]
return None
[docs]
def make_video_cache(self, days=None):
"""Save videos on _cache_videos to avoid dups."""
if days is None:
days = self._min_days_vdo_cache
self._cached_videos = self.videos(days)
[docs]
def videos(self, days=None):
"""
Return all <ArloVideo> objects from camera given days range
:param days: number of days to retrieve
"""
if days is None:
days = self._min_days_vdo_cache
library = ArloMediaLibrary(self._session, preload=False)
try:
return library.load(only_cameras=[self], days=days)
except (AttributeError, IndexError):
# make sure we are returning an empty list istead of None
# returning an empty list, cache will be forced only when calling
# the update method. Changing this can impact badly
# in the Home Assistant performance
return []
@property
def captured_today(self):
"""Return list of <ArloVideo> object captured today."""
if self._cached_videos is None:
self.make_video_cache()
return [vdo for vdo in self._cached_videos if vdo.created_today]
[docs]
def play_last_video(self):
"""Play last <ArloVideo> recorded from camera."""
video = self.last_video
return video.download_video()
@property
def xcloud_id(self):
"""Return X-Cloud-ID attribute."""
if self._attrs is not None:
return self._attrs.get('xCloudId')
return None
@property
def base_station(self):
"""Return the base_station assigned for the given camera."""
try:
return list(filter(lambda x: x.device_id == self.parent_id,
self._session.base_stations))[0]
except (IndexError, AttributeError):
return None
def _get_camera_properties(self):
"""Lookup camera properties from base station."""
if self.base_station and self.base_station.camera_properties:
for cam in self.base_station.camera_properties:
if cam["serialNumber"] == self.device_id:
return cam
return None
@property
def properties(self):
"""Get this camera's properties."""
return self._get_camera_properties()
@property
def capabilities(self):
"""Get a camera's capabilities."""
properties = self.properties
return properties.get("capabilities") if properties else None
@property
def triggers(self):
"""Get a camera's triggers."""
capabilities = self.capabilities
if not capabilities:
return None
for capability in capabilities:
if not isinstance(capability, dict):
continue
triggers = capability.get("Triggers")
if triggers:
return triggers
return None
@property
def battery_level(self):
"""Get the camera battery level."""
properties = self.properties
return properties.get("batteryLevel") if properties else None
@property
def signal_strength(self):
"""Get the camera Signal strength."""
properties = self.properties
return properties.get("signalStrength") if properties else None
@property
def brightness(self):
"""Get the brightness property of camera."""
properties = self.properties
return properties.get("brightness") if properties else None
@property
def mirror_state(self):
"""Get the mirror state of camera image."""
properties = self.properties
return properties.get("mirror") if properties else None
@property
def flip_state(self):
"""Get the flipped state of camera image."""
properties = self.properties
return properties.get("flip") if properties else None
@property
def powersave_mode(self):
"""Get the power mode (stream quality) of camera."""
properties = self.properties
return properties.get("powerSaveMode") if properties else None
@property
def is_camera_connected(self):
"""Connectivity status of Cam with Base Station."""
properties = self.properties
return bool(properties.get("connectionState") == "available") \
if properties else None
@property
def motion_detection_sensitivity(self):
"""Sensitivity level of Camera motion detection."""
if not self.triggers:
return None
for trigger in self.triggers:
if trigger.get("type") != "pirMotionActive":
continue
sensitivity = trigger.get("sensitivity")
if sensitivity:
return sensitivity.get("default")
return None
@property
def audio_detection_sensitivity(self):
"""Sensitivity level of Camera audio detection."""
if not self.triggers:
return None
for trigger in self.triggers:
if trigger.get("type") != "audioAmplitude":
continue
sensitivity = trigger.get("sensitivity")
if sensitivity:
return sensitivity.get("default")
return None
[docs]
def live_streaming(self):
"""Return live streaming generator."""
url = STREAM_ENDPOINT
# override params
params = STREAMING_BODY
params['from'] = "{0}_web".format(self.user_id)
params['to'] = self.device_id
params['resource'] = "cameras/{0}".format(self.device_id)
params['transId'] = "web!{0}".format(self.xcloud_id)
# override headers
headers = {'xCloudId': self.xcloud_id}
_LOGGER.debug("Streaming device %s", self.name)
_LOGGER.debug("Device params %s", params)
_LOGGER.debug("Device headers %s", headers)
ret = self._session.query(url,
method='POST',
extra_params=params,
extra_headers=headers)
_LOGGER.debug("Streaming results %s", ret)
if ret.get('success'):
return ret.get('data').get('url')
return ret.get('data')
@property
def snapshot_url(self):
"""Return the snapshot url."""
# Snapshot should be scheduled first. It will
# available a couple seconds after.
# If a GET request fails on this URL, trying
# again is logical since the snapshot isn't
# taken immediately. Snapshots will be cached for a
# predefined amount of time.
return self._attrs.get('presignedFullFrameSnapshotUrl')
[docs]
def schedule_snapshot(self):
"""Trigger snapshot to be uploaded to AWS.
Return success state."""
# Notes:
# - Snapshots are not immediate.
# - Snapshots will be cached for predefined amount
# of time.
# - Snapshots are not balanced. To get a better
# image, it must be taken from the stream, a few
# seconds after stream start.
url = SNAPSHOTS_ENDPOINT
params = SNAPSHOTS_BODY
params['from'] = "{0}_web".format(self.user_id)
params['to'] = self.device_id
params['resource'] = "cameras/{0}".format(self.device_id)
params['transId'] = "web!{0}".format(self.xcloud_id)
# override headers
headers = {'xCloudId': self.xcloud_id}
_LOGGER.debug("Snapshot device %s", self.name)
_LOGGER.debug("Device params %s", params)
_LOGGER.debug("Device headers %s", headers)
ret = self._session.query(url,
method='POST',
extra_params=params,
extra_headers=headers)
_LOGGER.debug("Snapshot results %s", ret)
return ret is not None and ret.get('success')
[docs]
def update(self):
"""Update object properties."""
self._attrs = self._session.refresh_attributes(self.name)
self._attrs = assert_is_dict(self._attrs)
# force base_state to update properties
if self.base_station:
self.base_station.update()
# vim:sw=4:ts=4:et: