pavel.muhortov/cctv-scheduler
1
0
Fork 0
generated from pavel.muhortov/template-bash
Code Issues Pull Requests Projects Releases Wiki Activity

add additional headers to Connect.http()

Browse Source
This commit is contained in:
Pavel Muhortov 2023-06-18 10:44:43 +03:00
parent db76bb8e0b
commit 699492f77f
1 changed files with 218 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

331
cctv-scheduler.py
View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# pylint: disable=C0103,C0302,C0114,W0621
import base64
import json import json
import logging import logging
import urllib.request import urllib.request
@ -18,6 +20,7 @@ from paramiko import SSHClient, AutoAddPolicy
class Parse: class Parse:
"""Parser of configs, arguments, parameters. """Parser of configs, arguments, parameters.
""" """
# pylint: disable=C0123
def __init__(self, parameters, block: str = None) -> None: def __init__(self, parameters, block: str = None) -> None:
"""Object constructor. """Object constructor.
@ -65,6 +68,7 @@ class Parse:
""" """
self.data.update(dictionary) self.data.update(dictionary)
# pylint: disable=C0206
def expand(self, store: str = None) -> dict: def expand(self, store: str = None) -> dict:
"""Expand dictionary "key":"name.conf" to dictionary "key":{subkey: subval}. """Expand dictionary "key":"name.conf" to dictionary "key":{subkey: subval}.
@ -79,7 +83,7 @@ class Parse:
config = store + sep + self.data[key] config = store + sep + self.data[key]
else: else:
config = self.data[key] config = self.data[key]
with open(config) as file: with open(config, encoding='UTF-8') as file:
self.data[key] = Parse(file.read()).data self.data[key] = Parse(file.read()).data
return self.data return self.data
@ -106,7 +110,7 @@ class Parse:
Returns: Returns:
str: string as "var1=val1;\nvar2=val2;". str: string as "var1=val1;\nvar2=val2;".
""" """
with open(config) as file: with open(config, encoding='UTF-8') as file:
raw = file.read() raw = file.read()
strs = '' strs = ''
for line in raw.splitlines(): for line in raw.splitlines():
@ -130,7 +134,9 @@ class Parse:
strings = cls.block(blockname, strings) strings = cls.block(blockname, strings)
for line in strings.replace('\n', ';').split(';'): for line in strings.replace('\n', ';').split(';'):
if not line.lstrip().startswith('#') and "=" in line: if not line.lstrip().startswith('#') and "=" in line:
dictionary[line.split('=')[0].strip()] = line.split('=')[1].strip().split(';')[0].strip() dictionary[line.split('=')[0].strip()] = (
line.split('=')[1].strip().split(';')[0].strip()
)
return dictionary return dictionary
@classmethod @classmethod
@ -171,66 +177,93 @@ class Parse:
class Connect: class Connect:
# pylint: disable=W0105
"""Set of connection methods (functions) for various protocols. """Set of connection methods (functions) for various protocols.
""" """
@staticmethod @staticmethod
# pylint: disable=W0102, W0718
def http( def http(
url: str, method: str = 'GET', url: str,
username: str = '', password: str = '', authtype: str = None, method: str = 'GET',
contenttype: str = 'text/plain', contentdata: str = '' username: str = '',
) -> str: password: str = '',
authtype: (str, type(None)) = None,
contenttype: str = 'text/plain',
contentdata: (str, bytes) = '',
headers: dict = {}
) -> dict:
"""Handling HTTP request. """Handling HTTP request.
Args: Args:
url (str): request url. url (str): Handling HTTP request.
method (str, optional): HTTP request method. Defaults to 'GET'. method (str, optional): HTTP request method. Defaults to 'GET'.
username (str, optional): username for url authentication. Defaults to ''. username (str, optional): username for url authentication. Defaults to ''.
password (str, optional): password for url authentication. Defaults to ''. password (str, optional): password for url authentication. Defaults to ''.
authtype (str, optional): digest|basic authentication type. Defaults to None. authtype (str, None, optional): digest|basic authentication type. Defaults to None.
contenttype (str, optional): 'Content-Type' header. Defaults to 'text/plain'. contenttype (str, optional): 'Content-Type' header. Defaults to 'text/plain'.
contentdata (str, optional): content data. Defaults to ''. contentdata (str, bytes, optional): content data. Defaults to ''.
headers (dict, optional): additional headers. Defaults to {}.
Returns: Returns:
str: HTTP response or 'ERROR'. dict: {'success':bool,'result':HTTP response or 'ERROR'}.
""" """
if Do.args_valid(locals(), Connect.http.__annotations__):
if contentdata != '':
headers['Content-Type'] = contenttype
if isinstance(contentdata, str):
contentdata = bytes(contentdata.encode('utf-8'))
# Preparing authorization # Preparing authorization
if authtype: if authtype:
pswd = urllib.request.HTTPPasswordMgrWithDefaultRealm() pswd = urllib.request.HTTPPasswordMgrWithDefaultRealm()
pswd.add_password(None, url, username, password) pswd.add_password(None, url, username, password)
if authtype == 'basic': if authtype == 'basic':
auth = urllib.request.HTTPBasicAuthHandler(pswd) auth = urllib.request.HTTPBasicAuthHandler(pswd)
if authtype == 'digest': token = base64.b64encode((username + ':' + password).encode())
auth = urllib.request.HTTPDigestAuthHandler(pswd) headers['Authorization'] = 'Basic ' + token.decode('utf-8')
urllib.request.install_opener(urllib.request.build_opener(auth)) if authtype == 'digest':
auth = urllib.request.HTTPDigestAuthHandler(pswd)
urllib.request.install_opener(urllib.request.build_opener(auth))
# Preparing request # Preparing request
request = urllib.request.Request(url=url, data=bytes(contentdata.encode('utf-8')), method=method) request = urllib.request.Request(
request.add_header('Content-Type', contenttype) url=url,
data=contentdata,
# Response method=method
try:
response = urllib.request.urlopen(request).read()
logging.debug(
msg=''
+ '\n' + 'uri: ' + url
+ '\n' + 'method: ' + method
+ '\n' + 'username: ' + username
+ '\n' + 'password: ' + password
+ '\n' + 'authtype: ' + authtype
+ '\n' + 'content-type: ' + contenttype
+ '\n' + 'content-data: ' + contentdata
) )
if response.startswith(b'\xff\xd8'): for key, val in headers.items():
return response request.add_header(key, val)
else: if len(contentdata) > 128:
return str(response.decode('utf-8')) contentdata = contentdata[:64] + b' ... ' + contentdata[-64:]
except Exception as error: logging.debug(msg=''
logging.debug(msg='\n' + 'error: ' + str(error)) + '\n' + 'uri: ' + url
return 'ERROR' + '\n' + 'method: ' + method
+ '\n' + 'username: ' + username
+ '\n' + 'password: ' + password
+ '\n' + 'authtype: ' + str(authtype)
+ '\n' + 'headers: ' + json.dumps(headers, indent=2)
+ '\n' + 'content-data: ' + str(contentdata)
)
# Response
try:
response = urllib.request.urlopen(request).read()
if not response.startswith(b'\xff\xd8'):
response = str(response.decode('utf-8'))
return {"success": True, "result": response}
except Exception as error:
logging.debug(msg='\n' + 'error: ' + str(error))
return {"success": False, "result": "ERROR"}
@staticmethod @staticmethod
def ssh_commands(command: str, hostname: str, username: str, password: str, port: int = 22) -> str: # pylint: disable=W0718
def ssh_commands(
command: str,
hostname: str,
username: str,
password: str,
port: int = 22
) -> str:
"""Handling SSH command executing. """Handling SSH command executing.
Args: Args:
@ -267,7 +300,15 @@ class Connect:
return 'ERROR' return 'ERROR'
@staticmethod @staticmethod
def ssh_put_file(src_file: str, dst_file: str, hostname: str, username: str, password: str, port: int = 22) -> str: # pylint: disable=W0718
def ssh_put_file(
src_file: str,
dst_file: str,
hostname: str,
username: str,
password: str,
port: int = 22
) -> str:
"""Handling SFTP upload file. """Handling SFTP upload file.
Args: Args:
@ -346,7 +387,14 @@ class Connect:
return 'ERROR' return 'ERROR'
''' '''
@staticmethod @staticmethod
def ftp_put_file(src_file: str, dst_file: str, hostname: str, username: str, password: str) -> bool: # pylint: disable=W0718,C0116
def ftp_put_file(
src_file: str,
dst_file: str,
hostname: str,
username: str,
password: str
) -> bool:
dst_path = dst_file.split('/')[:-1] dst_path = dst_file.split('/')[:-1]
ftp = FTP(host=hostname) ftp = FTP(host=hostname)
try: try:
@ -397,7 +445,7 @@ class HikISAPI(Connect):
authtype: str = 'digest', authtype: str = 'digest',
hostport: int = 80, protocol: str = 'http', hostport: int = 80, protocol: str = 'http',
channel: int = 101, videoid: int = 1 channel: int = 101, videoid: int = 1
) -> None: ) -> None:
"""Object constructor. """Object constructor.
Args: Args:
@ -424,23 +472,28 @@ class HikISAPI(Connect):
url: str, method: str = 'GET', url: str, method: str = 'GET',
contenttype: str = 'application/x-www-form-urlencoded', contenttype: str = 'application/x-www-form-urlencoded',
contentdata: str = '' contentdata: str = ''
) -> str: ) -> str:
"""Send request to camera. """Send request to camera.
Args: Args:
url (str): API path for request. url (str): API path for request.
method (str, optional): HTTP request method. Defaults to 'GET'. method (str, optional): HTTP request method. Defaults to 'GET'.
contenttype (str, optional): Content-Type header. Defaults to 'application/x-www-form-urlencoded'. contenttype (str, optional): Content-Type header.
Defaults to 'application/x-www-form-urlencoded'.
contentdata (str, optional): data for send with request. Defaults to ''. contentdata (str, optional): data for send with request. Defaults to ''.
Returns: Returns:
str: HTTP response content. str: HTTP response content or 'ERROR'.
""" """
return self.http( response = self.http(
url=url, method=method, url=url, method=method,
username=self._user, password=self._pswd, authtype=self._auth, username=self._user, password=self._pswd, authtype=self._auth,
contenttype=contenttype, contentdata=contentdata contenttype=contenttype, contentdata=contentdata
) )
if response['success']:
return response['result']
else:
return 'ERROR'
def capabilities(self) -> bool: def capabilities(self) -> bool:
"""Get camera capabilities. """Get camera capabilities.
@ -451,7 +504,7 @@ class HikISAPI(Connect):
url = ( url = (
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._viid) + "/capabilities" + "/ISAPI/PTZCtrl/channels/" + str(self._viid) + "/capabilities"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.info(msg='\n' + response + '\n') logging.info(msg='\n' + response + '\n')
@ -459,11 +512,16 @@ class HikISAPI(Connect):
else: else:
return False return False
def downloadjpeg(self, dst_file: str = path.splitext(__file__)[0] + '.jpeg', x: int = 1920, y: int = 1080) -> bool: def downloadjpeg(
self,
dst_file: str = path.splitext(__file__)[0] + '.jpeg',
x: int = 1920,
y: int = 1080
) -> bool:
"""Get static picture from camera. """Get static picture from camera.
Args: Args:
dst_file (str, optional): absolute path of picture to save. Defaults to scriptname+'.jpeg'. dst_file (str, optional): abs picture's path to save. Defaults to scriptname+'.jpeg'.
x (int, optional): picture width. Defaults to 1920. x (int, optional): picture width. Defaults to 1920.
y (int, optional): picture height. Defaults to 1080. y (int, optional): picture height. Defaults to 1080.
@ -475,7 +533,7 @@ class HikISAPI(Connect):
+ "/Streaming/channels/" + str(self._viid) + "/Streaming/channels/" + str(self._viid)
+ "/picture?snapShotImageType=JPEG&videoResolutionWidth=" + "/picture?snapShotImageType=JPEG&videoResolutionWidth="
+ str(x) + "&videoResolutionHeight=" + str(y) + str(x) + "&videoResolutionHeight=" + str(y)
) )
with open(dst_file, "wb") as file: with open(dst_file, "wb") as file:
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
@ -491,8 +549,10 @@ class HikISAPI(Connect):
Returns: Returns:
bool: True if successed. Printing a response with a logger at the INFO level. bool: True if successed. Printing a response with a logger at the INFO level.
""" """
url = (self._prot + '://' + self._host + ':' + str(self._port) url = (
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/status") self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/status"
)
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.info(msg='\n' + response + '\n') logging.info(msg='\n' + response + '\n')
@ -509,7 +569,7 @@ class HikISAPI(Connect):
url = ( url = (
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/System/reboot" + "/ISAPI/System/reboot"
) )
response = self.__call(url=url, method="PUT") response = self.__call(url=url, method="PUT")
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -531,7 +591,7 @@ class HikISAPI(Connect):
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=TILT_UP&speed=" + str(speed) + "/PTZControl?command=TILT_UP&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -553,7 +613,7 @@ class HikISAPI(Connect):
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=TILT_DOWN&speed=" + str(speed) + "/PTZControl?command=TILT_DOWN&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -575,7 +635,7 @@ class HikISAPI(Connect):
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=PAN_LEFT&speed=" + str(speed) + "/PTZControl?command=PAN_LEFT&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -597,7 +657,7 @@ class HikISAPI(Connect):
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=PAN_RIGHT&speed=" + str(speed) + "/PTZControl?command=PAN_RIGHT&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -619,7 +679,7 @@ class HikISAPI(Connect):
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=ZOOM_OUT&speed=" + str(speed) + "/PTZControl?command=ZOOM_OUT&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -641,7 +701,7 @@ class HikISAPI(Connect):
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=ZOOM_IN&speed=" + str(speed) + "/PTZControl?command=ZOOM_IN&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -665,7 +725,7 @@ class HikISAPI(Connect):
+ "/PTZControl?command=GOTO_PRESET&presetNo=" + str(preset) + "/PTZControl?command=GOTO_PRESET&presetNo=" + str(preset)
+ "&speed=" + str(speed) + "&speed=" + str(speed)
+ "&mode=start" + "&mode=start"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -683,7 +743,7 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/PTZ/channels/" + str(self._viid) + "/PTZ/channels/" + str(self._viid)
+ "/PTZControl?command=GOTO_PRESET&mode=stop" + "/PTZControl?command=GOTO_PRESET&mode=stop"
) )
response = self.__call(url=url, method='GET') response = self.__call(url=url, method='GET')
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -706,7 +766,7 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/ISAPI/PTZCtrl/channels/" + str(self._chan)
+ "/absolute" + "/absolute"
) )
xml = ''.join( xml = ''.join(
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
+ '<PTZData><AbsoluteHigh>' + '<PTZData><AbsoluteHigh>'
@ -726,9 +786,12 @@ class HikISAPI(Connect):
"""Set camera moving to direction until other signal or 180 seconds elapse. """Set camera moving to direction until other signal or 180 seconds elapse.
Args: Args:
x (int, optional): acceleration of horizontal camera movement from -100 to 100. Defaults to 0. x (int, optional): acceleration of horizontal camera movement from -100 to 100.
y (int, optional): acceleration of vertical camera movement from -100 to 100. Defaults to 0. Defaults to 0.
z (int, optional): acceleration of zoom camera movement from -100 to 100. Defaults to 0. y (int, optional): acceleration of vertical camera movement from -100 to 100.
Defaults to 0.
z (int, optional): acceleration of zoom camera movement from -100 to 100.
Defaults to 0.
Returns: Returns:
bool: True if successed. bool: True if successed.
@ -737,7 +800,7 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/ISAPI/PTZCtrl/channels/" + str(self._chan)
+ "/continuous" + "/continuous"
) )
xml = ''.join( xml = ''.join(
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
+ '<PTZData>' + '<PTZData>'
@ -745,7 +808,7 @@ class HikISAPI(Connect):
+ '<tilt>' + str(y) + '</tilt>' + '<tilt>' + str(y) + '</tilt>'
+ '<zoom>' + str(z) + '</zoom>' + '<zoom>' + str(z) + '</zoom>'
+ '</PTZData>' + '</PTZData>'
) )
response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml) response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml)
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -757,10 +820,14 @@ class HikISAPI(Connect):
"""Set camera moving to direction until other signal or duration elapse. """Set camera moving to direction until other signal or duration elapse.
Args: Args:
x (int, optional): acceleration of horizontal camera movement from -100 to 100. Defaults to 0. x (int, optional): acceleration of horizontal camera movement from -100 to 100.
y (int, optional): acceleration of vertical camera movement from -100 to 100. Defaults to 0. Defaults to 0.
z (int, optional): acceleration of zoom camera movement from -100 to 100. Defaults to 0. y (int, optional): acceleration of vertical camera movement from -100 to 100.
t (int, optional): duration in ms of acceleration from 0 to 180000. Defaults to 180000. Defaults to 0.
z (int, optional): acceleration of zoom camera movement from -100 to 100.
Defaults to 0.
t (int, optional): duration in ms of acceleration from 0 to 180000.
Defaults to 180000.
Returns: Returns:
bool: True if successed. bool: True if successed.
@ -769,7 +836,7 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/ISAPI/PTZCtrl/channels/" + str(self._chan)
+ "/momentary" + "/momentary"
) )
xml = ''.join( xml = ''.join(
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
+ '<PTZData>' + '<PTZData>'
@ -778,7 +845,7 @@ class HikISAPI(Connect):
+ '<zoom>' + str(z) + '</zoom>' + '<zoom>' + str(z) + '</zoom>'
+ '<Momentary><duration>' + str(t) + '</duration></Momentary>' + '<Momentary><duration>' + str(t) + '</duration></Momentary>'
+ '</PTZData>' + '</PTZData>'
) )
response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml) response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml)
if response != 'ERROR': if response != 'ERROR':
sleep(t/1000) sleep(t/1000)
@ -791,10 +858,14 @@ class HikISAPI(Connect):
"""Set camera moving to direction (polymorph abstraction). """Set camera moving to direction (polymorph abstraction).
Args: Args:
x (int, optional): acceleration of horizontal camera movement from -100 to 100. Defaults to 0. x (int, optional): acceleration of horizontal camera movement from -100 to 100.
y (int, optional): acceleration of vertical camera movement from -100 to 100. Defaults to 0. Defaults to 0.
z (int, optional): acceleration of zoom camera movement from -100 to 100. Defaults to 0. y (int, optional): acceleration of vertical camera movement from -100 to 100.
t (int, optional): duration in ms of acceleration from 0 to 180000. Defaults to 0. Defaults to 0.
z (int, optional): acceleration of zoom camera movement from -100 to 100.
Defaults to 0.
t (int, optional): duration in ms of acceleration from 0 to 180000.
Defaults to 0.
Returns: Returns:
bool: True if successed. bool: True if successed.
@ -814,11 +885,11 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/ISAPI/PTZCtrl/channels/" + str(self._chan)
+ "/homeposition/goto" + "/homeposition/goto"
) )
xml = ''.join( xml = ''.join(
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
+ '<PTZData></PTZData>' + '<PTZData></PTZData>'
) )
response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml) response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml)
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -836,11 +907,11 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/PTZCtrl/channels/" + str(self._chan) + "/ISAPI/PTZCtrl/channels/" + str(self._chan)
+ "/homeposition" + "/homeposition"
) )
xml = ''.join( xml = ''.join(
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
+ '<PTZData></PTZData>' + '<PTZData></PTZData>'
) )
response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml) response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml)
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -848,7 +919,13 @@ class HikISAPI(Connect):
else: else:
return False return False
def settextonosd(self, enabled: str = "true", x: int = 0, y: int = 0, message: str = "") -> bool: def settextonosd(
self,
enabled: str = "true",
x: int = 0,
y: int = 0,
message: str = ""
) -> bool:
"""Set message as video overlay text. """Set message as video overlay text.
Args: Args:
@ -866,7 +943,7 @@ class HikISAPI(Connect):
self._prot + '://' + self._host + ':' + str(self._port) self._prot + '://' + self._host + ':' + str(self._port)
+ "/ISAPI/System/Video/inputs/channels/" + str(self._chan) + "/ISAPI/System/Video/inputs/channels/" + str(self._chan)
+ "/overlays/text" + "/overlays/text"
) )
xml = ''.join( xml = ''.join(
'<?xml version="1.0" encoding="UTF-8"?>' '<?xml version="1.0" encoding="UTF-8"?>'
+ '<TextOverlayList version="1.0" xmlns="http://www.hikvision.com/ver10/XMLSchema">' + '<TextOverlayList version="1.0" xmlns="http://www.hikvision.com/ver10/XMLSchema">'
@ -878,7 +955,7 @@ class HikISAPI(Connect):
+ '<displayText>' + message + '</displayText>' + '<displayText>' + message + '</displayText>'
+ '</TextOverlay>' + '</TextOverlay>'
+ '</TextOverlayList>' + '</TextOverlayList>'
) )
response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml) response = self.__call(url=url, method="PUT", contenttype="text/xml", contentdata=xml)
if response != 'ERROR': if response != 'ERROR':
logging.debug(msg='\n' + response + '\n') logging.debug(msg='\n' + response + '\n')
@ -896,7 +973,7 @@ class Sensor(Connect):
hostname: str, username: str, userpass: str, hostname: str, username: str, userpass: str,
nodetype: str, nodename: str, nodetype: str, nodename: str,
hostport: int = 22 hostport: int = 22
) -> None: ) -> None:
"""Object constructor. """Object constructor.
Args: Args:
@ -929,6 +1006,7 @@ class Sensor(Connect):
username=self._user, password=self._pswd username=self._user, password=self._pswd
) )
# pylint: disable=W0718
def __temperature(self, nodename: str) -> str: def __temperature(self, nodename: str) -> str:
"""Preparating request for ds18b20 sensor type. """Preparating request for ds18b20 sensor type.
@ -969,21 +1047,25 @@ class Sequence:
"""Sequence handling. """Sequence handling.
""" """
@staticmethod @staticmethod
# pylint: disable=W0718
def run( def run(
device: HikISAPI, sensors: dict, sequence: dict, device: HikISAPI, sensors: dict, sequence: dict,
records_root_path: str = None, records_root_path: str = None,
records_root_user: str = None, records_root_user: str = None,
records_root_pass: str = None records_root_pass: str = None
) -> None: ) -> None:
"""Sequences executor. """Sequences executor.
Args: Args:
device (HikISAPI): HikISAPI object. device (HikISAPI): HikISAPI object.
sensors (dict): collection as key=sensorname:value=Sensor object. sensors (dict): collection as key=sensorname:value=Sensor object.
sequence (dict): sequence steps collection. sequence (dict): sequence steps collection.
records_root_path (str, optional): path (local|smb|ftp,sftp) to records directory. Defaults to None. records_root_path (str, optional): path (local|smb|ftp,sftp) to records directory.
records_root_user (str, optional): username if path on remote host. Defaults to None. Defaults to None.
records_root_pass (str, optional): password if path on remote host. Defaults to None. records_root_user (str, optional): username if path on remote host.
Defaults to None.
records_root_pass (str, optional): password if path on remote host.
Defaults to None.
""" """
for key, value in sequence.items(): for key, value in sequence.items():
action = value.split(',')[0].strip() action = value.split(',')[0].strip()
@ -1002,12 +1084,12 @@ class Sequence:
m = sensor_value m = sensor_value
else: else:
m = '' m = ''
logging.info( logging.info(msg=''
msg=' action:' + key + ' = ' + action + ' action:' + key + ' = ' + action
+ ',' + x + ',' + y + ',' + z + ',' + x + ',' + y + ',' + z
+ ',' + p + ',' + s + ',' + t + ',' + p + ',' + s + ',' + t
+ ',' + w + ',' + m + ',' + w + ',' + m
) )
if action == 'capabilities': if action == 'capabilities':
response = device.capabilities() response = device.capabilities()
elif action == 'getcamerapos': elif action == 'getcamerapos':
@ -1052,8 +1134,14 @@ class Sequence:
th = datetime.now().strftime('%H') th = datetime.now().strftime('%H')
tm = datetime.now().strftime('%M') tm = datetime.now().strftime('%M')
ts = datetime.now().strftime('%S') ts = datetime.now().strftime('%S')
records_file_name = (key + '_' + dy + '-' + dm + '-' + dd + '_' + th + '.' + tm + '.' + ts + '.jpeg') records_file_name = (
if device.downloadjpeg(x=int(x), y=int(y), dst_file=records_root_temp + sep + records_file_name): key + '_' + dy + '-' + dm + '-' + dd + '_' + th + '.' + tm + '.' + ts + '.jpeg'
)
if device.downloadjpeg(
x=int(x),
y=int(y),
dst_file=records_root_temp + sep + records_file_name
):
hostname = 'localhost' hostname = 'localhost'
hostport, hosttype = None, None hostport, hosttype = None, None
username = records_root_user username = records_root_user
@ -1078,7 +1166,11 @@ class Sequence:
hostname = hostname.split(':')[0] hostname = hostname.split(':')[0]
if hosttype == 'ftp': if hosttype == 'ftp':
src_file = records_root_temp + sep + records_file_name src_file = records_root_temp + sep + records_file_name
dst_file = hostpath + '/' + dy + '/' + dm + '/' + dv + '/' + dd + '/' + records_file_name dst_file = (
hostpath
+ '/' + dy + '/' + dm + '/' + dv + '/' + dd + '/'
+ records_file_name
)
if Connect.ftp_put_file( if Connect.ftp_put_file(
src_file=src_file, src_file=src_file,
dst_file=dst_file, dst_file=dst_file,
@ -1092,7 +1184,11 @@ class Sequence:
pass pass
elif hosttype == 'sftp': elif hosttype == 'sftp':
src_file = records_root_temp + sep + records_file_name src_file = records_root_temp + sep + records_file_name
dst_file = hostpath + '/' + dy + '/' + dm + '/' + dv + '/' + dd + '/' + records_file_name dst_file = (
hostpath
+ '/' + dy + '/' + dm + '/' + dv + '/' + dd + '/'
+ records_file_name
)
response = Connect.ssh_put_file( response = Connect.ssh_put_file(
src_file=src_file, dst_file=dst_file, src_file=src_file, dst_file=dst_file,
hostname=hostname, port=hostport, hostname=hostname, port=hostport,
@ -1107,18 +1203,23 @@ class Sequence:
response = False response = False
else: else:
src_file = records_root_temp + sep + records_file_name src_file = records_root_temp + sep + records_file_name
dst_file = hostpath + sep + dy + sep + dm + sep + dv + sep + dd + sep + records_file_name dst_file = (
hostpath
+ sep + dy + sep + dm + sep + dv + sep + dd + sep
+ records_file_name)
try: try:
makedirs(hostpath + sep + dy + sep + dm + sep + dv + sep + dd, exist_ok=True) makedirs(
hostpath + sep + dy + sep + dm + sep + dv + sep + dd,
exist_ok=True
)
replace(src=src_file, dst=dst_file) replace(src=src_file, dst=dst_file)
response = True response = True
except Exception as error: except Exception as error:
logging.debug( logging.debug(msg=''
msg=''
+ '\n' + 'src_file: ' + src_file + '\n' + 'src_file: ' + src_file
+ '\n' + 'dst_file: ' + dst_file + '\n' + 'dst_file: ' + dst_file
+ '\n' + 'error: ' + str(error) + '\n' + 'error: ' + str(error)
) )
response = False response = False
else: else:
response = False response = False
@ -1134,6 +1235,7 @@ class Proc:
"""Find a running process from Python. """Find a running process from Python.
""" """
@classmethod @classmethod
# pylint: disable=W0612
def _list_windows(cls) -> list: def _list_windows(cls) -> list:
"""Find all running process with wmi. """Find all running process with wmi.
@ -1174,6 +1276,7 @@ class Proc:
return execlist return execlist
@classmethod @classmethod
# pylint: disable=W0612
def _list_linux(cls) -> list: def _list_linux(cls) -> list:
"""Find all running process with ps. """Find all running process with ps.
@ -1219,6 +1322,7 @@ class Proc:
return None return None
@classmethod @classmethod
# pylint: disable=W0150
def search(cls, find: str, exclude: str = None) -> list: def search(cls, find: str, exclude: str = None) -> list:
"""Find specified processes. """Find specified processes.
@ -1301,12 +1405,12 @@ class FFmpeg:
int: ffmpeg return code int: ffmpeg return code
""" """
if not raw: if not raw:
process = ('' process = ([]
+ cls._bin(ffpath).split() + cls._bin(ffpath).split()
+ cls._src(src).split() + cls._src(src).split()
+ cls._preset(preset, fps).split() + cls._preset(preset, fps).split()
+ cls._dst(dst).split() + cls._dst(dst).split()
) )
else: else:
process = cls._bin(ffpath).split() + raw.split() process = cls._bin(ffpath).split() + raw.split()
@ -1328,7 +1432,7 @@ class FFmpeg:
print(line, flush=True) print(line, flush=True)
else: else:
que.put(line) que.put(line)
return proc.returncode return proc.returncode
@classmethod @classmethod
def _bin(cls, ffpath: str, tool: str = 'ffmpeg') -> str: def _bin(cls, ffpath: str, tool: str = 'ffmpeg') -> str:
@ -1350,11 +1454,11 @@ class FFmpeg:
'\tTarget: /usr/bin/ffmpeg\n' '\tTarget: /usr/bin/ffmpeg\n'
'\n' '\n'
'Install on Windows:\n' 'Install on Windows:\n'
'\tDownload and extract archive from: https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-full.7z\n' '\tDownload and extract: https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-full.7z\n'
'\tTarget: "%PROGRAMFILES%\\ffmpeg\\bin\\ffmpeg.exe"\n' '\tTarget: "%PROGRAMFILES%\\ffmpeg\\bin\\ffmpeg.exe"\n'
'\n' '\n'
'Install on MacOS:\n' 'Install on MacOS:\n'
'\tDownload and extract archive from: https://evermeet.cx/ffmpeg/\n' '\tDownload and extract: https://evermeet.cx/ffmpeg/\n'
'\tTarget: /usr/bin/ffmpeg\n' '\tTarget: /usr/bin/ffmpeg\n'
) )
if not ffpath: if not ffpath:
@ -1565,7 +1669,8 @@ if __name__ == "__main__":
'- Python 3 (tested version 3.9.5), ' '- Python 3 (tested version 3.9.5), '
'- Python 3 modules: paramiko ' '- Python 3 modules: paramiko '
) )
args.add_argument('--config', type=str, default=path.splitext(__file__)[0] + '.conf', required=False, args.add_argument('--config', type=str, default=path.splitext(__file__)[0] + '.conf',
required=False,
help='custom configuration file path') help='custom configuration file path')
args.add_argument('-b', '--broadcast', action='store_true', required=False, args.add_argument('-b', '--broadcast', action='store_true', required=False,
help='streaming media to destination') help='streaming media to destination')