From 0d7996d53dd3931bcd20b4d1eea467fe4758203e Mon Sep 17 00:00:00 2001 From: Adam Reeve Date: Wed, 10 Jun 2026 00:01:42 -0700 Subject: [PATCH] Add new version of snapshot method, with basic tests. Mark old version deprecated. --- ring_doorbell/const.py | 2 ++ ring_doorbell/doorbot.py | 38 +++++++++++++++++++++ tests/test_ring.py | 73 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 112 insertions(+), 1 deletion(-) diff --git a/ring_doorbell/const.py b/ring_doorbell/const.py index 79c88ff..71c7584 100644 --- a/ring_doorbell/const.py +++ b/ring_doorbell/const.py @@ -66,6 +66,7 @@ def from_name(name: str) -> RingCapability: # API endpoints API_VERSION = "11" API_URI = "https://api.ring.com" +API_URI_SNAPSHOT = "https://app-snaps.ring.com" APP_API_URI = "https://prd-api-us.prd.rings.solutions" USER_AGENT = "android:com.ringapp" @@ -101,6 +102,7 @@ def from_name(name: str) -> RingCapability: SIREN_ENDPOINT = DOORBELLS_ENDPOINT + "/siren_{1}" SNAPSHOT_ENDPOINT = "/clients_api/snapshots/image/{0}" SNAPSHOT_TIMESTAMP_ENDPOINT = "/clients_api/snapshots/timestamps" +TAKE_SNAPSHOT_ENDPOINT = "/snapshots/next/{0}" TESTSOUND_CHIME_ENDPOINT = CHIMES_ENDPOINT + "/play_sound" URL_DOORBELL_HISTORY = DOORBELLS_ENDPOINT + "/history" URL_RECORDING = "/clients_api/dings/{0}/recording" diff --git a/ring_doorbell/doorbot.py b/ring_doorbell/doorbot.py index 7fa6809..d98554a 100644 --- a/ring_doorbell/doorbot.py +++ b/ring_doorbell/doorbot.py @@ -8,10 +8,12 @@ import time from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar +from warnings import deprecated import aiofiles from ring_doorbell.const import ( + API_URI_SNAPSHOT, DEFAULT_VIDEO_DOWNLOAD_TIMEOUT, DINGS_ENDPOINT, DOORBELL_2_KINDS, @@ -44,6 +46,7 @@ SETTINGS_ENDPOINT, SNAPSHOT_ENDPOINT, SNAPSHOT_TIMESTAMP_ENDPOINT, + TAKE_SNAPSHOT_ENDPOINT, URL_RECORDING, URL_RECORDING_SHARE_PLAY, RingCapability, @@ -399,6 +402,7 @@ def connection_status(self) -> str | None: return alerts.get("connection") return None + @deprecated("Use async_take_snapshot() instead") async def async_get_snapshot( self, retries: int = 3, delay: int = 1, filename: str | None = None ) -> bytes | None: @@ -423,6 +427,40 @@ async def async_get_snapshot( return snapshot return None + async def async_take_snapshot( + self, max_age: int = 30, max_wait: int = 10, filename: str | None = None + ) -> bytes | None: + """ + Take a snapshot and download it. + + If a snapshot already exists which is less that max_age seconds old, it will + be returned immediately. Otherwise, we'll wait at most max_wait seconds for + a fresh one before giving up. A 404 response from this method typically + indicates a server-side timeout (try increasing max_wait). + """ + request_time_s = time.time() + oldest_ms = int(request_time_s - max_age) * 1000 + + params = { + "after-ms": oldest_ms, + "max-wait-ms": max_wait * 1000, + "extras": "force", + } + + resp = await self._ring.async_query( + TAKE_SNAPSHOT_ENDPOINT.format(self._attrs.get("id")), + extra_params=params, + base_uri=API_URI_SNAPSHOT, + timeout=max_wait + 1, + ) + + snapshot = resp.content + if filename: + async with aiofiles.open(filename, "wb") as jpg: + await jpg.write(snapshot) + return None + return snapshot + def _motion_detection_state(self) -> bool | None: if settings := self._attrs.get("settings"): return settings.get("motion_detection_enabled") diff --git a/tests/test_ring.py b/tests/test_ring.py index def8f47..36961fd 100644 --- a/tests/test_ring.py +++ b/tests/test_ring.py @@ -1,15 +1,18 @@ """The tests for the Ring platform.""" import asyncio +import re +import time from datetime import datetime, timezone import pytest +from freezegun import freeze_time from freezegun.api import FrozenDateTimeFactory from ring_doorbell import Auth, Ring, RingError from ring_doorbell.const import MSG_EXISTING_TYPE, USER_AGENT from ring_doorbell.util import parse_datetime -from .conftest import json_request_kwargs, load_fixture_as_dict +from .conftest import json_request_kwargs, load_fixture_as_dict, nojson_request_kwargs def test_basic_attributes(ring): @@ -146,6 +149,74 @@ async def test_stickup_cam_controls(ring, aioresponses_mock): ) +async def test_doorbot_snapshot(ring, aioresponses_mock): + dev = ring.devices()["doorbots"][0] + kwargs = nojson_request_kwargs() + + freezer = freeze_time("2026-01-01 00:00:01") + freezer.start() + pattern = re.compile(r"^https://app-snaps.ring.com.*$") + aioresponses_mock.get( + pattern, + content_type="image/jpeg", + ) + + # defaults - max_age:30, max_wait:10 + params = { + "after-ms": (int(time.time()) - 30) * 1000, + "max-wait-ms": 10_000, + "extras": "force", + } + kwargs["params"] = params + kwargs["timeout"] = 11 + + snapshot = await dev.async_take_snapshot() + aioresponses_mock.assert_called_with( + url="https://app-snaps.ring.com/snapshots/next/987652", + method="GET", + **kwargs, + ) + assert isinstance(snapshot, bytes) + + # custom values - max_age:0, max_wait:1 + aioresponses_mock.get( + pattern, + content_type="image/jpeg", + ) + + params = { + "after-ms": (int(time.time())) * 1000, + "max-wait-ms": 1_000, + "extras": "force", + } + kwargs["params"] = params + kwargs["timeout"] = 2 + + snapshot = await dev.async_take_snapshot(max_age=0, max_wait=1) + aioresponses_mock.assert_called_with( + url="https://app-snaps.ring.com/snapshots/next/987652", + method="GET", + **kwargs, + ) + assert isinstance(snapshot, bytes) + + # simulate failure, should throw + aioresponses_mock.get( + pattern, + status=404, + ) + with pytest.raises(RingError): + snapshot = await dev.async_take_snapshot(max_age=0, max_wait=1) + + aioresponses_mock.assert_called_with( + url="https://app-snaps.ring.com/snapshots/next/987652", + method="GET", + **kwargs, + ) + + freezer.stop() + + async def test_light_groups(ring): group = ring.groups()["mock-group-id"]