Files
reolink_api/APIHandler.py
Alano Terblanche 5bc90f7c60 Added some API's and Doc strings
OSD -> set and get
System -> get_performance, get_information
Recording -> get
2019-08-11 15:35:52 +02:00

863 lines
31 KiB
Python

import io
import json
import random
import string
from urllib.request import urlopen
from PIL import Image
from resthandle import Request
class APIHandler:
"""
The APIHandler class is the backend part of the API.
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):
"""
Initialise the Camera API Handler (maps api calls into python)
:param ip:
:param username:
:param password:
"""
self.url = "http://" + ip + "/cgi-bin/api.cgi"
self.token = None
self.username = username
self.password = password
###########
# Token
###########
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 = json.loads(response.text)[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
###########
# 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
"""
try:
if self.token is None:
raise ValueError("Login first")
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
}}}]
param = {"token": self.token}
response = Request.post(self.url, data=body, params=param)
if response is not None:
if response.status_code == 200:
print("Successfully Set Network Ports")
return True
else:
print("Something went wront\nStatus Code:", response.status_code)
return False
return False
except Exception as e:
print("Setting Network Port Error\n", e)
raise
def set_wifi(self, ssid, password) -> json or None:
try:
if self.token is None:
raise ValueError("Login first")
body = [{"cmd": "SetWifi", "action": 0, "param": {
"Wifi": {
"ssid": ssid,
"password": password
}}}]
param = {"cmd": "SetWifi", "token": self.token}
response = Request.post(self.url, data=body, params=param)
return json.loads(response.text)
except Exception as e:
print("Could not Set Wifi details", e)
raise
###########
# GET
###########
def get_net_ports(self) -> json or None:
"""
Get network ports
:return:
"""
try:
if self.token is not None:
raise ValueError("Login first")
body = [{"cmd": "GetNetPort", "action": 1, "param": {}},
{"cmd": "GetUpnp", "action": 0, "param": {}},
{"cmd": "GetP2p", "action": 0, "param": {}}]
param = {"token": self.token}
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not get network ports data. Status:", response.status_code)
return None
except Exception as e:
print("Get Network Ports", e)
def get_link_local(self):
"""
Get General network data
This includes IP address, Device mac, Gateway and DNS
:return:
"""
try:
if self.token is None:
raise ValueError("Login first")
body = [{"cmd": "GetLocalLink", "action": 1, "param": {}}]
param = {"cmd": "GetLocalLink", "token": self.token}
request = Request.post(self.url, data=body, params=param)
if request.status_code == 200:
return json.loads(request.text)
print("Could not get ")
except Exception as e:
print("Could not get Link Local", e)
raise
def get_wifi(self):
try:
if self.token is None:
raise ValueError("Login first")
body = [{"cmd": "GetWifi", "action": 1, "param": {}}]
param = {"cmd": "GetWifi", "token": self.token}
response = Request.post(self.url, data=body, params=param)
return json.loads(response.text)
except Exception as e:
print("Could not get Wifi\n", e)
raise
def scan_wifi(self):
try:
if self.token is None:
raise ValueError("Login first")
body = [{"cmd": "ScanWifi", "action": 1, "param": {}}]
param = {"cmd": "ScanWifi", "token": self.token}
response = Request.post(self.url, data=body, params=param)
return json.loads(response.text)
except Exception as e:
print("Could not Scan wifi\n", e)
raise
###########
# Display
###########
###########
# GET
###########
def get_osd(self) -> json or None:
"""
Get OSD information.
Response data format is as follows:
[{"cmd" : "GetOsd","code" : 0, "initial" : {
"Osd" : {"bgcolor" : 0,"channel" : 0,"osdChannel" : {"enable" : 1,"name" : "Camera1","pos" : "Lower Right"},
"osdTime" : {"enable" : 1,"pos" : "Top Center"}
}},"range" : {"Osd" : {"bgcolor" : "boolean","channel" : 0,"osdChannel" : {"enable" : "boolean","name" : {"maxLen" : 31},
"pos" : ["Upper Left","Top Center","Upper Right","Lower Left","Bottom Center","Lower Right"]
},
"osdTime" : {"enable" : "boolean","pos" : ["Upper Left","Top Center","Upper Right","Lower Left","Bottom Center","Lower Right"]
}
}},"value" : {"Osd" : {"bgcolor" : 0,"channel" : 0,"osdChannel" : {"enable" : 0,"name" : "FarRight","pos" : "Lower Right"},
"osdTime" : {"enable" : 0,"pos" : "Top Center"}
}}}]
:return: json or None
"""
try:
param = {"cmd": "GetOsd", "token": self.token}
body = [{"cmd": "GetOsd", "action": 1, "param": {"channel": 0}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve OSD from camera successfully. Status:", response.status_code)
return None
except Exception as e:
print("Could not get OSD", e)
raise
def get_mask(self) -> json or None:
"""
Get the camera mask information
Response data format is as follows:
[{"cmd" : "GetMask","code" : 0,"initial" : {
"Mask" : {
"area" : [{"block" : {"height" : 0,"width" : 0,"x" : 0,"y" : 0},"screen" : {"height" : 0,"width" : 0}}],
"channel" : 0,
"enable" : 0
}
},"range" : {"Mask" : {"channel" : 0,"enable" : "boolean","maxAreas" : 4}},"value" : {
"Mask" : {
"area" : null,
"channel" : 0,
"enable" : 0}
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetMask", "token": self.token}
body = [{"cmd": "GetMask", "action": 1, "param": {"channel": 0}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not get Mask from camera successfully. Status:", response.status_code)
return None
except Exception as e:
print("Could not get mask", e)
raise
###########
# 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:
"""
try:
param = {"cmd": "setOsd", "token": self.token}
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}
}
}
}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
r_data = json.loads(response.text)
if r_data["value"]["rspCode"] == "200":
return True
print("Could not set OSD. Camera responded with status:", r_data["value"])
return False
print("Could not set OSD. Status:", response.status_code)
return False
except Exception as e:
print("Could not set OSD", e)
raise
###########
# SYSTEM
###########
###########
# GET
###########
def get_general_system(self) -> json or None:
try:
if self.token is None:
raise ValueError("Login first")
body = [{"cmd": "GetTime", "action": 1, "param": {}}, {"cmd": "GetNorm", "action": 1, "param": {}}]
param = {"token": self.token}
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve general information from camera successfully. Status:", response.status_code)
return None
except Exception as e:
print("Could not get General System settings\n", e)
raise
def get_performance(self) -> json or None:
"""
Get a snapshot of the current performance of the camera.
Response data format is as follows:
[{"cmd" : "GetPerformance",
"code" : 0,
"value" : {
"Performance" : {
"codecRate" : 2154,
"cpuUsed" : 14,
"netThroughput" : 0
}
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetPerformance", "token": self.token}
body = [{"cmd": "GetPerformance", "action": 0, "param": {}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Cound not retrieve performance information from camera successfully. Status:", response.status_code)
return None
except Exception as e:
print("Could not get performance", e)
raise
def get_information(self) -> json or None:
"""
Get the camera information
Response data format is as follows:
[{"cmd" : "GetDevInfo","code" : 0,"value" : {
"DevInfo" : {
"B485" : 0,
"IOInputNum" : 0,
"IOOutputNum" : 0,
"audioNum" : 0,
"buildDay" : "build 18081408",
"cfgVer" : "v2.0.0.0",
"channelNum" : 1,
"detail" : "IPC_3816M100000000100000",
"diskNum" : 1,
"firmVer" : "v2.0.0.1389_18081408",
"hardVer" : "IPC_3816M",
"model" : "RLC-411WS",
"name" : "Camera1_withpersonality",
"serial" : "00000000000000",
"type" : "IPC",
"wifi" : 1
}
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetDevInfo", "token": self.token}
body = [{"cmd": "GetDevInfo", "action": 0, "param": {}}]
response = Request.post(self.url, data=body, params=param)
if response == 200:
return json.loads(response.text)
print("Could not retrieve camera information. Status:", response.status_code)
return None
except Exception as e:
print("Could not get device information", e)
raise
##########
# User
##########
##########
# GET
##########
def get_online_user(self) -> json or None:
"""
Return a list of current logged-in users in json format
Response data format is as follows:
[{"cmd" : "GetOnline","code" : 0,"value" : {
"User" : [{
"canbeDisconn" : 0,
"ip" : "192.168.1.100",
"level" : "admin",
"sessionId" : 1000,
"userName" : "admin"
}]
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetOnline", "token": self.token}
body = [{"cmd": "GetOnline", "action": 1, "param": {}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve online user from camera. Status:", response.status_code)
return None
except Exception as e:
print("Could not get online user", e)
raise
def get_users(self) -> json or None:
"""
Return a list of user accounts from the camera in json format
Response data format is as follows:
[{"cmd" : "GetUser","code" : 0,"initial" : {
"User" : {
"level" : "guest"
}},
"range" : {"User" : {
"level" : [ "guest", "admin" ],
"password" : {
"maxLen" : 31,
"minLen" : 6
},
"userName" : {
"maxLen" : 31,
"minLen" : 1
}}
},"value" : {
"User" : [
{
"level" : "admin",
"userName" : "admin"
}]
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetUser", "token": self.token}
body = [{"cmd": "GetUser", "action": 1, "param": {}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve users from camera. Status:", response.status_code)
return None
except Exception as e:
print("Could not get users", e)
raise
##########
# 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: bool
"""
try:
param = {"cmd": "AddUser", "token": self.token}
body = [{"cmd": "AddUser", "action": 0,
"param": {"User": {"userName": username, "password": password, "level": level}}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
r_data = json.loads(response.text)
if r_data["value"]["rspCode"] == "200":
return True
print("Could not add user. Camera responded with:", r_data["value"])
return False
print("Something went wrong. Could not add user. Status:", response.status_code)
return False
except Exception as e:
print("Could not add user", e)
raise
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: bool
"""
try:
param = {"cmd": "ModifyUser", "token": self.token}
body = [{"cmd": "ModifyUser", "action": 0, "param": {"User": {"userName": username, "password": password}}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
r_data = json.loads(response.text)
if r_data["value"]["rspCode"] == "200":
return True
print("Could not modify user:", username, "\nCamera responded with:", r_data["value"])
print("Something went wrong. Could not modify user. Status:", response.status_code)
return False
except Exception as e:
print("Could not modify user", e)
raise
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: bool
"""
try:
param = {"cmd": "DelUser", "token": self.token}
body = [{"cmd": "DelUser", "action": 0, "param": {"User": {"userName": username}}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
r_data = json.loads(response.text)
if r_data["value"]["rspCode"] == "200":
return True
print("Could not delete user:", username, "\nCamera responded with:", r_data["value"])
return False
except Exception as e:
print("Could not delete user", e)
raise
##########
# 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 = "?cmd=Snap&channel=0&rs=" \
+ randomstr \
+ "&user=" + self.username \
+ "&password=" + self.password
reader = urlopen(self.url + snap, 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 or None:
"""
Gets all HDD and SD card information from Camera
Response data format is as follows:
[{"cmd" : "GetHddInfo",
"code" : 0,
"value" : {
"HddInfo" : [{
"capacity" : 15181,
"format" : 1,
"id" : 0,
"mount" : 1,
"size" : 15181
}]
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetHddInfo", "token": self.token}
body = [{"cmd": "GetHddInfo", "action": 0, "param": {}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve HDD/SD info from camera successfully. Status:", response.status_code)
return None
except Exception as e:
print("Could not get HDD/SD card information", e)
raise
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
"""
try:
param = {"cmd": "Format", "token": self.token}
body = [{"cmd": "Format", "action": 0, "param": {"HddInfo": {"id": hdd_id}}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
r_data = json.loads(response.text)
if r_data["value"]["rspCode"] == "200":
return True
print("Could not format HDD/SD. Camera responded with:", r_data["value"])
return False
print("Could not format HDD/SD. Status:", response.status_code)
return False
except Exception as e:
print("Could not format HDD/SD", e)
raise
###########
# Recording
###########
###########
# SET
###########
###########
# GET
###########
def get_recording_encoding(self) -> json or None:
"""
Get the current camera encoding settings for "Clear" and "Fluent" profiles.
Response data format is as follows:
[{
"cmd" : "GetEnc",
"code" : 0,
"initial" : {
"Enc" : {
"audio" : 0,
"channel" : 0,
"mainStream" : {
"bitRate" : 4096,
"frameRate" : 15,
"profile" : "High",
"size" : "3072*1728"
},
"subStream" : {
"bitRate" : 160,
"frameRate" : 7,
"profile" : "High",
"size" : "640*360"
}
}
},
"range" : {
"Enc" : [
{
"audio" : "boolean",
"mainStream" : {
"bitRate" : [ 1024, 1536, 2048, 3072, 4096, 5120, 6144, 7168, 8192 ],
"default" : {
"bitRate" : 4096,
"frameRate" : 15
},
"frameRate" : [ 20, 18, 16, 15, 12, 10, 8, 6, 4, 2 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "3072*1728"
},
"subStream" : {
"bitRate" : [ 64, 128, 160, 192, 256, 384, 512 ],
"default" : {
"bitRate" : 160,
"frameRate" : 7
},
"frameRate" : [ 15, 10, 7, 4 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "640*360"
}
},
{
"audio" : "boolean",
"mainStream" : {
"bitRate" : [ 1024, 1536, 2048, 3072, 4096, 5120, 6144, 7168, 8192 ],
"default" : {
"bitRate" : 4096,
"frameRate" : 15
},
"frameRate" : [ 20, 18, 16, 15, 12, 10, 8, 6, 4, 2 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "2592*1944"
},
"subStream" : {
"bitRate" : [ 64, 128, 160, 192, 256, 384, 512 ],
"default" : {
"bitRate" : 160,
"frameRate" : 7
},
"frameRate" : [ 15, 10, 7, 4 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "640*360"
}
},
{
"audio" : "boolean",
"mainStream" : {
"bitRate" : [ 1024, 1536, 2048, 3072, 4096, 5120, 6144, 7168, 8192 ],
"default" : {
"bitRate" : 3072,
"frameRate" : 15
},
"frameRate" : [ 30, 22, 20, 18, 16, 15, 12, 10, 8, 6, 4, 2 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "2560*1440"
},
"subStream" : {
"bitRate" : [ 64, 128, 160, 192, 256, 384, 512 ],
"default" : {
"bitRate" : 160,
"frameRate" : 7
},
"frameRate" : [ 15, 10, 7, 4 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "640*360"
}
},
{
"audio" : "boolean",
"mainStream" : {
"bitRate" : [ 1024, 1536, 2048, 3072, 4096, 5120, 6144, 7168, 8192 ],
"default" : {
"bitRate" : 3072,
"frameRate" : 15
},
"frameRate" : [ 30, 22, 20, 18, 16, 15, 12, 10, 8, 6, 4, 2 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "2048*1536"
},
"subStream" : {
"bitRate" : [ 64, 128, 160, 192, 256, 384, 512 ],
"default" : {
"bitRate" : 160,
"frameRate" : 7
},
"frameRate" : [ 15, 10, 7, 4 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "640*360"
}
},
{
"audio" : "boolean",
"mainStream" : {
"bitRate" : [ 1024, 1536, 2048, 3072, 4096, 5120, 6144, 7168, 8192 ],
"default" : {
"bitRate" : 3072,
"frameRate" : 15
},
"frameRate" : [ 30, 22, 20, 18, 16, 15, 12, 10, 8, 6, 4, 2 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "2304*1296"
},
"subStream" : {
"bitRate" : [ 64, 128, 160, 192, 256, 384, 512 ],
"default" : {
"bitRate" : 160,
"frameRate" : 7
},
"frameRate" : [ 15, 10, 7, 4 ],
"profile" : [ "Base", "Main", "High" ],
"size" : "640*360"
}
}
]
},
"value" : {
"Enc" : {
"audio" : 0,
"channel" : 0,
"mainStream" : {
"bitRate" : 2048,
"frameRate" : 20,
"profile" : "Main",
"size" : "3072*1728"
},
"subStream" : {
"bitRate" : 64,
"frameRate" : 4,
"profile" : "High",
"size" : "640*360"
}
}
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetEnc", "token": self.token}
body = [{"cmd": "GetEnc", "action": 1, "param": {"channel": 0}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve recording encoding data. Status:", response.status_code)
return None
except Exception as e:
print("Could not get recording encoding", e)
raise
def get_recording_advanced(self) -> json or None:
"""
Get recording advanced setup data
Response data format is as follows:
[{
"cmd" : "GetRec",
"code" : 0,
"initial" : {
"Rec" : {
"channel" : 0,
"overwrite" : 1,
"postRec" : "15 Seconds",
"preRec" : 1,
"schedule" : {
"enable" : 1,
"table" : "111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111"
}
}
},
"range" : {
"Rec" : {
"channel" : 0,
"overwrite" : "boolean",
"postRec" : [ "15 Seconds", "30 Seconds", "1 Minute" ],
"preRec" : "boolean",
"schedule" : {
"enable" : "boolean"
}
}
},
"value" : {
"Rec" : {
"channel" : 0,
"overwrite" : 1,
"postRec" : "15 Seconds",
"preRec" : 1,
"schedule" : {
"enable" : 1,
"table" : "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
}
}
}
}]
:return: json or None
"""
try:
param = {"cmd": "GetRec", "token": self.token}
body = [{"cmd": "GetRec", "action": 1, "param": {"channel": 0}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return json.loads(response.text)
print("Could not retrieve advanced recording. Status:", response.status_code)
return None
except Exception as e:
print("Could not get advanced recoding", e)