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

Enforce lowercase standard on all submodule files, improve type hinting throughout, complete first pass for logic issues

Browse Source
This commit is contained in:
Bobrock
2020-12-18 09:05:56 -06:00
parent 17bc207e3b
commit 63537f9daf
16 changed files with 456 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
reolink_api/__init__.py
View File

@@ -1,4 +1,4 @@
from .APIHandler import APIHandler
from .Camera import Camera
from .api_handler import APIHandler
from .camera import Camera
__version__ = "0.1.1"
__version__ = "0.1.2"

5
reolink_api/alarm.py
View File

@@ -1,7 +1,10 @@
from typing import Dict
class AlarmAPIMixin:
"""API calls for getting device alarm information."""
def get_alarm_motion(self) -> object:
def get_alarm_motion(self) -> Dict:
"""
Gets the device alarm motion
See examples/response/GetAlarmMotion.json for example response data.

138
reolink_api/api_handler.py Normal file
View File

@@ -0,0 +1,138 @@
import requests
from typing import Dict, List, Optional, Union
from reolink_api.alarm import AlarmAPIMixin
from reolink_api.device import DeviceAPIMixin
from reolink_api.display import DisplayAPIMixin
from reolink_api.download import DownloadAPIMixin
from reolink_api.image import ImageAPIMixin
from reolink_api.motion import MotionAPIMixin
from reolink_api.network import NetworkAPIMixin
from reolink_api.ptz import PtzAPIMixin
from reolink_api.recording import RecordingAPIMixin
from reolink_api.resthandle import Request
from reolink_api.system import SystemAPIMixin
from reolink_api.user import UserAPIMixin
from reolink_api.zoom import ZoomAPIMixin
class APIHandler(AlarmAPIMixin,
DeviceAPIMixin,
DisplayAPIMixin,
DownloadAPIMixin,
ImageAPIMixin,
MotionAPIMixin,
NetworkAPIMixin,
PtzAPIMixin,
RecordingAPIMixin,
SystemAPIMixin,
UserAPIMixin,
ZoomAPIMixin):
"""
The APIHandler class is the backend part of the API, the actual API calls
are implemented in Mixins.
This handles communication directly with the camera.
Current camera's tested: RLC-411WS
All Code will try to follow the PEP 8 standard as described here: https://www.python.org/dev/peps/pep-0008/
"""
def __init__(self, ip: str, username: str, password: str, https: bool = False, **kwargs):
"""
Initialise the Camera API Handler (maps api calls into python)
:param ip:
:param username:
:param password:
:param proxy: Add a proxy dict for requests to consume.
eg: {"http":"socks5://[username]:[password]@[host]:[port], "https": ...}
More information on proxies in requests: https://stackoverflow.com/a/15661226/9313679
"""
scheme = 'https' if https else 'http'
self.url = f"{scheme}://{ip}/cgi-bin/api.cgi"
self.ip = ip
self.token = None
self.username = username
self.password = password
Request.proxies = kwargs.get("proxy") # Defaults to None if key isn't found
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 = response.json()[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:
# TODO: Verify this change w/ owner. Delete old code if acceptable.
# A this point, response is NoneType. There won't be a status code property.
# print("Failed to login\nStatus Code:", response.status_code)
print("Failed to login\nResponse was null.")
return False
except Exception as e:
print("Error Login\n", e)
raise
def logout(self) -> bool:
"""
Logout of the camera
:return: bool
"""
try:
data = [{"cmd": "Logout", "action": 0}]
self._execute_command('Logout', data)
# print(ret)
return True
except Exception as e:
print("Error Logout\n", e)
return False
def _execute_command(self, command: str, data: List[Dict], multi: bool = False) -> \
Optional[Union[Dict, bool]]:
"""
Send a POST request to the IP camera with given data.
:param command: name of the command to send
:param data: object to send to the camera (send as json)
:param multi: whether the given command name should be added to the
url parameters of the request. Defaults to False. (Some multi-step
commands seem to not have a single command name)
:return: response JSON as python object
"""
params = {"token": self.token, 'cmd': command}
if multi:
del params['cmd']
try:
if self.token is None:
raise ValueError("Login first")
if command == 'Download':
# Special handling for downloading an mp4
# Pop the filepath from data
tgt_filepath = data[0].pop('filepath')
# Apply the data to the params
params.update(data[0])
with requests.get(self.url, params=params, stream=True) as req:
if req.status_code == 200:
with open(tgt_filepath, 'wb') as f:
f.write(req.content)
return True
else:
print(f'Error received: {req.status_code}')
return False
else:
response = Request.post(self.url, data=data, params=params)
return response.json()
except Exception as e:
print(f"Command {command} failed: {e}")
raise

24
reolink_api/camera.py Normal file
View File

@@ -0,0 +1,24 @@
from .api_handler import APIHandler
class Camera(APIHandler):
def __init__(self, ip: str, username: str = "admin", password: str = "", https: bool = False):
"""
Initialise the Camera object by passing the ip address.
The default details {"username":"admin", "password":""} will be used if nothing passed
:param ip:
:param username:
:param password:
"""
# For when you need to connect to a camera behind a proxy, pass
# a proxy argument: proxy={"http": "socks5://127.0.0.1:8000"}
APIHandler.__init__(self, ip, username, password, https=https)
# Normal call without proxy:
# APIHandler.__init__(self, ip, username, password)
self.ip = ip
self.username = username
self.password = password
super().login()

17
reolink_api/config_handler.py Normal file
View File

@@ -0,0 +1,17 @@
import io
import yaml
from typing import Optional, Dict
class ConfigHandler:
camera_settings = {}
@staticmethod
def load() -> Optional[Dict]:
try:
stream = io.open("config.yml", 'r', encoding='utf8')
data = yaml.safe_load(stream)
return data
except Exception as e:
print("Config Property Error\n", e)
return None

35
reolink_api/display.py
View File

@@ -1,7 +1,10 @@
from typing import Dict
class DisplayAPIMixin:
"""API calls related to the current image (osd, on screen display)."""
def get_osd(self) -> object:
def get_osd(self) -> Dict:
"""
Get OSD information.
See examples/response/GetOsd.json for example response data.
@@ -10,7 +13,7 @@ class DisplayAPIMixin:
body = [{"cmd": "GetOsd", "action": 1, "param": {"channel": 0}}]
return self._execute_command('GetOsd', body)
def get_mask(self) -> object:
def get_mask(self) -> Dict:
"""
Get the camera mask information.
See examples/response/GetMask.json for example response data.
@@ -19,8 +22,8 @@ class DisplayAPIMixin:
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,
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
@@ -28,18 +31,24 @@ class DisplayAPIMixin:
: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_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"]
: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}
}
}}]
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

34
reolink_api/image.py
View File

@@ -1,3 +1,5 @@
from typing import Dict
class ImageAPIMixin:
"""API calls for image settings."""
@@ -5,20 +7,20 @@ class ImageAPIMixin:
def set_adv_image_settings(self,
anti_flicker: str = 'Outdoor',
exposure: str = 'Auto',
gain_min: int = 1,
gain_max: int = 62,
shutter_min: int = 1,
shutter_max: int = 125,
blue_gain: int = 128,
red_gain: int = 128,
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: int = 128,
drc: int = 128,
rotation: int = 0,
mirroring: int = 0,
nr3d: int = 1) -> object:
blc: float = 128,
drc: float = 128,
rotation: float = 0,
mirroring: float = 0,
nr3d: float = 1) -> Dict:
"""
Sets the advanced camera settings.
@@ -66,11 +68,11 @@ class ImageAPIMixin:
return self._execute_command('SetIsp', body)
def set_image_settings(self,
brightness: int = 128,
contrast: int = 62,
hue: int = 1,
saturation: int = 125,
sharpness: int = 128) -> object:
brightness: float = 128,
contrast: float = 62,
hue: float = 1,
saturation: float = 125,
sharpness: float = 128) -> Dict:
"""
Sets the camera image settings.

25
reolink_api/network.py
View File

@@ -1,3 +1,6 @@
from typing import Dict
class NetworkAPIMixin:
"""API calls for network settings."""
def set_net_port(self, http_port: int = 80, https_port: int = 443, media_port: int = 9000,
@@ -25,7 +28,7 @@ class NetworkAPIMixin:
print("Successfully Set Network Ports")
return True
def set_wifi(self, ssid: str, password: str) -> object:
def set_wifi(self, ssid: str, password: str) -> Dict:
body = [{"cmd": "SetWifi", "action": 0, "param": {
"Wifi": {
"ssid": ssid,
@@ -33,7 +36,7 @@ class NetworkAPIMixin:
}}}]
return self._execute_command('SetWifi', body)
def get_net_ports(self) -> object:
def get_net_ports(self) -> Dict:
"""
Get network ports
See examples/response/GetNetworkAdvanced.json for example response data.
@@ -44,15 +47,15 @@ class NetworkAPIMixin:
{"cmd": "GetP2p", "action": 0, "param": {}}]
return self._execute_command('GetNetPort', body, multi=True)
def get_wifi(self):
def get_wifi(self) -> Dict:
body = [{"cmd": "GetWifi", "action": 1, "param": {}}]
return self._execute_command('GetWifi', body)
def scan_wifi(self):
def scan_wifi(self) -> Dict:
body = [{"cmd": "ScanWifi", "action": 1, "param": {}}]
return self._execute_command('ScanWifi', body)
def get_network_general(self) -> object:
def get_network_general(self) -> Dict:
"""
Get the camera information
See examples/response/GetNetworkGeneral.json for example response data.
@@ -61,7 +64,7 @@ class NetworkAPIMixin:
body = [{"cmd": "GetLocalLink", "action": 0, "param": {}}]
return self._execute_command('GetLocalLink', body)
def get_network_ddns(self) -> object:
def get_network_ddns(self) -> Dict:
"""
Get the camera DDNS network information
See examples/response/GetNetworkDDNS.json for example response data.
@@ -70,7 +73,7 @@ class NetworkAPIMixin:
body = [{"cmd": "GetDdns", "action": 0, "param": {}}]
return self._execute_command('GetDdns', body)
def get_network_ntp(self) -> object:
def get_network_ntp(self) -> Dict:
"""
Get the camera NTP network information
See examples/response/GetNetworkNTP.json for example response data.
@@ -79,7 +82,7 @@ class NetworkAPIMixin:
body = [{"cmd": "GetNtp", "action": 0, "param": {}}]
return self._execute_command('GetNtp', body)
def get_network_email(self) -> object:
def get_network_email(self) -> Dict:
"""
Get the camera email network information
See examples/response/GetNetworkEmail.json for example response data.
@@ -88,7 +91,7 @@ class NetworkAPIMixin:
body = [{"cmd": "GetEmail", "action": 0, "param": {}}]
return self._execute_command('GetEmail', body)
def get_network_ftp(self) -> object:
def get_network_ftp(self) -> Dict:
"""
Get the camera FTP network information
See examples/response/GetNetworkFtp.json for example response data.
@@ -97,7 +100,7 @@ class NetworkAPIMixin:
body = [{"cmd": "GetFtp", "action": 0, "param": {}}]
return self._execute_command('GetFtp', body)
def get_network_push(self) -> object:
def get_network_push(self) -> Dict:
"""
Get the camera push network information
See examples/response/GetNetworkPush.json for example response data.
@@ -106,7 +109,7 @@ class NetworkAPIMixin:
body = [{"cmd": "GetPush", "action": 0, "param": {}}]
return self._execute_command('GetPush', body)
def get_network_status(self) -> object:
def get_network_status(self) -> Dict:
"""
Get the camera status network information
See examples/response/GetNetworkGeneral.json for example response data.

49
reolink_api/ptz.py
View File

@@ -1,46 +1,49 @@
from typing import Dict
class PtzAPIMixin:
"""
API for PTZ functions.
"""
def _send_operation(self, operation, speed, index=None):
if index is None:
data = [{"cmd": "PtzCtrl", "action": 0, "param": {"channel": 0, "op": operation, "speed": speed}}]
else:
data = [{"cmd": "PtzCtrl", "action": 0, "param": {
"channel": 0, "op": operation, "speed": speed, "id": index}}]
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):
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, operation, enable, preset=1, name='pos1'):
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=60, index=1):
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=1, name='pos1'):
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('PtzPreset', enable=1, preset=preset, name=name)
return self._send_set_preset(enable=1, preset=preset, name=name)
def remove_preset(self, preset=1, name='pos1'):
def remove_preset(self, preset: float = 1, name: str = 'pos1') -> Dict:
"""
Removes the specified preset
:return: response json
"""
return self._send_set_preset('PtzPreset', enable=0, preset=preset, name=name)
return self._send_set_preset(enable=0, preset=preset, name=name)
def move_right(self, speed=25):
def move_right(self, speed: float = 25) -> Dict:
"""
Move the camera to the right
The camera moves self.stop_ptz() is called.
@@ -48,7 +51,7 @@ class PtzAPIMixin:
"""
return self._send_operation('Right', speed=speed)
def move_right_up(self, speed=25):
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.
@@ -56,7 +59,7 @@ class PtzAPIMixin:
"""
return self._send_operation('RightUp', speed=speed)
def move_right_down(self, speed=25):
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.
@@ -64,7 +67,7 @@ class PtzAPIMixin:
"""
return self._send_operation('RightDown', speed=speed)
def move_left(self, speed=25):
def move_left(self, speed: float = 25) -> Dict:
"""
Move the camera to the left
The camera moves self.stop_ptz() is called.
@@ -72,7 +75,7 @@ class PtzAPIMixin:
"""
return self._send_operation('Left', speed=speed)
def move_left_up(self, speed=25):
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.
@@ -80,7 +83,7 @@ class PtzAPIMixin:
"""
return self._send_operation('LeftUp', speed=speed)
def move_left_down(self, speed=25):
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.
@@ -88,7 +91,7 @@ class PtzAPIMixin:
"""
return self._send_operation('LeftDown', speed=speed)
def move_up(self, speed=25):
def move_up(self, speed: float = 25) -> Dict:
"""
Move the camera up.
The camera moves self.stop_ptz() is called.
@@ -96,7 +99,7 @@ class PtzAPIMixin:
"""
return self._send_operation('Up', speed=speed)
def move_down(self, speed=25):
def move_down(self, speed: float = 25) -> Dict:
"""
Move the camera down.
The camera moves self.stop_ptz() is called.
@@ -104,14 +107,14 @@ class PtzAPIMixin:
"""
return self._send_operation('Down', speed=speed)
def stop_ptz(self):
def stop_ptz(self) -> Dict:
"""
Stops the cameras current action.
:return: response json
"""
return self._send_noparm_operation('Stop')
def auto_movement(self, speed=25):
def auto_movement(self, speed: float = 25) -> Dict:
"""
Move the camera in a clockwise rotation.
The camera moves self.stop_ptz() is called.

91
reolink_api/recording.py
View File

@@ -3,14 +3,15 @@ import random
import string
from urllib import parse
from io import BytesIO
from PIL import Image
from reolink_api.RtspClient import RtspClient
from typing import Dict, Any, Optional
from PIL.Image import Image, open as open_image
from reolink_api.rtsp_client import RtspClient
class RecordingAPIMixin:
"""API calls for recording/streaming image or video."""
def get_recording_encoding(self) -> object:
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.
@@ -19,7 +20,7 @@ class RecordingAPIMixin:
body = [{"cmd": "GetEnc", "action": 1, "param": {"channel": 0}}]
return self._execute_command('GetEnc', body)
def get_recording_advanced(self) -> object:
def get_recording_advanced(self) -> Dict:
"""
Get recording advanced setup data
See examples/response/GetRec.json for example response data.
@@ -29,15 +30,15 @@ class RecordingAPIMixin:
return self._execute_command('GetRec', body)
def set_recording_encoding(self,
audio=0,
main_bit_rate=8192,
main_frame_rate=8,
main_profile='High',
main_size="2560*1440",
sub_bit_rate=160,
sub_frame_rate=7,
sub_profile='High',
sub_size='640*480') -> object:
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
@@ -51,59 +52,67 @@ class RecordingAPIMixin:
: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}}
}}]
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)
###########
# RTSP Stream
###########
def open_video_stream(self, callback=None, profile: str = "main", proxies=None):
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 profile: profile is "main" or "sub"
: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: int = 3, proxies=None) -> Image or None:
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 = {}
data['cmd'] = 'Snap'
data['channel'] = 0
data['rs'] = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10))
data['user'] = self.username
data['password'] = self.password
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 Image.open(BytesIO(response.content))
print("Could not retrieve data from camera successfully. Status:", response.stats_code)
return open_image(BytesIO(response.content))
print("Could not retrieve data from camera successfully. Status:", response.status_code)
return None
except Exception as e:

11
reolink_api/resthandle.py
View File

@@ -1,12 +1,14 @@
import json
import requests
from typing import List, Dict, Union, Optional
class Request:
proxies = None
@staticmethod
def post(url: str, data, params=None) -> requests.Response or None:
def post(url: str, data: List[Dict], params: Dict[str, Union[str, float]] = None) -> \
Optional[requests.Response]:
"""
Post request
:param params:
@@ -18,10 +20,6 @@ class Request:
headers = {'content-type': 'application/json'}
r = requests.post(url, verify=False, params=params, json=data, headers=headers,
proxies=Request.proxies)
# if params is not None:
# r = requests.post(url, params=params, json=data, headers=headers, proxies=proxies)
# else:
# r = requests.post(url, json=data)
if r.status_code == 200:
return r
else:
@@ -31,7 +29,7 @@ class Request:
raise
@staticmethod
def get(url, params, timeout=1) -> json or None:
def get(url: str, params: Dict[str, Union[str, float]], timeout: float = 1) -> Optional[requests.Response]:
"""
Get request
:param url:
@@ -41,7 +39,6 @@ class Request:
"""
try:
data = requests.get(url=url, verify=False, params=params, timeout=timeout, proxies=Request.proxies)
return data
except Exception as e:
print("Get Error\n", e)

110
reolink_api/rtsp_client.py Normal file
View File

@@ -0,0 +1,110 @@
import os
from threading import ThreadError
from typing import Any
import cv2
from reolink_api.util import threaded
class RtspClient:
"""
Inspiration from:
- https://benhowell.github.io/guide/2015/03/09/opencv-and-web-cam-streaming
- https://stackoverflow.com/questions/19846332/python-threading-inside-a-class
- https://stackoverflow.com/questions/55828451/video-streaming-from-ip-camera-in-python-using-opencv-cv2-videocapture
"""
def __init__(self, ip: str, username: str, password: str, port: float = 554, profile: str = "main",
use_udp: bool = True, callback: Any = None, **kwargs):
"""
RTSP client is used to retrieve frames from the camera in a stream
:param ip: Camera IP
:param username: Camera Username
:param password: Camera User Password
:param port: RTSP port
:param profile: "main" or "sub"
:param use_upd: True to use UDP, False to use TCP
:param proxies: {"host": "localhost", "port": 8000}
"""
self.capture = None
self.thread_cancelled = False
self.callback = callback
capture_options = 'rtsp_transport;'
self.ip = ip
self.username = username
self.password = password
self.port = port
self.proxy = kwargs.get("proxies")
self.url = f'rtsp://{self.username}:{self.password}@{self.ip}:{self.port}//h264Preview_01_{profile}'
if use_udp:
capture_options = capture_options + 'udp'
else:
capture_options = capture_options + 'tcp'
os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = capture_options
# opens the stream capture, but does not retrieve any frames yet.
self._open_video_capture()
def _open_video_capture(self):
# To CAP_FFMPEG or not To ?
self.capture = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
def _stream_blocking(self):
while True:
try:
if self.capture.isOpened():
ret, frame = self.capture.read()
if ret:
yield frame
else:
print("stream closed")
self.capture.release()
return
except Exception as e:
print(e)
self.capture.release()
return
@threaded
def _stream_non_blocking(self):
while not self.thread_cancelled:
try:
if self.capture.isOpened():
ret, frame = self.capture.read()
if ret:
self.callback(frame)
else:
print("stream is closed")
self.stop_stream()
except ThreadError as e:
print(e)
self.stop_stream()
def stop_stream(self):
self.capture.release()
self.thread_cancelled = True
def open_stream(self):
"""
Opens OpenCV Video stream and returns the result according to the OpenCV documentation
https://docs.opencv.org/3.4/d8/dfe/classcv_1_1VideoCapture.html#a473055e77dd7faa4d26d686226b292c1
:param callback: The function to callback the cv::mat frame to if required to be non-blocking. If this is left
as None, then the function returns a generator which is blocking.
"""
# Reset the capture object
if self.capture is None or not self.capture.isOpened():
self._open_video_capture()
print("opening stream")
if self.callback is None:
return self._stream_blocking()
else:
# reset the thread status if the object was not re-created
if not self.thread_cancelled:
self.thread_cancelled = False
return self._stream_non_blocking()

15
reolink_api/system.py
View File

@@ -1,11 +1,14 @@
from typing import Dict
class SystemAPIMixin:
"""API for accessing general system information of the camera."""
def get_general_system(self) -> object:
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) -> object:
def get_performance(self) -> Dict:
"""
Get a snapshot of the current performance of the camera.
See examples/response/GetPerformance.json for example response data.
@@ -14,7 +17,7 @@ class SystemAPIMixin:
body = [{"cmd": "GetPerformance", "action": 0, "param": {}}]
return self._execute_command('GetPerformance', body)
def get_information(self) -> object:
def get_information(self) -> Dict:
"""
Get the camera information
See examples/response/GetDevInfo.json for example response data.
@@ -23,7 +26,7 @@ class SystemAPIMixin:
body = [{"cmd": "GetDevInfo", "action": 0, "param": {}}]
return self._execute_command('GetDevInfo', body)
def reboot_camera(self) -> object:
def reboot_camera(self) -> Dict:
"""
Reboots the camera
:return: response json
@@ -31,11 +34,11 @@ class SystemAPIMixin:
body = [{"cmd": "Reboot", "action": 0, "param": {}}]
return self._execute_command('Reboot', body)
def get_dst(self) -> object:
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)
return self._execute_command('GetTime', body)

7
reolink_api/user.py
View File

@@ -1,6 +1,9 @@
from typing import Dict
class UserAPIMixin:
"""User-related API calls."""
def get_online_user(self) -> object:
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.
@@ -9,7 +12,7 @@ class UserAPIMixin:
body = [{"cmd": "GetOnline", "action": 1, "param": {}}]
return self._execute_command('GetOnline', body)
def get_users(self) -> object:
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.

2
reolink_api/util.py
View File

@@ -8,4 +8,4 @@ def threaded(fn):
thread.start()
return thread
return wrapper
return wrapper

19
reolink_api/zoom.py
View File

@@ -1,54 +1,57 @@
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, speed):
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):
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=60):
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=60):
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):
def stop_zooming(self) -> Dict:
"""
Stop zooming.
:return: response json
"""
return self._stop_zooming_or_focusing()
def start_focusing_in(self, speed=32):
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=32):
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):
def stop_focusing(self) -> Dict:
"""
Stop focusing.
:return: response json