Merge pull request #724 from fronzbot/dev

0.21.0

Author: fronzbot

CHANGES.rst

@@ -4,6 +4,29 @@ Changelog
 
 A list of changes between each release
 
+0.21.0 (2023-05-28)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+**Bugfixes**
+
+- None
+
+**New Features**
+
+- Add get_videos_metadata function (`@rhhayward #685 <https://github.com/fronzbot/blinkpy/pull/685>`__)
+- Add night vision toggling support (`@jrhunger #717 <https://github.com/fronzbot/blinkpy/pull/717>`__)
+- Add doorbell arming functionality (`@mkmer #719 <https://github.com/fronzbot/blinkpy/pull/719>`__)
+
+**Other Changes**
+
+- Upgrade pylint to 2.17.4
+- Upgrade coverage to 7.2.5
+- Upgrade pygments to 2.15.1
+- Upgrade pytest to 7.3.1
+- Upgrade pytest-sugar to 0.9.7
+- Upgrade black to 23.3.0
+
+
 0.20.0 (2023-01-29)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

blinkpy/api.py

@@ -352,6 +352,51 @@ def request_local_storage_clip(blink, network, sync_id, manifest_id, clip_id):
     return http_post(blink, url)
 
 
+def request_get_config(blink, network, camera_id, product_type="owl"):
+    """Get camera configuration.
+
+    :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"
+    """
+    if product_type == "owl":
+        url = f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}/networks/{network}/owls/{camera_id}/config"
+    elif product_type == "catalina":
+        url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/config"
+    else:
+        _LOGGER.info(
+            "Camera %s with product type %s config get not implemented.",
+            camera_id,
+            product_type,
+        )
+        return None
+    return http_get(blink, url)
+
+
+def request_update_config(blink, network, camera_id, product_type="owl", data=None):
+    """Update camera configuration.
+
+    :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 data: string w/JSON dict of parameters/values to update
+    """
+    if product_type == "owl":
+        url = f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}/networks/{network}/owls/{camera_id}/update"
+    elif product_type == "catalina":
+        url = f"{blink.urls.base_url}/network/{network}/camera/{camera_id}/update"
+    else:
+        _LOGGER.info(
+            "Camera %s with product type %s config update not implemented.",
+            camera_id,
+            product_type,
+        )
+        return None
+    return http_post(blink, url, json=False, data=data)
+
+
 def http_get(blink, url, stream=False, json=True, is_retry=False, timeout=TIMEOUT):
     """Perform an http get request.
 

blinkpy/blinkpy.py

@@ -330,6 +330,22 @@ def download_videos(
         :param debug: Set to TRUE to prevent downloading of items.
                       Instead of downloading, entries will be printed to log.
         """
+        if not isinstance(camera, list):
+            camera = [camera]
+
+        results = self.get_videos_metadata(since=since, stop=stop)
+        self._parse_downloaded_items(results, camera, path, delay, debug)
+
+    def get_videos_metadata(self, since=None, camera="all", stop=10):
+        """
+        Fetch and return video metadata.
+
+        :param since: Date and time to get videos from.
+                      Ex: "2018/07/28 12:33:00" to retrieve videos since
+                      July 28th 2018 at 12:33:00
+        :param stop: Page to stop on (~25 items per page. Default page 10).
+        """
+        videos = []
         if since is None:
             since_epochs = self.last_refresh
         else:
@@ -339,21 +355,33 @@ def download_videos(
         formatted_date = util.get_time(time_to_convert=since_epochs)
         _LOGGER.info("Retrieving videos since %s", formatted_date)
 
-        if not isinstance(camera, list):
-            camera = [camera]
-
         for page in range(1, stop):
             response = api.request_videos(self, time=since_epochs, page=page)
             _LOGGER.debug("Processing page %s", page)
             try:
                 result = response["media"]
                 if not result:
                     raise KeyError
+                videos.extend(result)
             except (KeyError, TypeError):
                 _LOGGER.info("No videos found on page %s. Exiting.", page)
                 break
+        return videos
 
-            self._parse_downloaded_items(result, camera, path, delay, debug)
+    def do_http_get(self, address):
+        """
+        Do an http_get on address.
+
+        :param address: address to be added to base_url.
+        """
+        response = api.http_get(
+            self,
+            url=f"{self.urls.base_url}{address}",
+            stream=True,
+            json=False,
+            timeout=TIMEOUT_MEDIA,
+        )
+        return response
 
     def _parse_downloaded_items(self, result, camera, path, delay, debug):
         """Parse downloaded videos."""
@@ -375,7 +403,6 @@ def _parse_downloaded_items(self, result, camera, path, delay, debug):
                 _LOGGER.debug("%s: %s is marked as deleted.", camera_name, address)
                 continue
 
-            clip_address = f"{self.urls.base_url}{address}"
             filename = f"{camera_name}-{created_at}"
             filename = f"{slugify(filename)}.mp4"
             filename = os.path.join(path, filename)
@@ -385,13 +412,7 @@ def _parse_downloaded_items(self, result, camera, path, delay, debug):
                     _LOGGER.info("%s already exists, skipping...", filename)
                     continue
 
-                response = api.http_get(
-                    self,
-                    url=clip_address,
-                    stream=True,
-                    json=False,
-                    timeout=TIMEOUT_MEDIA,
-                )
+                response = self.do_http_get(address)
                 with open(filename, "wb") as vidfile:
                     copyfileobj(response.raw, vidfile)
 

blinkpy/camera.py

@@ -109,6 +109,49 @@ def arm(self, value):
             self.sync.blink, self.network_id, self.camera_id
         )
 
+    @property
+    def night_vision(self):
+        """Return night_vision status."""
+        res = api.request_get_config(
+            self.sync.blink,
+            self.network_id,
+            self.camera_id,
+            product_type=self.product_type,
+        )
+        if res is None:
+            return None
+        if self.product_type == "catalina":
+            res = res.get("camera", [{}])[0]
+        if res["illuminator_enable"] in [0, 1, 2]:
+            res["illuminator_enable"] = ["off", "on", "auto"][
+                res.get("illuminator_enable")
+            ]
+        nv_keys = [
+            "night_vision_control",
+            "illuminator_enable",
+            "illuminator_enable_v2",
+        ]
+        return {key: res.get(key) for key in nv_keys}
+
+    @night_vision.setter
+    def night_vision(self, value):
+        """Set camera night_vision status."""
+        if value not in ["on", "off", "auto"]:
+            return None
+        if self.product_type == "catalina":
+            value = {"off": 0, "on": 1, "auto": 2}.get(value, None)
+        data = dumps({"illuminator_enable": value})
+        res = api.request_update_config(
+            self.sync.blink,
+            self.network_id,
+            self.camera_id,
+            product_type=self.product_type,
+            data=data,
+        )
+        if res.ok:
+            return res.json()
+        return None
+
     def record(self):
         """Initiate clip recording."""
         return api.request_new_video(self.sync.blink, self.network_id, self.camera_id)
@@ -440,14 +483,17 @@ def __init__(self, sync):
     @property
     def arm(self):
         """Return camera arm status."""
-        return self.sync.arm
+        return self.motion_enabled
 
     @arm.setter
     def arm(self, value):
         """Set camera arm status."""
-        _LOGGER.warning(
-            "Individual camera motion detection enable/disable for Blink Doorbell is unsupported at this time."
-        )
+        url = f"{self.sync.urls.base_url}/api/v1/accounts/{self.sync.blink.account_id}/networks/{self.sync.network_id}/doorbells/{self.camera_id}"
+        if value:
+            url = f"{url}/enable"
+        else:
+            url = f"{url}/disable"
+        return api.http_post(self.sync.blink, url)
 
     def snap_picture(self):
         """Snap picture for a blink doorbell camera."""

blinkpy/helpers/constants.py

@@ -3,7 +3,7 @@
 import os
 
 MAJOR_VERSION = 0
-MINOR_VERSION = 20
+MINOR_VERSION = 21
 PATCH_VERSION = 0
 
 __version__ = f"{MAJOR_VERSION}.{MINOR_VERSION}.{PATCH_VERSION}"

requirements_test.txt

@@ -1,15 +1,15 @@
-black==22.12.0
-coverage==7.1.0
+black==23.3.0
+coverage==7.2.5
 flake8==6.0.0
-pre-commit==3.0.2
+pre-commit==3.0.4
 flake8-docstrings==1.7.0
-pylint==2.15.10
+pylint==2.17.4
 pydocstyle==6.3.0
-pytest==7.2.1
+pytest==7.3.1
 pytest-cov==3.0.0
-pytest-sugar==0.9.6
+pytest-sugar==0.9.7
 pytest-timeout==2.1.0
 restructuredtext-lint==1.4.0
-pygments==2.14.0
+pygments==2.15.1
 testtools>=2.4.0
 sortedcontainers~=2.4.0

tests/test_blink_functions.py

@@ -9,6 +9,8 @@
 from blinkpy.camera import BlinkCamera
 from blinkpy.helpers.util import get_time, BlinkURLHandler
 
+from requests import Response
+
 
 class MockSyncModule(BlinkSyncModule):
     """Mock blink sync module object."""
@@ -117,6 +119,41 @@ def test_parse_downloaded_throttle(self, mock_req):
         delta = now - start
         self.assertTrue(delta >= 0.1)
 
+    @mock.patch("blinkpy.blinkpy.api.request_videos")
+    def test_get_videos_metadata(self, mock_req):
+        """Test ability to fetch videos metadata."""
+        blink = blinkpy.Blink()
+        generic_entry = {
+            "created_at": "1970",
+            "device_name": "foo",
+            "deleted": True,
+            "media": "/bar.mp4",
+        }
+        result = [generic_entry]
+        mock_req.return_value = {"media": result}
+        blink.last_refresh = 0
+
+        results = blink.get_videos_metadata(stop=2)
+        expected_results = [
+            {
+                "created_at": "1970",
+                "device_name": "foo",
+                "deleted": True,
+                "media": "/bar.mp4",
+            }
+        ]
+        self.assertListEqual(results, expected_results)
+
+    @mock.patch("blinkpy.blinkpy.api.http_get")
+    def test_do_http_get(self, mock_req):
+        """Test ability to do_http_get."""
+        blink = blinkpy.Blink()
+        blink.urls = BlinkURLHandler("test")
+
+        mock_req.return_value = Response()
+        response = blink.do_http_get("/path/to/request")
+        self.assertTrue(response is not None)
+
     @mock.patch("blinkpy.blinkpy.api.request_videos")
     def test_parse_camera_not_in_list(self, mock_req):
         """Test ability to parse downloaded items list."""

tests/test_cameras.py

@@ -46,12 +46,29 @@ def tearDown(self):
     def test_camera_arm_status(self, mock_resp):
         """Test arming and disarming camera."""
         self.camera.motion_enabled = None
+        self.camera.arm = None
         self.assertFalse(self.camera.arm)
+        self.camera.arm = False
         self.camera.motion_enabled = False
         self.assertFalse(self.camera.arm)
+        self.camera.arm = True
         self.camera.motion_enabled = True
         self.assertTrue(self.camera.arm)
 
+    def test_doorbell_camera_arm(self, mock_resp):
+        """Test arming and disarming camera."""
+        self.blink.sync.arm = False
+        doorbell_camera = BlinkDoorbell(self.blink.sync["test"])
+        doorbell_camera.motion_enabled = None
+        doorbell_camera.arm = None
+        self.assertFalse(doorbell_camera.arm)
+        doorbell_camera.arm = False
+        doorbell_camera.motion_enabled = False
+        self.assertFalse(doorbell_camera.arm)
+        doorbell_camera.arm = True
+        doorbell_camera.motion_enabled = True
+        self.assertTrue(doorbell_camera.arm)
+
     def test_missing_attributes(self, mock_resp):
         """Test that attributes return None if missing."""
         self.camera.temperature = None