129 lines
5.1 KiB
Python
129 lines
5.1 KiB
Python
from flask import Flask, request, jsonify
|
|
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
|
|
import imutils
|
|
app = Flask(__name__)
|
|
#establish connection to minio for this instance of the api
|
|
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():
|
|
#save demo object to reserve id in minio
|
|
id = None
|
|
#five tries to generate unique id
|
|
for limited_try in range(0,5):
|
|
id = str(uuid.uuid4())
|
|
if check_id(id) == False:
|
|
break
|
|
demo_object = ['test']
|
|
#dump demo object and save it like its a face encoding to reserve id
|
|
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 jsonify({'id':id})
|
|
|
|
@app.route('/init_face')
|
|
#call like https://face.sguba.de/init_face?id=123&encoded_string=base64encoded_urlsafe_string
|
|
def new_face():
|
|
id = request.args.get('id', None)
|
|
img = request.args.get('encoded_string', None)
|
|
temporary_img_path = '/tmp/'+str(id)+'.jpg'
|
|
temporary_pkl_path = '/tmp/'+str(id)+'.pkl'
|
|
#transcode base64_string to img and generate encoding
|
|
save_img(img, temporary_img_path)
|
|
face_encoding_response = encode_face(temporary_img_path)
|
|
#check if encoding was successfull, if yes save encoding in minio
|
|
if face_encoding_response['success']==True:
|
|
with open(temporary_pkl_path, 'wb') as file:
|
|
pickle.dump(face_encoding_response['encoding'], file)
|
|
minioClient.fput_object('users', str(id), temporary_pkl_path)
|
|
return jsonify({'success':face_encoding_response['success']})
|
|
|
|
@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)
|
|
temporary_img_path = '/tmp/'+str(id)+'.jpg'
|
|
temporary_pkl_path = '/tmp/'+str(id)+'.pkl'
|
|
#transcode base64_string to img and generate encoding
|
|
save_img(img, temporary_img_path)
|
|
face_encoding_response = encode_face(temporary_img_path)
|
|
#check if encoding was successfull, if yes get initial encoding from minio and compare encodings
|
|
if face_encoding_response['success']==True:
|
|
minioClient.fget_object('users', str(id), temporary_pkl_path)
|
|
with open(temporary_pkl_path, 'rb') as file:
|
|
face_encoding = pickle.load(file)
|
|
face_encoding_response['result'] = str(face_recognition.compare_faces([face_encoding], face_encoding_response['encoding'])[0])
|
|
else:
|
|
face_encoding_response['result'] = str(False)
|
|
return jsonify({'result':face_encoding_response['result'],'success':face_encoding_response['success']})
|
|
|
|
|
|
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):
|
|
#length of base64 string is always a multiple of 4, if not its padded with =, here we add padding that might have been lost so there are no errors when decoding the img
|
|
encoded_data += "=" * ((4 - len(encoded_data) % 4) % 4)
|
|
#convert string to binary, decode base64, decode image and rotate it 90 degree because android studio rotates the camera, then save image to temp
|
|
encoded_data = bytes(encoded_data, encoding='utf-8')
|
|
nparr = np.fromstring(base64.urlsafe_b64decode(encoded_data), np.uint8)
|
|
img = cv2.imdecode(nparr, cv2.IMREAD_ANYCOLOR)
|
|
img = imutils.rotate(img, 90)
|
|
return cv2.imwrite(filename, img)
|
|
|
|
def encode_face(path):
|
|
#create face encoding from data, and check wether it was succesfull. If it was unsuccesfull the encodings list is empty
|
|
success = None
|
|
face = face_recognition.load_image_file(path)
|
|
face_encoding = face_recognition.face_encodings(face)
|
|
if len(face_encoding)==0:
|
|
response = {'success':False, 'encoding':None}
|
|
else:
|
|
response = {'success':True, 'encoding':face_encoding[0]}
|
|
return response
|
|
|
|
def setup():
|
|
#create minio buckets that are used at startup if they dont exist already
|
|
try:
|
|
minioClient.make_bucket("users")
|
|
except BucketAlreadyOwnedByYou as err:
|
|
pass
|
|
except BucketAlreadyExists as err:
|
|
pass
|
|
except ResponseError as err:
|
|
raise
|
|
|
|
#check if the script is run primarily and not imported etc.
|
|
if __name__ == '__main__':
|
|
setup()
|
|
#start running the api
|
|
app.run(host='0.0.0.0') |