Updated project structure and some file names.
Restored `requirements.txt` Updated `setup.py` to include new repository url and contact details. Moved the rtsp code from `record` to `stream`. Updated project structure to make it more readable and developer friendly - moved mixins to the `mixins` package, moved handlers to the `handlers` package. Moved files not belonging to anything in particular to the `util` package. Updated `camera` class to also defer login call. Deleted unused files like `config_handler`.
This commit is contained in:
Alano Terblanche
0
reolinkapi/mixins/__init__.py
Normal file
0
reolinkapi/mixins/__init__.py
Normal file
14
reolinkapi/mixins/alarm.py
Normal file
14
reolinkapi/mixins/alarm.py
Normal file
@@ -0,0 +1,14 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class AlarmAPIMixin:
|
||||
"""API calls for getting device alarm information."""
|
||||
|
||||
def get_alarm_motion(self) -> Dict:
|
||||
"""
|
||||
Gets the device alarm motion
|
||||
See examples/response/GetAlarmMotion.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetAlarm", "action": 1, "param": {"Alarm": {"channel": 0, "type": "md"}}}]
|
||||
return self._execute_command('GetAlarm', body)
|
||||
30
reolinkapi/mixins/device.py
Normal file
30
reolinkapi/mixins/device.py
Normal file
@@ -0,0 +1,30 @@
|
||||
from typing import List, Dict
|
||||
|
||||
|
||||
class DeviceAPIMixin:
|
||||
"""API calls for getting device information."""
|
||||
DEFAULT_HDD_ID = [0]
|
||||
|
||||
def get_hdd_info(self) -> Dict:
|
||||
"""
|
||||
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: List[float] = None) -> 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
|
||||
"""
|
||||
if hdd_id is None:
|
||||
hdd_id = self.DEFAULT_HDD_ID
|
||||
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
|
||||
56
reolinkapi/mixins/display.py
Normal file
56
reolinkapi/mixins/display.py
Normal file
@@ -0,0 +1,56 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class DisplayAPIMixin:
|
||||
"""API calls related to the current image (osd, on screen display)."""
|
||||
|
||||
def get_osd(self) -> Dict:
|
||||
"""
|
||||
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) -> Dict:
|
||||
"""
|
||||
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: float = 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
|
||||
18
reolinkapi/mixins/download.py
Normal file
18
reolinkapi/mixins/download.py
Normal file
@@ -0,0 +1,18 @@
|
||||
class DownloadAPIMixin:
|
||||
"""API calls for downloading video files."""
|
||||
def get_file(self, filename: str, output_path: str) -> bool:
|
||||
"""
|
||||
Download the selected video file
|
||||
:return: response json
|
||||
"""
|
||||
body = [
|
||||
{
|
||||
"cmd": "Download",
|
||||
"source": filename,
|
||||
"output": filename,
|
||||
"filepath": output_path
|
||||
}
|
||||
]
|
||||
resp = self._execute_command('Download', body)
|
||||
|
||||
return resp
|
||||
103
reolinkapi/mixins/image.py
Normal file
103
reolinkapi/mixins/image.py
Normal file
@@ -0,0 +1,103 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class ImageAPIMixin:
|
||||
"""API calls for image settings."""
|
||||
|
||||
def set_adv_image_settings(self,
|
||||
anti_flicker: str = 'Outdoor',
|
||||
exposure: str = 'Auto',
|
||||
gain_min: float = 1,
|
||||
gain_max: float = 62,
|
||||
shutter_min: float = 1,
|
||||
shutter_max: float = 125,
|
||||
blue_gain: float = 128,
|
||||
red_gain: float = 128,
|
||||
white_balance: str = 'Auto',
|
||||
day_night: str = 'Auto',
|
||||
back_light: str = 'DynamicRangeControl',
|
||||
blc: float = 128,
|
||||
drc: float = 128,
|
||||
rotation: float = 0,
|
||||
mirroring: float = 0,
|
||||
nr3d: float = 1) -> Dict:
|
||||
"""
|
||||
Sets the advanced camera settings.
|
||||
|
||||
:param anti_flicker: string
|
||||
:param exposure: string
|
||||
:param gain_min: int
|
||||
:param gain_max: string
|
||||
:param shutter_min: int
|
||||
:param shutter_max: int
|
||||
:param blue_gain: int
|
||||
:param red_gain: int
|
||||
:param white_balance: string
|
||||
:param day_night: string
|
||||
:param back_light: string
|
||||
:param blc: int
|
||||
:param drc: int
|
||||
:param rotation: int
|
||||
:param mirroring: int
|
||||
:param nr3d: int
|
||||
:return: response
|
||||
"""
|
||||
body = [{
|
||||
"cmd": "SetIsp",
|
||||
"action": 0,
|
||||
"param": {
|
||||
"Isp": {
|
||||
"channel": 0,
|
||||
"antiFlicker": anti_flicker,
|
||||
"exposure": exposure,
|
||||
"gain": {"min": gain_min, "max": gain_max},
|
||||
"shutter": {"min": shutter_min, "max": shutter_max},
|
||||
"blueGain": blue_gain,
|
||||
"redGain": red_gain,
|
||||
"whiteBalance": white_balance,
|
||||
"dayNight": day_night,
|
||||
"backLight": back_light,
|
||||
"blc": blc,
|
||||
"drc": drc,
|
||||
"rotation": rotation,
|
||||
"mirroring": mirroring,
|
||||
"nr3d": nr3d
|
||||
}
|
||||
}
|
||||
}]
|
||||
return self._execute_command('SetIsp', body)
|
||||
|
||||
def set_image_settings(self,
|
||||
brightness: float = 128,
|
||||
contrast: float = 62,
|
||||
hue: float = 1,
|
||||
saturation: float = 125,
|
||||
sharpness: float = 128) -> Dict:
|
||||
"""
|
||||
Sets the camera image settings.
|
||||
|
||||
:param brightness: int
|
||||
:param contrast: string
|
||||
:param hue: int
|
||||
:param saturation: int
|
||||
:param sharpness: int
|
||||
:return: response
|
||||
"""
|
||||
body = [
|
||||
{
|
||||
"cmd": "SetImage",
|
||||
"action": 0,
|
||||
"param": {
|
||||
"Image": {
|
||||
"bright": brightness,
|
||||
"channel": 0,
|
||||
"contrast": contrast,
|
||||
"hue": hue,
|
||||
"saturation": saturation,
|
||||
"sharpen": sharpness
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
return self._execute_command('SetImage', body)
|
||||
80
reolinkapi/mixins/motion.py
Normal file
80
reolinkapi/mixins/motion.py
Normal file
@@ -0,0 +1,80 @@
|
||||
from typing import Union, List, Dict
|
||||
from datetime import datetime as dt
|
||||
|
||||
|
||||
# Type hints for input and output of the motion api response
|
||||
RAW_MOTION_LIST_TYPE = List[Dict[str, Union[str, float, Dict[str, str]]]]
|
||||
PROCESSED_MOTION_LIST_TYPE = List[Dict[str, Union[str, dt]]]
|
||||
|
||||
|
||||
class MotionAPIMixin:
|
||||
"""API calls for past motion alerts."""
|
||||
def get_motion_files(self, start: dt, end: dt = dt.now(),
|
||||
streamtype: str = 'sub') -> PROCESSED_MOTION_LIST_TYPE:
|
||||
"""
|
||||
Get the timestamps and filenames of motion detection events for the time range provided.
|
||||
|
||||
Args:
|
||||
start: the starting time range to examine
|
||||
end: the end time of the time range to examine
|
||||
streamtype: 'main' or 'sub' - the stream to examine
|
||||
:return: response json
|
||||
"""
|
||||
search_params = {
|
||||
'Search': {
|
||||
'channel': 0,
|
||||
'streamType': streamtype,
|
||||
'onlyStatus': 0,
|
||||
'StartTime': {
|
||||
'year': start.year,
|
||||
'mon': start.month,
|
||||
'day': start.day,
|
||||
'hour': start.hour,
|
||||
'min': start.minute,
|
||||
'sec': start.second
|
||||
},
|
||||
'EndTime': {
|
||||
'year': end.year,
|
||||
'mon': end.month,
|
||||
'day': end.day,
|
||||
'hour': end.hour,
|
||||
'min': end.minute,
|
||||
'sec': end.second
|
||||
}
|
||||
}
|
||||
}
|
||||
body = [{"cmd": "Search", "action": 1, "param": search_params}]
|
||||
|
||||
resp = self._execute_command('Search', body)[0]
|
||||
result = resp['value']['SearchResult']
|
||||
files = result.get('File', [])
|
||||
if len(files) > 0:
|
||||
# Begin processing files
|
||||
processed_files = self._process_motion_files(files)
|
||||
return processed_files
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _process_motion_files(motion_files: RAW_MOTION_LIST_TYPE) -> PROCESSED_MOTION_LIST_TYPE:
|
||||
"""Processes raw list of dicts containing motion timestamps
|
||||
and the filename associated with them"""
|
||||
# Process files
|
||||
processed_motions = []
|
||||
replace_fields = {'mon': 'month', 'sec': 'second', 'min': 'minute'}
|
||||
for file in motion_files:
|
||||
time_range = {}
|
||||
for x in ['Start', 'End']:
|
||||
# Get raw dict
|
||||
raw = file[f'{x}Time']
|
||||
# Replace certain keys
|
||||
for k, v in replace_fields.items():
|
||||
if k in raw.keys():
|
||||
raw[v] = raw.pop(k)
|
||||
time_range[x.lower()] = dt(**raw)
|
||||
start, end = time_range.values()
|
||||
processed_motions.append({
|
||||
'start': start,
|
||||
'end': end,
|
||||
'filename': file['name']
|
||||
})
|
||||
return processed_motions
|
||||
118
reolinkapi/mixins/network.py
Normal file
118
reolinkapi/mixins/network.py
Normal file
@@ -0,0 +1,118 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class NetworkAPIMixin:
|
||||
"""API calls for network settings."""
|
||||
def set_net_port(self, http_port: float = 80, https_port: float = 443, media_port: float = 9000,
|
||||
onvif_port: float = 8000, rtmp_port: float = 1935, rtsp_port: float = 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: str, password: str) -> Dict:
|
||||
body = [{"cmd": "SetWifi", "action": 0, "param": {
|
||||
"Wifi": {
|
||||
"ssid": ssid,
|
||||
"password": password
|
||||
}}}]
|
||||
return self._execute_command('SetWifi', body)
|
||||
|
||||
def get_net_ports(self) -> Dict:
|
||||
"""
|
||||
Get network ports
|
||||
See examples/response/GetNetworkAdvanced.json for example response data.
|
||||
: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) -> Dict:
|
||||
body = [{"cmd": "GetWifi", "action": 1, "param": {}}]
|
||||
return self._execute_command('GetWifi', body)
|
||||
|
||||
def scan_wifi(self) -> Dict:
|
||||
body = [{"cmd": "ScanWifi", "action": 1, "param": {}}]
|
||||
return self._execute_command('ScanWifi', body)
|
||||
|
||||
def get_network_general(self) -> Dict:
|
||||
"""
|
||||
Get the camera information
|
||||
See examples/response/GetNetworkGeneral.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetLocalLink", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetLocalLink', body)
|
||||
|
||||
def get_network_ddns(self) -> Dict:
|
||||
"""
|
||||
Get the camera DDNS network information
|
||||
See examples/response/GetNetworkDDNS.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetDdns", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetDdns', body)
|
||||
|
||||
def get_network_ntp(self) -> Dict:
|
||||
"""
|
||||
Get the camera NTP network information
|
||||
See examples/response/GetNetworkNTP.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetNtp", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetNtp', body)
|
||||
|
||||
def get_network_email(self) -> Dict:
|
||||
"""
|
||||
Get the camera email network information
|
||||
See examples/response/GetNetworkEmail.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetEmail", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetEmail', body)
|
||||
|
||||
def get_network_ftp(self) -> Dict:
|
||||
"""
|
||||
Get the camera FTP network information
|
||||
See examples/response/GetNetworkFtp.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetFtp", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetFtp', body)
|
||||
|
||||
def get_network_push(self) -> Dict:
|
||||
"""
|
||||
Get the camera push network information
|
||||
See examples/response/GetNetworkPush.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetPush", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetPush', body)
|
||||
|
||||
def get_network_status(self) -> Dict:
|
||||
"""
|
||||
Get the camera status network information
|
||||
See examples/response/GetNetworkGeneral.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
return self.get_network_general()
|
||||
123
reolinkapi/mixins/ptz.py
Normal file
123
reolinkapi/mixins/ptz.py
Normal file
@@ -0,0 +1,123 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class PtzAPIMixin:
|
||||
"""
|
||||
API for PTZ functions.
|
||||
"""
|
||||
def _send_operation(self, operation: str, speed: float, index: float = None) -> Dict:
|
||||
# Refactored to reduce redundancy
|
||||
param = {"channel": 0, "op": operation, "speed": speed}
|
||||
if index is not None:
|
||||
param['id'] = index
|
||||
data = [{"cmd": "PtzCtrl", "action": 0, "param": param}]
|
||||
return self._execute_command('PtzCtrl', data)
|
||||
|
||||
def _send_noparm_operation(self, operation: str) -> Dict:
|
||||
data = [{"cmd": "PtzCtrl", "action": 0, "param": {"channel": 0, "op": operation}}]
|
||||
return self._execute_command('PtzCtrl', data)
|
||||
|
||||
def _send_set_preset(self, enable: float, preset: float = 1, name: str = 'pos1') -> Dict:
|
||||
data = [{"cmd": "SetPtzPreset", "action": 0, "param": {
|
||||
"channel": 0, "enable": enable, "id": preset, "name": name}}]
|
||||
return self._execute_command('PtzCtrl', data)
|
||||
|
||||
def go_to_preset(self, speed: float = 60, index: float = 1) -> Dict:
|
||||
"""
|
||||
Move the camera to a preset location
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('ToPos', speed=speed, index=index)
|
||||
|
||||
def add_preset(self, preset: float = 1, name: str = 'pos1') -> Dict:
|
||||
"""
|
||||
Adds the current camera position to the specified preset.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_set_preset(enable=1, preset=preset, name=name)
|
||||
|
||||
def remove_preset(self, preset: float = 1, name: str = 'pos1') -> Dict:
|
||||
"""
|
||||
Removes the specified preset
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_set_preset(enable=0, preset=preset, name=name)
|
||||
|
||||
def move_right(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera to the right
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('Right', speed=speed)
|
||||
|
||||
def move_right_up(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera to the right and up
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('RightUp', speed=speed)
|
||||
|
||||
def move_right_down(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera to the right and down
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('RightDown', speed=speed)
|
||||
|
||||
def move_left(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera to the left
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('Left', speed=speed)
|
||||
|
||||
def move_left_up(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera to the left and up
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('LeftUp', speed=speed)
|
||||
|
||||
def move_left_down(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera to the left and down
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('LeftDown', speed=speed)
|
||||
|
||||
def move_up(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera up.
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('Up', speed=speed)
|
||||
|
||||
def move_down(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera down.
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('Down', speed=speed)
|
||||
|
||||
def stop_ptz(self) -> Dict:
|
||||
"""
|
||||
Stops the cameras current action.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_noparm_operation('Stop')
|
||||
|
||||
def auto_movement(self, speed: float = 25) -> Dict:
|
||||
"""
|
||||
Move the camera in a clockwise rotation.
|
||||
The camera moves self.stop_ptz() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('Auto', speed=speed)
|
||||
72
reolinkapi/mixins/record.py
Normal file
72
reolinkapi/mixins/record.py
Normal file
@@ -0,0 +1,72 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class RecordAPIMixin:
|
||||
"""API calls for the recording settings"""
|
||||
|
||||
def get_recording_encoding(self) -> Dict:
|
||||
"""
|
||||
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) -> Dict:
|
||||
"""
|
||||
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)
|
||||
|
||||
def set_recording_encoding(self,
|
||||
audio: float = 0,
|
||||
main_bit_rate: float = 8192,
|
||||
main_frame_rate: float = 8,
|
||||
main_profile: str = 'High',
|
||||
main_size: str = "2560*1440",
|
||||
sub_bit_rate: float = 160,
|
||||
sub_frame_rate: float = 7,
|
||||
sub_profile: str = 'High',
|
||||
sub_size: str = '640*480') -> Dict:
|
||||
"""
|
||||
Sets the current camera encoding settings for "Clear" and "Fluent" profiles.
|
||||
:param audio: int Audio on or off
|
||||
:param main_bit_rate: int Clear Bit Rate
|
||||
:param main_frame_rate: int Clear Frame Rate
|
||||
:param main_profile: string Clear Profile
|
||||
:param main_size: string Clear Size
|
||||
:param sub_bit_rate: int Fluent Bit Rate
|
||||
:param sub_frame_rate: int Fluent Frame Rate
|
||||
:param sub_profile: string Fluent Profile
|
||||
:param sub_size: string Fluent Size
|
||||
:return: response
|
||||
"""
|
||||
body = [
|
||||
{
|
||||
"cmd": "SetEnc",
|
||||
"action": 0,
|
||||
"param": {
|
||||
"Enc": {
|
||||
"audio": audio,
|
||||
"channel": 0,
|
||||
"mainStream": {
|
||||
"bitRate": main_bit_rate,
|
||||
"frameRate": main_frame_rate,
|
||||
"profile": main_profile,
|
||||
"size": main_size
|
||||
},
|
||||
"subStream": {
|
||||
"bitRate": sub_bit_rate,
|
||||
"frameRate": sub_frame_rate,
|
||||
"profile": sub_profile,
|
||||
"size": sub_size
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
return self._execute_command('SetEnc', body)
|
||||
52
reolinkapi/mixins/stream.py
Normal file
52
reolinkapi/mixins/stream.py
Normal file
@@ -0,0 +1,52 @@
|
||||
import string
|
||||
from random import random
|
||||
from typing import Any, Optional
|
||||
from urllib import parse
|
||||
from io import BytesIO
|
||||
|
||||
import requests
|
||||
from PIL.Image import Image, open as open_image
|
||||
|
||||
from reolinkapi.utils.rtsp_client import RtspClient
|
||||
|
||||
|
||||
class StreamAPIMixin:
|
||||
""" API calls for opening a video stream or capturing an image from the camera."""
|
||||
|
||||
def open_video_stream(self, callback: Any = None, proxies: Any = None) -> Any:
|
||||
"""
|
||||
'https://support.reolink.com/hc/en-us/articles/360007010473-How-to-Live-View-Reolink-Cameras-via-VLC-Media-Player'
|
||||
Blocking function creates a generator and returns the frames as it is spawned
|
||||
:param callback:
|
||||
:param proxies: Default is none, example: {"host": "localhost", "port": 8000}
|
||||
"""
|
||||
rtsp_client = RtspClient(
|
||||
ip=self.ip, username=self.username, password=self.password, proxies=proxies, callback=callback)
|
||||
return rtsp_client.open_stream()
|
||||
|
||||
def get_snap(self, timeout: float = 3, proxies: Any = None) -> Optional[Image]:
|
||||
"""
|
||||
Gets a "snap" of the current camera video data and returns a Pillow Image or None
|
||||
:param timeout: Request timeout to camera in seconds
|
||||
:param proxies: http/https proxies to pass to the request object.
|
||||
:return: Image or None
|
||||
"""
|
||||
data = {
|
||||
'cmd': 'Snap',
|
||||
'channel': 0,
|
||||
'rs': ''.join(random.choices(string.ascii_uppercase + string.digits, k=10)),
|
||||
'user': self.username,
|
||||
'password': self.password,
|
||||
}
|
||||
parms = parse.urlencode(data).encode("utf-8")
|
||||
|
||||
try:
|
||||
response = requests.get(self.url, proxies=proxies, params=parms, timeout=timeout)
|
||||
if response.status_code == 200:
|
||||
return open_image(BytesIO(response.content))
|
||||
print("Could not retrieve data from camera successfully. Status:", response.status_code)
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print("Could not get Image data\n", e)
|
||||
raise
|
||||
44
reolinkapi/mixins/system.py
Normal file
44
reolinkapi/mixins/system.py
Normal file
@@ -0,0 +1,44 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class SystemAPIMixin:
|
||||
"""API for accessing general system information of the camera."""
|
||||
def get_general_system(self) -> Dict:
|
||||
""":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) -> Dict:
|
||||
"""
|
||||
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) -> Dict:
|
||||
"""
|
||||
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) -> Dict:
|
||||
"""
|
||||
Reboots the camera
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "Reboot", "action": 0, "param": {}}]
|
||||
return self._execute_command('Reboot', body)
|
||||
|
||||
def get_dst(self) -> Dict:
|
||||
"""
|
||||
Get the camera DST information
|
||||
See examples/response/GetDSTInfo.json for example response data.
|
||||
:return: response json
|
||||
"""
|
||||
body = [{"cmd": "GetTime", "action": 0, "param": {}}]
|
||||
return self._execute_command('GetTime', body)
|
||||
65
reolinkapi/mixins/user.py
Normal file
65
reolinkapi/mixins/user.py
Normal file
@@ -0,0 +1,65 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class UserAPIMixin:
|
||||
"""User-related API calls."""
|
||||
def get_online_user(self) -> Dict:
|
||||
"""
|
||||
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) -> Dict:
|
||||
"""
|
||||
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)[0]
|
||||
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)[0]
|
||||
if r_data["value"]["rspCode"] == 200:
|
||||
return True
|
||||
print(f"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)[0]
|
||||
if r_data["value"]["rspCode"] == 200:
|
||||
return True
|
||||
print(f"Could not delete user: {username}\nCamera responded with: {r_data['value']}")
|
||||
return False
|
||||
59
reolinkapi/mixins/zoom.py
Normal file
59
reolinkapi/mixins/zoom.py
Normal file
@@ -0,0 +1,59 @@
|
||||
from typing import Dict
|
||||
|
||||
|
||||
class ZoomAPIMixin:
|
||||
"""
|
||||
API for zooming and changing focus.
|
||||
Note that the API does not allow zooming/focusing by absolute
|
||||
values rather that changing focus/zoom for a given time.
|
||||
"""
|
||||
def _start_operation(self, operation: str, speed: float) -> Dict:
|
||||
data = [{"cmd": "PtzCtrl", "action": 0, "param": {"channel": 0, "op": operation, "speed": speed}}]
|
||||
return self._execute_command('PtzCtrl', data)
|
||||
|
||||
def _stop_zooming_or_focusing(self) -> Dict:
|
||||
"""This command stops any ongoing zooming or focusing actions."""
|
||||
data = [{"cmd": "PtzCtrl", "action": 0, "param": {"channel": 0, "op": "Stop"}}]
|
||||
return self._execute_command('PtzCtrl', data)
|
||||
|
||||
def start_zooming_in(self, speed: float = 60) -> Dict:
|
||||
"""
|
||||
The camera zooms in until self.stop_zooming() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._start_operation('ZoomInc', speed=speed)
|
||||
|
||||
def start_zooming_out(self, speed: float = 60) -> Dict:
|
||||
"""
|
||||
The camera zooms out until self.stop_zooming() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._start_operation('ZoomDec', speed=speed)
|
||||
|
||||
def stop_zooming(self) -> Dict:
|
||||
"""
|
||||
Stop zooming.
|
||||
:return: response json
|
||||
"""
|
||||
return self._stop_zooming_or_focusing()
|
||||
|
||||
def start_focusing_in(self, speed: float = 32) -> Dict:
|
||||
"""
|
||||
The camera focuses in until self.stop_focusing() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._start_operation('FocusInc', speed=speed)
|
||||
|
||||
def start_focusing_out(self, speed: float = 32) -> Dict:
|
||||
"""
|
||||
The camera focuses out until self.stop_focusing() is called.
|
||||
:return: response json
|
||||
"""
|
||||
return self._start_operation('FocusDec', speed=speed)
|
||||
|
||||
def stop_focusing(self) -> Dict:
|
||||
"""
|
||||
Stop focusing.
|
||||
:return: response json
|
||||
"""
|
||||
return self._stop_zooming_or_focusing()
|
||||
Reference in New Issue
Block a user