Skip to content

Integrating Watcher with access control systemsΒΆ

Flussonic Watcher can send commands to access control systems (ACS) so that doors can be open after recognizing a person from the list. Below you can find a script for integration with the Sigur access control system, with comments on its use.


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()

How to use the script for access control integration:

1) Change a script listing according to your Sigur installation. You'll need to set Sigur login and password by editing the line:

message = "LOGIN 1.8 Administrator\r\n"

2) Set the ID of the door you want to open:

message = "ALLOWPASS 1 2 IN\r\n"

3) Change a systemd configuration file for your service to provide autostart:

[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

You'll need to set the {IP} of a Sigur server and the {serverport} on the server with installed integration module that will listen to events from Flussonic Watcher Face Recognition module and send it to Sigur.

Note: You will need to specify the {serverport} port when configuring events notification subscription in a face recognition module.

4) Reload and restart your new systemd service.

5) Subscribe to face recognition events by using Flussonic Watcher API, which sends events on the server port that you chose in a .service file. The face recognition events have person_detected type, but if you are implementing your access control system based on LPR then use car_detected.

Example of the request to create the subscription to the notifications about the person with 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

Please note that the mode parameter can take one of the two values: single (one person) or list (list of persons); depending on that, id is the identifier of a person or a person list correspondingly.

In the url parameter, specify the address of the server to which the notifications should be sent. The notification_type supports only http at the moment.

6) Test how recognition works and make sure that the integration module is working correctly.