Hawk Camera Support

blinkpy/api.py

@@ -568,14 +568,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 "owl", "hawk", "catalina", or "sedona"
     """
     if product_type == "owl":
         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 == "hawk":
+        url = (
+            f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
+            f"/networks/{network}/hawks/{camera_id}/config"
+        )
+    elif product_type in ["catalina", "sedona"]:
         url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/config"
     else:
         _LOGGER.info(
@@ -596,15 +601,20 @@ async def request_update_config(
     :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 "owl", "hawk", "catalina", or "sedona"
     :param data: string w/JSON dict of parameters/values to update
     """
     if product_type == "owl":
         url = (
             f"{blink.urls.base_url}/api/v1/accounts/"
             f"{blink.account_id}/networks/{network}/owls/{camera_id}/config"
         )
-    elif product_type == "catalina":
+    elif product_type == "hawk":
+        url = (
+            f"{blink.urls.base_url}/api/v1/accounts/"
+            f"{blink.account_id}/networks/{network}/hawks/{camera_id}/config"
+        )
+    elif product_type in ["catalina", "sedona"]:
         url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/update"
     else:
         _LOGGER.info(

blinkpy/blinkpy.py

@@ -23,7 +23,7 @@
 from slugify import slugify
 
 from blinkpy import api
-from blinkpy.sync_module import BlinkSyncModule, BlinkOwl, BlinkLotus
+from blinkpy.sync_module import BlinkSyncModule, BlinkOwl, BlinkHawk, BlinkLotus
 from blinkpy.helpers import util
 from blinkpy.helpers.constants import (
     DEFAULT_MOTION_INTERVAL,
@@ -225,7 +225,7 @@ async def setup_owls(self):
                 network_id = str(owl["network_id"])
                 if network_id in self.network_ids:
                     camera_list.append(
-                        {network_id: {"name": name, "id": network_id, "type": "mini"}}
+                        {network_id: {"name": name, "id": owl["id"], "type": "mini"}}
                     )
                     continue
                 if owl["onboarded"]:
@@ -252,7 +252,7 @@ async def setup_lotus(self):
                         {
                             network_id: {
                                 "name": name,
-                                "id": network_id,
+                                "id": lotus["id"],
                                 "type": "doorbell",
                             }
                         }
@@ -269,6 +269,30 @@ async def setup_lotus(self):
         self.network_ids.extend(network_list)
         return camera_list
 
+    async def setup_hawks(self):
+        """Check for blink arc cameras."""
+        network_list = []
+        camera_list = []
+        try:
+            for hawk in self.homescreen["hawks"]:
+                name = hawk["name"]
+                network_id = str(hawk["network_id"])
+                if network_id in self.network_ids:
+                    camera_list.append(
+                        {network_id: {"name": name, "id": hawk["id"], "type": "hawk"}}
+                    )
+                    continue
+                if hawk["onboarded"]:
+                    network_list.append(str(network_id))
+                    self.sync[name] = BlinkHawk(self, name, network_id, hawk)
+                    await self.sync[name].start()
+        except (KeyError, TypeError):
+            # No sync-less devices found
+            pass
+
+        self.network_ids.extend(network_list)
+        return camera_list
+
     async def setup_camera_list(self):
         """Create camera list for onboarded networks."""
         all_cameras = {}
@@ -284,10 +308,14 @@ async def setup_camera_list(self):
                         {"name": camera["name"], "id": camera["id"], "type": "default"}
                     )
             mini_cameras = await self.setup_owls()
+            hawk_cameras = await self.setup_hawks()
             lotus_cameras = await self.setup_lotus()
             for camera in mini_cameras:
                 for network, camera_info in camera.items():
                     all_cameras[network].append(camera_info)
+            for camera in hawk_cameras:
+                for network, camera_info in camera.items():
+                    all_cameras[network].append(camera_info)
             for camera in lotus_cameras:
                 for network, camera_info in camera.items():
                     all_cameras[network].append(camera_info)

blinkpy/camera.py

@@ -570,3 +570,75 @@ def __init__(self, sync):
 
     async def get_sensor_info(self):
         """Get sensor info for blink doorbell camera."""
+
+
+class BlinkCameraHawk(BlinkCamera):
+    """Define a class for a Blink Arc camera."""
+
+    def __init__(self, sync):
+        """Initialize a Blink Arc (Hawk)."""
+        super().__init__(sync)
+        self.camera_type = "hawk"
+        self.product_type = "hawk"
+
+    @property
+    def arm(self):
+        """Return camera arm status."""
+        return self.sync.arm
+
+    async def async_arm(self, value):
+        """Set camera arm status."""
+        url = (
+            f"{self.sync.urls.base_url}/api/v1/accounts/"
+            f"{self.sync.blink.account_id}/networks/"
+            f"{self.network_id}/hawks/{self.camera_id}/config"
+        )
+        data = dumps({"enabled": value})
+        response = await api.http_post(self.sync.blink, url, data=data)
+        await api.wait_for_command(self.sync.blink, response)
+        return response
+
+    async def record(self):
+        """Initiate clip recording for a blink hawk camera."""
+        url = (
+            f"{self.sync.urls.base_url}/api/v1/accounts/"
+            f"{self.sync.blink.account_id}/networks/"
+            f"{self.network_id}/hawks/{self.camera_id}/clip"
+        )
+        response = await api.http_post(self.sync.blink, url)
+        await api.wait_for_command(self.sync.blink, response)
+        return response
+
+    async def snap_picture(self):
+        """Snap picture for a blink hawk camera."""
+        url = (
+            f"{self.sync.urls.base_url}/api/v1/accounts/"
+            f"{self.sync.blink.account_id}/networks/"
+            f"{self.network_id}/hawks/{self.camera_id}/thumbnail"
+        )
+        response = await api.http_post(self.sync.blink, url)
+        await api.wait_for_command(self.sync.blink, response)
+        return response
+
+    async def get_sensor_info(self):
+        """Get sensor info for blink hawk camera."""
+
+    def extract_config_info(self, config):
+        """Extract config info, preserving hawk product type."""
+        super().extract_config_info(config)
+        self.product_type = "hawk"
+
+    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.network_id}/hawks/{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"]
+        server_split = server.split(":")
+        server_split[0] = "rtsps"
+        link = ":".join(server_split)
+        return link

blinkpy/sync_module.py

@@ -9,7 +9,7 @@
 from sortedcontainers import SortedSet
 from requests.structures import CaseInsensitiveDict
 from blinkpy import api
-from blinkpy.camera import BlinkCamera, BlinkCameraMini, BlinkDoorbell
+from blinkpy.camera import BlinkCamera, BlinkCameraHawk, BlinkCameraMini, BlinkDoorbell
 from blinkpy.helpers.util import (
     time_to_seconds,
     backoff_seconds,
@@ -54,6 +54,7 @@ def __init__(self, blink, network_name, network_id, camera_list):
         # Outdoor cameras have their own URL API which must be queried.
         self.type_key_map = {
             "mini": "owls",
+            "hawk": "hawks",
             "doorbell": "doorbells",
         }
         self._names_table = {}
@@ -194,6 +195,7 @@ async def update_cameras(self, camera_type=BlinkCamera):
         """Update cameras from server."""
         type_map = {
             "mini": BlinkCameraMini,
+            "hawk": BlinkCameraHawk,
             "doorbell": BlinkDoorbell,
             "default": BlinkCamera,
         }
@@ -207,12 +209,22 @@ async def update_cameras(self, camera_type=BlinkCamera):
                 name = camera_config["name"]
                 self.motion[name] = False
                 unique_info = self.get_unique_info(name)
-                if blink_camera_type in type_map:
-                    camera_type = type_map[blink_camera_type]
-                self.cameras[name] = camera_type(self)
+
+                # Get camera info first to check product_type
                 camera_info = await self.get_camera_info(
                     camera_config["id"], unique_info=unique_info
                 )
+
+                # Check product_type from camera_info to determine correct class
+                product_type = camera_info.get("type") if camera_info else None
+                if product_type == "hawk":
+                    resolved_camera_type = BlinkCameraHawk
+                elif blink_camera_type in type_map:
+                    resolved_camera_type = type_map[blink_camera_type]
+                else:
+                    resolved_camera_type = camera_type
+
+                self.cameras[name] = resolved_camera_type(self)
                 self._names_table[to_alphanumeric(name)] = name
                 await self.cameras[name].update(
                     camera_info, force_cache=True, force=True
@@ -644,6 +656,69 @@ def network_info(self, value):
         """Set network_info property."""
 
 
+class BlinkHawk(BlinkSyncModule):
+    """Representation of a sync-less device."""
+
+    def __init__(self, blink, name, network_id, response):
+        """Initialize a sync-less object."""
+        cameras = [{"name": name, "id": response["id"], "type": "hawk"}]
+        super().__init__(blink, name, network_id, cameras)
+        self.sync_id = response["id"]
+        self.serial = response["serial"]
+        self.status = response["enabled"]
+        if not self.serial:
+            self.serial = f"{network_id}-{self.sync_id}"
+
+    async def sync_initialize(self):
+        """Initialize a sync-less module."""
+        self.summary = {
+            "id": self.sync_id,
+            "name": self.name,
+            "serial": self.serial,
+            "status": self.status,
+            "onboarded": True,
+            "account_id": self.blink.account_id,
+            "network_id": self.network_id,
+        }
+        return self.summary
+
+    async def update_cameras(self, camera_type=None):
+        """Update sync-less cameras."""
+        return await super().update_cameras(camera_type=BlinkCameraHawk)
+
+    async def get_camera_info(self, camera_id, **kwargs):
+        """Retrieve camera information."""
+        try:
+            for hawk in self.blink.homescreen["hawks"]:
+                if hawk["name"] == self.name:
+                    self.status = hawk["enabled"]
+                    return hawk
+        except (TypeError, KeyError):
+            pass
+        return None
+
+    async def get_network_info(self):
+        """Get network info for sync-less module."""
+        return True
+
+    @property
+    def network_info(self):
+        """Format hawk response to resemble sync module."""
+        return {
+            "network": {
+                "id": self.network_id,
+                "name": self.name,
+                "armed": self.status,
+                "sync_module_error": False,
+                "account_id": self.blink.account_id,
+            }
+        }
+
+    @network_info.setter
+    def network_info(self, value):
+        """Set network_info property."""
+
+
 class LocalStorageMediaItem:
     """Metadata of media item in the local storage manifest."""
 

tests/test_api.py

@@ -171,6 +171,10 @@ async def test_request_get_config(self, mock_resp):
             ),
             {"config": "values"},
         )
+        self.assertEqual(
+            await api.request_get_config(self.blink, "network", "camera_id", "hawk"),
+            {"config": "values"},
+        )
 
     async def test_request_update_config(self, mock_resp):
         """Test Motion detect enable."""
@@ -183,6 +187,10 @@ async def test_request_update_config(self, mock_resp):
             self.blink, "network", "camera_id", "catalina"
         )
         self.assertEqual(response.status, 200)
+        response = await api.request_update_config(
+            self.blink, "network", "camera_id", "hawk"
+        )
+        self.assertEqual(response.status, 200)
         self.assertIsNone(
             await api.request_update_config(
                 self.blink, "network", "camera_id", "other_camera"