diff --git a/abodepy/devices/camera.py b/abodepy/devices/camera.py index c4e8f13..907a0a9 100644 --- a/abodepy/devices/camera.py +++ b/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']: diff --git a/abodepy/helpers/constants.py b/abodepy/helpers/constants.py index b43135d..1c3cb18 100644 --- a/abodepy/helpers/constants.py +++ b/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): diff --git a/requirements_test.txt b/requirements_test.txt index 4072852..fa06a10 100644 --- a/requirements_test.txt +++ b/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 diff --git a/tests/mock/devices/ir_camera.py b/tests/mock/devices/ir_camera.py index 045fbd5..665ecd8 100644 --- a/tests/mock/devices/ir_camera.py +++ b/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 + '''", diff --git a/tests/test_camera.py b/tests/test_camera.py index ce21ee8..abc5087 100644 --- a/tests/test_camera.py +++ b/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