Интеграция Flussonic Watcher с системами контроля и управления доступом¶
Flussonic Watcher умеет отправлять команды системам контроля доступа для открытия дверей после того, как Watcher распознал человека из списка. Ниже приведен скрипт для интеграции с системой контроля доступа Sigur и с комментариями по его использованию.
import socket
import sys
import argparse
import http.server
import socketserver
import cgi
import json
import requests
import logging
import logging.config
from logging.handlers import TimedRotatingFileHandler
fh = TimedRotatingFileHandler("ACS_integration.log", when='midnight')
sh = logging.StreamHandler()
logging.basicConfig(handlers=(fh, sh),
format='[%(asctime)s.%(msecs)03d | %(levelname)s]: %(message)s',
datefmt='%d.%m.%Y %H:%M:%S',
level=logging.INFO)
class ACS_Sigur:
@staticmethod
def connect(ip, port):
sigur = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sigur.connect((ip, int(port)))
except:
print("Connection error: ", sys.exc_info())
else:
print("Sigur server connected!")
return sigur
@staticmethod
def login(sigur):
message = "LOGIN 1.8 Administrator\r\n" #Change your credentials to connect to Sigur
sigur.send(bytes(message, 'utf-8'))
reply = sigur.recv(1024)
data_reply = reply.decode('ascii')
data_reply.replace('\n','')
if "OK" in data_reply:
print("Login in server successfull")
else:
print("Sigur server is not connected. That is why:", data_reply)
@staticmethod
def open_door(sigur):
message = "ALLOWPASS 1 2 IN\r\n" #Change ID of a door you want to open. You can find ID in a Sigur app.
sigur.send(bytes(message, 'utf-8'))
reply = sigur.recv(1024)
data_reply = reply.decode('ascii')
data_reply.replace('\n','')
if "OK" in data_reply:
print("Door is opened")
else:
print("Something went wrong. That is why:", data_reply)
def create_cmd_parser():
parser = argparse.ArgumentParser()
parser.add_argument('-ACS', action="store", dest="ACS")
parser.add_argument('-ip', action="store", dest="ip")
parser.add_argument('-port', action="store", dest="port")
parser.add_argument('-serverport', action="store", dest="serverport")
return parser
parser = create_cmd_parser()
args = parser.parse_args()
class MyHandler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
logging.info("New request from {client}".format(client = self.client_address))
content_length = self.headers.get('content-length')
if content_length == None:
result = ""
else:
body = self.rfile.read(int(content_length))
result = json.loads(body, encoding='utf-8')
logging.info("Request JSON is: {result}".format(result = result))
if args.ACS == "Sigur":
logging.info("Trying connect to the Sigur server")
try:
sigur = ACS_Sigur.connect(args.ip, args.port)
logging.info("Sigur server connected")
ACS_Sigur.login(sigur)
logging.info("Success login on a Siger server")
try:
ACS_Sigur.open_door(sigur)
logging.info("Success login on a Sigur server")
except:
logging.warning("There was a problem while opening door")
sigur.close()
logging.info("Connection with the Sigur server closed")
except:
logging.warning("Something went wrong. Basic request is nor JSON: {body}".format(body = body))
if args.ACS == "Beward":
logging.info("Trying connect to the Beward")
uri = 'http://'+args.ip+'/cgi-bin/io/port.cgi?action=O0:/'
try:
response = requests.get(uri, auth=('admin', 'admin'))
if response.status_code == 200:
logging.info("Door was opened")
else:
logging.warning("Something went wrong")
except:
logging.warning("There was an error while sending command to Beward")
def main():
logging.info("Application started ACS - {ACS}, IP address - {ip}, port - {port}, serverport - {serverport}".format(ACS = args.ACS, ip = args.ip, port = args.port, serverport = args.serverport))
if args.ACS == "Sigur":
logging.info("ACS Sigur chosen")
elif args.ACS == "Beward":
logging.info("ACS Beward chosen")
else:
logging.info("ACS is unknown. Please use \"Sigur\" or \"Beward\".\nIntegrated module shutted down.")
sys.exit()
try:
with socketserver.TCPServer(("127.0.0.1", int(args.serverport)), MyHandler) as httpd:
logging.info("Started listening port {port}".format(port = args.serverport))
httpd.serve_forever()
except:
logging.warning("Cannot connect to the listening port or someone shut application down")
sys.exit()
if __name__ == "__main__":
main()
Как использовать этот скрипт для интеграции Watcher co СКУД:
1) Измените список сценариев в соответствии с вашей установкой Sigur. Вам нужно будет установить логин и пароль к Sigur, отредактировав строку:
message = "LOGIN 1.8 Administrator\r\n"
2) Измените ID двери, которую необходимо открывать:
message = "ALLOWPASS 1 2 IN\r\n"
3) Отредактируйте файл systemd сервиса, чтобы добавить автозапуск:
[Unit]
Description=ACS Unlocker
After=network.target
[Service]
Type=simple
Restart=always
RestartSec=3
User=root
Group=root
WorkingDirectory=/opt/acs-unlocker
ExecStart=/opt/flussonic/bin/python3 runner.py -ACS Sigur -ip {IP} -port 3312 -serverport {serverport}
[Install]
WantedBy=multi-user.target
Вам необходимо указать {IP} сервера Sigur и {serverport} на сервере с установленным модулем интеграции, который будет прослушивать события от модуля распознавания лиц Flussonic Watcher и отправлять их в Sigur.
Замечание: Порт {serverport} следует использовать при настройке подписки на уведомления о событиях в модуле распознавания лиц.
4) Перезагрузите и перезапустите новую службу systemd.
5) Подпишитесь на события распознавания лиц с помощью Flussonic Watcher API, который отправляет событие на порт сервера, который вы выбрали в файле .service. События распознавания лиц имеют тип person_detected, а если вы строите СКУД на распознавании номеров авто, то используйте тип car_detected.
Пример скрипта для создания подписки на уведомления о персоне с id=1:
curl --header "Content-Type: application/json" \
--request POST \
-d '{"camera_id": "person.detection.test.camera-7d8ea4ebf2",
"notification_type": "http",
"event_type": "person_detected",
"webhook_params": {
"url": "http://example.com",
"method": "post",
"params": {
"mode": "single",
"id": 1
}}}' \
http://localhost/vsaas/api/v2/my/subscriptions
Обратите внимание, что параметр mode может принимать значения single (одна персона) и list (список персон), и в зависимости от этого id будет соответственно идентификатором конкретной персоны или списка персон.
В параметре url укажите адрес сервера, на который отправлять уведомления.
Для notification_type на данный момент поддерживается только значение http.
6) Произведите распознавание и проверьте правильность работы модуля интеграции.