Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 37b929fcbe |
@@ -2,12 +2,27 @@ from reolinkapi.handlers.api_handler import APIHandler
|
||||
|
||||
|
||||
class Camera(APIHandler):
|
||||
def __exit__(self):
|
||||
if self._token is not None:
|
||||
print('Logging out')
|
||||
self.logout()
|
||||
else:
|
||||
print('not logged in, logging out')
|
||||
|
||||
|
||||
def __del__(self):
|
||||
if self._token is not None:
|
||||
print('Logging out')
|
||||
self.logout()
|
||||
else:
|
||||
print('not logged in, logging out')
|
||||
|
||||
|
||||
def __init__(self, ip: str,
|
||||
username: str = "admin",
|
||||
password: str = "",
|
||||
https: bool = False,
|
||||
defer_login: bool = False,
|
||||
defer_login: bool = True,
|
||||
profile: str = "main",
|
||||
**kwargs):
|
||||
"""
|
||||
|
||||
@@ -52,11 +52,28 @@ class APIHandler(AlarmAPIMixin,
|
||||
scheme = 'https' if https else 'http'
|
||||
self.url = f"{scheme}://{ip}/cgi-bin/api.cgi"
|
||||
self.ip = ip
|
||||
self.token = None
|
||||
self._token = None
|
||||
self.username = username
|
||||
self.password = password
|
||||
Request.proxies = kwargs.get("proxy") # Defaults to None if key isn't found
|
||||
|
||||
def __getattr__(self, name):
|
||||
if name == 'token':
|
||||
if self._token is None:
|
||||
self.login()
|
||||
return self._token
|
||||
|
||||
else:
|
||||
return getattr(self, name)
|
||||
|
||||
def __del__(self):
|
||||
if self._token is not None:
|
||||
print('Logging out')
|
||||
self.logout()
|
||||
else:
|
||||
print('not logged in, logging out')
|
||||
|
||||
|
||||
def login(self) -> bool:
|
||||
"""
|
||||
Get login token
|
||||
@@ -72,7 +89,7 @@ class APIHandler(AlarmAPIMixin,
|
||||
data = response.json()[0]
|
||||
code = data["code"]
|
||||
if int(code) == 0:
|
||||
self.token = data["value"]["Token"]["name"]
|
||||
self._token = data["value"]["Token"]["name"]
|
||||
print("Login success")
|
||||
return True
|
||||
print(self.token)
|
||||
@@ -95,6 +112,7 @@ class APIHandler(AlarmAPIMixin,
|
||||
try:
|
||||
data = [{"cmd": "Logout", "action": 0}]
|
||||
self._execute_command('Logout', data)
|
||||
self._token = None
|
||||
# print(ret)
|
||||
return True
|
||||
except Exception as e:
|
||||
|
||||
@@ -12,3 +12,7 @@ class AlarmAPIMixin:
|
||||
"""
|
||||
body = [{"cmd": "GetAlarm", "action": 1, "param": {"Alarm": {"channel": 0, "type": "md"}}}]
|
||||
return self._execute_command('GetAlarm', body)
|
||||
|
||||
def get_ai_state(self) -> Dict:
|
||||
body = [{"cmd": "GetAIState"}]
|
||||
return self._execute_command('GetAlarm', body)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from typing import Dict
|
||||
|
||||
import time
|
||||
|
||||
class PtzAPIMixin:
|
||||
"""
|
||||
@@ -121,3 +121,65 @@ class PtzAPIMixin:
|
||||
:return: response json
|
||||
"""
|
||||
return self._send_operation('Auto', speed=speed)
|
||||
|
||||
def get_zoom_focus_pos(self) -> Dict:
|
||||
body = [{"cmd": "GetZoomFocus", "action": 0, "param": {'channel':0}}]
|
||||
return self._execute_command('GetZoomFocus',body)
|
||||
|
||||
def set_auto_focus_state(self, state: bool = True) -> Dict:
|
||||
body = [{"cmd": "SetAutoFocus", "action": 0, "param": {'AutoFocus':{'channel':0,'disable':int(state)}}}]
|
||||
return self._execute_command('SetAutoFocus',body)
|
||||
|
||||
def get_auto_focus_state(self) -> bool:
|
||||
body = [{"cmd": "GetAutoFocus", "action": 1, "param": {'channel':0}}]
|
||||
return bool(self._execute_command('GetAutoFocus',body)[0]['value']['AutoFocus']['disable'])
|
||||
|
||||
|
||||
def set_zoom_pos(self, pos: int) -> Dict:
|
||||
body = [{"cmd": "StartZoomFocus", "action": 0, "param": {'ZoomFocus':{'channel':0,'pos':pos,'op':'ZoomPos'}}}]
|
||||
self._execute_command('StartZoomFocus',body)
|
||||
|
||||
|
||||
def set_focus_pos(self, pos: int) -> Dict:
|
||||
body = [{"cmd": "StartZoomFocus", "action": 0, "param": {'ZoomFocus':{'channel':0,'pos':pos,'op':'FocusPos'}}}]
|
||||
self._execute_command('StartZoomFocus',body)
|
||||
|
||||
|
||||
def generic_set_block(self, pos:int, get_func, set_func) -> int:
|
||||
|
||||
|
||||
change_index = 0
|
||||
last_pos = get_func()
|
||||
pos = min([pos,254])
|
||||
pos = max([pos, 0])
|
||||
|
||||
target_pos = pos
|
||||
set_func(pos)
|
||||
for i in range(100):
|
||||
cpos = get_func()
|
||||
if cpos != last_pos:
|
||||
change_index = i
|
||||
|
||||
if cpos == target_pos:
|
||||
break
|
||||
|
||||
|
||||
if (i - change_index) > 8:
|
||||
break
|
||||
|
||||
last_pos = cpos
|
||||
time.sleep(0.25)
|
||||
|
||||
return cpos
|
||||
|
||||
def set_focus_pos_block(self, pos: int) -> int:
|
||||
return self.generic_set_block(pos, self.get_focus_pos, self.set_focus_pos)
|
||||
|
||||
def set_zoom_pos_block(self, pos: int) -> int:
|
||||
return self.generic_set_block(pos, self.get_zoom_pos, self.set_zoom_pos)
|
||||
|
||||
def get_zoom_pos(self) -> int:
|
||||
return self.get_zoom_focus_pos()[0]['value']['ZoomFocus']['zoom']['pos']
|
||||
|
||||
def get_focus_pos(self) -> int:
|
||||
return self.get_zoom_focus_pos()[0]['value']['ZoomFocus']['focus']['pos']
|
||||
|
||||
@@ -1 +1,2 @@
|
||||
.
|
||||
Pillow
|
||||
opencv-python
|
||||
|
||||
Reference in New Issue
Block a user