-
Notifications
You must be signed in to change notification settings - Fork 171
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
319 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,295 @@ | ||
| # coding: utf-8 | ||
| # vim:sw=4:ts=4:et: | ||
| """Python Ring Other (Intercom) wrapper.""" | ||
| import json | ||
| import logging | ||
| import uuid | ||
|
|
||
| from ring_doorbell.const import ( | ||
| DOORBELLS_ENDPOINT, | ||
| HEALTH_DOORBELL_ENDPOINT, | ||
| INTERCOM_ALLOWED_USERS, | ||
| INTERCOM_INVITATIONS_DELETE_ENDPOINT, | ||
| INTERCOM_INVITATIONS_ENDPOINT, | ||
| INTERCOM_KINDS, | ||
| INTERCOM_OPEN_ENDPOINT, | ||
| MIC_VOL_MAX, | ||
| MIC_VOL_MIN, | ||
| MSG_VOL_OUTBOUND, | ||
| OTHER_DOORBELL_VOL_MAX, | ||
| OTHER_DOORBELL_VOL_MIN, | ||
| SETTINGS_ENDPOINT, | ||
| VOICE_VOL_MAX, | ||
| VOICE_VOL_MIN, | ||
| ) | ||
| from ring_doorbell.exceptions import RingError | ||
| from ring_doorbell.generic import RingGeneric | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class Other(RingGeneric): | ||
| """Implementation for Ring Intercom.""" | ||
|
|
||
| def __init__(self, ring, device_id, shared=False): | ||
| super().__init__(ring, device_id) | ||
| self.shared = shared | ||
|
|
||
| @property | ||
| def family(self): | ||
| """Return Ring device family type.""" | ||
| return "other" | ||
|
|
||
| def update_health_data(self): | ||
| """Update health attrs.""" | ||
| self._health_attrs = ( | ||
| self._ring.query(HEALTH_DOORBELL_ENDPOINT.format(self.id)) | ||
| .json() | ||
| .get("device_health", {}) | ||
| ) | ||
|
|
||
| @property | ||
| def model(self): | ||
| """Return Ring device model name.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return "Intercom" | ||
| return None | ||
|
|
||
| def has_capability(self, capability): | ||
| """Return if device has specific capability.""" | ||
| if capability == "open": | ||
| return self.kind in INTERCOM_KINDS | ||
| return False | ||
|
|
||
| @property | ||
| def battery_life(self): | ||
| """Return battery life.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| if self._attrs.get("battery_life") is None: | ||
| return None | ||
|
|
||
| value = int(self._attrs.get("battery_life", 0)) | ||
| if value and value > 100: | ||
| value = 100 | ||
|
|
||
| return value | ||
| return None | ||
|
|
||
| @property | ||
| def subscribed(self): | ||
| """Return if is online.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| result = self._attrs.get("subscribed") | ||
| if result is None: | ||
| return False | ||
| return True | ||
| return None | ||
|
|
||
| @property | ||
| def subscriptions(self): | ||
| """Return event type subscriptions.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("subscriptions", []).get("event_types", []) | ||
| return None | ||
|
|
||
| @property | ||
| def has_subscription(self): | ||
| """Return boolean if the account has subscription.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("features").get("show_recordings") | ||
| return None | ||
|
|
||
| @property | ||
| def unlock_duration(self): | ||
| """Return time unlock switch is held closed""" | ||
| json.loads( | ||
| self._attrs.get("settings").get("intercom_settings").get("config") | ||
| ).get("analog").get("unlock_duration") | ||
|
|
||
| @property | ||
| def doorbell_volume(self): | ||
| """Return doorbell volume.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("settings").get("doorbell_volume") | ||
| return None | ||
|
|
||
| @doorbell_volume.setter | ||
| def doorbell_volume(self, value): | ||
| if not ( | ||
| (isinstance(value, int)) | ||
| and (OTHER_DOORBELL_VOL_MIN <= value <= OTHER_DOORBELL_VOL_MAX) | ||
| ): | ||
| _LOGGER.error( | ||
| "%s", | ||
| MSG_VOL_OUTBOUND.format(OTHER_DOORBELL_VOL_MIN, OTHER_DOORBELL_VOL_MAX), | ||
| ) | ||
| return False | ||
|
|
||
| params = { | ||
| "doorbot[description]": self.name, | ||
| "doorbot[settings][doorbell_volume]": str(value), | ||
| } | ||
| url = DOORBELLS_ENDPOINT.format(self.id) | ||
| self._ring.query(url, extra_params=params, method="PUT") | ||
| self._ring.update_devices() | ||
| return True | ||
|
|
||
| @property | ||
| def keep_alive_auto(self): | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("settings").get("keep_alive_auto") | ||
| return None | ||
|
|
||
| @keep_alive_auto.setter | ||
| def keep_alive_auto(self, value): | ||
| url = SETTINGS_ENDPOINT.format(self.id) | ||
| payload = {"keep_alive_settings": {"keep_alive_auto": value}} | ||
|
|
||
| self._ring.query(url, method="PATCH", json=payload) | ||
| self._ring.update_devices() | ||
| return True | ||
|
|
||
| @property | ||
| def mic_volume(self): | ||
| """Return mic volume.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("settings").get("mic_volume") | ||
| return None | ||
|
|
||
| @mic_volume.setter | ||
| def mic_volume(self, value): | ||
| if not ((isinstance(value, int)) and (MIC_VOL_MIN <= value <= MIC_VOL_MAX)): | ||
| _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(MIC_VOL_MIN, MIC_VOL_MAX)) | ||
| return False | ||
|
|
||
| url = SETTINGS_ENDPOINT.format(self.id) | ||
| payload = {"volume_settings": {"mic_volume": value}} | ||
|
|
||
| self._ring.query(url, method="PATCH", json=payload) | ||
| self._ring.update_devices() | ||
| return True | ||
|
|
||
| @property | ||
| def voice_volume(self): | ||
| """Return voice volume.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("settings").get("voice_volume") | ||
| return None | ||
|
|
||
| @voice_volume.setter | ||
| def voice_volume(self, value): | ||
| if not ((isinstance(value, int)) and (VOICE_VOL_MIN <= value <= VOICE_VOL_MAX)): | ||
| _LOGGER.error("%s", MSG_VOL_OUTBOUND.format(VOICE_VOL_MIN, VOICE_VOL_MAX)) | ||
| return False | ||
|
|
||
| url = SETTINGS_ENDPOINT.format(self.id) | ||
| payload = {"volume_settings": {"voice_volume": value}} | ||
|
|
||
| self._ring.query(url, method="PATCH", json=payload) | ||
| self._ring.update_devices() | ||
| return True | ||
|
|
||
| @property | ||
| def clip_length_max(self): | ||
| # this value sets an effective refractory period on consecutive rigns | ||
| # eg if set to default value of 60, rings occuring with 60 seconds of | ||
| # first will not be detected | ||
|
|
||
| url = SETTINGS_ENDPOINT.format(self.id) | ||
|
|
||
| return ( | ||
| self._ring.query(url, method="GET") | ||
| .json() | ||
| .get("video_settings") | ||
| .get("clip_length_max") | ||
| ) | ||
|
|
||
| @clip_length_max.setter | ||
| def clip_length_max(self, value): | ||
| url = SETTINGS_ENDPOINT.format(self.id) | ||
| payload = {"video_settings": {"clip_length_max": value}} | ||
| try: | ||
| self._ring.query(url, method="PATCH", json=payload) | ||
| self._ring.update_devices() | ||
| return True | ||
| except Exception as ex: | ||
| raise RingError(f"Unknown error during query of url {url}: {ex}") from ex | ||
|
|
||
| @property | ||
| def connection_status(self): | ||
| """Return connection status.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("alerts").get("connection") | ||
| return None | ||
|
|
||
| @property | ||
| def location_id(self): | ||
| """Return location id.""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| return self._attrs.get("location_id", None) | ||
| return None | ||
|
|
||
| @property | ||
| def allowed_users(self): | ||
| """Return list of users allowed or invited to access""" | ||
| if self.kind in INTERCOM_KINDS: | ||
| url = INTERCOM_ALLOWED_USERS.format(self.location_id) | ||
| return self._ring.query(url, method="GET").json() | ||
|
|
||
| return None | ||
|
|
||
| def open_door(self): | ||
| """Open the door""" | ||
|
|
||
| if self.kind in INTERCOM_KINDS: | ||
| url = INTERCOM_OPEN_ENDPOINT.format(self.id) | ||
| request_id = str(uuid.uuid4()) | ||
| # request_timestamp = int(time.time() * 1000) | ||
| payload = { | ||
| "command_name": "device_rpc", | ||
| "request": { | ||
| "id": request_id, | ||
| "jsonrpc": "2.0", | ||
| "method": "unlock_door", | ||
| "params": { | ||
| # "command_timeout": 5, | ||
| "door_id": 0, | ||
| # "issue_time": request_timestamp, | ||
| "user_id": 64658594, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| response = self._ring.query(url, method="PUT", json=payload).json() | ||
| self._ring.update_devices() | ||
| if response.get("result", -1).get("code", -1) == 0: | ||
| return True | ||
|
|
||
| return False | ||
|
|
||
| def invite_access(self, email): | ||
| """Invite user""" | ||
|
|
||
| if self.kind in INTERCOM_KINDS: | ||
| url = INTERCOM_INVITATIONS_ENDPOINT.format(self.location_id) | ||
| payload = { | ||
| "invitation": { | ||
| "doorbot_ids": [self.id], | ||
| "invited_email": email, | ||
| "group_ids": [], | ||
| } | ||
| } | ||
| self._ring.query(url, method="POST", json=payload) | ||
| return True | ||
|
|
||
| return False | ||
|
|
||
| def remove_access(self, user_id): | ||
| """Remove user access or invitation""" | ||
|
|
||
| if self.kind in INTERCOM_KINDS: | ||
| url = INTERCOM_INVITATIONS_DELETE_ENDPOINT.format(self.location_id, user_id) | ||
| self._ring.query(url, method="DELETE") | ||
| return True | ||
|
|
||
| return False |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters