YOUreka/tools-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

* Implement ClockIN service for IDID Biometrics

Browse Source 
* Make Biometrics service configurable to domain
* Use native threading with flask (remove 'eventlet')
* Other minor UI adjustment.
This commit is contained in:
adebisi-fa
2022-01-18 09:17:26 +01:00
parent 1ec205e3d5
commit e48beab2f5
11 changed files with 408 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+175 -53
View File
@@ -1,21 +1,36 @@
import eventlet
import json
import socketio
import uuid
import base64
import os
import requests
from flask import Flask
import gi
gi.require_version("FPrint", "2.0")
gi.require_version("Gio", "2.0")
from gi.repository import FPrint, GLib, Gio
sio = socketio.Server()
static_files = {
'/assets': './public',
'/': {'filename': './public/index.html'}
config = {
'domain': {
'name': 'UNICAL',
'code': 'uc'
},
'mode': 'capture'
}
app = socketio.WSGIApp(sio, static_files=static_files)
configFileExists = os.path.exists('.config.json');
if configFileExists:
with open('.config.json') as reader:
config = json.loads(reader.read());
sio = socketio.Server(async_mode="threading")
app = Flask(__name__)
static_files = {
'/assets': './public/{}'.format(config['mode']),
'/': { 'filename': './public/{}/index.html'.format(config['mode']) }
}
app.wsgi_app = socketio.WSGIApp(sio, static_files=static_files)
stopEnrollment = False
def progress(device, completed_stages, prnt, sessionId, error):
@@ -57,29 +72,20 @@ def enroll(sessionId, data):
sio.emit('fprint-enroll-done', { 'ok': False, 'message': e})
finally:
device.close_sync()
is_verify_active = False
device = None;
def open_dev(sessionId):
device = None;
ctx = FPrint.Context()
devices = ctx.get_devices()
if (len(devices) == 0):
print("There is no attached device.")
else:
device = devices[0]
if (not device):
print("There is no attached device")
if (not device.is_open()):
device.open_sync()
while (not device.is_open()):
print('Opening...')
return device
def verify(sessionId, data):
def verify_start(sessionId, data):
global device;
global is_verify_active;
ctx = FPrint.Context()
devices = ctx.get_devices()
if (len(devices) == 0):
print("There is no attached device.")
sio.emit('fprint-verify-start-error', "There is no attached device.");
return;
else:
device = devices[0]
if (not device):
@@ -90,36 +96,127 @@ def verify(sessionId, data):
while (not device.is_open()):
print('Opening...')
with open('adebisi-fa.jpg', 'r+b') as f:
img = f.read()
prtRecord = FPrint.Print.deserialize(img)
canceller = Gio.Cancellable.new()
(match, prtFound) = device.verify_sync(prtRecord, canceller, None)
if (match):
sio.emit('fprint-verify-done', { 'ok': True, 'data': data }, room=sessionId)
else:
sio.emit('fprint-verify-done', { 'ok': False }, room=sessionId)
device.close_sync()
def verify_short(sessionId, data):
d = open_dev(sessionId)
with open('adebisi-fa.jpg', 'r+b') as f:
img = f.read()
prtRecord = FPrint.Print.deserialize(img)
(match, print) = d.verify_sync(prtRecord, None, None)
if (match):
sio.emit('fprint-verify-done', { 'ok': True, 'data': data }, room=sessionId)
while True:
with open('adebisi-fa.jpg', 'r+b') as f:
img = f.read()
prtRecord = FPrint.Print.deserialize(img)
(match, prtFound) = device.verify_sync(prtRecord, Gio.Cancellable.new(), None)
if (match):
sio.emit('fprint-verify-start-done', { 'ok': True, 'data': data }, room=sessionId)
else:
sio.emit('fprint-verify-start-done', { 'ok': False }, room=sessionId)
if not is_verify_active:
break;
if device.is_open():
device.close_sync()
pass
is_identify_active = False
identify_canceller = None
raw_prints = None
def identify_start(sessionId, data):
global device;
global is_identify_active;
global identify_canceller;
global raw_prints;
ctx = FPrint.Context()
devices = ctx.get_devices()
if (len(devices) == 0):
print("There is no attached device.")
sio.emit('fprint-identify-start-error', "There is no attached device.");
return;
else:
sio.emit('fprint-verify-done', { 'ok': False }, room=sessionId)
d.close_sync()
device = devices[0]
if (not device):
print("There is no attached device")
if (not device.is_open()):
device.open_sync()
while (not device.is_open()):
print('Opening...')
sio.emit('fprint-identify-start-done', room=sessionId)
is_identify_active = True
while True:
prtRecords = [];
with open('adebisi-fa.jpg', 'r+b') as f:
img = f.read()
prtRecords = [FPrint.Print.deserialize(img)]
try:
identify_canceller = Gio.Cancellable.new();
(match, prtFound) = device.identify_sync(prtRecords, identify_canceller, None)
if (match):
sio.emit('fprint-identify-done', { 'ok': True, 'name': 'Adebisi Foluso A.', 'phone': "+2348068287222" }, room=sessionId)
print('Found.')
else:
sio.emit('fprint-identify-done', { 'ok': False }, room=sessionId)
print('Not found.')
except:
print("Identification stopped due to error.")
sio.emit("fprint-identify-stop-done", { 'ok': True, 'message': 'Identification stopped due to error.' })
is_identify_active = False;
finally:
if not is_identify_active:
print('Identification is active. Quiting...')
break;
pass
if device.is_open():
try:
device.close_sync()
except:
print('An error occured while closing device.');
finally:
print('It appears device has already been removed.')
# events
@sio.on("fprint-verify")
def fprint_verify(sid, data):
verify_short(sid, data)
@sio.on("fprint-verify-stop")
def fprint_verify_stop(sid, data):
global is_verify_active;
if (device.is_open()):
try:
device.close_sync();
sio.emit("fprint-verify-stop-done", { 'ok': True, 'message': 'Verifycation stopped.' })
return;
except:
print('An error occured while closing device.');
finally:
print("It appears device has already been removed.")
else:
sio.emit("fprint-verify-stop-done", { 'ok': False, 'message': 'Verifycation was inactive.' })
is_verify_active = False;
pass
@sio.on("fprint-verify-start")
def fprint_verify_start(sid, data):
verify_start(sid, data)
def stop_fprint_identify():
global is_identify_active;
global identify_canceller;
if identify_canceller:
identify_canceller.cancel()
sio.emit("fprint-identify-stop-done", { 'ok': True, 'message': 'Identification stopped.' })
is_identify_active = False;
@sio.on("fprint-identify-start")
def fprint_identify_start(sid, data):
identify_start(sid, data)
@sio.on("fprint-identify-stop")
def fprint_identify_stop(sid, data):
stop_fprint_identify()
@sio.on("fprint-enroll")
def fprint_enroll(sid, data):
print('Enrolling {}'.format(data['finger']))
@@ -133,7 +230,32 @@ def capture(sid, data):
@sio.event
def connect(sid, data):
# print({ 'sid': sid, 'event': 'connect', 'data': data});
global config;
if config['mode'] == 'clockio':
global raw_prints;
stop_fprint_identify();
config['started'] = is_identify_active
domain_code = config['domain']['code'];
response = requests.get("https://apis.a2loft.net/identity/{}/units".format(domain_code))
responseJson = json.loads(response.text)
if responseJson['ok']:
config['departments'] = responseJson['data'];
domain_fprint_file = '{}/fprint-sync/.{}-fprint-data.json'.format(os.getenv("HOME"), domain_code);
print(domain_fprint_file);
if os.path.exists(domain_fprint_file):
with open(domain_fprint_file) as f:
records = json.loads(f.read());
raw_prints = records['profiles'];
prints_dict = {};
for profile in raw_prints:
for fp in profile['fingerprints']:
prints_dict[fp] = profile
config['records'] = {
'length': len(records['profiles']),
'timestamp': records['timestamp']
};
sio.emit('config', config);
pass
@sio.event
@@ -141,7 +263,6 @@ def disconnect(sid):
print({ 'sid': sid, 'event': 'disconnect'});
pass
@sio.on('capture-other')
def capture_other(sid, data):
print({ "event": "capture_other", "data": data })
@@ -150,4 +271,5 @@ def capture_other(sid, data):
with open('/tmp/libfprint-server', 'w') as f:
f.write(str(os.getpid()));
eventlet.wsgi.server(eventlet.listen(('127.0.0.1', 9000), reuse_port=False), app)
if __name__ == '__main__':
app.run(port=9000)