Replace print statements with log, add PyYAML to requirements, fix test and add more type hints

Author: barretobrock

.gitignore

@@ -1,3 +1,4 @@
+*.bak
 secrets.cfg
 # Byte-compiled / optimized / DLL files
 __pycache__/

CHANGELOG.md

@@ -15,6 +15,17 @@ _Note: 'Unreleased' section below is used for untagged changes that will be issu
 #### Security
 __BEGIN-CHANGELOG__
 
+### [1.0.1] - 2022-04-24
+#### Added
+ - More type hints (albeit they'll need to be made more precise later)
+ - `PyYAML` was left out of requirements upon scanning the older `setup.py` file, so that was added & pinned
+#### Changed
+ - `print` statements are now log outputs
+ - Some dictionaries were simplified and optimized
+#### Fixed
+ - Minor variable misspellings
+ - `test_camera.py` now doesn't test a real camera connection
+ 
 ### [1.0.0] - 2022-04-23
 #### Added
  - `CHANGELOG.md`

poetry.lock

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "reolink_api"
-version = '1.0.0'
+version = '1.0.1'
 description = "Unofficial Reolink API fork for personal use in Python 3.10"
 authors = ["bobrock <[email protected]>"]
 license = "GPL-3.0"
@@ -19,6 +19,8 @@ include = ["CHANGELOG.md"]
 python = "^3.10"
 Pillow = "9.1.0"
 PySocks = "1.7.1"
+PyYAML = "6.0"
+loguru = "0.6.0"
 numpy = "^1.22.3"
 opencv-python = "4.5.5.64"
 requests = "^2.27.1"

pyproject.toml

@@ -1,17 +1,23 @@
+from typing import (
+    Dict,
+    List,
+    Union
+)
 import requests
+from loguru import logger
+from reolink_api.alarm import AlarmAPIMixin
+from reolink_api.device import DeviceAPIMixin
+from reolink_api.display import DisplayAPIMixin
+from reolink_api.download import DownloadAPIMixin
+from reolink_api.image import ImageAPIMixin
+from reolink_api.motion import MotionAPIMixin
+from reolink_api.network import NetworkAPIMixin
+from reolink_api.ptz import PtzAPIMixin
+from reolink_api.recording import RecordingAPIMixin
 from reolink_api.resthandle import Request
-from .alarm import AlarmAPIMixin
-from .device import DeviceAPIMixin
-from .display import DisplayAPIMixin
-from .download import DownloadAPIMixin
-from .image import ImageAPIMixin
-from .motion import MotionAPIMixin
-from .network import NetworkAPIMixin
-from .ptz import PtzAPIMixin
-from .recording import RecordingAPIMixin
-from .system import SystemAPIMixin
-from .user import UserAPIMixin
-from .zoom import ZoomAPIMixin
+from reolink_api.system import SystemAPIMixin
+from reolink_api.user import UserAPIMixin
+from reolink_api.zoom import ZoomAPIMixin
 
 
 class APIHandler(AlarmAPIMixin,
@@ -35,7 +41,7 @@ class APIHandler(AlarmAPIMixin,
     All Code will try to follow the PEP 8 standard as described here: https://www.python.org/dev/peps/pep-0008/
     """
 
-    def __init__(self, ip: str, username: str, password: str, https=False, **kwargs):
+    def __init__(self, ip: str, username: str, password: str, https: bool = False, **kwargs):
         """
         Initialise the Camera API Handler (maps api calls into python)
         :param ip:
@@ -63,21 +69,22 @@ def login(self) -> bool:
             body = [{"cmd": "Login", "action": 0,
                      "param": {"User": {"userName": self.username, "password": self.password}}}]
             param = {"cmd": "Login", "token": "null"}
-            response = Request.post(self.url, data=body, params=param)
+            response = Request.post(self.url, data=body, params=param)  # type: requests.Response
             if response is not None:
                 data = response.json()[0]
                 code = data["code"]
                 if int(code) == 0:
                     self.token = data["value"]["Token"]["name"]
-                    print("Login success")
+                    logger.debug("Login success")
                     return True
-                print(self.token)
+                logger.debug(self.token)
                 return False
             else:
-                print("Failed to login\nStatus Code:", response.status_code)
+                # Response object was NoneType (empty)
+                logger.warning("Failed to login\nError in request.")
                 return False
         except Exception as e:
-            print("Error Login\n", e)
+            logger.error("Error Login\n", e)
             raise
 
     def logout(self) -> bool:
@@ -88,13 +95,12 @@ def logout(self) -> bool:
         try:
             data = [{"cmd": "Logout", "action": 0}]
             self._execute_command('Logout', data)
-            # print(ret)
             return True
         except Exception as e:
-            print("Error Logout\n", e)
+            logger.error("Error Logout\n", e)
             return False
 
-    def _execute_command(self, command, data, multi=False):
+    def _execute_command(self, command: str, data: List[Dict], multi: bool = False) -> Union[bool, Dict]:
         """
         Send a POST request to the IP camera with given data.
         :param command: name of the command to send
@@ -122,12 +128,12 @@ def _execute_command(self, command, data, multi=False):
                             f.write(req.content)
                         return True
                     else:
-                        print(f'Error received: {req.status_code}')
+                        logger.error(f'Error received: {req.status_code}')
                         return False
 
             else:
                 response = Request.post(self.url, data=data, params=params)
-            return response.json()
+                return response.json()
         except Exception as e:
-            print(f"Command {command} failed: {e}")
+            logger.error(f"Command {command} failed: {e}")
             raise

reolink_api/APIHandler.py

@@ -1,16 +1,21 @@
 import io
+from typing import (
+    Dict,
+    Optional
+)
+from loguru import logger
 import yaml
 
 
 class ConfigHandler:
     camera_settings = {}
 
     @staticmethod
-    def load() -> yaml or None:
+    def load() -> Optional[Dict]:
         try:
             stream = io.open("config.yml", 'r', encoding='utf8')
             data = yaml.safe_load(stream)
             return data
         except Exception as e:
-            print("Config Property Error\n", e)
+            logger.error("Config Property Error\n", e)
         return None

reolink_api/ConfigHandler.py

@@ -1,5 +1,6 @@
 import os
 from threading import ThreadError
+from loguru import logger
 import cv2
 from reolink_api.util import threaded
 
@@ -12,7 +13,8 @@ class RtspClient:
         - https://stackoverflow.com/questions/55828451/video-streaming-from-ip-camera-in-python-using-opencv-cv2-videocapture
     """
 
-    def __init__(self, ip, username, password, port=554, profile="main", use_udp=True, callback=None, **kwargs):
+    def __init__(self, ip: str, username: str, password: str, port: int = 554, profile: str = "main",
+                 use_udp: bool = True, callback=None, **kwargs):
         """
         RTSP client is used to retrieve frames from the camera in a stream
 
@@ -34,12 +36,8 @@ def __init__(self, ip, username, password, port=554, profile="main", use_udp=Tru
         self.password = password
         self.port = port
         self.proxy = kwargs.get("proxies")
-        self.url = "rtsp://" + self.username + ":" + self.password + "@" + \
-                   self.ip + ":" + str(self.port) + "//h264Preview_01_" + profile
-        if use_udp:
-            capture_options = capture_options + 'udp'
-        else:
-            capture_options = capture_options + 'tcp'
+        self.url = f'rtsp://{self.username}:{self.password}@{self.ip}:{self.port}//h264Preview_01_{profile}'
+        capture_options += 'udp' if use_udp else 'tcp'
 
         os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = capture_options
 
@@ -50,19 +48,19 @@ def _open_video_capture(self):
         # To CAP_FFMPEG or not To ?
         self.capture = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
 
-    def _stream_blocking(self):
+    def _stream_blocking(self) -> object:
         while True:
             try:
                 if self.capture.isOpened():
                     ret, frame = self.capture.read()
                     if ret:
                         yield frame
                 else:
-                    print("stream closed")
+                    logger.info("stream closed")
                     self.capture.release()
                     return
             except Exception as e:
-                print(e)
+                logger.error(e)
                 self.capture.release()
                 return
 
@@ -75,17 +73,17 @@ def _stream_non_blocking(self):
                     if ret:
                         self.callback(frame)
                 else:
-                    print("stream is closed")
+                    logger.info("stream is closed")
                     self.stop_stream()
             except ThreadError as e:
-                print(e)
+                logger.error(e)
                 self.stop_stream()
 
     def stop_stream(self):
         self.capture.release()
         self.thread_cancelled = True
 
-    def open_stream(self):
+    def open_stream(self) -> object:
         """
         Opens OpenCV Video stream and returns the result according to the OpenCV documentation
         https://docs.opencv.org/3.4/d8/dfe/classcv_1_1VideoCapture.html#a473055e77dd7faa4d26d686226b292c1
@@ -95,7 +93,7 @@ def open_stream(self):
         if self.capture is None or not self.capture.isOpened():
             self._open_video_capture()
 
-        print("opening stream")
+        logger.info("opening stream")
 
         if self.callback is None:
             return self._stream_blocking()

reolink_api/RtspClient.py

@@ -1,5 +1,5 @@
 from .APIHandler import APIHandler
 from .Camera import Camera
 
-__version__ = "1.0.0"
-__update_date__ = '2022-04-23_10:56:59'
+__version__ = '1.0.1'
+__update_date__ = '2022-04-24_10:52:20'