Improvements to motion detection and download methods, add to examples

Author: Bobrock

examples/download_motions.py

@@ -0,0 +1,43 @@
+import os
+from configparser import RawConfigParser
+from datetime import datetime as dt, timedelta
+from reolink_api import Camera
+
+
+def read_config(props_path: str) -> dict:
+    """Reads in a properties file into variables.
+
+    NB! this config file is kept out of commits with .gitignore. The structure of this file is such:
+    # secrets.cfg
+        [camera]
+        ip={ip_address}
+        username={username}
+        password={password}
+    """
+    config = RawConfigParser()
+    assert os.path.exists(props_path), f"Path does not exist: {props_path}"
+    config.read(props_path)
+    return config
+
+
+# Read in your ip, username, & password
+#   (NB! you'll likely have to create this file. See tests/test_camera.py for details on structure)
+config = read_config('../secrets.cfg')
+
+ip = config.get('camera', 'ip')
+un = config.get('camera', 'username')
+pw = config.get('camera', 'password')
+
+# Connect to camera
+cam = Camera(ip, un, pw)
+
+start = (dt.now() - timedelta(hours=1))
+end = dt.now()
+# Collect motion events between these timestamps for substream
+processed_motions = cam.get_motion_files(start=start, end=end, streamtype='sub')
+
+dl_dir = os.path.join(os.path.expanduser('~'), 'Downloads')
+for i, motion in enumerate(processed_motions):
+    fname = motion['filename']
+    # Download the mp4
+    resp = cam.get_file(fname, output_path=os.path.join(dl_dir, f'motion_event_{i}.mp4'))

reolink_api/APIHandler.py

@@ -1,3 +1,4 @@
+import requests
 from reolink_api.resthandle import Request
 from .alarm import AlarmAPIMixin
 from .device import DeviceAPIMixin
@@ -109,7 +110,23 @@ def _execute_command(self, command, data, multi=False):
         try:
             if self.token is None:
                 raise ValueError("Login first")
-            response = Request.post(self.url, data=data, params=params)
+            if command == 'Download':
+                # Special handling for downloading an mp4
+                # Pop the filepath from data
+                tgt_filepath = data[0].pop('filepath')
+                # Apply the data to the params
+                params.update(data[0])
+                with requests.get(self.url, params=params, stream=True) as req:
+                    if req.status_code == 200:
+                        with open(tgt_filepath, 'wb') as f:
+                            f.write(req.content)
+                        return True
+                    else:
+                        print(f'Error received: {req.status_code}')
+                        return False
+
+            else:
+                response = Request.post(self.url, data=data, params=params)
             return response.json()
         except Exception as e:
             print(f"Command {command} failed: {e}")

reolink_api/download.py

@@ -1,6 +1,6 @@
 class DownloadAPIMixin:
     """API calls for downloading video files."""
-    def get_file(self, filename: str) -> object:
+    def get_file(self, filename: str, output_path: str) -> bool:
         """
         Download the selected video file
         :return: response json
@@ -9,7 +9,10 @@ def get_file(self, filename: str) -> object:
             {
                 "cmd": "Download",
                 "source": filename,
-                "output": filename
+                "output": filename,
+                "filepath": output_path
             }
         ]
-        return self._execute_command('Download', body)
+        resp = self._execute_command('Download', body)
+
+        return resp

reolink_api/motion.py

@@ -1,10 +1,16 @@
+from typing import Union, List, Dict
 from datetime import datetime as dt
 
 
+# Type hints for input and output of the motion api response
+RAW_MOTION_LIST_TYPE = List[Dict[str, Union[str, int, Dict[str, str]]]]
+PROCESSED_MOTION_LIST_TYPE = List[Dict[str, Union[str, dt]]]
+
+
 class MotionAPIMixin:
     """API calls for past motion alerts."""
     def get_motion_files(self, start: dt, end: dt = dt.now(),
-                         streamtype: str = 'main') -> object:
+                         streamtype: str = 'sub') -> PROCESSED_MOTION_LIST_TYPE:
         """
         Get the timestamps and filenames of motion detection events for the time range provided.
 
@@ -38,4 +44,36 @@ def get_motion_files(self, start: dt, end: dt = dt.now(),
             }
         }
         body = [{"cmd": "Search", "action": 1, "param": search_params}]
-        return self._execute_command('Search', body)
+
+        resp = self._execute_command('Search', body)[0]
+        files = resp['value']['SearchResult']['File']
+        if len(files) > 0:
+            # Begin processing files
+            processed_files = self._process_motion_files(files)
+            return processed_files
+        return []
+
+    @staticmethod
+    def _process_motion_files(motion_files: RAW_MOTION_LIST_TYPE) -> PROCESSED_MOTION_LIST_TYPE:
+        """Processes raw list of dicts containing motion timestamps
+        and the filename associated with them"""
+        # Process files
+        processed_motions = []
+        replace_fields = {'mon': 'month', 'sec': 'second', 'min': 'minute'}
+        for file in motion_files:
+            time_range = {}
+            for x in ['Start', 'End']:
+                # Get raw dict
+                raw = file[f'{x}Time']
+                # Replace certain keys
+                for k, v in replace_fields.items():
+                    if k in raw.keys():
+                        raw[v] = raw.pop(k)
+                time_range[x.lower()] = dt(**raw)
+            start, end = time_range.values()
+            processed_motions.append({
+                'start': start,
+                'end': end,
+                'filename': file['name']
+            })
+        return processed_motions