diff --git a/docs/source/index.rst b/docs/source/index.rst index 6e016c5a..70c533f5 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -134,17 +134,23 @@ Developing :undoc-members: :show-inheritance: -.. autoclass:: ring_doorbell.RingGeneric +.. autoclass:: ring_doorbell.generic.RingGeneric :members: :undoc-members: :show-inheritance: -.. autoclass:: ring_doorbell.RingChime +.. autoclass:: ring_doorbell.chime.RingChime :members: :undoc-members: :show-inheritance: -.. autoclass:: ring_doorbell.RingDoorBell +.. autoclass:: ring_doorbell.doorbot.RingDoorBell + :members: + :undoc-members: + :show-inheritance: + :inherited-members: + +.. autoclass:: ring_doorbell.stickup_cam.RingStickUpCam :members: :undoc-members: :show-inheritance: diff --git a/ring_doorbell/__init__.py b/ring_doorbell/__init__.py index dfa6af0c..fe37317a 100644 --- a/ring_doorbell/__init__.py +++ b/ring_doorbell/__init__.py @@ -1,31 +1,24 @@ # coding: utf-8 # vim:sw=4:ts=4:et: """Python Ring Doorbell wrapper.""" -from datetime import datetime - try: from urllib.parse import urlencode except ImportError: from urllib import urlencode -import os import logging import requests -import pytz -from ring_doorbell.utils import ( - _locator, _exists_cache, _save_cache, _read_cache) +from ring_doorbell.utils import _exists_cache, _save_cache, _read_cache + from ring_doorbell.const import ( - API_VERSION, API_URI, CACHE_ATTRS, CACHE_FILE, CHIMES_ENDPOINT, - CHIME_VOL_MIN, CHIME_VOL_MAX, - DEVICES_ENDPOINT, DOORBELLS_ENDPOINT, DOORBELL_VOL_MIN, DOORBELL_VOL_MAX, - DOORBELL_EXISTING_TYPE, DINGS_ENDPOINT, FILE_EXISTS, - HEADERS, HEALTH_CHIMES_ENDPOINT, HEALTH_DOORBELL_ENDPOINT, - LINKED_CHIMES_ENDPOINT, LIVE_STREAMING_ENDPOINT, NEW_SESSION_ENDPOINT, - MSG_BOOLEAN_REQUIRED, MSG_EXISTING_TYPE, MSG_GENERIC_FAIL, - MSG_VOL_OUTBOUND, NOT_FOUND, URL_DOORBELL_HISTORY, URL_RECORDING, - POST_DATA, PERSIST_TOKEN_ENDPOINT, PERSIST_TOKEN_DATA, - RETRY_TOKEN, TESTSOUND_CHIME_ENDPOINT, CHIME_TEST_SOUND_KINDS, KIND_DING) + API_VERSION, API_URI, CACHE_ATTRS, CACHE_FILE, + DEVICES_ENDPOINT, HEADERS, NEW_SESSION_ENDPOINT, MSG_GENERIC_FAIL, + POST_DATA, PERSIST_TOKEN_ENDPOINT, PERSIST_TOKEN_DATA, RETRY_TOKEN) + +from ring_doorbell.doorbot import RingDoorBell +from ring_doorbell.chime import RingChime +from ring_doorbell.stickup_cam import RingStickUpCam _LOGGER = logging.getLogger(__name__) @@ -254,454 +247,3 @@ def stickup_cams(self): def doorbells(self): """Return a list of RingDoorBell objects.""" return self.__devices('doorbells') - - -class RingGeneric(object): - """Generic Implementation for Ring Chime/Doorbell.""" - - def __init__(self, ring, name, shared=False): - """Initialize Ring Generic.""" - self._ring = ring - self.debug = self._ring.debug - self.name = name - self.shared = shared - self._attrs = None - self._health_attrs = None - - # alerts notifications - self.alert_expires_at = None - - # force update - self.update() - - def __repr__(self): - """Return __repr__.""" - return "<{0}: {1}>".format(self.__class__.__name__, self.name) - - @property - def family(self): - """Return Ring device family type.""" - return None - - def update(self): - """Refresh attributes.""" - self._get_attrs() - self._get_health_attrs() - self._update_alert() - - @property - def alert(self): - """Return alert attribute.""" - return self._ring.cache['alerts'] - - @alert.setter - def alert(self, value): - """Set attribute to alert.""" - self._ring.cache['alerts'] = value - _save_cache(self._ring.cache, self._ring.cache_file) - return True - - def _update_alert(self): - """Verify if alert received is still valid.""" - # alert is no longer valid - if self.alert and self.alert_expires_at: - if datetime.now() >= self.alert_expires_at: - self.alert = None - self.alert_expires_at = None - _save_cache(self._ring.cache, self._ring.cache_file) - - def _get_attrs(self): - """Return attributes.""" - url = API_URI + DEVICES_ENDPOINT - try: - if self.family == 'doorbots' and self.shared: - lst = self._ring.query(url).get('authorized_doorbots') - else: - lst = self._ring.query(url).get(self.family) - index = _locator(lst, 'description', self.name) - if index == NOT_FOUND: - return None - except AttributeError: - return None - - self._attrs = lst[index] - return True - - def _get_health_attrs(self): - """Return health attributes.""" - if self.family == 'doorbots' or self.family == 'stickup_cams': - url = API_URI + HEALTH_DOORBELL_ENDPOINT.format(self.account_id) - elif self.family == 'chimes': - url = API_URI + HEALTH_CHIMES_ENDPOINT.format(self.account_id) - self._health_attrs = self._ring.query(url).get('device_health') - - @property - def account_id(self): - """Return account ID.""" - return self._attrs.get('id') - - @property - def address(self): - """Return address.""" - return self._attrs.get('address') - - @property - def firmware(self): - """Return firmware.""" - return self._attrs.get('firmware_version') - - # pylint: disable=invalid-name - @property - def id(self): - """Return ID.""" - return self._attrs.get('device_id') - - @property - def latitude(self): - """Return latitude attr.""" - return self._attrs.get('latitude') - - @property - def longitude(self): - """Return longitude attr.""" - return self._attrs.get('longitude') - - @property - def kind(self): - """Return kind attr.""" - return self._attrs.get('kind') - - @property - def timezone(self): - """Return timezone.""" - return self._attrs.get('time_zone') - - @property - def wifi_name(self): - """Return wifi ESSID name.""" - return self._health_attrs.get('wifi_name') - - @property - def wifi_signal_strength(self): - """Return wifi RSSI.""" - return self._health_attrs.get('latest_signal_strength') - - @property - def wifi_signal_category(self): - """Return wifi signal category.""" - return self._health_attrs.get('latest_signal_category') - - -class RingChime(RingGeneric): - """Implementation for Ring Chime.""" - - @property - def family(self): - """Return Ring device family type.""" - return 'chimes' - - @property - def battery_life(self): - """Return battery life.""" - return self._health_attrs.get('battery_percentage') - - @property - def volume(self): - """Return if chime volume.""" - return self._attrs.get('settings').get('volume') - - @volume.setter - def volume(self, value): - if not ((isinstance(value, int)) and - (value >= CHIME_VOL_MIN and value <= CHIME_VOL_MAX)): - _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(CHIME_VOL_MIN, - CHIME_VOL_MAX)) - return False - - params = { - 'chime[description]': self.name, - 'chime[settings][volume]': str(value)} - url = API_URI + CHIMES_ENDPOINT.format(self.account_id) - self._ring.query(url, extra_params=params, method='PUT') - self.update() - return True - - @property - def linked_tree(self): - """Return doorbell data linked to chime.""" - url = API_URI + LINKED_CHIMES_ENDPOINT.format(self.account_id) - return self._ring.query(url) - - def test_sound(self, kind=KIND_DING): - """Play chime to test sound.""" - if kind not in CHIME_TEST_SOUND_KINDS: - return False - url = API_URI + TESTSOUND_CHIME_ENDPOINT.format(self.account_id) - self._ring.query(url, method='POST', extra_params={"kind": kind}) - return True - - -class RingDoorBell(RingGeneric): - """Implementation for Ring Doorbell.""" - - @property - def family(self): - """Return Ring device family type.""" - return 'doorbots' - - @property - def battery_life(self): - """Return battery life.""" - value = int(self._attrs.get('battery_life')) - if value > 100: - value = 100 - return value - - def check_alerts(self): - """Return JSON when motion or ring is detected.""" - url = API_URI + DINGS_ENDPOINT - self.update() - - try: - resp = self._ring.query(url)[0] - except (IndexError, TypeError): - return None - - if resp: - timestamp = resp.get('now') + resp.get('expires_in') - self.alert = resp - self.alert_expires_at = datetime.fromtimestamp(timestamp) - - # save to a pickle data - if self.alert: - _save_cache(self._ring.cache, self._ring.cache_file) - return True - return None - - @property - def existing_doorbell_type(self): - """ - Return existing doorbell type. - - 0: Mechanical - 1: Digital - 2: Not Present - """ - try: - return DOORBELL_EXISTING_TYPE[ - self._attrs.get('settings').get('chime_settings').get('type')] - except AttributeError: - return None - - @existing_doorbell_type.setter - def existing_doorbell_type(self, value): - """ - Return existing doorbell type. - - 0: Mechanical - 1: Digital - 2: Not Present - """ - if value not in DOORBELL_EXISTING_TYPE.keys(): - _LOGGER.error("%s", MSG_EXISTING_TYPE) - return False - params = { - 'doorbot[description]': self.name, - 'doorbot[settings][chime_settings][type]': value} - if self.existing_doorbell_type: - url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) - self._ring.query(url, extra_params=params, method='PUT') - self.update() - return True - return None - - @property - def existing_doorbell_type_enabled(self): - """Return if existing doorbell type is enabled.""" - if self.existing_doorbell_type: - if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[2]: - return None - return \ - self._attrs.get('settings').get('chime_settings').get('enable') - return False - - @existing_doorbell_type_enabled.setter - def existing_doorbell_type_enabled(self, value): - """Enable/disable the existing doorbell if Digital/Mechanical.""" - if self.existing_doorbell_type: - - if not isinstance(value, bool): - _LOGGER.error("%s", MSG_BOOLEAN_REQUIRED) - return None - - if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[2]: - return None - - params = { - 'doorbot[description]': self.name, - 'doorbot[settings][chime_settings][enable]': value} - url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) - self._ring.query(url, extra_params=params, method='PUT') - self.update() - return True - return False - - @property - def existing_doorbell_type_duration(self): - """Return duration for Digital chime.""" - if self.existing_doorbell_type: - if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[1]: - return self._attrs.get('settings').\ - get('chime_settings').get('duration') - return None - - @existing_doorbell_type_duration.setter - def existing_doorbell_type_duration(self, value): - """Set duration for Digital chime.""" - if self.existing_doorbell_type: - - if not ((isinstance(value, int)) and - (value >= DOORBELL_VOL_MIN and value <= DOORBELL_VOL_MAX)): - _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(DOORBELL_VOL_MIN, - DOORBELL_VOL_MAX)) - return False - - if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[1]: - params = { - 'doorbot[description]': self.name, - 'doorbot[settings][chime_settings][duration]': value} - url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) - self._ring.query(url, extra_params=params, method='PUT') - self.update() - return True - return None - - def history(self, limit=30, timezone=None, kind=None): - """Return history with datetime objects.""" - # allow modify the items to return - params = {'limit': str(limit)} - - url = API_URI + URL_DOORBELL_HISTORY.format(self.account_id) - response = self._ring.query(url, extra_params=params) - - # convert for specific timezone - utc = pytz.utc - if timezone: - mytz = pytz.timezone(timezone) - - for entry in response: - dt_at = datetime.strptime(entry['created_at'], - '%Y-%m-%dT%H:%M:%S.000Z') - utc_dt = datetime(dt_at.year, dt_at.month, dt_at.day, dt_at.hour, - dt_at.minute, dt_at.second, tzinfo=utc) - if timezone: - tz_dt = utc_dt.astimezone(mytz) - entry['created_at'] = tz_dt - else: - entry['created_at'] = utc_dt - - if kind: - return list(filter(lambda array: array['kind'] == kind, response)) - - return response - - @property - def last_recording_id(self): - """Return the last recording ID.""" - try: - return self.history(limit=1)[0]['id'] - except (IndexError, TypeError): - return None - - @property - def live_streaming_json(self): - """Return JSON for live streaming.""" - url = API_URI + LIVE_STREAMING_ENDPOINT.format(self.account_id) - req = self._ring.query((url), method='POST', raw=True) - if req.status_code == 204: - url = API_URI + DINGS_ENDPOINT - try: - return self._ring.query(url)[0] - except (IndexError, TypeError): - pass - return None - - def recording_download(self, recording_id, filename=None, override=False): - """Save a recording in MP4 format to a file or return raw.""" - url = API_URI + URL_RECORDING.format(recording_id) - try: - req = self._ring.query(url, raw=True) - if req.status_code == 200: - - if filename: - if os.path.isfile(filename) and not override: - _LOGGER.error("%s", FILE_EXISTS.format(filename)) - return False - - with open(filename, 'wb') as recording: - recording.write(req.content) - return True - else: - return req.content - except IOError as error: - _LOGGER.error("%s", error) - raise - - def recording_url(self, recording_id): - """Return HTTPS recording URL.""" - url = API_URI + URL_RECORDING.format(recording_id) - req = self._ring.query(url, raw=True) - if req.status_code == 200: - return req.url - return False - - @property - def subscribed(self): - """Return if is online.""" - result = self._attrs.get('subscribed') - if result is None: - return False - return True - - @property - def subscribed_motion(self): - """Return if is subscribed_motion.""" - result = self._attrs.get('subscribed_motions') - if result is None: - return False - return True - - @property - def volume(self): - """Return volume.""" - return self._attrs.get('settings').get('doorbell_volume') - - @volume.setter - def volume(self, value): - if not ((isinstance(value, int)) and - (value >= DOORBELL_VOL_MIN and value <= DOORBELL_VOL_MAX)): - _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(DOORBELL_VOL_MIN, - DOORBELL_VOL_MAX)) - return False - - params = { - 'doorbot[description]': self.name, - 'doorbot[settings][doorbell_volume]': str(value)} - url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) - self._ring.query(url, extra_params=params, method='PUT') - self.update() - return True - - @property - def connection_status(self): - """Return connection status.""" - return self._attrs.get('alerts').get('connection') - - -class RingStickUpCam(RingDoorBell): - """Implementation for Ring RingStickUpCam.""" - - @property - def family(self): - """Return Ring device family type.""" - return 'stickup_cams' diff --git a/ring_doorbell/chime.py b/ring_doorbell/chime.py new file mode 100644 index 00000000..9fa38b23 --- /dev/null +++ b/ring_doorbell/chime.py @@ -0,0 +1,61 @@ +# coding: utf-8 +# vim:sw=4:ts=4:et: +"""Python Ring Chime wrapper.""" +import logging + +from ring_doorbell.generic import RingGeneric +from ring_doorbell.const import ( + API_URI, CHIMES_ENDPOINT, CHIME_VOL_MIN, CHIME_VOL_MAX, + LINKED_CHIMES_ENDPOINT, MSG_VOL_OUTBOUND, TESTSOUND_CHIME_ENDPOINT, + CHIME_TEST_SOUND_KINDS, KIND_DING) + +_LOGGER = logging.getLogger(__name__) + + +class RingChime(RingGeneric): + """Implementation for Ring Chime.""" + + @property + def family(self): + """Return Ring device family type.""" + return 'chimes' + + @property + def battery_life(self): + """Return battery life.""" + return self._health_attrs.get('battery_percentage') + + @property + def volume(self): + """Return if chime volume.""" + return self._attrs.get('settings').get('volume') + + @volume.setter + def volume(self, value): + if not ((isinstance(value, int)) and + (value >= CHIME_VOL_MIN and value <= CHIME_VOL_MAX)): + _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(CHIME_VOL_MIN, + CHIME_VOL_MAX)) + return False + + params = { + 'chime[description]': self.name, + 'chime[settings][volume]': str(value)} + url = API_URI + CHIMES_ENDPOINT.format(self.account_id) + self._ring.query(url, extra_params=params, method='PUT') + self.update() + return True + + @property + def linked_tree(self): + """Return doorbell data linked to chime.""" + url = API_URI + LINKED_CHIMES_ENDPOINT.format(self.account_id) + return self._ring.query(url) + + def test_sound(self, kind=KIND_DING): + """Play chime to test sound.""" + if kind not in CHIME_TEST_SOUND_KINDS: + return False + url = API_URI + TESTSOUND_CHIME_ENDPOINT.format(self.account_id) + self._ring.query(url, method='POST', extra_params={"kind": kind}) + return True diff --git a/ring_doorbell/doorbot.py b/ring_doorbell/doorbot.py new file mode 100644 index 00000000..9231029d --- /dev/null +++ b/ring_doorbell/doorbot.py @@ -0,0 +1,276 @@ +# coding: utf-8 +# vim:sw=4:ts=4:et: +"""Python Ring Doorbell wrapper.""" +import logging +from datetime import datetime +import os +import pytz + + +from ring_doorbell.generic import RingGeneric + +from ring_doorbell.utils import _save_cache +from ring_doorbell.const import ( + API_URI, DOORBELLS_ENDPOINT, DOORBELL_VOL_MIN, DOORBELL_VOL_MAX, + DOORBELL_EXISTING_TYPE, DINGS_ENDPOINT, FILE_EXISTS, + LIVE_STREAMING_ENDPOINT, MSG_BOOLEAN_REQUIRED, MSG_EXISTING_TYPE, + MSG_VOL_OUTBOUND, URL_DOORBELL_HISTORY, URL_RECORDING) + +_LOGGER = logging.getLogger(__name__) + + +class RingDoorBell(RingGeneric): + """Implementation for Ring Doorbell.""" + + @property + def family(self): + """Return Ring device family type.""" + return 'doorbots' + + @property + def battery_life(self): + """Return battery life.""" + value = int(self._attrs.get('battery_life')) + if value > 100: + value = 100 + return value + + def check_alerts(self): + """Return JSON when motion or ring is detected.""" + url = API_URI + DINGS_ENDPOINT + self.update() + + try: + resp = self._ring.query(url)[0] + except (IndexError, TypeError): + return None + + if resp: + timestamp = resp.get('now') + resp.get('expires_in') + self.alert = resp + self.alert_expires_at = datetime.fromtimestamp(timestamp) + + # save to a pickle data + if self.alert: + _save_cache(self._ring.cache, self._ring.cache_file) + return True + return None + + @property + def existing_doorbell_type(self): + """ + Return existing doorbell type. + + 0: Mechanical + 1: Digital + 2: Not Present + """ + try: + return DOORBELL_EXISTING_TYPE[ + self._attrs.get('settings').get('chime_settings').get('type')] + except AttributeError: + return None + + @existing_doorbell_type.setter + def existing_doorbell_type(self, value): + """ + Return existing doorbell type. + + 0: Mechanical + 1: Digital + 2: Not Present + """ + if value not in DOORBELL_EXISTING_TYPE.keys(): + _LOGGER.error("%s", MSG_EXISTING_TYPE) + return False + params = { + 'doorbot[description]': self.name, + 'doorbot[settings][chime_settings][type]': value} + if self.existing_doorbell_type: + url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) + self._ring.query(url, extra_params=params, method='PUT') + self.update() + return True + return None + + @property + def existing_doorbell_type_enabled(self): + """Return if existing doorbell type is enabled.""" + if self.existing_doorbell_type: + if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[2]: + return None + return \ + self._attrs.get('settings').get('chime_settings').get('enable') + return False + + @existing_doorbell_type_enabled.setter + def existing_doorbell_type_enabled(self, value): + """Enable/disable the existing doorbell if Digital/Mechanical.""" + if self.existing_doorbell_type: + + if not isinstance(value, bool): + _LOGGER.error("%s", MSG_BOOLEAN_REQUIRED) + return None + + if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[2]: + return None + + params = { + 'doorbot[description]': self.name, + 'doorbot[settings][chime_settings][enable]': value} + url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) + self._ring.query(url, extra_params=params, method='PUT') + self.update() + return True + return False + + @property + def existing_doorbell_type_duration(self): + """Return duration for Digital chime.""" + if self.existing_doorbell_type: + if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[1]: + return self._attrs.get('settings').\ + get('chime_settings').get('duration') + return None + + @existing_doorbell_type_duration.setter + def existing_doorbell_type_duration(self, value): + """Set duration for Digital chime.""" + if self.existing_doorbell_type: + + if not ((isinstance(value, int)) and + (value >= DOORBELL_VOL_MIN and value <= DOORBELL_VOL_MAX)): + _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(DOORBELL_VOL_MIN, + DOORBELL_VOL_MAX)) + return False + + if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[1]: + params = { + 'doorbot[description]': self.name, + 'doorbot[settings][chime_settings][duration]': value} + url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) + self._ring.query(url, extra_params=params, method='PUT') + self.update() + return True + return None + + def history(self, limit=30, timezone=None, kind=None): + """Return history with datetime objects.""" + # allow modify the items to return + params = {'limit': str(limit)} + + url = API_URI + URL_DOORBELL_HISTORY.format(self.account_id) + response = self._ring.query(url, extra_params=params) + + # convert for specific timezone + utc = pytz.utc + if timezone: + mytz = pytz.timezone(timezone) + + for entry in response: + dt_at = datetime.strptime(entry['created_at'], + '%Y-%m-%dT%H:%M:%S.000Z') + utc_dt = datetime(dt_at.year, dt_at.month, dt_at.day, dt_at.hour, + dt_at.minute, dt_at.second, tzinfo=utc) + if timezone: + tz_dt = utc_dt.astimezone(mytz) + entry['created_at'] = tz_dt + else: + entry['created_at'] = utc_dt + + if kind: + return list(filter(lambda array: array['kind'] == kind, response)) + + return response + + @property + def last_recording_id(self): + """Return the last recording ID.""" + try: + return self.history(limit=1)[0]['id'] + except (IndexError, TypeError): + return None + + @property + def live_streaming_json(self): + """Return JSON for live streaming.""" + url = API_URI + LIVE_STREAMING_ENDPOINT.format(self.account_id) + req = self._ring.query((url), method='POST', raw=True) + if req.status_code == 204: + url = API_URI + DINGS_ENDPOINT + try: + return self._ring.query(url)[0] + except (IndexError, TypeError): + pass + return None + + def recording_download(self, recording_id, filename=None, override=False): + """Save a recording in MP4 format to a file or return raw.""" + url = API_URI + URL_RECORDING.format(recording_id) + try: + req = self._ring.query(url, raw=True) + if req.status_code == 200: + + if filename: + if os.path.isfile(filename) and not override: + _LOGGER.error("%s", FILE_EXISTS.format(filename)) + return False + + with open(filename, 'wb') as recording: + recording.write(req.content) + return True + else: + return req.content + except IOError as error: + _LOGGER.error("%s", error) + raise + + def recording_url(self, recording_id): + """Return HTTPS recording URL.""" + url = API_URI + URL_RECORDING.format(recording_id) + req = self._ring.query(url, raw=True) + if req.status_code == 200: + return req.url + return False + + @property + def subscribed(self): + """Return if is online.""" + result = self._attrs.get('subscribed') + if result is None: + return False + return True + + @property + def subscribed_motion(self): + """Return if is subscribed_motion.""" + result = self._attrs.get('subscribed_motions') + if result is None: + return False + return True + + @property + def volume(self): + """Return volume.""" + return self._attrs.get('settings').get('doorbell_volume') + + @volume.setter + def volume(self, value): + if not ((isinstance(value, int)) and + (value >= DOORBELL_VOL_MIN and value <= DOORBELL_VOL_MAX)): + _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(DOORBELL_VOL_MIN, + DOORBELL_VOL_MAX)) + return False + + params = { + 'doorbot[description]': self.name, + 'doorbot[settings][doorbell_volume]': str(value)} + url = API_URI + DOORBELLS_ENDPOINT.format(self.account_id) + self._ring.query(url, extra_params=params, method='PUT') + self.update() + return True + + @property + def connection_status(self): + """Return connection status.""" + return self._attrs.get('alerts').get('connection') diff --git a/ring_doorbell/generic.py b/ring_doorbell/generic.py new file mode 100644 index 00000000..edcfb5c6 --- /dev/null +++ b/ring_doorbell/generic.py @@ -0,0 +1,148 @@ +# coding: utf-8 +# vim:sw=4:ts=4:et: +"""Python Ring RingGeneric wrapper.""" +import logging +from datetime import datetime + +from ring_doorbell.utils import _locator, _save_cache +from ring_doorbell.const import ( + API_URI, DEVICES_ENDPOINT, NOT_FOUND, + HEALTH_CHIMES_ENDPOINT, HEALTH_DOORBELL_ENDPOINT) + +_LOGGER = logging.getLogger(__name__) + + +class RingGeneric(object): + """Generic Implementation for Ring Chime/Doorbell.""" + + def __init__(self, ring, name, shared=False): + """Initialize Ring Generic.""" + self._ring = ring + self.debug = self._ring.debug + self.name = name + self.shared = shared + self._attrs = None + self._health_attrs = None + + # alerts notifications + self.alert_expires_at = None + + # force update + self.update() + + def __repr__(self): + """Return __repr__.""" + return "<{0}: {1}>".format(self.__class__.__name__, self.name) + + @property + def family(self): + """Return Ring device family type.""" + return None + + def update(self): + """Refresh attributes.""" + self._get_attrs() + self._get_health_attrs() + self._update_alert() + + @property + def alert(self): + """Return alert attribute.""" + return self._ring.cache['alerts'] + + @alert.setter + def alert(self, value): + """Set attribute to alert.""" + self._ring.cache['alerts'] = value + _save_cache(self._ring.cache, self._ring.cache_file) + return True + + def _update_alert(self): + """Verify if alert received is still valid.""" + # alert is no longer valid + if self.alert and self.alert_expires_at: + if datetime.now() >= self.alert_expires_at: + self.alert = None + self.alert_expires_at = None + _save_cache(self._ring.cache, self._ring.cache_file) + + def _get_attrs(self): + """Return attributes.""" + url = API_URI + DEVICES_ENDPOINT + try: + if self.family == 'doorbots' and self.shared: + lst = self._ring.query(url).get('authorized_doorbots') + else: + lst = self._ring.query(url).get(self.family) + index = _locator(lst, 'description', self.name) + if index == NOT_FOUND: + return None + except AttributeError: + return None + + self._attrs = lst[index] + return True + + def _get_health_attrs(self): + """Return health attributes.""" + if self.family == 'doorbots' or self.family == 'stickup_cams': + url = API_URI + HEALTH_DOORBELL_ENDPOINT.format(self.account_id) + elif self.family == 'chimes': + url = API_URI + HEALTH_CHIMES_ENDPOINT.format(self.account_id) + self._health_attrs = self._ring.query(url).get('device_health') + + @property + def account_id(self): + """Return account ID.""" + return self._attrs.get('id') + + @property + def address(self): + """Return address.""" + return self._attrs.get('address') + + @property + def firmware(self): + """Return firmware.""" + return self._attrs.get('firmware_version') + + # pylint: disable=invalid-name + @property + def id(self): + """Return ID.""" + return self._attrs.get('device_id') + + @property + def latitude(self): + """Return latitude attr.""" + return self._attrs.get('latitude') + + @property + def longitude(self): + """Return longitude attr.""" + return self._attrs.get('longitude') + + @property + def kind(self): + """Return kind attr.""" + return self._attrs.get('kind') + + @property + def timezone(self): + """Return timezone.""" + return self._attrs.get('time_zone') + + @property + def wifi_name(self): + """Return wifi ESSID name.""" + return self._health_attrs.get('wifi_name') + + @property + def wifi_signal_strength(self): + """Return wifi RSSI.""" + return self._health_attrs.get('latest_signal_strength') + + @property + def wifi_signal_category(self): + """Return wifi signal category.""" + return self._health_attrs.get('latest_signal_category') diff --git a/ring_doorbell/stickup_cam.py b/ring_doorbell/stickup_cam.py new file mode 100644 index 00000000..7ad45c5b --- /dev/null +++ b/ring_doorbell/stickup_cam.py @@ -0,0 +1,17 @@ +# coding: utf-8 +# vim:sw=4:ts=4:et: +"""Python Ring Doorbell wrapper.""" +import logging + +from ring_doorbell import RingDoorBell + +_LOGGER = logging.getLogger(__name__) + + +class RingStickUpCam(RingDoorBell): + """Implementation for RingStickUpCam.""" + + @property + def family(self): + """Return Ring device family type.""" + return 'stickup_cams'