Per camera snooze

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
 ):

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

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")

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)