Merge pull request #67 from tchellomello/0.1.1

Version 0.1.1

pyarlo/__init__.py

@@ -198,6 +198,7 @@ def lookup_camera_by_id(self, device_id):
             lambda cam: cam.device_id == device_id, self.cameras))[0]
         if camera:
             return camera
+        return None
 
     def refresh_attributes(self, name):
         """Refresh attributes from a given Arlo object."""
@@ -206,6 +207,7 @@ def refresh_attributes(self, name):
         for device in data:
             if device.get('deviceName') == name:
                 return device
+        return None
 
     @property
     def unseen_videos_reset(self):

pyarlo/camera.py

@@ -2,7 +2,8 @@
 """Generic Python Class file for Netgear Arlo camera module."""
 import logging
 from pyarlo.const import (
-    RESET_CAM_ENDPOINT, STREAM_ENDPOINT, STREAMING_BODY)
+    RESET_CAM_ENDPOINT, STREAM_ENDPOINT, STREAMING_BODY,
+    SNAPSHOTS_ENDPOINT, SNAPSHOTS_BODY)
 from pyarlo.media import ArloMediaLibrary
 from pyarlo.utils import http_get
 
@@ -278,6 +279,50 @@ def live_streaming(self):
             return ret.get('data').get('url')
         return ret.get('data')
 
+    @property
+    def snapshot_url(self):
+        """Return the snapshot url."""
+        # Snapshot should be scheduled first.  It will
+        # available a couple seconds after.
+        # If a GET request fails on this URL, trying
+        # again is logical since the snapshot isn't
+        # taken immediately.  Snapshots will be cached for a
+        # predefined amount of time.
+        return self._attrs.get('presignedFullFrameSnapshotUrl')
+
+    def schedule_snapshot(self):
+        """Trigger snapshot to be uploaded to AWS.
+        Return success state."""
+        # Notes:
+        #  - Snapshots are not immediate.
+        #  - Snapshots will be cached for predefined amount
+        #    of time.
+        #  - Snapshots are not balanced. To get a better
+        #    image, it must be taken from the stream, a few
+        #    seconds after stream start.
+        url = SNAPSHOTS_ENDPOINT
+        params = SNAPSHOTS_BODY
+        params['from'] = "{0}_web".format(self.user_id)
+        params['to'] = self.device_id
+        params['resource'] = "cameras/{0}".format(self.device_id)
+        params['transId'] = "web!{0}".format(self.xcloud_id)
+
+        # override headers
+        headers = {'xCloudId': self.xcloud_id}
+
+        _LOGGER.debug("Snapshot device %s", self.name)
+        _LOGGER.debug("Device params %s", params)
+        _LOGGER.debug("Device headers %s", headers)
+
+        ret = self._session.query(url,
+                                  method='POST',
+                                  extra_params=params,
+                                  extra_headers=headers)
+
+        _LOGGER.debug("Snapshot results %s", ret)
+
+        return ret is not None and ret.get('success')
+
     def update(self):
         """Update object properties."""
         self._attrs = self._session.refresh_attributes(self.name)

pyarlo/const.py

@@ -17,6 +17,7 @@
 RESET_ENDPOINT = LIBRARY_ENDPOINT + "/reset"
 RESET_CAM_ENDPOINT = RESET_ENDPOINT + "/?uniqueId={0}"
 STREAM_ENDPOINT = API_URL + "/users/devices/startStream"
+SNAPSHOTS_ENDPOINT = API_URL + "/users/devices/fullFrameSnapshot"
 
 # number of days to preload video
 PRELOAD_DAYS = 30
@@ -56,4 +57,16 @@
     'transId': "",
 }
 
+
+# define body used for live_streaming
+SNAPSHOTS_BODY = {
+    'action': 'set',
+    'from': None,
+    'properties': {'activityState': 'fullFrameSnapshot'},
+    'publishResponse': 'true',
+    'resource': None,
+    'to': None,
+    'transId': ""
+}
+
 # vim:sw=4:ts=4:et:

setup.py

@@ -5,7 +5,7 @@
 setup(
     name='pyarlo',
     packages=['pyarlo'],
-    version='0.1.0',
+    version='0.1.1',
     description='Python Arlo is a library written in Python 2.7/3x ' +
                 'that exposes the Netgear Arlo cameras as Python objects.',
     author='Marcelo Moreira de Mello',

tests/common.py

@@ -46,3 +46,8 @@ def load_camera_schedule(*args, **kwargs):
 def load_camera_live_streaming(*args, **kwargs):
     """Load camera live streaming response as dict."""
     return load_fixture_json("pyarlo_camera_live_streaming.json")
+
+
+def load_camera_schedule_snapshot(*args, **kwargs):
+    """Load camera live streaming response as dict."""
+    return load_fixture_json("pyarlo_success.json")

tests/fixtures/pyarlo_camera_properties.json

@@ -90,7 +90,8 @@
             "idleLedEnable": true,
             "batteryTech": "Rechargeable",
             "resolution": {"width": 1280, "height": 720},
-            "chargeNotificationLedEnable": true
+            "chargeNotificationLedEnable": true,
+            "presignedFullFrameSnapshotUrl": "https://www.example.com/1"
         },
         {
             "signalStrength": 4,
@@ -128,7 +129,8 @@
             "idleLedEnable": true,
             "batteryTech": "Rechargeable",
             "resolution": {"width": 1280, "height": 720},
-            "chargeNotificationLedEnable": true
+            "chargeNotificationLedEnable": true,
+            "presignedFullFrameSnapshotUrl": "https://www.example.com/2"
         }
     ]
 }

tests/fixtures/pyarlo_failure.json

@@ -0,0 +1 @@
+{"success": false}

tests/fixtures/pyarlo_success.json

@@ -1 +1 @@
-{"success": false}
+{"success": true}

tests/test_camera.py

@@ -7,6 +7,7 @@
     load_fixture_json,
     load_camera_live_streaming,
     load_camera_properties as load_camera_props,
+    load_camera_schedule_snapshot,
     open_fixture
 )
 
@@ -16,7 +17,7 @@
 from pyarlo.const import (
     DEVICES_ENDPOINT, LIBRARY_ENDPOINT, LOGIN_ENDPOINT,
     NOTIFY_ENDPOINT, RESET_CAM_ENDPOINT, STREAM_ENDPOINT,
-    UNSUBSCRIBE_ENDPOINT
+    UNSUBSCRIBE_ENDPOINT, SNAPSHOTS_ENDPOINT
 )
 
 BASE_STATION_ID = "48B14CBBBBBBB"
@@ -150,3 +151,16 @@ def test_live_streaming(self, mock):
         self.assertEqual(
             streaming_url, mocked_streaming_response["data"]["url"]
         )
+
+    @requests_mock.Mocker()
+    def test_schedule_snapshot(self, mock):
+        """Test ArloCamera.live_streaming."""
+        arlo = self.load_arlo(mock)
+        camera = arlo.cameras[0]
+        response_text = load_fixture("pyarlo_success.json")
+        mock.post(SNAPSHOTS_ENDPOINT, text=response_text)
+        mocked_snapshot_response = load_camera_schedule_snapshot()
+        status = camera.schedule_snapshot()
+        self.assertEqual(
+            status, mocked_snapshot_response["success"]
+        )

tests/test_modes.py

@@ -79,7 +79,7 @@ def test_set_mode(self, mock):
         """Test PyArlo BaseStation.mode property."""
         notify_url = NOTIFY_ENDPOINT.format("48b14cbbbbbbb")
 
-        mock.post(notify_url, text=load_fixture("pyarlo_success.json"))
+        mock.post(notify_url, text=load_fixture("pyarlo_failure.json"))
         base_station = self.load_base_station(mock)
 
         base_station.mode = "Inside"