Merge branch 'Benehiko-master'

Author: Max Ziermann

APIHandler.py

@@ -1,4 +1,4 @@
-from APIHandler import APIHandler
+from api import APIHandler
 
 
 class Camera(APIHandler):
@@ -11,8 +11,9 @@ def __init__(self, ip, username="admin", password="", https=False):
         :param username:
         :param password:
         """
-        # For when you need to connect to a camera behind a proxy
-        APIHandler.__init__(self, ip, username, password, proxy={"http": "socks5://127.0.0.1:8000"}, https=https)
+        # For when you need to connect to a camera behind a proxy, pass
+        # a proxy argument: proxy={"http": "socks5://127.0.0.1:8000"}
+        APIHandler.__init__(self, ip, username, password, https=https)
 
         # Normal call without proxy:
         # APIHandler.__init__(self, ip, username, password)

Camera.py

@@ -0,0 +1,13 @@
+[[source]]
+name = "pypi"
+url = "https://pypi.org/simple"
+verify_ssl = true
+
+[dev-packages]
+
+[packages]
+pillow = "*"
+pyyaml = "*"
+requests = "*"
+
+[requires]

Pipfile

@@ -0,0 +1,90 @@
+from api.recording import RecordingAPIMixin
+from .device import DeviceAPIMixin
+from .display import DisplayAPIMixin
+from .network import NetworkAPIMixin
+from .system import SystemAPIMixin
+from .user import UserAPIMixin
+from resthandle import Request
+
+
+class APIHandler(SystemAPIMixin,
+                 NetworkAPIMixin,
+                 UserAPIMixin,
+                 DeviceAPIMixin,
+                 DisplayAPIMixin,
+                 RecordingAPIMixin):
+    """
+    The APIHandler class is the backend part of the API, the actual API calls
+    are implemented in Mixins.
+    This handles communication directly with the camera.
+    Current camera's tested: RLC-411WS
+
+    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):
+        """
+        Initialise the Camera API Handler (maps api calls into python)
+        :param ip:
+        :param username:
+        :param password:
+        :param proxy: Add a proxy dict for requests to consume.
+        eg: {"http":"socks5://[username]:[password]@[host]:[port], "https": ...}
+        More information on proxies in requests: https://stackoverflow.com/a/15661226/9313679
+        """
+        scheme = 'https' if https else 'http'
+        self.url = f"{scheme}://{ip}/cgi-bin/api.cgi"
+        self.ip = ip
+        self.token = None
+        self.username = username
+        self.password = password
+        Request.proxies = kwargs.get("proxy")  # Defaults to None if key isn't found
+
+    def login(self) -> bool:
+        """
+        Get login token
+        Must be called first, before any other operation can be performed
+        :return: bool
+        """
+        try:
+            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)
+            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")
+                    return True
+                print(self.token)
+                return False
+            else:
+                print("Failed to login\nStatus Code:", response.status_code)
+                return False
+        except Exception as e:
+            print("Error Login\n", e)
+            raise
+        
+    def _execute_command(self, command, data, multi=False):
+        """
+        Send a POST request to the IP camera with given data.
+        :param command: name of the command to send
+        :param data: object to send to the camera (send as json)
+        :param multi: whether the given command name should be added to the
+        url parameters of the request. Defaults to False. (Some multi-step
+        commands seem to not have a single command name)
+        :return: response JSON as python object
+        """
+        params = {"token": self.token, 'cmd': command}
+        if multi:
+            del params['cmd']
+        try:
+            if self.token is None:
+                raise ValueError("Login first")
+            response = Request.post(self.url, data=data, params=params)
+            return response.json()
+        except Exception as e:
+            print(f"Command {command} failed: {e}")
+            raise

Pipfile.lock

@@ -0,0 +1 @@
+from .APIHandler import APIHandler

api/APIHandler.py

@@ -0,0 +1,23 @@
+class DeviceAPIMixin:
+    """API calls for getting device information."""
+    def get_hdd_info(self) -> object:
+        """
+        Gets all HDD and SD card information from Camera
+        See examples/response/GetHddInfo.json for example response data.
+        :return: response json
+        """
+        body = [{"cmd": "GetHddInfo", "action": 0, "param": {}}]
+        return self._execute_command('GetHddInfo', body)
+
+    def format_hdd(self, hdd_id: [int] = [0]) -> bool:
+        """
+        Format specified HDD/SD cards with their id's
+        :param hdd_id: List of id's specified by the camera with get_hdd_info api. Default is 0 (SD card)
+        :return: bool
+        """
+        body = [{"cmd": "Format", "action": 0, "param": {"HddInfo": {"id": hdd_id}}}]
+        r_data = self._execute_command('Format', body)[0]
+        if r_data["value"]["rspCode"] == 200:
+            return True
+        print("Could not format HDD/SD. Camera responded with:", r_data["value"])
+        return False

api/__init__.py

@@ -0,0 +1,47 @@
+class DisplayAPIMixin:
+    """API calls related to the current image (osd, on screen display)."""
+
+    def get_osd(self) -> object:
+        """
+        Get OSD information.
+        See examples/response/GetOsd.json for example response data.
+        :return: response json
+        """
+        body = [{"cmd": "GetOsd", "action": 1, "param": {"channel": 0}}]
+        return self._execute_command('GetOsd', body)
+
+    def get_mask(self) -> object:
+        """
+        Get the camera mask information.
+        See examples/response/GetMask.json for example response data.
+        :return: response json
+        """
+        body = [{"cmd": "GetMask", "action": 1, "param": {"channel": 0}}]
+        return self._execute_command('GetMask', body)
+
+    def set_osd(self, bg_color: bool = 0, channel: int = 0, osd_channel_enabled: bool = 0, osd_channel_name: str = "",
+                osd_channel_pos: str = "Lower Right", osd_time_enabled: bool = 0,
+                osd_time_pos: str = "Lower Right") -> bool:
+        """
+        Set OSD
+        :param bg_color: bool
+        :param channel: int channel id
+        :param osd_channel_enabled: bool
+        :param osd_channel_name: string channel name
+        :param osd_channel_pos: string channel position ["Upper Left","Top Center","Upper Right","Lower Left","Bottom Center","Lower Right"]
+        :param osd_time_enabled: bool
+        :param osd_time_pos: string time position ["Upper Left","Top Center","Upper Right","Lower Left","Bottom Center","Lower Right"]
+        :return: whether the action was successful
+        """
+        body = [{"cmd": "SetOsd", "action": 1, "param": {
+            "Osd": {"bgcolor": bg_color, "channel": channel,
+                    "osdChannel": {"enable": osd_channel_enabled, "name": osd_channel_name,
+                                   "pos": osd_channel_pos},
+                    "osdTime": {"enable": osd_time_enabled, "pos": osd_time_pos}
+                    }
+        }}]
+        r_data = self._execute_command('SetOsd', body)[0]
+        if r_data["value"]["rspCode"] == 200:
+            return True
+        print("Could not set OSD. Camera responded with status:", r_data["value"])
+        return False

api/device.py

@@ -0,0 +1,52 @@
+class NetworkAPIMixin:
+    """API calls for network settings."""
+    def set_net_port(self, http_port=80, https_port=443, media_port=9000, onvif_port=8000, rtmp_port=1935,
+                     rtsp_port=554) -> bool:
+        """
+        Set network ports
+        If nothing is specified, the default values will be used
+        :param rtsp_port: int
+        :param rtmp_port: int
+        :param onvif_port: int
+        :param media_port: int
+        :param https_port: int
+        :type http_port: int
+        :return: bool
+        """
+        body = [{"cmd": "SetNetPort", "action": 0, "param": {"NetPort": {
+            "httpPort": http_port,
+            "httpsPort": https_port,
+            "mediaPort": media_port,
+            "onvifPort": onvif_port,
+            "rtmpPort": rtmp_port,
+            "rtspPort": rtsp_port
+        }}}]
+        self._execute_command('SetNetPort', body, multi=True)
+        print("Successfully Set Network Ports")
+        return True
+
+    def set_wifi(self, ssid, password) -> object:
+        body = [{"cmd": "SetWifi", "action": 0, "param": {
+            "Wifi": {
+                "ssid": ssid,
+                "password": password
+            }}}]
+        return self._execute_command('SetWifi', body)
+
+    def get_net_ports(self) -> object:
+        """
+        Get network ports
+        :return: response json
+        """
+        body = [{"cmd": "GetNetPort", "action": 1, "param": {}},
+                {"cmd": "GetUpnp", "action": 0, "param": {}},
+                {"cmd": "GetP2p", "action": 0, "param": {}}]
+        return self._execute_command('GetNetPort', body, multi=True)
+
+    def get_wifi(self):
+        body = [{"cmd": "GetWifi", "action": 1, "param": {}}]
+        return self._execute_command('GetWifi', body)
+
+    def scan_wifi(self):
+        body = [{"cmd": "ScanWifi", "action": 1, "param": {}}]
+        return self._execute_command('ScanWifi', body)