face-recognition-app/face_recognition/app/app.py

from flask import Flask, request
from minio import Minio
from minio.error import (ResponseError, BucketAlreadyOwnedByYou, BucketAlreadyExists)
import os
import uuid
import pickle
import face_recognition
from werkzeug.datastructures import ImmutableMultiDict
import base64
import cv2
import numpy as np
app = Flask(__name__)
minioClient = Minio('minio:9000', access_key=os.environ['s3-name'], secret_key=os.environ['s3-password'], secure=False)
 
@app.route('/')
def hello_docker():
    return 'Hello, I run in a docker container'

@app.route('/acces_minio')
def access_minio():
    for bucket in minioClient.list_buckets():
        print(bucket.name, bucket.creation_date, flush=True)
    return('Connection Succesfull')
  
@app.route('/new_user_id')
def new_user_id():
#demo objekt  speichern um id zu 'speichern'. Später mit if (type(object)==list oä überschreiben mit face_encoding.pkl
    id = None
    for limited_try in range(0,5):
        id = str(uuid.uuid4())
        if check_id(id) == False:
            break
    demo_object = ['test']
    with open('/tmp/demo_object.pkl', 'wb') as f:
        pickle.dump(demo_object, f)
    minioClient.fput_object('users', str(id), '/tmp/demo_object.pkl')
    return(id)
  
@app.route('/init_face')
#call like https://face.sguba.de/init_face?id=123&encoded_string=abc
def new_face():
    id  = request.args.get('id', None)
    img = request.args.get('encoded_string', None)
    img += "=" * ((4 - len(img) % 4) % 4)
    img = bytes(img, encoding='utf-8')
    #print(img, flush=True)
    #imgdata = base64.b64decode(img)
    #with open('/tmp/'+str(id), 'wb') as file:
    #    file.write(bytearray(imgdata))
    save_img(img, '/tmp/'+str(id)+'.jpg')
    #to debug send img
    #minioClient.fput_object('users', str(id)+'.jpg', '/tmp/'+str(id)+'.jpg')
    #face = face_recognition.load_image_file('/tmp/'+str(id)+'.jpg')
    #face_encoding = face_recognition.face_encodings(face)[0]
    #with open('/tmp/'+str(id)+'.pkl', 'wb') as file:
    #    pickle.dump(face_encoding, file)
    #minioClient.fput_object('users', str(id), '/tmp/'+str(id)+'.pkl')
    return str(id)
  
@app.route('/check_face')
#call like https://face.sguba.de/check_face?id=123&encoded_string=abc
def check_face():
    id  = request.args.get('id', None)
    img = request.args.get('encoded_string', None)
    imgdata = base64.b64decode(img)
    with open('/tmp/'+str(id), 'wb') as file:
        file.write(imgdata)
    face = face_recognition.load_image_file('/tmp/'+str(id))
    face_encoding = face_recognition.face_encodings(face)[0]
    minioClient.fget_object('users', str(id), '/tmp/'+str(id))
    with open('/tmp/'+str(id), 'rb') as file:
        face2_encoding = pickle.load(file)
    return str(face_recognition.compare_faces([face_encoding], face2_encoding))
    
                
def check_id(id):
    #return True -> id bereits verwendet
    #return False -> id noch nicht verwendet
    users = minioClient.list_objects('users')
    known = False
    for user in users:
        if id == user.object_name:
            known = True
        else:
            pass
    return known

def save_img(encoded_data, filename):
    nparr = np.fromstring(base64.b64decode(encoded_data), np.uint8)
    img = cv2.imdecode(nparr, cv2.IMREAD_ANYCOLOR)
    return cv2.imwrite(filename, img)
          
def setup():
    try:
        minioClient.make_bucket("users")
    except BucketAlreadyOwnedByYou as err:
        pass
    except BucketAlreadyExists as err:
        pass
    except ResponseError as err:
        raise
    try:
        minioClient.make_bucket("uploads")
    except BucketAlreadyOwnedByYou as err:
        pass
    except BucketAlreadyExists as err:
        pass
    except ResponseError as err:
        raise
  
 
if __name__ == '__main__':
    setup()
    app.run(host='0.0.0.0')