[LIST][*][*] class knxuf_12296_hikvison_alerts(object):
BASE="/ISAPI/Event/notification/alertStream"
def __init__(self,local):
import re
self.logik = local.get("pItem")
self.MC = self.logik.MC
EN = local.get("EN")
self.outputs = local.get("AN")
self._thread = None
self.host = EN[1]
self.user = EN[2]
self.password = EN[3]
self.deadtime = EN[4]
self._regex = re.compile("<eventType>(?P<type>.*?)</eventType>")
self.motion = 0
self.linecrossing = 0
self.intrusion = 0
self.url = "https://{0}{1}".format(EN[1],self.BASE)
self.run()
def run(self):
import threading
_cthread = threading.Thread(target=self._connect,name='hikvision_connect_{0}'.format(self.host))
_cthread.setDaemon(True)
_cthread.start()
def _connect(self):
import urllib2
import socket
import ssl
import time
self.connect_error = 0
_connect_wait_time = { 0:0, 1:15, 2:30, 3:45, 4:30, 5:30, 6:60, 7:60, 8:120, 9:120 }
p_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
p_mgr.add_password(None,self.url,self.user,self.password)
auth = urllib2.HTTPBasicAuthHandler(p_mgr)
if hasattr(ssl,"create_default_context"):
_ssl_context = ssl.create_default_context()
_ssl_context.check_hostname = False
_ssl_context.verify_mode = ssl.CERT_NONE
opener = urllib2.build_opener(urllib2.HTTPSHandler(context=_ssl_context),auth)
else:
opener = urllib2.build_opener(auth)
urllib2.install_opener(opener)
while True:
try:
self._read(opener.open(self.url,timeout=1))
except (urllib2.URLError,socket.timeout,ssl.SSLError):
pass
except:
self.MC.Debug.setErr(__import__('sys').exc_info(),"")
self.send_to_output(4,0)
self.connect_error += 1
time.sleep(_connect_wait_time.get(self.connect_error,360))
def _read(self,sockfile):
self.send_to_output(4,1)
_connectmsg = True
self.connect_error = 0
_msg = ""
while True:
_line = sockfile.readline()
if _line.startswith("<EventNotificationAlert"):
_msg = ""
_msg += _line
if _line.startswith("</EventNotificationAlert>"):
#if _msg.find("videoloss")== -1:
self.parse(_msg)
_msg = ""
def parse(self,msg):
_match = self._regex.search(msg)
if _match:
_type = _match.group("type")
if _type == "VMD":
self.set_output(1,1)
elif _type == "linedetection":
self.set_output(2,1)
elif _type == "fielddetection":
self.set_output(3,1)
def set_output(self,output,value):
if value == 1:
if self.outputs[output] != value:
self.send_to_output(output,value)
self.set_timer(output,self.deadtime)
self.outputs[output] = value
return value
## send_out: Wert auf einen Ausgang senden
## Parameter out: Ausgang auf den gesendet werden soll analog zu AN[x]
## Parameter value: Wert der an den Ausgang gesendet werden soll
def send_to_output(self,out,value):
import time
out -= 1 ## Intern starten die Ausgänge bei 0 und nicht bei 1
## Sendeintervall wird beachtet
if self.logik.SendIntervall == 0 or time.time() >= self.logik.Ausgang[out][0]:
## Wert an allen iKOs an den Ausgängen setzen
for iko in self.logik.Ausgang[out][1]:
self.logik.MC.TagList.setWert(FLAG_EIB_LOGIK,iko, value)
## Wenn value nicht 0 / "" / None etc ist dann die Befehle ausführen
if value:
for cmd in self.logik.Ausgang[out][2]:
cmd.execBefehl()
## Direkte Verbindungen und Connectoren
for con in self.logik.Ausgang[out][3]:
self.logik.MC.LogikList.ConnectList.append(con + [ value ])
## Wenn Sendeintervall gesetzt, nächste mögliche Ausführungszeit setzen
if self.logik.SendIntervall > 0:
self.logik.Ausgang[out][0] = time.time() + self.logik.SendIntervall
## Ausgangswert auch in der Logik setzen
self.logik.OutWert[out] = value
## set_timer: Einen Timer der schon definiert wurde in der Logik setzen
## Parameter offset: Timernummer analog zu OC[x]
## Parameter seconds: Sekunden bis zur Ausführung des Timers
## Parameter value (optional): Wert der ON[x] enthalten soll
def set_timer(self,offset,seconds,value=None):
import time
##
offset -= 1
_nextstart = time.time() + seconds
self.logik.OutOfset[offset][0] = _nextstart
self.logik.OutOfset[offset][1] = value
_min = min(filter(lambda x: x[0] > 0,self.logik.OutOfset) + [_nextstart])
if _min > 0 and self.logik.NextStart <> _min:
self.logik.NextStart = _min
self.logik.MC.ZyklusTimer.addTimerList(self.logik,self.logik.NextStart)[*][/LIST]
Kommentar