From 28c6f3f2a1ed4b596310bd624c415baece981a8a Mon Sep 17 00:00:00 2001 From: Max Ziermann Date: Fri, 6 Mar 2020 17:22:40 +0100 Subject: [PATCH] Split APIHandler into single files --- api/APIHandler.py | 365 ++-------------------------------------------- api/device.py | 23 +++ api/display.py | 47 ++++++ api/network.py | 52 +++++++ api/recording.py | 69 +++++++++ api/system.py | 32 ++++ api/user.py | 62 ++++++++ 7 files changed, 299 insertions(+), 351 deletions(-) create mode 100644 api/device.py create mode 100644 api/display.py create mode 100644 api/network.py create mode 100644 api/recording.py create mode 100644 api/system.py create mode 100644 api/user.py diff --git a/api/APIHandler.py b/api/APIHandler.py index 2417fbf..61274e7 100644 --- a/api/APIHandler.py +++ b/api/APIHandler.py @@ -1,26 +1,25 @@ -import io -import json -import random -import string -import sys -from urllib import request - -import numpy -import rtsp -from PIL import Image - -from RtspClient import RtspClient +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: +class APIHandler(SystemAPIMixin, + NetworkAPIMixin, + UserAPIMixin, + DeviceAPIMixin, + DisplayAPIMixin, + RecordingAPIMixin): """ - The APIHandler class is the backend part of the API. + 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): @@ -41,10 +40,6 @@ class APIHandler: self.password = password Request.proxies = kwargs.get("proxy") # Defaults to None if key isn't found - ########### - # Token - ########### - def login(self) -> bool: """ Get login token @@ -93,335 +88,3 @@ class APIHandler: except Exception as e: print(f"Command {command} failed: {e}") raise - - ########### - # NETWORK - ########### - - ########### - # SET Network - ########### - 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) -> json: - body = [{"cmd": "SetWifi", "action": 0, "param": { - "Wifi": { - "ssid": ssid, - "password": password - }}}] - return self._execute_command('SetWifi', body) - - ########### - # GET - ########### - def get_net_ports(self) -> json: - """ - 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) - - ########### - # Display - ########### - - ########### - # GET - ########### - def get_osd(self) -> json: - """ - 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) -> json: - """ - 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) - - ########### - # SET - ########### - 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) - if r_data["value"]["rspCode"] == "200": - return True - print("Could not set OSD. Camera responded with status:", r_data["value"]) - return False - - ########### - # SYSTEM - ########### - - ########### - # GET - ########### - def get_general_system(self) -> json: - """:return: response json""" - body = [{"cmd": "GetTime", "action": 1, "param": {}}, {"cmd": "GetNorm", "action": 1, "param": {}}] - return self._execute_command('get_general_system', body, multi=True) - - def get_performance(self) -> json: - """ - Get a snapshot of the current performance of the camera. - See examples/response/GetPerformance.json for example response data. - :return: response json - """ - body = [{"cmd": "GetPerformance", "action": 0, "param": {}}] - return self._execute_command('GetPerformance', body) - - def get_information(self) -> json: - """ - Get the camera information - See examples/response/GetDevInfo.json for example response data. - :return: response json - """ - body = [{"cmd": "GetDevInfo", "action": 0, "param": {}}] - return self._execute_command('GetDevInfo', body) - - ########### - # SET - ########### - def reboot_camera(self) -> json: - """ - Reboots the camera - :return: response json - """ - body = [{"cmd": "Reboot", "action": 0, "param": {}}] - return self._execute_command('Reboot', body) - - ########## - # User - ########## - - ########## - # GET - ########## - def get_online_user(self) -> json: - """ - Return a list of current logged-in users in json format - See examples/response/GetOnline.json for example response data. - :return: response json - """ - body = [{"cmd": "GetOnline", "action": 1, "param": {}}] - return self._execute_command('GetOnline', body) - - def get_users(self) -> json: - """ - Return a list of user accounts from the camera in json format. - See examples/response/GetUser.json for example response data. - :return: response json - """ - body = [{"cmd": "GetUser", "action": 1, "param": {}}] - return self._execute_command('GetUser', body) - - ########## - # SET - ########## - def add_user(self, username: str, password: str, level: str = "guest") -> bool: - """ - Add a new user account to the camera - :param username: The user's username - :param password: The user's password - :param level: The privilege level 'guest' or 'admin'. Default is 'guest' - :return: whether the user was added successfully - """ - body = [{"cmd": "AddUser", "action": 0, - "param": {"User": {"userName": username, "password": password, "level": level}}}] - r_data = self._execute_command('AddUser', body) - if r_data["value"]["rspCode"] == "200": - return True - print("Could not add user. Camera responded with:", r_data["value"]) - return False - - def modify_user(self, username: str, password: str) -> bool: - """ - Modify the user's password by specifying their username - :param username: The user which would want to be modified - :param password: The new password - :return: whether the user was modified successfully - """ - body = [{"cmd": "ModifyUser", "action": 0, "param": {"User": {"userName": username, "password": password}}}] - r_data = self._execute_command('ModifyUser', body) - if r_data["value"]["rspCode"] == "200": - return True - print("Could not modify user:", username, "\nCamera responded with:", r_data["value"]) - return False - - def delete_user(self, username: str) -> bool: - """ - Delete a user by specifying their username - :param username: The user which would want to be deleted - :return: whether the user was deleted successfully - """ - body = [{"cmd": "DelUser", "action": 0, "param": {"User": {"userName": username}}}] - r_data = self._execute_command('DelUser', body) - if r_data["value"]["rspCode"] == "200": - return True - print("Could not delete user:", username, "\nCamera responded with:", r_data["value"]) - return False - - ########## - # Image Data - ########## - def get_snap(self, timeout: int = 3) -> Image or None: - """ - Gets a "snap" of the current camera video data and returns a Pillow Image or None - :param timeout: Request timeout to camera in seconds - :return: Image or None - """ - try: - randomstr = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10)) - snap = self.url + "?cmd=Snap&channel=0&rs=" \ - + randomstr \ - + "&user=" + self.username \ - + "&password=" + self.password - req = request.Request(snap) - req.set_proxy(Request.proxies, 'http') - reader = request.urlopen(req, timeout) - if reader.status == 200: - b = bytearray(reader.read()) - return Image.open(io.BytesIO(b)) - print("Could not retrieve data from camera successfully. Status:", reader.status) - return None - - except Exception as e: - print("Could not get Image data\n", e) - raise - - ######### - # Device - ######### - def get_hdd_info(self) -> json: - """ - 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) - if r_data["value"]["rspCode"] == "200": - return True - print("Could not format HDD/SD. Camera responded with:", r_data["value"]) - return False - - ########### - # Recording - ########### - - ########### - # SET - ########### - - ########### - # GET - ########### - def get_recording_encoding(self) -> json: - """ - Get the current camera encoding settings for "Clear" and "Fluent" profiles. - See examples/response/GetEnc.json for example response data. - :return: response json - """ - body = [{"cmd": "GetEnc", "action": 1, "param": {"channel": 0}}] - return self._execute_command('GetEnc', body) - - def get_recording_advanced(self) -> json: - """ - Get recording advanced setup data - See examples/response/GetRec.json for example response data. - :return: response json - """ - body = [{"cmd": "GetRec", "action": 1, "param": {"channel": 0}}] - return self._execute_command('GetRec', body) - - ########### - # RTSP Stream - ########### - def open_video_stream(self, profile: str = "main") -> Image: - """ - profile is "main" or "sub" - https://support.reolink.com/hc/en-us/articles/360007010473-How-to-Live-View-Reolink-Cameras-via-VLC-Media-Player - :param profile: - :return: - """ - with RtspClient(ip=self.ip, username=self.username, password=self.password, - proxies={"host": "127.0.0.1", "port": 8000}) as rtsp_client: - rtsp_client.preview() - # with rtsp.Client( - # rtsp_server_uri="rtsp://" - # + self.username + ":" - # + self.password + "@" - # + self.ip - # + ":554//h264Preview_01_" - # + profile) as client: - # return client diff --git a/api/device.py b/api/device.py new file mode 100644 index 0000000..263a999 --- /dev/null +++ b/api/device.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) + if r_data["value"]["rspCode"] == "200": + return True + print("Could not format HDD/SD. Camera responded with:", r_data["value"]) + return False diff --git a/api/display.py b/api/display.py new file mode 100644 index 0000000..72380b4 --- /dev/null +++ b/api/display.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) + if r_data["value"]["rspCode"] == "200": + return True + print("Could not set OSD. Camera responded with status:", r_data["value"]) + return False diff --git a/api/network.py b/api/network.py new file mode 100644 index 0000000..8e1ecc5 --- /dev/null +++ b/api/network.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) diff --git a/api/recording.py b/api/recording.py new file mode 100644 index 0000000..c204708 --- /dev/null +++ b/api/recording.py @@ -0,0 +1,69 @@ +import io +import random +import string +from urllib import request + +from PIL import Image + +from RtspClient import RtspClient +from resthandle import Request + + +class RecordingAPIMixin: + """API calls for recording/streaming image or video.""" + def get_recording_encoding(self) -> object: + """ + Get the current camera encoding settings for "Clear" and "Fluent" profiles. + See examples/response/GetEnc.json for example response data. + :return: response json + """ + body = [{"cmd": "GetEnc", "action": 1, "param": {"channel": 0}}] + return self._execute_command('GetEnc', body) + + def get_recording_advanced(self) -> object: + """ + Get recording advanced setup data + See examples/response/GetRec.json for example response data. + :return: response json + """ + body = [{"cmd": "GetRec", "action": 1, "param": {"channel": 0}}] + return self._execute_command('GetRec', body) + + ########### + # RTSP Stream + ########### + def open_video_stream(self, profile: str = "main") -> Image: + """ + profile is "main" or "sub" + https://support.reolink.com/hc/en-us/articles/360007010473-How-to-Live-View-Reolink-Cameras-via-VLC-Media-Player + :param profile: + :return: + """ + with RtspClient(ip=self.ip, username=self.username, password=self.password, + proxies={"host": "127.0.0.1", "port": 8000}) as rtsp_client: + rtsp_client.preview() + + def get_snap(self, timeout: int = 3) -> Image or None: + """ + Gets a "snap" of the current camera video data and returns a Pillow Image or None + :param timeout: Request timeout to camera in seconds + :return: Image or None + """ + randomstr = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10)) + snap = self.url + "?cmd=Snap&channel=0&rs=" \ + + randomstr \ + + "&user=" + self.username \ + + "&password=" + self.password + try: + req = request.Request(snap) + req.set_proxy(Request.proxies, 'http') + reader = request.urlopen(req, timeout) + if reader.status == 200: + b = bytearray(reader.read()) + return Image.open(io.BytesIO(b)) + print("Could not retrieve data from camera successfully. Status:", reader.status) + return None + + except Exception as e: + print("Could not get Image data\n", e) + raise diff --git a/api/system.py b/api/system.py new file mode 100644 index 0000000..244b849 --- /dev/null +++ b/api/system.py @@ -0,0 +1,32 @@ +class SystemAPIMixin: + """API for accessing general system information of the camera.""" + def get_general_system(self) -> object: + """:return: response json""" + body = [{"cmd": "GetTime", "action": 1, "param": {}}, {"cmd": "GetNorm", "action": 1, "param": {}}] + return self._execute_command('get_general_system', body, multi=True) + + def get_performance(self) -> object: + """ + Get a snapshot of the current performance of the camera. + See examples/response/GetPerformance.json for example response data. + :return: response json + """ + body = [{"cmd": "GetPerformance", "action": 0, "param": {}}] + return self._execute_command('GetPerformance', body) + + def get_information(self) -> object: + """ + Get the camera information + See examples/response/GetDevInfo.json for example response data. + :return: response json + """ + body = [{"cmd": "GetDevInfo", "action": 0, "param": {}}] + return self._execute_command('GetDevInfo', body) + + def reboot_camera(self) -> object: + """ + Reboots the camera + :return: response json + """ + body = [{"cmd": "Reboot", "action": 0, "param": {}}] + return self._execute_command('Reboot', body) diff --git a/api/user.py b/api/user.py new file mode 100644 index 0000000..7473901 --- /dev/null +++ b/api/user.py @@ -0,0 +1,62 @@ +class UserAPIMixin: + """User-related API calls.""" + def get_online_user(self) -> object: + """ + Return a list of current logged-in users in json format + See examples/response/GetOnline.json for example response data. + :return: response json + """ + body = [{"cmd": "GetOnline", "action": 1, "param": {}}] + return self._execute_command('GetOnline', body) + + def get_users(self) -> object: + """ + Return a list of user accounts from the camera in json format. + See examples/response/GetUser.json for example response data. + :return: response json + """ + body = [{"cmd": "GetUser", "action": 1, "param": {}}] + return self._execute_command('GetUser', body) + + def add_user(self, username: str, password: str, level: str = "guest") -> bool: + """ + Add a new user account to the camera + :param username: The user's username + :param password: The user's password + :param level: The privilege level 'guest' or 'admin'. Default is 'guest' + :return: whether the user was added successfully + """ + body = [{"cmd": "AddUser", "action": 0, + "param": {"User": {"userName": username, "password": password, "level": level}}}] + r_data = self._execute_command('AddUser', body) + if r_data["value"]["rspCode"] == "200": + return True + print("Could not add user. Camera responded with:", r_data["value"]) + return False + + def modify_user(self, username: str, password: str) -> bool: + """ + Modify the user's password by specifying their username + :param username: The user which would want to be modified + :param password: The new password + :return: whether the user was modified successfully + """ + body = [{"cmd": "ModifyUser", "action": 0, "param": {"User": {"userName": username, "password": password}}}] + r_data = self._execute_command('ModifyUser', body) + if r_data["value"]["rspCode"] == "200": + return True + print("Could not modify user:", username, "\nCamera responded with:", r_data["value"]) + return False + + def delete_user(self, username: str) -> bool: + """ + Delete a user by specifying their username + :param username: The user which would want to be deleted + :return: whether the user was deleted successfully + """ + body = [{"cmd": "DelUser", "action": 0, "param": {"User": {"userName": username}}}] + r_data = self._execute_command('DelUser', body) + if r_data["value"]["rspCode"] == "200": + return True + print("Could not delete user:", username, "\nCamera responded with:", r_data["value"]) + return False