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

Merge pull request #37 from barretobrock/develop

Browse Source 
Refactor package to be imported as a single package
This commit is contained in:
Alano Terblanche
2020-12-19 22:34:37 +02:00
committed by GitHub
parent 0221215e33 0fdd0df468
commit e1cb9bda3e
35 changed files with 601 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,3 +1,4 @@
secrets.cfg
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]

24
Camera.py
View File

@@ -1,24 +0,0 @@
from api import APIHandler
class Camera(APIHandler):
def __init__(self, ip, username="admin", password="", https=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
ConfigHandler.py
View File

@@ -1,17 +0,0 @@
import io
import yaml
class ConfigHandler:
camera_settings = {}
@staticmethod
def load() -> yaml or None:
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

6
README.md
View File

@@ -2,9 +2,9 @@
<p align="center">
<img alt="Reolink Approval" src="https://img.shields.io/badge/reolink-approved-blue?style=flat-square">
<img alt="GitHub" src="https://img.shields.io/github/license/ReolinkCameraApi/reolink-python-api?style=flat-square">
<img alt="GitHub tag (latest SemVer)" src="https://img.shields.io/github/v/tag/ReolinkCameraApi/reolink-python-api?style=flat-square">
<img alt="PyPI" src="https://img.shields.io/pypi/v/reolink-api?style=flat-square">
<img alt="GitHub" src="https://img.shields.io/github/license/ReolinkCameraAPI/reolinkapipy?style=flat-square">
<img alt="GitHub tag (latest SemVer)" src="https://img.shields.io/github/v/tag/ReolinkCameraAPI/reolinkapipy?style=flat-square">
<img alt="PyPI" src="https://img.shields.io/pypi/v/reolinkapi?style=flat-square">
<img alt="Discord" src="https://img.shields.io/discord/773257004911034389?style=flat-square">
</p>

4
api/__init__.py
View File

@@ -1,4 +0,0 @@
from .APIHandler import APIHandler
__version__ = "0.0.5"
VERSION = __version__

111
api/recording.py
View File

@@ -1,111 +0,0 @@
import requests
import random
import string
from urllib import parse
from io import BytesIO
from PIL import Image
from RtspClient import RtspClient
class RecordingAPIMixin:
"""API calls for recording/streaming image or video."""
def get_recording_encoding(self) -> object:
"""
Get the current camera encoding settings for "Clear" and "Fluent" profiles.
See examples/response/GetEnc.json for example response data.
:return: response json
"""
body = [{"cmd": "GetEnc", "action": 1, "param": {"channel": 0}}]
return self._execute_command('GetEnc', body)
def get_recording_advanced(self) -> object:
"""
Get recording advanced setup data
See examples/response/GetRec.json for example response data.
:return: response json
"""
body = [{"cmd": "GetRec", "action": 1, "param": {"channel": 0}}]
return self._execute_command('GetRec', body)
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:
"""
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)
###########
# RTSP Stream
###########
def open_video_stream(self, callback=None, profile: str = "main", proxies=None):
"""
'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 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:
"""
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
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 None
except Exception as e:
print("Could not get Image data\n", e)
raise

11
examples/basic_usage.py Normal file
View File

@@ -0,0 +1,11 @@
import reolinkapi
if __name__ == "__main__":
cam = reolinkapi.Camera("192.168.0.102", defer_login=True)
# must first login since I defer have deferred the login process
cam.login()
dst = cam.get_dst()
ok = cam.add_user("foo", "bar", "admin")
alarm = cam.get_alarm_motion()

44
examples/download_motions.py Normal file
View File

@@ -0,0 +1,44 @@
"""Downloads all motion events from camera from the past hour."""
import os
from configparser import RawConfigParser
from datetime import datetime as dt, timedelta
from reolinkapi import Camera
def read_config(props_path: str) -> dict:
"""Reads in a properties file into variables.
NB! this config file is kept out of commits with .gitignore. The structure of this file is such:
# secrets.cfg
[camera]
ip={ip_address}
username={username}
password={password}
"""
config = RawConfigParser()
assert os.path.exists(props_path), f"Path does not exist: {props_path}"
config.read(props_path)
return config
# Read in your ip, username, & password
# (NB! you'll likely have to create this file. See tests/test_camera.py for details on structure)
config = read_config('../secrets.cfg')
ip = config.get('camera', 'ip')
un = config.get('camera', 'username')
pw = config.get('camera', 'password')
# Connect to camera
cam = Camera(ip, un, pw)
start = (dt.now() - timedelta(hours=1))
end = dt.now()
# Collect motion events between these timestamps for substream
processed_motions = cam.get_motion_files(start=start, end=end, streamtype='sub')
dl_dir = os.path.join(os.path.expanduser('~'), 'Downloads')
for i, motion in enumerate(processed_motions):
fname = motion['filename']
# Download the mp4
resp = cam.get_file(fname, output_path=os.path.join(dl_dir, f'motion_event_{i}.mp4'))

3
examples/streaming_video.py
View File

@@ -1,6 +1,5 @@
import cv2
from Camera import Camera
from reolinkapi import Camera
def non_blocking():

4
reolinkapi/__init__.py Normal file
View File

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

38
reolinkapi/camera.py Normal file
View File

@@ -0,0 +1,38 @@
from reolinkapi.handlers.api_handler import APIHandler
class Camera(APIHandler):
def __init__(self, ip: str,
username: str = "admin",
password: str = "",
https: bool = False,
defer_login: bool = False,
**kwargs):
"""
Initialise the Camera object by passing the ip address.
The default details {"username":"admin", "password":""} will be used if nothing passed
For deferring the login to the camera, just pass defer_login = True.
For connecting to the camera behind a proxy pass a proxy argument: proxy={"http": "socks5://127.0.0.1:8000"}
:param ip:
:param username:
:param password:
:param https: connect to the camera over https
:param defer_login: defer the login process
: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
"""
# 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, **kwargs)
# Normal call without proxy:
# APIHandler.__init__(self, ip, username, password)
self.ip = ip
self.username = username
self.password = password
if not defer_login:
super().login()

0
reolinkapi/handlers/__init__.py Normal file
View File

71
api/APIHandler.py → reolinkapi/handlers/api_handler.py
View File

@@ -1,26 +1,34 @@
from .recording import RecordingAPIMixin
from .zoom import ZoomAPIMixin
from .device import DeviceAPIMixin
from .display import DisplayAPIMixin
from .network import NetworkAPIMixin
from .system import SystemAPIMixin
from .user import UserAPIMixin
from .ptz import PtzAPIMixin
from .alarm import AlarmAPIMixin
from .image import ImageAPIMixin
from resthandle import Request
import requests
from typing import Dict, List, Optional, Union
from reolinkapi.mixins.alarm import AlarmAPIMixin
from reolinkapi.mixins.device import DeviceAPIMixin
from reolinkapi.mixins.display import DisplayAPIMixin
from reolinkapi.mixins.download import DownloadAPIMixin
from reolinkapi.mixins.image import ImageAPIMixin
from reolinkapi.mixins.motion import MotionAPIMixin
from reolinkapi.mixins.network import NetworkAPIMixin
from reolinkapi.mixins.ptz import PtzAPIMixin
from reolinkapi.mixins.record import RecordAPIMixin
from reolinkapi.handlers.rest_handler import Request
from reolinkapi.mixins.stream import StreamAPIMixin
from reolinkapi.mixins.system import SystemAPIMixin
from reolinkapi.mixins.user import UserAPIMixin
from reolinkapi.mixins.zoom import ZoomAPIMixin
class APIHandler(SystemAPIMixin,
NetworkAPIMixin,
UserAPIMixin,
class APIHandler(AlarmAPIMixin,
DeviceAPIMixin,
DisplayAPIMixin,
RecordingAPIMixin,
ZoomAPIMixin,
DownloadAPIMixin,
ImageAPIMixin,
MotionAPIMixin,
NetworkAPIMixin,
PtzAPIMixin,
AlarmAPIMixin,
ImageAPIMixin):
RecordAPIMixin,
SystemAPIMixin,
UserAPIMixin,
ZoomAPIMixin,
StreamAPIMixin):
"""
The APIHandler class is the backend part of the API, the actual API calls
are implemented in Mixins.
@@ -30,12 +38,13 @@ class APIHandler(SystemAPIMixin,
All Code will try to follow the PEP 8 standard as described here: https://www.python.org/dev/peps/pep-0008/
"""
def __init__(self, ip: str, username: str, password: str, https=False, **kwargs):
def __init__(self, ip: str, username: str, password: str, https: bool = False, **kwargs):
"""
Initialise the Camera API Handler (maps api calls into python)
:param ip:
:param username:
:param password:
:param https: connect over https
: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
@@ -69,7 +78,10 @@ class APIHandler(SystemAPIMixin,
print(self.token)
return False
else:
print("Failed to login\nStatus Code:", response.status_code)
# 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)
@@ -89,7 +101,8 @@ class APIHandler(SystemAPIMixin,
print("Error Logout\n", e)
return False
def _execute_command(self, command, data, multi=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
@@ -105,6 +118,22 @@ class APIHandler(SystemAPIMixin,
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:

16
resthandle.py → reolinkapi/handlers/rest_handler.py
View File

@@ -1,13 +1,13 @@
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:
@@ -17,11 +17,8 @@ class Request:
"""
try:
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)
r = requests.post(url, verify=False, params=params, json=data, headers=headers,
proxies=Request.proxies)
if r.status_code == 200:
return r
else:
@@ -31,7 +28,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 +38,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)

0
reolinkapi/mixins/__init__.py Normal file
View File

5
api/alarm.py → reolinkapi/mixins/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.

11
api/device.py → reolinkapi/mixins/device.py
View File

@@ -1,6 +1,11 @@
from typing import List, Dict
class DeviceAPIMixin:
"""API calls for getting device information."""
def get_hdd_info(self) -> object:
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.
@@ -9,12 +14,14 @@ class DeviceAPIMixin:
body = [{"cmd": "GetHddInfo", "action": 0, "param": {}}]
return self._execute_command('GetHddInfo', body)
def format_hdd(self, hdd_id: [int] = [0]) -> bool:
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:

33
api/display.py → reolinkapi/mixins/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: 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
@@ -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},
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

18
reolinkapi/mixins/download.py Normal file
View 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

44
api/image.py → reolinkapi/mixins/image.py
View File

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

80
reolinkapi/mixins/motion.py Normal file
View 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

29
api/network.py → reolinkapi/mixins/network.py
View File

@@ -1,7 +1,10 @@
from typing import Dict
class NetworkAPIMixin:
"""API calls for network settings."""
def set_net_port(self, http_port=80, https_port=443, media_port=9000, onvif_port=8000, rtmp_port=1935,
rtsp_port=554) -> bool:
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
@@ -25,7 +28,7 @@ class NetworkAPIMixin:
print("Successfully Set Network Ports")
return True
def set_wifi(self, ssid, password) -> 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
api/ptz.py → reolinkapi/mixins/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.

72
reolinkapi/mixins/record.py Normal file
View 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
View 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

13
api/system.py → reolinkapi/mixins/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,7 +34,7 @@ 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.

11
api/user.py → reolinkapi/mixins/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.
@@ -45,7 +48,7 @@ class UserAPIMixin:
r_data = self._execute_command('ModifyUser', body)[0]
if r_data["value"]["rspCode"] == 200:
return True
print("Could not modify user:", username, "\nCamera responded with:", r_data["value"])
print(f"Could not modify user: {username}\nCamera responded with: {r_data['value']}")
return False
def delete_user(self, username: str) -> bool:
@@ -58,5 +61,5 @@ class UserAPIMixin:
r_data = self._execute_command('DelUser', body)[0]
if r_data["value"]["rspCode"] == 200:
return True
print("Could not delete user:", username, "\nCamera responded with:", r_data["value"])
print(f"Could not delete user: {username}\nCamera responded with: {r_data['value']}")
return False

19
api/zoom.py → reolinkapi/mixins/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

0
reolinkapi/utils/__init__.py Normal file
View File

20
RtspClient.py → reolinkapi/utils/rtsp_client.py
View File

@@ -1,20 +1,21 @@
import os
from threading import ThreadError
from typing import Any
import cv2
from util import threaded
from reolinkapi.utils.util import threaded
class RtspClient:
"""
This is a wrapper of the OpenCV VideoCapture method
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, username, password, port=554, profile="main", use_udp=True, callback=None, **kwargs):
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
@@ -36,12 +37,8 @@ class RtspClient:
self.password = password
self.port = port
self.proxy = kwargs.get("proxies")
self.url = "rtsp://" + self.username + ":" + self.password + "@" + \
self.ip + ":" + str(self.port) + "//h264Preview_01_" + profile
if use_udp:
capture_options = capture_options + 'udp'
else:
capture_options = capture_options + 'tcp'
self.url = f'rtsp://{self.username}:{self.password}@{self.ip}:{self.port}//h264Preview_01_{profile}'
capture_options += 'udp' if use_udp else 'tcp'
os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = capture_options
@@ -91,9 +88,6 @@ class RtspClient:
"""
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

0
util.py → reolinkapi/utils/util.py
View File

10
requirements.txt
View File

@@ -1,4 +1,6 @@
requests
opencv-python
numpy
socks
numpy==1.19.4
opencv-python==4.4.0.46
Pillow==8.0.1
PySocks==1.7.1
PyYaml==5.3.1
requests>=2.18.4

73
setup.py
View File

@@ -1,31 +1,8 @@
#!/usr/bin/python3
import os
import re
import codecs
from setuptools import setup
# Package meta-data.
NAME = 'reolink-api'
DESCRIPTION = 'Reolink Camera API written in Python 3.6'
URL = 'https://github.com/Benehiko/ReolinkCameraAPI'
AUTHOR_EMAIL = ''
AUTHOR = 'Benehiko'
LICENSE = 'GPL-3.0'
INSTALL_REQUIRES = [
'pillow',
'pyyaml',
'requests>=2.18.4',
'numpy',
'opencv-python',
'pysocks'
]
here = os.path.abspath(os.path.dirname(__file__))
# read the contents of your README file
with open(os.path.join(here, 'README.md'), encoding='utf-8') as f:
long_description = f.read()
from setuptools import setup, find_packages
def read(*parts):
@@ -41,9 +18,33 @@ def find_version(*file_paths):
raise RuntimeError("Unable to find version string.")
setup(name=NAME,
# Package meta-data.
NAME = 'reolinkapi'
DESCRIPTION = 'Reolink Camera API client written in Python 3'
URL = 'https://github.com/ReolinkCameraAPI/reolinkapipy'
AUTHOR_EMAIL = 'alanoterblanche@gmail.com'
AUTHOR = 'Benehiko'
LICENSE = 'GPL-3.0'
INSTALL_REQUIRES = [
'numpy==1.19.4',
'opencv-python==4.4.0.46',
'Pillow==8.0.1',
'PySocks==1.7.1',
'PyYaml==5.3.1',
'requests>=2.18.4',
]
here = os.path.abspath(os.path.dirname(__file__))
# read the contents of your README file
with open(os.path.join(here, 'README.md'), encoding='utf-8') as f:
long_description = f.read()
setup(
name=NAME,
python_requires='>=3.6.0',
version=find_version('api', '__init__.py'),
version=find_version('reolinkapi', '__init__.py'),
description=DESCRIPTION,
long_description=long_description,
long_description_content_type='text/markdown',
@@ -52,21 +53,5 @@ setup(name=NAME,
url=URL,
license=LICENSE,
install_requires=INSTALL_REQUIRES,
py_modules=[
'Camera',
'ConfigHandler',
'RtspClient',
'resthandle',
'api.APIHandler',
'api.device',
'api.display',
'api.network',
'api.ptz',
'api.recording',
'api.system',
'api.user',
'api.zoom',
'api.alarm',
'api.image'
]
)
packages=find_packages(exclude=['examples', 'tests'])
)

5
test.py
View File

@@ -1,5 +0,0 @@
from Camera import Camera
c = Camera("192.168.1.112", "admin", "jUa2kUzi")
# print("Getting information", c.get_information())
c.open_video_stream()

40
tests/test_camera.py Normal file
View File

@@ -0,0 +1,40 @@
import os
from configparser import RawConfigParser
import unittest
from reolinkapi import Camera
def read_config(props_path: str) -> dict:
"""Reads in a properties file into variables.
NB! this config file is kept out of commits with .gitignore. The structure of this file is such:
# secrets.cfg
[camera]
ip={ip_address}
username={username}
password={password}
"""
config = RawConfigParser()
assert os.path.exists(props_path), f"Path does not exist: {props_path}"
config.read(props_path)
return config
class TestCamera(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.config = read_config('../secrets.cfg')
def setUp(self) -> None:
self.cam = Camera(self.config.get('camera', 'ip'), self.config.get('camera', 'username'),
self.config.get('camera', 'password'))
def test_camera(self):
"""Test that camera connects and gets a token"""
self.assertTrue(self.cam.ip == self.config.get('camera', 'ip'))
self.assertTrue(self.cam.token != '')
if __name__ == '__main__':
unittest.main()