diff --git a/blinkpy/api.py b/blinkpy/api.py index 24f2b9f8..8ea2f7d0 100644 --- a/blinkpy/api.py +++ b/blinkpy/api.py @@ -553,14 +553,19 @@ async def request_get_config(blink, network, camera_id, product_type="owl"): :param blink: Blink instance. :param network: Sync module network id. :param camera_id: ID of camera - :param product_type: Camera product type "owl" or "catalina" + :param product_type: Camera product type """ - if product_type == "owl": + if product_type in ["owl", "hawk"]: url = ( f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}" f"/networks/{network}/owls/{camera_id}/config" ) - elif product_type == "catalina": + elif product_type in ["doorbell", "lotus"]: + url = ( + f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}" + f"/networks/{network}/doorbells/{camera_id}/config" + ) + elif product_type in ["catalina", "sedona"]: url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/config" else: _LOGGER.info( @@ -584,7 +589,7 @@ async def request_update_config( :param product_type: Camera product type "owl" or "catalina" :param data: string w/JSON dict of parameters/values to update """ - if product_type == "owl": + if product_type in ["owl", "hawk"]: url = ( f"{blink.urls.base_url}/api/v1/accounts/" f"{blink.account_id}/networks/{network}/owls/{camera_id}/config" @@ -601,6 +606,59 @@ async def request_update_config( return await http_post(blink, url, json=False, data=data) +async def request_camera_snooze( + blink, network, camera_id, product_type="owl", data=None +): + """ + Update camera snooze configuration. + + :param blink: Blink instance. + :param network: Sync module network id. + :param camera_id: ID of camera + :param product_type: Camera product type "owl", "catalina", + "doorbell", "hawk", "lotus", or "sedona" + :param data: string w/JSON dict of parameters/values to update + """ + product_lookup = { + "catalina": "cameras", + "sedona": "cameras", # Older outdoor cameras use same endpoint as catalina + "owl": "owls", + "hawk": "owls", # Hawk uses same endpoint as owl + "doorbell": "doorbells", + "lotus": "doorbells", # Lotus is internal name for doorbells + } + + if product_type not in product_lookup: + _LOGGER.info( + "Camera %s with product type %s snooze update not implemented.", + camera_id, + product_type, + ) + return None + + url_root = ( + f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}" + f"/networks/{network}" + ) + url = f"{url_root}/{product_lookup[product_type]}/{camera_id}/snooze" + return await http_post(blink, url, json=True, data=data) + + +async def request_sync_snooze(blink, network, data=None): + """ + Update sync snooze configuration. + + :param blink: Blink instance. + :param network: Sync module network id. + :param data: string w/JSON dict of parameters/values to update + """ + url = ( + f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}" + f"/networks/{network}/snooze" + ) + return await http_post(blink, url, json=True, data=data) + + async def http_get( blink, url, stream=False, json=True, is_retry=False, timeout=TIMEOUT ): diff --git a/blinkpy/camera.py b/blinkpy/camera.py index d8332e67..5c759c53 100644 --- a/blinkpy/camera.py +++ b/blinkpy/camera.py @@ -177,6 +177,71 @@ async def async_set_night_vision(self, value): return await res.json() return None + @property + async def snoozed(self): + """Return snooze status as boolean.""" + response_data = None + try: + if self.product_type in ["catalina", "sedona"]: + # Catalina and Sedona cameras have snooze timestamp in config + res = await api.request_get_config( + self.sync.blink, + self.network_id, + self.camera_id, + product_type=self.product_type, + ) + response_data = res + snooze_value = res["camera"][0].get("snooze_till") + # Return True if timestamp exists and is not empty + return bool(snooze_value) + else: + # Owl/hawk/mini/doorbell/lotus cameras get snooze from homescreen + response_data = self.sync.blink.homescreen + + # Determine which homescreen collection to search + if self.product_type in ["doorbell", "lotus"]: + collection_key = "doorbells" + else: # owl, hawk, mini + collection_key = "owls" + + for device in self.sync.blink.homescreen.get(collection_key, []): + # Compare as integers to handle type mismatch + if int(device.get("id")) == int(self.camera_id): + snooze_value = device.get("snooze") + # Return True if snooze value exists and is truthy + return bool(snooze_value) + return False + except TypeError: + return False + except (IndexError, KeyError, ValueError) as e: + _LOGGER.warning( + "Exception %s: Encountered a likely malformed response " + "from the snooze API endpoint. Response: %s", + e, + response_data, + ) + return False + + async def async_snooze(self, snooze_time=3600): + """ + Set camera snooze status. + + :param snooze_time: Time in seconds to snooze camera. Default is 3600 (1 hour). + """ + data = dumps({"snooze_time": snooze_time}) + res = await api.request_camera_snooze( + self.sync.blink, + self.network_id, + self.camera_id, + product_type=self.product_type, + data=data, + ) + # Refresh homescreen to update snooze status + if res and self.product_type not in ["catalina", "sedona"]: + # Owl/hawk/mini/doorbell/lotus cameras use homescreen for snooze status + await self.sync.blink.get_homescreen() + return res + async def record(self): """Initiate clip recording.""" return await api.request_new_video( @@ -540,3 +605,16 @@ def __init__(self, sync): async def get_sensor_info(self): """Get sensor info for blink doorbell camera.""" + + async def get_liveview(self): + """Get liveview link.""" + url = ( + f"{self.sync.urls.base_url}/api/v1/accounts/" + f"{self.sync.blink.account_id}/networks/" + f"{self.sync.network_id}/doorbells/{self.camera_id}/liveview" + ) + response = await api.http_post(self.sync.blink, url) + await api.wait_for_command(self.sync.blink, response) + server = response["server"] + link = server.replace("immis://", "rtsps://") + return link diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 63b6aef6..56305f44 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -127,6 +127,43 @@ async def async_arm(self, value): return await api.request_system_arm(self.blink, self.network_id) return await api.request_system_disarm(self.blink, self.network_id) + @property + async def snoozed(self): + """Return snooze status as boolean.""" + res = None + try: + res = await api.request_sync_snooze( + self.blink, + self.network_id, + ) + if res is None: + return False + snooze_value = res.get("snooze_till") + # Return True if timestamp exists and is not empty + return bool(snooze_value) + except TypeError: + return False + except KeyError as e: + _LOGGER.warning( + "Exception %s: Encountered a likely malformed response " + "from the snooze API endpoint. Response: %s", + e, + res, + ) + return False + + async def async_snooze(self, snooze_time=240): + """Set sync snooze status.""" + data = json_dumps({"snooze_time": snooze_time}) + res = await api.request_sync_snooze( + self.blink, + self.network_id, + data=data, + ) + # Refresh homescreen is not needed for sync-level snooze + # as it's retrieved directly from the sync endpoint + return res + async def start(self): """Initialize the system.""" _LOGGER.debug("Initializing the sync module") diff --git a/tests/test_api.py b/tests/test_api.py index f8f77b89..7574938c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -189,6 +189,54 @@ async def test_request_update_config(self, mock_resp): ) ) + async def test_request_camera_snooze(self, mock_resp): + """Test camera snooze request.""" + mock_resp.return_value = {"message": "Camera snoozed"} + # Test catalina camera + response = await api.request_camera_snooze( + self.blink, "network", "camera_id", "catalina", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Camera snoozed"}) + # Test sedona camera + response = await api.request_camera_snooze( + self.blink, "network", "camera_id", "sedona", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Camera snoozed"}) + # Test owl camera + response = await api.request_camera_snooze( + self.blink, "network", "camera_id", "owl", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Camera snoozed"}) + # Test hawk camera + response = await api.request_camera_snooze( + self.blink, "network", "camera_id", "hawk", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Camera snoozed"}) + # Test doorbell camera + response = await api.request_camera_snooze( + self.blink, "network", "camera_id", "doorbell", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Camera snoozed"}) + # Test lotus camera + response = await api.request_camera_snooze( + self.blink, "network", "camera_id", "lotus", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Camera snoozed"}) + # Test unsupported camera type + self.assertIsNone( + await api.request_camera_snooze( + self.blink, "network", "camera_id", "unsupported_type" + ) + ) + + async def test_request_sync_snooze(self, mock_resp): + """Test sync snooze request.""" + mock_resp.return_value = {"message": "Sync snoozed"} + response = await api.request_sync_snooze( + self.blink, "network", '{"snooze": 300}' + ) + self.assertEqual(response, {"message": "Sync snoozed"}) + async def test_wait_for_command(self, mock_resp): """Test Motion detect enable.""" mock_resp.side_effect = (COMMAND_NOT_COMPLETE, COMMAND_COMPLETE) diff --git a/tests/test_cameras.py b/tests/test_cameras.py index 3437ea8d..b494256a 100644 --- a/tests/test_cameras.py +++ b/tests/test_cameras.py @@ -102,6 +102,271 @@ async def test_doorbell_arm_disarm(self, mock_resp): self.assertEqual(await self.doorbell.async_arm(True), "arm") self.assertEqual(await self.doorbell.async_arm(False), "disarm") + @mock.patch( + "blinkpy.api.request_camera_snooze", + mock.AsyncMock(return_value={"status": 200}), + ) + async def test_camera_snooze(self, mock_resp): + """Test camera snooze.""" + self.camera.product_type = "catalina" + with mock.patch.object( + self.blink, "get_homescreen", mock.AsyncMock() + ) as mock_homescreen: + result = await self.camera.async_snooze(300) + # Catalina cameras should NOT refresh homescreen + mock_homescreen.assert_not_called() + self.assertEqual(result, {"status": 200}) + + @mock.patch( + "blinkpy.api.request_camera_snooze", + mock.AsyncMock(return_value={"status": 400}), + ) + async def test_camera_snooze_failure(self, mock_resp): + """Test camera snooze failure.""" + self.camera.product_type = "owl" + with mock.patch.object( + self.blink, "get_homescreen", mock.AsyncMock() + ) as mock_homescreen: + result = await self.camera.async_snooze(300) + # Non-catalina/sedona cameras refresh homescreen even on failure + mock_homescreen.assert_called_once() + self.assertEqual(result, {"status": 400}) + + @mock.patch( + "blinkpy.api.request_camera_snooze", + mock.AsyncMock(return_value=None), + ) + async def test_camera_snooze_none_response(self, mock_resp): + """Test camera snooze with None response.""" + self.camera.product_type = "doorbell" + with mock.patch.object( + self.blink, "get_homescreen", mock.AsyncMock() + ) as mock_homescreen: + result = await self.camera.async_snooze(240) + # get_homescreen should not be called when response is None + mock_homescreen.assert_not_called() + self.assertIsNone(result) + + @mock.patch( + "blinkpy.api.request_camera_snooze", + mock.AsyncMock(return_value={"status": 200}), + ) + async def test_camera_snooze_mini(self, mock_resp): + """Test mini camera snooze refreshes homescreen.""" + self.camera.product_type = "mini" + with mock.patch.object( + self.blink, "get_homescreen", mock.AsyncMock() + ) as mock_homescreen: + result = await self.camera.async_snooze(300) + # Mini cameras should refresh homescreen + mock_homescreen.assert_called_once() + self.assertEqual(result, {"status": 200}) + + @mock.patch( + "blinkpy.api.request_camera_snooze", + mock.AsyncMock(return_value={"status": 200}), + ) + async def test_camera_snooze_sedona(self, mock_resp): + """Test sedona camera snooze does not refresh homescreen.""" + self.camera.product_type = "sedona" + with mock.patch.object( + self.blink, "get_homescreen", mock.AsyncMock() + ) as mock_homescreen: + result = await self.camera.async_snooze(300) + # Sedona cameras should NOT refresh homescreen + mock_homescreen.assert_not_called() + self.assertEqual(result, {"status": 200}) + + @mock.patch( + "blinkpy.api.request_get_config", + mock.AsyncMock( + return_value={"camera": [{"snooze_till": "2026-02-15T12:00:00+00:00"}]} + ), + ) + async def test_camera_snoozed_catalina(self, mock_resp): + """Test getting snoozed status for catalina camera.""" + self.camera.product_type = "catalina" + self.camera.network_id = "5678" + self.camera.camera_id = "1234" + result = await self.camera.snoozed + self.assertTrue(result) + + async def test_camera_snoozed_owl(self, mock_resp): + """Test getting snoozed status for owl camera.""" + self.camera.product_type = "owl" + self.camera.camera_id = "1234" + self.blink.homescreen = { + "owls": [{"id": "1234", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertTrue(result) + # Verify we're reading from homescreen, not calling API + expected_snooze = "2026-02-15T12:00:00+00:00" + self.assertEqual(self.blink.homescreen["owls"][0]["snooze"], expected_snooze) + + async def test_camera_snoozed_owl_false(self, mock_resp): + """Test getting snoozed status returns False when not snoozed.""" + self.camera.product_type = "owl" + self.camera.camera_id = "9999" + self.blink.homescreen = { + "owls": [{"id": "1234", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertFalse(result) + # Verify camera ID 9999 is not in homescreen + camera_ids = [str(cam["id"]) for cam in self.blink.homescreen["owls"]] + self.assertNotIn("9999", camera_ids) + + async def test_camera_snoozed_doorbell(self, mock_resp): + """Test getting snoozed status for doorbell camera.""" + self.camera.product_type = "doorbell" + self.camera.camera_id = "5678" + self.blink.homescreen = { + "doorbells": [{"id": "5678", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertTrue(result) + # Verify we're reading from the doorbells collection + self.assertIn("doorbells", self.blink.homescreen) + self.assertEqual(self.blink.homescreen["doorbells"][0]["id"], "5678") + + async def test_camera_snoozed_lotus(self, mock_resp): + """Test getting snoozed status for lotus (doorbell) camera.""" + self.camera.product_type = "lotus" + self.camera.camera_id = "5678" + self.blink.homescreen = { + "doorbells": [{"id": "5678", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertTrue(result) + # Verify lotus uses doorbells collection + self.assertIn("doorbells", self.blink.homescreen) + + async def test_camera_snoozed_hawk(self, mock_resp): + """Test getting snoozed status for hawk camera.""" + self.camera.product_type = "hawk" + self.camera.camera_id = "1234" + self.blink.homescreen = { + "owls": [{"id": "1234", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertTrue(result) + # Verify hawk uses owls collection + self.assertIn("owls", self.blink.homescreen) + + @mock.patch( + "blinkpy.api.request_get_config", + mock.AsyncMock( + return_value={"camera": [{"snooze_till": "2026-02-15T12:00:00+00:00"}]} + ), + ) + async def test_camera_snoozed_sedona(self, mock_resp): + """Test getting snoozed status for sedona camera.""" + self.camera.product_type = "sedona" + self.camera.network_id = "5678" + self.camera.camera_id = "1234" + result = await self.camera.snoozed + self.assertTrue(result) + + async def test_camera_snoozed_boolean_true(self, mock_resp): + """Test getting snoozed status when snooze is boolean True.""" + self.camera.product_type = "owl" + self.camera.camera_id = "1234" + self.blink.homescreen = {"owls": [{"id": "1234", "snooze": True}]} + result = await self.camera.snoozed + self.assertTrue(result) + + async def test_camera_snoozed_boolean_false(self, mock_resp): + """Test getting snoozed status when snooze is boolean False.""" + self.camera.product_type = "owl" + self.camera.camera_id = "1234" + self.blink.homescreen = {"owls": [{"id": "1234", "snooze": False}]} + result = await self.camera.snoozed + self.assertFalse(result) + # Verify the snooze value is actually False, not missing + camera = next(c for c in self.blink.homescreen["owls"] if c["id"] == "1234") + self.assertIn("snooze", camera) + self.assertFalse(camera["snooze"]) + + async def test_camera_snoozed_mini(self, mock_resp): + """Test getting snoozed status for mini camera from homescreen.""" + self.camera.product_type = "mini" + self.camera.camera_id = "1234" + self.blink.homescreen = { + "owls": [{"id": "1234", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertTrue(result) + + async def test_camera_snoozed_mini_false(self, mock_resp): + """Test getting snoozed status for mini camera returns False.""" + self.camera.product_type = "mini" + self.camera.camera_id = "9999" + self.blink.homescreen = { + "owls": [{"id": "1234", "snooze": "2026-02-15T12:00:00+00:00"}] + } + result = await self.camera.snoozed + self.assertFalse(result) + # Verify mini camera ID not found in owls collection + camera_ids = [str(cam["id"]) for cam in self.blink.homescreen["owls"]] + self.assertNotIn("9999", camera_ids) + + async def test_camera_snoozed_empty_homescreen(self, mock_resp): + """Test getting snoozed status with empty homescreen collection.""" + self.camera.product_type = "owl" + self.camera.camera_id = "1234" + self.blink.homescreen = {"owls": []} + result = await self.camera.snoozed + self.assertFalse(result) + # Verify owls collection is empty + self.assertEqual(len(self.blink.homescreen["owls"]), 0) + + async def test_camera_snoozed_missing_collection(self, mock_resp): + """Test getting snoozed status when homescreen collection is missing.""" + self.camera.product_type = "doorbell" + self.camera.camera_id = "5678" + self.blink.homescreen = {"owls": []} + result = await self.camera.snoozed + self.assertFalse(result) + # Verify doorbells collection is missing + self.assertNotIn("doorbells", self.blink.homescreen) + + @mock.patch( + "blinkpy.api.request_get_config", + mock.AsyncMock(return_value=None), + ) + async def test_camera_snoozed_none(self, mock_resp): + """Test getting snoozed status with None response.""" + self.camera.product_type = "catalina" + self.camera.network_id = "5678" + self.camera.camera_id = "1234" + result = await self.camera.snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_get_config", + mock.AsyncMock(return_value={"camera": [{}]}), + ) + async def test_camera_snoozed_malformed(self, mock_resp): + """Test getting snoozed status with malformed response.""" + self.camera.product_type = "catalina" + self.camera.network_id = "5678" + self.camera.camera_id = "1234" + result = await self.camera.snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_get_config", + mock.AsyncMock(return_value={"camera": [{"snooze_till": ""}]}), + ) + async def test_camera_snoozed_empty_string(self, mock_resp): + """Test getting snoozed status with empty string.""" + self.camera.product_type = "catalina" + self.camera.network_id = "5678" + self.camera.camera_id = "1234" + result = await self.camera.snoozed + self.assertFalse(result) + def test_missing_attributes(self, mock_resp): """Test that attributes return None if missing.""" self.camera.temperature = None diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index b35f56fd..040beb2c 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -82,6 +82,69 @@ def test_bad_arm(self, mock_resp) -> None: self.assertEqual(self.blink.sync["test"].arm, None) self.assertFalse(self.blink.sync["test"].available) + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"snooze_till": "2026-02-15T12:00:00+00:00"}), + ) + async def test_snoozed(self, mock_resp) -> None: + """Check that we get snoozed status.""" + result = await self.blink.sync["test"].snoozed + self.assertTrue(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value=None), + ) + async def test_snoozed_none(self, mock_resp) -> None: + """Check that we handle None response.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={}), + ) + async def test_snoozed_malformed(self, mock_resp) -> None: + """Check that we handle malformed response.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"snooze_till": ""}), + ) + async def test_snoozed_empty_string(self, mock_resp) -> None: + """Check that we handle empty string response.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"status": 200}), + ) + async def test_async_snooze(self, mock_resp) -> None: + """Check that we can set snooze.""" + result = await self.blink.sync["test"].async_snooze(300) + self.assertEqual(result, {"status": 200}) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"status": 400}), + ) + async def test_async_snooze_failure(self, mock_resp) -> None: + """Check that we handle snooze failure.""" + result = await self.blink.sync["test"].async_snooze(300) + self.assertEqual(result, {"status": 400}) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value=None), + ) + async def test_async_snooze_none_response(self, mock_resp) -> None: + """Check that we handle None response when setting snooze.""" + result = await self.blink.sync["test"].async_snooze(300) + self.assertIsNone(result) + def test_get_unique_info_valid_device(self, mock_resp) -> None: """Check that we get the correct info.""" device = {