Add support for camera snapshots

abodepy/devices/camera.py

@@ -1,4 +1,5 @@
 """Abode camera device."""
+import base64
 import json
 import logging
 from shutil import copyfileobj
@@ -20,6 +21,7 @@ def __init__(self, json_obj, abode):
         """Set up Abode alarm device."""
         AbodeDevice.__init__(self, json_obj, abode)
         self._image_url = None
+        self._snapshot_base64 = None
 
     def capture(self):
         """Request a new camera image."""
@@ -115,6 +117,47 @@ def image_to_file(self, path, get_image=True):
 
         return True
 
+    def snapshot(self):
+        """Request the current camera snapshot as a base64-encoded string."""
+        url = CONST.CAMERA_INTEGRATIONS_URL + self._device_uuid + '/snapshot'
+
+        try:
+            response = self._abode.send_request("post", url)
+            _LOGGER.debug("Camera snapshot response: %s", response.text)
+        except AbodeException as exc:
+            _LOGGER.warning("Failed to get camera snapshot image: %s", exc)
+            return False
+
+        self._snapshot_base64 = json.loads(response.text).get('base64Image')
+        if self._snapshot_base64 is None:
+            _LOGGER.warning("Camera snapshot data missing")
+            return False
+
+        return True
+
+    def snapshot_to_file(self, path, get_snapshot=True):
+        """Write the snapshot image to a file."""
+        if not self._snapshot_base64 or get_snapshot:
+            if not self.snapshot():
+                return False
+
+        try:
+            with open(path, 'wb') as imgfile:
+                imgfile.write(base64.b64decode(self._snapshot_base64))
+        except OSError as exc:
+            _LOGGER.warning("Failed to write snapshot image to file: %s", exc)
+            return False
+
+        return True
+
+    def snapshot_data_url(self, get_snapshot=True):
+        """Return the snapshot image as a data url."""
+        if not self._snapshot_base64 or get_snapshot:
+            if not self.snapshot():
+                return ''
+
+        return 'data:image/jpeg;base64,' + self._snapshot_base64
+
     def privacy_mode(self, enable):
         """Set camera privacy mode (camera on/off)."""
         if self._json_state['privacy']:

abodepy/helpers/constants.py

@@ -60,6 +60,7 @@
 PANEL_URL = BASE_URL + 'api/v1/panel'
 
 INTEGRATIONS_URL = BASE_URL + 'integrations/v1/devices/'
+CAMERA_INTEGRATIONS_URL = BASE_URL + 'integrations/v1/camera/'
 
 
 def get_panel_mode_url(area, mode):

requirements_test.txt

@@ -1,7 +1,7 @@
 flake8>=3.6.0
-flake8-docstrings==1.1.0
+flake8-docstrings==1.4.0
 pylint==2.4.2
-pydocstyle==2.0.0
+pydocstyle==2.1.0
 pytest==5.2.4
 pytest-cov>=2.3.1
 pytest-sugar==0.9.2

tests/mock/devices/ir_camera.py

@@ -32,6 +32,7 @@ def device(devid=DEVICE_ID, status=CONST.STATUS_ONLINE,
           "sresp_mode_3":"0",
           "sresp_entry_3":"0",
           "sresp_exit_3":"0",
+          "uuid": "1234567890",
           "version":"852_00.00.03.05TC",
           "origin":"abode",
           "control_url":"''' + CONTROL_URL + '''",

tests/test_camera.py

@@ -1,4 +1,5 @@
 """Test the Abode camera class."""
+import base64
 import os
 import unittest
 
@@ -344,6 +345,136 @@ def tests_camera_image_write(self, m):
             m.get(url, text="[]")
             self.assertFalse(device.image_to_file(path, get_image=True))
 
+    def tests_camera_snapshot(self, m):
+        """Tests that camera devices capture new snapshots."""
+        # Set up URL's
+        m.post(CONST.LOGIN_URL, text=LOGIN.post_response_ok())
+        m.get(CONST.OAUTH_TOKEN_URL, text=OAUTH_CLAIMS.get_response_ok())
+        m.post(CONST.LOGOUT_URL, text=LOGOUT.post_response_ok())
+        m.get(CONST.PANEL_URL,
+              text=PANEL.get_response_ok(mode=CONST.MODE_STANDBY))
+        m.get(CONST.DEVICES_URL, text=self.all_devices)
+
+        # Test our camera devices
+        for device in self.abode.get_devices():
+            # Skip alarm devices
+            if device.type_tag == CONST.DEVICE_ALARM:
+                continue
+
+            # Specify which device module to use based on type_tag
+            cam_type = set_cam_type(device.type_tag)
+
+            # Test that we have the camera devices
+            self.assertIsNotNone(device)
+            self.assertEqual(device.status, CONST.STATUS_ONLINE)
+
+            # Set up snapshot URL response
+            snapshot_url = (CONST.CAMERA_INTEGRATIONS_URL +
+                            device.device_uuid + '/snapshot')
+            m.post(snapshot_url, text='{"base64Image":"test"}')
+
+            # Retrieve a snapshot
+            self.assertTrue(device.snapshot())
+
+            # Failed snapshot retrieval due to timeout response
+            m.post(snapshot_url, text=cam_type.get_capture_timeout(),
+                   status_code=600)
+            self.assertFalse(device.snapshot())
+
+            # Failed snapshot retrieval due to missing data
+            m.post(snapshot_url, text="{}")
+            self.assertFalse(device.snapshot())
+
+    def tests_camera_snapshot_write(self, m):
+        """Tests that camera snapshots will write to a file."""
+        # Set up URL's
+        m.post(CONST.LOGIN_URL, text=LOGIN.post_response_ok())
+        m.get(CONST.OAUTH_TOKEN_URL, text=OAUTH_CLAIMS.get_response_ok())
+        m.post(CONST.LOGOUT_URL, text=LOGOUT.post_response_ok())
+        m.get(CONST.PANEL_URL,
+              text=PANEL.get_response_ok(mode=CONST.MODE_STANDBY))
+        m.get(CONST.DEVICES_URL, text=self.all_devices)
+
+        # Test our camera devices
+        for device in self.abode.get_devices():
+            # Skip alarm devices
+            if device.type_tag == CONST.DEVICE_ALARM:
+                continue
+
+            # Specify which device module to use based on type_tag
+            cam_type = set_cam_type(device.type_tag)
+
+            # Test that we have our device
+            self.assertIsNotNone(device)
+            self.assertEqual(device.status, CONST.STATUS_ONLINE)
+
+            # Set up snapshot URL and image response
+            snapshot_url = (CONST.CAMERA_INTEGRATIONS_URL +
+                            device.device_uuid + '/snapshot')
+            image_response = b'this is a beautiful jpeg image'
+            b64_image = str(base64.b64encode(image_response), 'utf-8')
+            m.post(snapshot_url,
+                   text='{"base64Image":"' + b64_image + '"}')
+
+            # Request the snapshot and write to file
+            path = "test.jpg"
+            self.assertTrue(device.snapshot_to_file(path, get_snapshot=True))
+
+            # Test the file written and cleanup
+            image_data = open(path, 'rb').read()
+            self.assertTrue(image_response, image_data)
+            os.remove(path)
+
+            # Test that bad response returns False
+            m.post(snapshot_url, text=cam_type.get_capture_timeout(),
+                   status_code=600)
+            self.assertFalse(device.snapshot_to_file(path, get_snapshot=True))
+
+    def tests_camera_snapshot_data_url(self, m):
+        """Tests that camera snapshots can be converted to a data url."""
+        # Set up URL's
+        m.post(CONST.LOGIN_URL, text=LOGIN.post_response_ok())
+        m.get(CONST.OAUTH_TOKEN_URL, text=OAUTH_CLAIMS.get_response_ok())
+        m.post(CONST.LOGOUT_URL, text=LOGOUT.post_response_ok())
+        m.get(CONST.PANEL_URL,
+              text=PANEL.get_response_ok(mode=CONST.MODE_STANDBY))
+        m.get(CONST.DEVICES_URL, text=self.all_devices)
+
+        # Test our camera devices
+        for device in self.abode.get_devices():
+            # Skip alarm devices
+            if device.type_tag == CONST.DEVICE_ALARM:
+                continue
+
+            # Specify which device module to use based on type_tag
+            cam_type = set_cam_type(device.type_tag)
+
+            # Test that we have our device
+            self.assertIsNotNone(device)
+            self.assertEqual(device.status, CONST.STATUS_ONLINE)
+
+            # Set up snapshot URL and image response
+            snapshot_url = (CONST.CAMERA_INTEGRATIONS_URL +
+                            device.device_uuid + '/snapshot')
+            image_response = b'this is a beautiful jpeg image'
+            b64_image = str(base64.b64encode(image_response), 'utf-8')
+            m.post(snapshot_url,
+                   text='{"base64Image":"' + b64_image + '"}')
+
+            # Request the snapshot as a data url
+            data_url = device.snapshot_data_url(get_snapshot=True)
+
+            # Test the data url matches the image response
+            header, encoded = data_url.split(',', 1)
+            decoded = base64.b64decode(encoded)
+            self.assertEqual(header, 'data:image/jpeg;base64')
+            self.assertEqual(decoded, image_response)
+
+            # Test that bad response returns an empty string
+            m.post(snapshot_url, text=cam_type.get_capture_timeout(),
+                   status_code=600)
+            self.assertEqual(device.snapshot_data_url(get_snapshot=True), '')
+
     def tests_camera_privacy_mode(self, m):
         """Tests camera privacy mode."""
         # Set up mock URLs