tools-legacy/app.py

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

config = {
    'domain': {
        'name': 'UNICAL',
        'code': 'uc'
    },
    'mode': 'capture'
}

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):
    print({'completedStages': completed_stages, 'sessionId': sessionId})
    global stopEnrollment
    if (error):
        stopEnrollment = True
        print({'error': error})
    else:
        if (completed_stages == 5):
            if (not stopEnrollment):
                sio.emit("fprint-enroll-status", { 'stage': completed_stages }, room=sessionId)
            else:
                print('Completed with an error.')

def enroll(sessionId, data):
    ctx = FPrint.Context()
    devices = ctx.get_devices()
    if (len(devices) == 0):
        print("There is no attached device.")
        sio.emit('fprint-enroll-error', "There is no attached device.");
        return;
    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...') 
    try:
        prt = FPrint.Print.new(device)
        prt = device.enroll_sync(prt, None, progress_cb=progress, progress_data=sessionId)
        img = prt.serialize()
        sio.emit('fprint-enroll-done', { 'ok': True, "fingerprint": base64.b64encode(img).decode('utf-8'), 'id': str(uuid.uuid4()), 'data': data })
    except Exception as e:
        print({'error': e});
        sio.emit('fprint-enroll-done', { 'ok': False, 'message': e})
    finally:
        device.close_sync()
             
is_verify_active = False
device = None;

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):
            print("There is no attached device")

        if (not device.is_open()):
            device.open_sync()
            while (not device.is_open()):
                print('Opening...') 

        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:
        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-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']))
    enroll(sid, data)
    pass

@sio.event
def capture(sid, data):
    print(data)
    pass

@sio.event
def connect(sid, 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
def disconnect(sid):
    print({ 'sid': sid, 'event': 'disconnect'});
    pass

@sio.on('capture-other')
def capture_other(sid, data):
    print({ "event": "capture_other", "data": data })
    pass

with open('/tmp/libfprint-server', 'w') as f:
    f.write(str(os.getpid()));
    
if __name__ == '__main__':
    app.run(port=9000)