Документация Flussonic Watcher

Содержание

Интеграция 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.

6) Произведите распознавание и проверьте правильность работы модуля интеграции.