Split APIHandler into single files
This commit is contained in:
Max Ziermann
@@ -1,26 +1,25 @@
|
|||||||
import io
|
from api.recording import RecordingAPIMixin
|
||||||
import json
|
from .device import DeviceAPIMixin
|
||||||
import random
|
from .display import DisplayAPIMixin
|
||||||
import string
|
from .network import NetworkAPIMixin
|
||||||
import sys
|
from .system import SystemAPIMixin
|
||||||
from urllib import request
|
from .user import UserAPIMixin
|
||||||
|
|
||||||
import numpy
|
|
||||||
import rtsp
|
|
||||||
from PIL import Image
|
|
||||||
|
|
||||||
from RtspClient import RtspClient
|
|
||||||
from resthandle import Request
|
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.
|
This handles communication directly with the camera.
|
||||||
Current camera's tested: RLC-411WS
|
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/
|
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=False, **kwargs):
|
||||||
@@ -41,10 +40,6 @@ class APIHandler:
|
|||||||
self.password = password
|
self.password = password
|
||||||
Request.proxies = kwargs.get("proxy") # Defaults to None if key isn't found
|
Request.proxies = kwargs.get("proxy") # Defaults to None if key isn't found
|
||||||
|
|
||||||
###########
|
|
||||||
# Token
|
|
||||||
###########
|
|
||||||
|
|
||||||
def login(self) -> bool:
|
def login(self) -> bool:
|
||||||
"""
|
"""
|
||||||
Get login token
|
Get login token
|
||||||
@@ -93,335 +88,3 @@ class APIHandler:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"Command {command} failed: {e}")
|
print(f"Command {command} failed: {e}")
|
||||||
raise
|
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
|
|
||||||
|
|||||||
23
api/device.py
Normal file
23
api/device.py
Normal file
@@ -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
|
||||||
47
api/display.py
Normal file
47
api/display.py
Normal file
@@ -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
|
||||||
52
api/network.py
Normal file
52
api/network.py
Normal file
@@ -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)
|
||||||
69
api/recording.py
Normal file
69
api/recording.py
Normal file
@@ -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
|
||||||
32
api/system.py
Normal file
32
api/system.py
Normal file
@@ -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)
|
||||||
62
api/user.py
Normal file
62
api/user.py
Normal file
@@ -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
|
||||||
Reference in New Issue
Block a user