6148597193

Committed 11 Sep 2023 03:29PM UTC coverage: 66.139%. First build

Build # 6148597193

Build Type

push

github

Committed by

sdb9696

Commit Message

Add motion detection enabled switch

Run Details

33 of 33 new or added lines in 4 files covered. (100.0%)

418 of 632 relevant lines covered (66.14%)

0.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.44

/ring_doorbell/doorbot.py

# coding: utf-8
# vim:sw=4:ts=4:et:
"""Python Ring Doorbell wrapper."""
import logging
from datetime import datetime
import os
import time
import pytz


from ring_doorbell.generic import RingGeneric

from ring_doorbell.const import (
    DOORBELLS_ENDPOINT,
    DOORBELL_VOL_MIN,
    DOORBELL_VOL_MAX,
    DOORBELL_EXISTING_TYPE,
    DINGS_ENDPOINT,
    DOORBELL_KINDS,
    DOORBELL_2_KINDS,
    DOORBELL_3_KINDS,
    DOORBELL_3_PLUS_KINDS,
    DOORBELL_PRO_KINDS,
    DOORBELL_ELITE_KINDS,
    FILE_EXISTS,
    LIVE_STREAMING_ENDPOINT,
    MSG_BOOLEAN_REQUIRED,
    MSG_EXISTING_TYPE,
    MSG_VOL_OUTBOUND,
    MSG_ALLOWED_VALUES,
    MSG_EXPECTED_ATTRIBUTE_NOT_FOUND,
    PEEPHOLE_CAM_KINDS,
    SNAPSHOT_ENDPOINT,
    SNAPSHOT_TIMESTAMP_ENDPOINT,
    URL_DOORBELL_HISTORY,
    URL_RECORDING,
    URL_RECORDING_SHARE_PLAY,
    DEFAULT_VIDEO_DOWNLOAD_TIMEOUT,
    HEALTH_DOORBELL_ENDPOINT,
    SETTINGS_ENDPOINT,
)

_LOGGER = logging.getLogger(__name__)


class RingDoorBell(RingGeneric):
    """Implementation for Ring Doorbell."""

    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 "authorized_doorbots" if self.shared else "doorbots"

    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 DOORBELL_KINDS:
            return "Doorbell"
        if self.kind in DOORBELL_2_KINDS:
            return "Doorbell 2"
        if self.kind in DOORBELL_3_KINDS:
            return "Doorbell 3"
        if self.kind in DOORBELL_3_PLUS_KINDS:
            return "Doorbell 3 Plus"
        if self.kind in DOORBELL_PRO_KINDS:
            return "Doorbell Pro"
        if self.kind in DOORBELL_ELITE_KINDS:
            return "Doorbell Elite"
        if self.kind in PEEPHOLE_CAM_KINDS:
            return "Peephole Cam"
        return None

    def has_capability(self, capability):
        """Return if device has specific capability."""
        if capability == "battery":
            return self.kind in (
                DOORBELL_KINDS
                + DOORBELL_2_KINDS
                + DOORBELL_3_KINDS
                + DOORBELL_3_PLUS_KINDS
                + PEEPHOLE_CAM_KINDS
            )
        if capability == "knock":
            return self.kind in PEEPHOLE_CAM_KINDS
        if capability == "pre-roll":
            return self.kind in DOORBELL_3_PLUS_KINDS
        if capability == "volume":
            return True
        if capability == "motion_detection":
            return self.kind in (
                DOORBELL_KINDS
                + DOORBELL_2_KINDS
                + DOORBELL_3_KINDS
                + DOORBELL_3_PLUS_KINDS
                + PEEPHOLE_CAM_KINDS
            )
        return False

    @property
    def battery_life(self):
        """Return battery life."""
        if (
            self._attrs.get("battery_life") is None
            and self._attrs.get("battery_life_2") is None
        ):
            return None

        value = 0
        if "battery_life_2" in self._attrs:
            # Camera has two battery bays
            if self._attrs.get("battery_life") is not None:
                # Bay 1
                value += int(self._attrs.get("battery_life"))
            if self._attrs.get("battery_life_2") is not None:
                # Bay 2
                value += int(self._attrs.get("battery_life_2"))
            return value

        # Camera has a single battery bay
        # Latest stickup cam can be externally powered
        value = int(self._attrs.get("battery_life"))
        if value and value > 100:
            value = 100

        return value

    @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
        """
        # pylint:disable=consider-iterating-dictionary
        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 = DOORBELLS_ENDPOINT.format(self.id)
            self._ring.query(url, extra_params=params, method="PUT")
            self._ring.update_devices()
            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 = DOORBELLS_ENDPOINT.format(self.id)
            self._ring.query(url, extra_params=params, method="PUT")
            self._ring.update_devices()
            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 (DOORBELL_VOL_MIN <= 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 = DOORBELLS_ENDPOINT.format(self.id)
                self._ring.query(url, extra_params=params, method="PUT")
                self._ring.update_devices()
                return True
        return None

    def history(
        self,
        limit=30,
        timezone=None,
        kind=None,
        enforce_limit=False,
        older_than=None,
        retry=8,
    ):
        """
        Return history with datetime objects.

        :param limit: specify number of objects to be returned
        :param timezone: determine which timezone to convert data objects
        :param kind: filter by kind (ding, motion, on_demand)
        :param enforce_limit: when True, this will enforce the limit and kind
        :param older_than: return older objects than the passed event_id
        :param retry: determine the max number of attempts to archive the limit
        """
        queries = 0
        original_limit = limit

        # set cap for max queries
        # pylint:disable=consider-using-min-builtin
        if retry > 10:
            retry = 10

        while True:
            params = {"limit": str(limit)}
            if older_than:
                params["older_than"] = older_than

            url = URL_DOORBELL_HISTORY.format(self.id)
            response = self._ring.query(url, extra_params=params).json()

            # cherrypick only the selected kind events
            if kind:
                response = list(filter(lambda array: array["kind"] == kind, response))

            # 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 enforce_limit:
                # return because already matched the number
                # of events by kind
                if len(response) >= original_limit:
                    return response[:original_limit]

                # ensure the loop will exit after max queries
                queries += 1
                if queries == retry:
                    _LOGGER.debug(
                        "Could not find total of %s of kind %s", original_limit, kind
                    )
                    break

                # ensure the kind objects returned to match limit
                limit = limit * 2

            else:
                break

        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 = LIVE_STREAMING_ENDPOINT.format(self.id)
        req = self._ring.query(url, method="POST")
        if req and req.status_code == 200:
            url = DINGS_ENDPOINT
            try:
                return self._ring.query(url).json()[0]
            except (IndexError, TypeError):
                pass
        return None

    def recording_download(
        self,
        recording_id,
        filename=None,
        override=False,
        timeout=DEFAULT_VIDEO_DOWNLOAD_TIMEOUT,
    ):
        """Save a recording in MP4 format to a file or return raw."""
        if not self.has_subscription:
            msg = "Your Ring account does not have an active subscription."
            _LOGGER.warning(msg)
            return False

        url = URL_RECORDING.format(recording_id)
        try:
            # Video download needs a longer timeout to get the large video file
            req = self._ring.query(url, timeout=timeout)
            if req and 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
        return False

    def recording_url(self, recording_id):
        """Return HTTPS recording URL."""
        if not self.has_subscription:
            msg = "Your Ring account does not have an active subscription."
            _LOGGER.warning(msg)
            return False

        url = URL_RECORDING_SHARE_PLAY.format(recording_id)
        req = self._ring.query(url)
        data = req.json()
        if req and req.status_code == 200 and data is not None:
            return data["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 has_subscription(self):
        """Return boolean if the account has subscription."""
        return self._attrs.get("features").get("show_recordings")

    @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 (DOORBELL_VOL_MIN <= 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 = DOORBELLS_ENDPOINT.format(self.id)
        self._ring.query(url, extra_params=params, method="PUT")
        self._ring.update_devices()
        return True

    @property
    def connection_status(self):
        """Return connection status."""
        return self._attrs.get("alerts").get("connection")

    def get_snapshot(self, retries=3, delay=1, filename=None):
        """Take a snapshot and download it"""
        url = SNAPSHOT_TIMESTAMP_ENDPOINT
        payload = {"doorbot_ids": [self._attrs.get("id")]}
        self._ring.query(url, method="POST", json=payload)
        request_time = time.time()
        for _ in range(retries):
            time.sleep(delay)
            response = self._ring.query(url, method="POST", json=payload).json()
            if response["timestamps"][0]["timestamp"] / 1000 > request_time:
                snapshot = self._ring.query(
                    SNAPSHOT_ENDPOINT.format(self._attrs.get("id")), raw=True
                ).content
                if filename:
                    with open(filename, "wb") as jpg:
                        jpg.write(snapshot)
                    return True
                return snapshot
        return False

    def _motion_detection_state(self):
        if "settings" in self._attrs and "motion_detection_enabled" in self._attrs.get(
            "settings"
        ):
            return self._attrs.get("settings")["motion_detection_enabled"]
        return None

    @property
    def motion_detection(self):
        """Return motion detection enabled state."""
        return self._motion_detection_state()

    @motion_detection.setter
    def motion_detection(self, state):
        """Set the motion detection enabled state."""
        values = [True, False]
        if state not in values:
            _LOGGER.error("%s", MSG_ALLOWED_VALUES.format(", ".join(values)))
            return False

        if self._motion_detection_state() is None:
            _LOGGER.warning(
                "%s",
                MSG_EXPECTED_ATTRIBUTE_NOT_FOUND.format(
                    "settings[motion_detection_enabled]"
                ),
            )
            return False

        url = SETTINGS_ENDPOINT.format(self.id)
        payload = {"motion_settings": {"motion_detection_enabled": state}}

        self._ring.query(url, method="PATCH", json=payload)
        self._ring.update_devices()
        return True

1	# coding: utf-8
2	# vim:sw=4:ts=4:et:
3	"""Python Ring Doorbell wrapper."""	1✔
4	import logging	1✔
5	from datetime import datetime	1✔
6	import os	1✔
7	import time	1✔
8	import pytz	1✔
9
10
11	from ring_doorbell.generic import RingGeneric	1✔
12
13	from ring_doorbell.const import (	1✔
14	DOORBELLS_ENDPOINT,
15	DOORBELL_VOL_MIN,
16	DOORBELL_VOL_MAX,
17	DOORBELL_EXISTING_TYPE,
18	DINGS_ENDPOINT,
19	DOORBELL_KINDS,
20	DOORBELL_2_KINDS,
21	DOORBELL_3_KINDS,
22	DOORBELL_3_PLUS_KINDS,
23	DOORBELL_PRO_KINDS,
24	DOORBELL_ELITE_KINDS,
25	FILE_EXISTS,
26	LIVE_STREAMING_ENDPOINT,
27	MSG_BOOLEAN_REQUIRED,
28	MSG_EXISTING_TYPE,
29	MSG_VOL_OUTBOUND,
30	MSG_ALLOWED_VALUES,
31	MSG_EXPECTED_ATTRIBUTE_NOT_FOUND,
32	PEEPHOLE_CAM_KINDS,
33	SNAPSHOT_ENDPOINT,
34	SNAPSHOT_TIMESTAMP_ENDPOINT,
35	URL_DOORBELL_HISTORY,
36	URL_RECORDING,
37	URL_RECORDING_SHARE_PLAY,
38	DEFAULT_VIDEO_DOWNLOAD_TIMEOUT,
39	HEALTH_DOORBELL_ENDPOINT,
40	SETTINGS_ENDPOINT,
41	)
42
43	_LOGGER = logging.getLogger(__name__)	1✔
44
45
46	class RingDoorBell(RingGeneric):	1✔
47	"""Implementation for Ring Doorbell."""
48
49	def __init__(self, ring, device_id, shared=False):	1✔
50	super().__init__(ring, device_id)	1✔
51	self.shared = shared	1✔
52
53	@property	1✔
54	def family(self):	1✔
55	"""Return Ring device family type."""
56	return "authorized_doorbots" if self.shared else "doorbots"	1✔
57
58	def update_health_data(self):	1✔
59	"""Update health attrs."""
60	self._health_attrs = (	1✔
61	self._ring.query(HEALTH_DOORBELL_ENDPOINT.format(self.id))
62	.json()
63	.get("device_health", {})
64	)
65
66	@property	1✔
67	def model(self):	1✔
68	"""Return Ring device model name."""
69	if self.kind in DOORBELL_KINDS:	1✔
70	return "Doorbell"	×
71	if self.kind in DOORBELL_2_KINDS:	1✔
72	return "Doorbell 2"	×
73	if self.kind in DOORBELL_3_KINDS:	1✔
74	return "Doorbell 3"	×
75	if self.kind in DOORBELL_3_PLUS_KINDS:	1✔
76	return "Doorbell 3 Plus"	×
77	if self.kind in DOORBELL_PRO_KINDS:	1✔
78	return "Doorbell Pro"	1✔
79	if self.kind in DOORBELL_ELITE_KINDS:	×
80	return "Doorbell Elite"	×
81	if self.kind in PEEPHOLE_CAM_KINDS:	×
82	return "Peephole Cam"	×
83	return None	×
84
85	def has_capability(self, capability):	1✔
86	"""Return if device has specific capability."""
87	if capability == "battery":	1✔
88	return self.kind in (	1✔
89	DOORBELL_KINDS
90	+ DOORBELL_2_KINDS
91	+ DOORBELL_3_KINDS
92	+ DOORBELL_3_PLUS_KINDS
93	+ PEEPHOLE_CAM_KINDS
94	)
95	if capability == "knock":	1✔
96	return self.kind in PEEPHOLE_CAM_KINDS	×
97	if capability == "pre-roll":	1✔
98	return self.kind in DOORBELL_3_PLUS_KINDS	×
99	if capability == "volume":	1✔
100	return True	1✔
101	if capability == "motion_detection":	×
102	return self.kind in (	×
103	DOORBELL_KINDS
104	+ DOORBELL_2_KINDS
105	+ DOORBELL_3_KINDS
106	+ DOORBELL_3_PLUS_KINDS
107	+ PEEPHOLE_CAM_KINDS
108	)
109	return False	×
110
111	@property	1✔
112	def battery_life(self):	1✔
113	"""Return battery life."""
114	if (	1✔
115	self._attrs.get("battery_life") is None
116	and self._attrs.get("battery_life_2") is None
117	):
118	return None	×
119
120	value = 0	1✔
121	if "battery_life_2" in self._attrs:	1✔
122	# Camera has two battery bays
123	if self._attrs.get("battery_life") is not None:	×
124	# Bay 1
125	value += int(self._attrs.get("battery_life"))	×
126	if self._attrs.get("battery_life_2") is not None:	×
127	# Bay 2
128	value += int(self._attrs.get("battery_life_2"))	×
129	return value	×
130
131	# Camera has a single battery bay
132	# Latest stickup cam can be externally powered
133	value = int(self._attrs.get("battery_life"))	1✔
134	if value and value > 100:	1✔
135	value = 100	×
136
137	return value	1✔
138
139	@property	1✔
140	def existing_doorbell_type(self):	1✔
141	"""
142	Return existing doorbell type.
143
144	0: Mechanical
145	1: Digital
146	2: Not Present
147	"""
148	try:	1✔
149	return DOORBELL_EXISTING_TYPE[	1✔
150	self._attrs.get("settings").get("chime_settings").get("type")
151	]
152	except AttributeError:	×
153	return None	×
154
155	@existing_doorbell_type.setter	1✔
156	def existing_doorbell_type(self, value):	1✔
157	"""
158	Return existing doorbell type.
159
160	0: Mechanical
161	1: Digital
162	2: Not Present
163	"""
164	# pylint:disable=consider-iterating-dictionary
165	if value not in DOORBELL_EXISTING_TYPE.keys():	×
166	_LOGGER.error("%s", MSG_EXISTING_TYPE)	×
167	return False	×
168	params = {	×
169	"doorbot[description]": self.name,
170	"doorbot[settings][chime_settings][type]": value,
171	}
172	if self.existing_doorbell_type:	×
173	url = DOORBELLS_ENDPOINT.format(self.id)	×
174	self._ring.query(url, extra_params=params, method="PUT")	×
175	self._ring.update_devices()	×
176	return True	×
177	return None	×
178
179	@property	1✔
180	def existing_doorbell_type_enabled(self):	1✔
181	"""Return if existing doorbell type is enabled."""
182	if self.existing_doorbell_type:	×
183	if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[2]:	×
184	return None	×
185	return self._attrs.get("settings").get("chime_settings").get("enable")	×
186	return False	×
187
188	@existing_doorbell_type_enabled.setter	1✔
189	def existing_doorbell_type_enabled(self, value):	1✔
190	"""Enable/disable the existing doorbell if Digital/Mechanical."""
191	if self.existing_doorbell_type:	×
192	if not isinstance(value, bool):	×
193	_LOGGER.error("%s", MSG_BOOLEAN_REQUIRED)	×
194	return None	×
195
196	if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[2]:	×
197	return None	×
198
199	params = {	×
200	"doorbot[description]": self.name,
201	"doorbot[settings][chime_settings][enable]": value,
202	}
203	url = DOORBELLS_ENDPOINT.format(self.id)	×
204	self._ring.query(url, extra_params=params, method="PUT")	×
205	self._ring.update_devices()	×
206	return True	×
207	return False	×
208
209	@property	1✔
210	def existing_doorbell_type_duration(self):	1✔
211	"""Return duration for Digital chime."""
212	if self.existing_doorbell_type:	×
213	if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[1]:	×
214	return self._attrs.get("settings").get("chime_settings").get("duration")	×
215	return None	×
216
217	@existing_doorbell_type_duration.setter	1✔
218	def existing_doorbell_type_duration(self, value):	1✔
219	"""Set duration for Digital chime."""
220	if self.existing_doorbell_type:	×
221	if not (	×
222	(isinstance(value, int))
223	and (DOORBELL_VOL_MIN <= value <= DOORBELL_VOL_MAX)
224	):
225	_LOGGER.error(	×
226	"%s", MSG_VOL_OUTBOUND.format(DOORBELL_VOL_MIN, DOORBELL_VOL_MAX)
227	)
228	return False	×
229
230	if self.existing_doorbell_type == DOORBELL_EXISTING_TYPE[1]:	×
231	params = {	×
232	"doorbot[description]": self.name,
233	"doorbot[settings][chime_settings][duration]": value,
234	}
235	url = DOORBELLS_ENDPOINT.format(self.id)	×
236	self._ring.query(url, extra_params=params, method="PUT")	×
237	self._ring.update_devices()	×
238	return True	×
239	return None	×
240
241	def history(	1✔
242	self,
243	limit=30,
244	timezone=None,
245	kind=None,
246	enforce_limit=False,
247	older_than=None,
248	retry=8,
249	):
250	"""
251	Return history with datetime objects.
252
253	:param limit: specify number of objects to be returned
254	:param timezone: determine which timezone to convert data objects
255	:param kind: filter by kind (ding, motion, on_demand)
256	:param enforce_limit: when True, this will enforce the limit and kind
257	:param older_than: return older objects than the passed event_id
258	:param retry: determine the max number of attempts to archive the limit
259	"""
260	queries = 0	1✔
261	original_limit = limit	1✔
262
263	# set cap for max queries
264	# pylint:disable=consider-using-min-builtin
265	if retry > 10:	1✔
266	retry = 10	1✔
267
268	while True:	1✔
269	params = {"limit": str(limit)}	1✔
270	if older_than:	1✔
271	params["older_than"] = older_than	×
272
273	url = URL_DOORBELL_HISTORY.format(self.id)	1✔
274	response = self._ring.query(url, extra_params=params).json()	1✔
275
276	# cherrypick only the selected kind events
277	if kind:	1✔
278	response = list(filter(lambda array: array["kind"] == kind, response))	1✔
279
280	# convert for specific timezone
281	utc = pytz.utc	1✔
282	if timezone:	1✔
283	mytz = pytz.timezone(timezone)	×
284
285	for entry in response:	1✔
286	dt_at = datetime.strptime(entry["created_at"], "%Y-%m-%dT%H:%M:%S.000Z")	1✔
287	utc_dt = datetime(	1✔
288	dt_at.year,
289	dt_at.month,
290	dt_at.day,
291	dt_at.hour,
292	dt_at.minute,
293	dt_at.second,
294	tzinfo=utc,
295	)
296	if timezone:	1✔
297	tz_dt = utc_dt.astimezone(mytz)	×
298	entry["created_at"] = tz_dt	×
299	else:
300	entry["created_at"] = utc_dt	1✔
301
302	if enforce_limit:	1✔
303	# return because already matched the number
304	# of events by kind
305	if len(response) >= original_limit:	1✔
306	return response[:original_limit]	×
307
308	# ensure the loop will exit after max queries
309	queries += 1	1✔
310	if queries == retry:	1✔
311	_LOGGER.debug(	1✔
312	"Could not find total of %s of kind %s", original_limit, kind
313	)
314	break	1✔
315
316	# ensure the kind objects returned to match limit
317	limit = limit * 2	1✔
318
319	else:
320	break	1✔
321
322	return response	1✔
323
324	@property	1✔
325	def last_recording_id(self):	1✔
326	"""Return the last recording ID."""
327	try:	×
328	return self.history(limit=1)[0]["id"]	×
329	except (IndexError, TypeError):	×
330	return None	×
331
332	@property	1✔
333	def live_streaming_json(self):	1✔
334	"""Return JSON for live streaming."""
335	url = LIVE_STREAMING_ENDPOINT.format(self.id)	×
336	req = self._ring.query(url, method="POST")	×
337	if req and req.status_code == 200:	×
338	url = DINGS_ENDPOINT	×
339	try:	×
340	return self._ring.query(url).json()[0]	×
341	except (IndexError, TypeError):	×
342	pass	×
343	return None	×
344
345	def recording_download(	1✔
346	self,
347	recording_id,
348	filename=None,
349	override=False,
350	timeout=DEFAULT_VIDEO_DOWNLOAD_TIMEOUT,
351	):
352	"""Save a recording in MP4 format to a file or return raw."""
353	if not self.has_subscription:	×
354	msg = "Your Ring account does not have an active subscription."	×
355	_LOGGER.warning(msg)	×
356	return False	×
357
358	url = URL_RECORDING.format(recording_id)	×
359	try:	×
360	# Video download needs a longer timeout to get the large video file
361	req = self._ring.query(url, timeout=timeout)	×
362	if req and req.status_code == 200:	×
363	if filename:	×
364	if os.path.isfile(filename) and not override:	×
365	_LOGGER.error("%s", FILE_EXISTS.format(filename))	×
366	return False	×
367
368	with open(filename, "wb") as recording:	×
369	recording.write(req.content)	×
370	return True	×
371	else:
372	return req.content	×
373	except IOError as error:	×
374	_LOGGER.error("%s", error)	×
375	raise	×
376	return False	×
377
378	def recording_url(self, recording_id):	1✔
379	"""Return HTTPS recording URL."""
380	if not self.has_subscription:	×
381	msg = "Your Ring account does not have an active subscription."	×
382	_LOGGER.warning(msg)	×
383	return False	×
384
385	url = URL_RECORDING_SHARE_PLAY.format(recording_id)	×
386	req = self._ring.query(url)	×
387	data = req.json()	×
388	if req and req.status_code == 200 and data is not None:	×
389	return data["url"]	×
390	return False	×
391
392	@property	1✔
393	def subscribed(self):	1✔
394	"""Return if is online."""
395	result = self._attrs.get("subscribed")	×
396	if result is None:	×
397	return False	×
398	return True	×
399
400	@property	1✔
401	def subscribed_motion(self):	1✔
402	"""Return if is subscribed_motion."""
403	result = self._attrs.get("subscribed_motions")	×
404	if result is None:	×
405	return False	×
406	return True	×
407
408	@property	1✔
409	def has_subscription(self):	1✔
410	"""Return boolean if the account has subscription."""
411	return self._attrs.get("features").get("show_recordings")	1✔
412
413	@property	1✔
414	def volume(self):	1✔
415	"""Return volume."""
416	return self._attrs.get("settings").get("doorbell_volume")	1✔
417
418	@volume.setter	1✔
419	def volume(self, value):	1✔
420	if not (	×
421	(isinstance(value, int)) and (DOORBELL_VOL_MIN <= value <= DOORBELL_VOL_MAX)
422	):
423	_LOGGER.error(	×
424	"%s", MSG_VOL_OUTBOUND.format(DOORBELL_VOL_MIN, DOORBELL_VOL_MAX)
425	)
426	return False	×
427
428	params = {	×
429	"doorbot[description]": self.name,
430	"doorbot[settings][doorbell_volume]": str(value),
431	}
432	url = DOORBELLS_ENDPOINT.format(self.id)	×
433	self._ring.query(url, extra_params=params, method="PUT")	×
434	self._ring.update_devices()	×
435	return True	×
436
437	@property	1✔
438	def connection_status(self):	1✔
439	"""Return connection status."""
440	return self._attrs.get("alerts").get("connection")	1✔
441
442	def get_snapshot(self, retries=3, delay=1, filename=None):	1✔
443	"""Take a snapshot and download it"""
444	url = SNAPSHOT_TIMESTAMP_ENDPOINT	×
445	payload = {"doorbot_ids": [self._attrs.get("id")]}	×
446	self._ring.query(url, method="POST", json=payload)	×
447	request_time = time.time()	×
448	for _ in range(retries):	×
449	time.sleep(delay)	×
450	response = self._ring.query(url, method="POST", json=payload).json()	×
451	if response["timestamps"][0]["timestamp"] / 1000 > request_time:	×
452	snapshot = self._ring.query(	×
453	SNAPSHOT_ENDPOINT.format(self._attrs.get("id")), raw=True
454	).content
455	if filename:	×
456	with open(filename, "wb") as jpg:	×
457	jpg.write(snapshot)	×
458	return True	×
459	return snapshot	×
460	return False	×
461
462	def _motion_detection_state(self):	1✔
463	if "settings" in self._attrs and "motion_detection_enabled" in self._attrs.get(	1✔
464	"settings"
465	):
466	return self._attrs.get("settings")["motion_detection_enabled"]	1✔
467	return None	×
468
469	@property	1✔
470	def motion_detection(self):	1✔
471	"""Return motion detection enabled state."""
472	return self._motion_detection_state()	×
473
474	@motion_detection.setter	1✔
475	def motion_detection(self, state):	1✔
476	"""Set the motion detection enabled state."""
477	values = [True, False]	1✔
478	if state not in values:	1✔
479	_LOGGER.error("%s", MSG_ALLOWED_VALUES.format(", ".join(values)))	×
480	return False	×
481
482	if self._motion_detection_state() is None:	1✔
483	_LOGGER.warning(	×
484	"%s",
485	MSG_EXPECTED_ATTRIBUTE_NOT_FOUND.format(
486	"settings[motion_detection_enabled]"
487	),
488	)
489	return False	×
490
491	url = SETTINGS_ENDPOINT.format(self.id)	1✔
492	payload = {"motion_settings": {"motion_detection_enabled": state}}	1✔
493
494	self._ring.query(url, method="PATCH", json=payload)	1✔
495	self._ring.update_devices()	1✔
496	return True	1✔

sdb9696 / python-ring-doorbell / 6148597193

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous