diff --git a/blinkpy/api.py b/blinkpy/api.py index 68063f92..c50a1497 100644 --- a/blinkpy/api.py +++ b/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,7 +601,7 @@ 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": @@ -604,7 +609,12 @@ async def request_update_config( 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( diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index 9171f987..55a9f98a 100644 --- a/blinkpy/blinkpy.py +++ b/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) diff --git a/blinkpy/camera.py b/blinkpy/camera.py index 33276963..69527315 100644 --- a/blinkpy/camera.py +++ b/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 diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 63b6aef6..31f84366 100644 --- a/blinkpy/sync_module.py +++ b/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.""" diff --git a/tests/test_api.py b/tests/test_api.py index f8f77b89..7e71700c 100644 --- a/tests/test_api.py +++ b/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" diff --git a/tests/test_blinkpy.py b/tests/test_blinkpy.py index 771189f7..cb15b17f 100644 --- a/tests/test_blinkpy.py +++ b/tests/test_blinkpy.py @@ -245,9 +245,7 @@ async def test_blink_mini_cameras_returned(self): } result = await self.blink.setup_owls() self.assertEqual(self.blink.network_ids, ["1234"]) - self.assertEqual( - result, [{"1234": {"name": "foo", "id": "1234", "type": "mini"}}] - ) + self.assertEqual(result, [{"1234": {"name": "foo", "id": 1, "type": "mini"}}]) self.blink.no_owls = True self.blink.network_ids = [] @@ -276,9 +274,7 @@ async def test_blink_mini_attached_to_sync(self, mock_usage): } mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} result = await self.blink.setup_camera_list() - self.assertEqual( - result, {"1234": [{"name": "foo", "id": "1234", "type": "mini"}]} - ) + self.assertEqual(result, {"1234": [{"name": "foo", "id": 1, "type": "mini"}]}) @mock.patch("blinkpy.blinkpy.BlinkLotus.start") async def test_initialize_blink_doorbells(self, mock_start): @@ -338,7 +334,7 @@ async def test_blink_doorbell_attached_to_sync(self, mock_usage): mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} result = await self.blink.setup_camera_list() self.assertEqual( - result, {"1234": [{"name": "foo", "id": "1234", "type": "doorbell"}]} + result, {"1234": [{"name": "foo", "id": 1, "type": "doorbell"}]} ) @mock.patch("blinkpy.api.request_camera_usage") @@ -371,8 +367,8 @@ async def test_blink_multi_doorbell(self, mock_usage): } expected = { "1234": [ - {"name": "foo", "id": "1234", "type": "doorbell"}, - {"name": "bar", "id": "1234", "type": "doorbell"}, + {"name": "foo", "id": 1, "type": "doorbell"}, + {"name": "bar", "id": 2, "type": "doorbell"}, ] } mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} @@ -409,8 +405,8 @@ async def test_blink_multi_mini(self, mock_usage): } expected = { "1234": [ - {"name": "foo", "id": "1234", "type": "mini"}, - {"name": "bar", "id": "1234", "type": "mini"}, + {"name": "foo", "id": 1, "type": "mini"}, + {"name": "bar", "id": 2, "type": "mini"}, ] } mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} @@ -469,10 +465,10 @@ async def test_blink_camera_mix(self, mock_usage): } expected = { "1234": [ - {"name": "foo", "id": "1234", "type": "doorbell"}, - {"name": "bar", "id": "1234", "type": "doorbell"}, - {"name": "dead", "id": "1234", "type": "mini"}, - {"name": "beef", "id": "1234", "type": "mini"}, + {"name": "foo", "id": 1, "type": "doorbell"}, + {"name": "bar", "id": 2, "type": "doorbell"}, + {"name": "dead", "id": 3, "type": "mini"}, + {"name": "beef", "id": 4, "type": "mini"}, {"name": "normal", "id": "1234", "type": "default"}, ] } diff --git a/tests/test_hawk_as_sync.py b/tests/test_hawk_as_sync.py new file mode 100644 index 00000000..d8fc5e21 --- /dev/null +++ b/tests/test_hawk_as_sync.py @@ -0,0 +1,49 @@ +"""Tests camera and system functions.""" + +from unittest import mock +from unittest import IsolatedAsyncioTestCase +import pytest +from blinkpy.blinkpy import Blink +from blinkpy.helpers.util import BlinkURLHandler +from blinkpy.sync_module import BlinkHawk +from blinkpy.camera import BlinkCameraHawk + + +@mock.patch("blinkpy.auth.Auth.query") +class TestBlinkSyncModule(IsolatedAsyncioTestCase): + """Test BlinkSyncModule functions in blinkpy.""" + + def setUp(self): + """Set up Blink module.""" + self.blink = Blink(motion_interval=0, session=mock.AsyncMock()) + self.blink.last_refresh = 0 + self.blink.urls = BlinkURLHandler("test") + response = { + "name": "test", + "id": 2, + "serial": "foobar123", + "enabled": True, + "network_id": 1, + "thumbnail": "/foo/bar", + } + self.blink.homescreen = {"hawks": [response]} + self.blink.sync["test"] = BlinkHawk(self.blink, "test", "1234", response) + self.blink.sync["test"].network_info = {"network": {"armed": True}} + + def tearDown(self): + """Clean up after test.""" + self.blink = None + + def test_sync_attributes(self, mock_resp): + """Test sync attributes.""" + self.assertEqual(self.blink.sync["test"].attributes["name"], "test") + self.assertEqual(self.blink.sync["test"].attributes["network_id"], "1234") + + @pytest.mark.asyncio + async def test_hawk_start(self, mock_resp): + """Test hawk camera instantiation.""" + self.blink.last_refresh = None + hawk = self.blink.sync["test"] + self.assertTrue(await hawk.start()) + self.assertTrue("test" in hawk.cameras) + self.assertEqual(hawk.cameras["test"].__class__, BlinkCameraHawk) diff --git a/tests/test_sync_functions.py b/tests/test_sync_functions.py index f31928f7..f740ac65 100644 --- a/tests/test_sync_functions.py +++ b/tests/test_sync_functions.py @@ -8,7 +8,7 @@ from blinkpy.blinkpy import Blink from blinkpy.helpers.util import BlinkURLHandler from blinkpy.sync_module import BlinkSyncModule -from blinkpy.camera import BlinkCamera, BlinkCameraMini, BlinkDoorbell +from blinkpy.camera import BlinkCamera, BlinkCameraMini, BlinkDoorbell, BlinkCameraHawk @mock.patch("blinkpy.auth.Auth.query") @@ -178,11 +178,13 @@ async def test_sync_with_mixed_cameras(self, mock_resp): {"name": "foo", "id": 10, "type": "default"}, {"name": "bar", "id": 11, "type": "mini"}, {"name": "fake", "id": 12, "type": "doorbell"}, + {"name": "hawk_cam", "id": 13, "type": "hawk"}, ] self.blink.homescreen = { "owls": [{"name": "bar", "id": 3}], "doorbells": [{"name": "fake", "id": 12}], + "hawks": [{"name": "hawk_cam", "id": 13}], } side_effect = [ @@ -195,6 +197,7 @@ async def test_sync_with_mixed_cameras(self, mock_resp): resp_empty, resp_empty, resp_empty, + resp_empty, ] mock_resp.side_effect = side_effect @@ -205,6 +208,7 @@ async def test_sync_with_mixed_cameras(self, mock_resp): self.assertEqual(test_sync.cameras["foo"].__class__, BlinkCamera) self.assertEqual(test_sync.cameras["bar"].__class__, BlinkCameraMini) self.assertEqual(test_sync.cameras["fake"].__class__, BlinkDoorbell) + self.assertEqual(test_sync.cameras["hawk_cam"].__class__, BlinkCameraHawk) # Now shuffle the cameras and see if it still works for i in range(0, 10): @@ -221,6 +225,9 @@ async def test_sync_with_mixed_cameras(self, mock_resp): self.assertEqual( test_sync.cameras["fake"].__class__, BlinkDoorbell, msg=debug_msg ) + self.assertEqual( + test_sync.cameras["hawk_cam"].__class__, BlinkCameraHawk, msg=debug_msg + ) @pytest.mark.asyncio async def test_init_local_storage(self, mock_resp): diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index b35f56fd..76a8ba1e 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -88,7 +88,7 @@ def test_get_unique_info_valid_device(self, mock_resp) -> None: "enabled": True, "name": "doorbell1", } - self.blink.homescreen = {"doorbells": [device], "owls": []} + self.blink.homescreen = {"doorbells": [device], "owls": [], "hawks": []} self.assertEqual(self.blink.sync["test"].get_unique_info("doorbell1"), device) def test_get_unique_info_invalid_device(self, mock_resp) -> None: @@ -97,7 +97,7 @@ def test_get_unique_info_invalid_device(self, mock_resp) -> None: "enabled": True, "name": "doorbell1", } - self.blink.homescreen = {"doorbells": [device], "owls": []} + self.blink.homescreen = {"doorbells": [device], "owls": [], "hawks": []} self.assertEqual(self.blink.sync["test"].get_unique_info("doorbell2"), None) async def test_get_events(self, mock_resp) -> None: