thebears/reolink_api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor simple commands

Browse Source
This commit is contained in:
Max Ziermann
2020-03-03 15:54:08 +01:00
parent a2c44f89de
commit 5d4a38f9a2
1 changed files with 85 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

330
APIHandler.py
View File

@@ -72,6 +72,22 @@ class APIHandler:
print("Error Login\n", e) print("Error Login\n", e)
raise raise
def _execute_command(self, command, data):
"""
Send a POST request to the IP camera with given JSON body and
:param command: name of the command to send
:param data: object to send to the camera
:return: response as python object
"""
try:
if self.token is None:
raise ValueError("Login first")
response = Request.post(self.url, data=data, params={"cmd": command, "token": self.token})
return json.loads(response.text)
except Exception as e:
print(f"Command {command} failed: {e}")
raise
########### ###########
# NETWORK # NETWORK
########### ###########
@@ -119,20 +135,12 @@ class APIHandler:
raise raise
def set_wifi(self, ssid, password) -> json or None: def set_wifi(self, ssid, password) -> json or None:
try: body = [{"cmd": "SetWifi", "action": 0, "param": {
if self.token is None: "Wifi": {
raise ValueError("Login first") "ssid": ssid,
body = [{"cmd": "SetWifi", "action": 0, "param": { "password": password
"Wifi": { }}}]
"ssid": ssid, return self._execute_command('SetWifi', body)
"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 # GET
@@ -158,49 +166,13 @@ class APIHandler:
except Exception as e: except Exception as e:
print("Get Network Ports", 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): def get_wifi(self):
try: body = [{"cmd": "GetWifi", "action": 1, "param": {}}]
if self.token is None: return self._execute_command('GetWifi', body)
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): def scan_wifi(self):
try: body = [{"cmd": "ScanWifi", "action": 1, "param": {}}]
if self.token is None: return self._execute_command('ScanWifi', body)
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 # Display
@@ -226,17 +198,8 @@ class APIHandler:
}}}] }}}]
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetOsd", "action": 1, "param": {"channel": 0}}]
param = {"cmd": "GetOsd", "token": self.token} return self._execute_command('GetOsd', body)
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: def get_mask(self) -> json or None:
""" """
@@ -257,17 +220,8 @@ class APIHandler:
}] }]
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetMask", "action": 1, "param": {"channel": 0}}]
param = {"cmd": "GetMask", "token": self.token} return self._execute_command('GetMask', body)
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 # SET
@@ -286,28 +240,19 @@ class APIHandler:
:param osd_time_pos: string time position ["Upper Left","Top Center","Upper Right","Lower Left","Bottom Center","Lower Right"] :param osd_time_pos: string time position ["Upper Left","Top Center","Upper Right","Lower Left","Bottom Center","Lower Right"]
:return: :return:
""" """
try: body = [{"cmd": "SetOsd", "action": 1, "param":
param = {"cmd": "setOsd", "token": self.token} {"Osd": {"bgcolor": bg_color, "channel": channel,
body = [{"cmd": "SetOsd", "action": 1, "param": "osdChannel": {"enable": osd_channel_enabled, "name": osd_channel_name,
{"Osd": {"bgcolor": bg_color, "channel": channel, "pos": osd_channel_pos},
"osdChannel": {"enable": osd_channel_enabled, "name": osd_channel_name, "osdTime": {"enable": osd_time_enabled, "pos": osd_time_pos}
"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":
response = Request.post(self.url, data=body, params=param) return True
if response.status_code == 200: print("Could not set OSD. Camera responded with status:", r_data["value"])
r_data = json.loads(response.text) return False
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 # SYSTEM
@@ -325,8 +270,6 @@ class APIHandler:
response = Request.post(self.url, data=body, params=param) response = Request.post(self.url, data=body, params=param)
if response.status_code == 200: if response.status_code == 200:
return json.loads(response.text) return json.loads(response.text)
print("Could not retrieve general information from camera successfully. Status:", response.status_code)
return None
except Exception as e: except Exception as e:
print("Could not get General System settings\n", e) print("Could not get General System settings\n", e)
raise raise
@@ -347,17 +290,8 @@ class APIHandler:
}] }]
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetPerformance", "action": 0, "param": {}}]
param = {"cmd": "GetPerformance", "token": self.token} return self._execute_command('GetPerformance', body)
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: def get_information(self) -> json or None:
""" """
@@ -386,17 +320,8 @@ class APIHandler:
}] }]
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetDevInfo", "action": 0, "param": {}}]
param = {"cmd": "GetDevInfo", "token": self.token} return self._execute_command('GetDevInfo', body)
body = [{"cmd": "GetDevInfo", "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 camera information. Status:", response.status_code)
return None
except Exception as e:
print("Could not get device information", e)
raise
########### ###########
# SET # SET
@@ -406,17 +331,8 @@ class APIHandler:
Reboots the camera Reboots the camera
:return: bool :return: bool
""" """
try: body = [{"cmd": "Reboot", "action": 0, "param": {}}]
param = {"cmd": "Reboot", "token": self.token} return self._execute_command('Reboot', body)
body = [{"cmd": "Reboot", "action": 0, "param": {}}]
response = Request.post(self.url, data=body, params=param)
if response.status_code == 200:
return True
print("Something went wrong. Could not reboot camera. Status:", response.status_code)
return False
except Exception as e:
print("Could not reboot camera", e)
raise
########## ##########
# User # User
@@ -441,17 +357,8 @@ class APIHandler:
}] }]
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetOnline", "action": 1, "param": {}}]
param = {"cmd": "GetOnline", "token": self.token} return self._execute_command('GetOnline', body)
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: def get_users(self) -> json or None:
""" """
@@ -481,17 +388,8 @@ class APIHandler:
}] }]
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetUser", "action": 1, "param": {}}]
param = {"cmd": "GetUser", "token": self.token} return self._execute_command('GetUser', body)
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 # SET
@@ -504,22 +402,13 @@ class APIHandler:
:param level: The privilege level 'guest' or 'admin'. Default is 'guest' :param level: The privilege level 'guest' or 'admin'. Default is 'guest'
:return: bool :return: bool
""" """
try: body = [{"cmd": "AddUser", "action": 0,
param = {"cmd": "AddUser", "token": self.token} "param": {"User": {"userName": username, "password": password, "level": level}}}]
body = [{"cmd": "AddUser", "action": 0, r_data = self._execute_command('AddUser', body)
"param": {"User": {"userName": username, "password": password, "level": level}}}] if r_data["value"]["rspCode"] == "200":
response = Request.post(self.url, data=body, params=param) return True
if response.status_code == 200: print("Could not add user. Camera responded with:", r_data["value"])
r_data = json.loads(response.text) return False
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: def modify_user(self, username: str, password: str) -> bool:
""" """
@@ -528,20 +417,12 @@ class APIHandler:
:param password: The new password :param password: The new password
:return: bool :return: bool
""" """
try: body = [{"cmd": "ModifyUser", "action": 0, "param": {"User": {"userName": username, "password": password}}}]
param = {"cmd": "ModifyUser", "token": self.token} r_data = self._execute_command('ModifyUser', body)
body = [{"cmd": "ModifyUser", "action": 0, "param": {"User": {"userName": username, "password": password}}}] if r_data["value"]["rspCode"] == "200":
response = Request.post(self.url, data=body, params=param) return True
if response.status_code == 200: print("Could not modify user:", username, "\nCamera responded with:", r_data["value"])
r_data = json.loads(response.text) return False
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: def delete_user(self, username: str) -> bool:
""" """
@@ -549,19 +430,12 @@ class APIHandler:
:param username: The user which would want to be deleted :param username: The user which would want to be deleted
:return: bool :return: bool
""" """
try: body = [{"cmd": "DelUser", "action": 0, "param": {"User": {"userName": username}}}]
param = {"cmd": "DelUser", "token": self.token} r_data = self._execute_command('DelUser', body)
body = [{"cmd": "DelUser", "action": 0, "param": {"User": {"userName": username}}}] if r_data["value"]["rspCode"] == "200":
response = Request.post(self.url, data=body, params=param) return True
if response.status_code == 200: print("Could not delete user:", username, "\nCamera responded with:", r_data["value"])
r_data = json.loads(response.text) return False
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 # Image Data
@@ -614,17 +488,8 @@ class APIHandler:
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetHddInfo", "action": 0, "param": {}}]
param = {"cmd": "GetHddInfo", "token": self.token} return self._execute_command('GetHddInfo', body)
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: def format_hdd(self, hdd_id: [int] = [0]) -> bool:
""" """
@@ -632,21 +497,13 @@ class APIHandler:
:param hdd_id: List of id's specified by the camera with get_hdd_info api. Default is 0 (SD card) :param hdd_id: List of id's specified by the camera with get_hdd_info api. Default is 0 (SD card)
:return: bool :return: bool
""" """
try: body = [{"cmd": "Format", "action": 0, "param": {"HddInfo": {"id": hdd_id}}}]
param = {"cmd": "Format", "token": self.token} r_data = self._execute_command('Format', body)
body = [{"cmd": "Format", "action": 0, "param": {"HddInfo": {"id": hdd_id}}}] if r_data["value"]["rspCode"] == "200":
response = Request.post(self.url, data=body, params=param) return True
if response.status_code == 200: print("Could not format HDD/SD. Camera responded with:", r_data["value"])
r_data = json.loads(response.text) return False
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 # Recording
@@ -825,17 +682,8 @@ class APIHandler:
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetEnc", "action": 1, "param": {"channel": 0}}]
param = {"cmd": "GetEnc", "token": self.token} return self._execute_command('GetEnc', body)
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: def get_recording_advanced(self) -> json or None:
""" """
@@ -883,16 +731,8 @@ class APIHandler:
:return: json or None :return: json or None
""" """
try: body = [{"cmd": "GetRec", "action": 1, "param": {"channel": 0}}]
param = {"cmd": "GetRec", "token": self.token} return self._execute_command('GetRec', body)
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)
########### ###########
# RTSP Stream # RTSP Stream