Untitled
unknown
plain_text
9 months ago
59 kB
11
Indexable
import os
from sqlite3 import Time, Timestamp
import numpy as np
import cv2
from PIL import Image
from numpy import dot, sqrt
import re
from functools import wraps
from aiohttp import web
from aiohttp import client
import aiohttp
import asyncio
from aiohttp_swagger import *
import threading
import argparse
import datetime
import torch
import torch.utils.data as data
import torchvision.datasets as datasets
import torch.nn.functional as F
import torchvision.transforms as transforms
import hnswlib
import jinja2
import aiohttp_jinja2
import traceback
from app.backbone import Backbone
from app.vision.ssd.config.fd_config import define_img_size
from app.vision.ssd.mb_tiny_RFB_fd import create_Mb_Tiny_RFB_fd, create_Mb_Tiny_RFB_fd_predictor
from sqlalchemy import func, delete
from sqlalchemy.sql import text
from sqlalchemy.orm import aliased
from create_app import db_session, engine, Base, DefineImages, People, Timeline, Users, ChildrenPicker, PeopleClasses, Classes, PickUp, verify_pass
from pubsub import pub
import base64
import requests
import uuid
import trimesh
import glob
import platform
import time
import face_alignment
import time
from functools import wraps
from torchvision import transforms
from PIL import Image
'''
from app.deep3d.util.load_mats import load_lm3d
from app.deep3d.data.flist_dataset import default_flist_reader
from app.deep3d.options.test_options import TestOptions
from app.deep3d.data import create_dataset
from app.deep3d.models import create_model
from app.deep3d.util.visualizer import MyVisualizer
from app.deep3d.util.preprocess import align_img
from app.deep3d.util.util import save_landmark
'''
from deep3d.util.load_mats import load_lm3d
from deep3d.data.flist_dataset import default_flist_reader
from deep3d.options.test_options import TestOptions
from deep3d.data import create_dataset
from deep3d.models import create_model
from deep3d.util.visualizer import MyVisualizer
from deep3d.util.preprocess import align_img
from deep3d.util.util import save_landmark
from face_dream.dream import *
from mtcnn import MTCNN
from scipy.io import loadmat, savemat
app = web.Application(client_max_size=200*1024**2)
aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader('templates'))
device = 'cuda' if torch.cuda.is_available() else 'cpu'
class_names = [name.strip() for name in open('./app/detect_RFB_640/voc-model-labels.txt').readlines()]
candidate_size = 1000
threshold = 0.7
input_img_size = 640
define_img_size(input_img_size)
model_path = "./app/detect_RFB_640/version-RFB-640.pth"
net = create_Mb_Tiny_RFB_fd(len(class_names), is_test=True, device=device)
predictor = create_Mb_Tiny_RFB_fd_predictor(net, candidate_size=candidate_size, device=device)
net.load(model_path)
input_size=[112, 112]
transform = transforms.Compose(
[
transforms.Resize(
[int(128 * input_size[0] / 112), int(128 * input_size[0] / 112)],
), # smaller side resized
transforms.CenterCrop([input_size[0], input_size[1]]),
# transforms.Resize([input_size[0], input_size[1]]),
transforms.ToTensor(),
transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5]),
],
)
backbone = Backbone(input_size)
backbone.load_state_dict(torch.load('./app/ms1mv3_arcface_r50_fp16/backbone_ir50_ms1m_epoch120.pth', map_location=torch.device(device)))
backbone.to(device)
backbone.eval()
def cosine_similarity(x, y):
return dot(x, y) / (sqrt(dot(x, x)) * sqrt(dot(y, y)))
def no_accent_vietnamese(utf8_str):
INTAB = "ạảãàáâậầấẩẫăắằặẳẵóòọõỏôộổỗồốơờớợởỡéèẻẹẽêếềệểễúùụủũưựữửừứíìịỉĩýỳỷỵỹđẠẢÃÀÁÂẬẦẤẨẪĂẮẰẶẲẴÓÒỌÕỎÔỘỔỖỒỐƠỜỚỢỞỠÉÈẺẸẼÊẾỀỆỂỄÚÙỤỦŨƯỰỮỬỪỨÍÌỊỈĨÝỲỶỴỸĐ"
OUTTAB = "a" * 17 + "o" * 17 + "e" * 11 + "u" * 11 + "i" * 5 + "y" * 5 + "d" + \
"A" * 17 + "O" * 17 + "E" * 11 + "U" * 11 + "I" * 5 + "Y" * 5 + "D"
r = re.compile("|".join(INTAB))
replaces_dict = dict(zip(INTAB, OUTTAB))
return r.sub(lambda m: replaces_dict[m.group(0)], utf8_str)
def loadBase64Img(uri):
encoded_data = uri.split(',')[1]
nparr = np.fromstring(base64.b64decode(encoded_data), np.uint8)
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
return img
def load_image(img):
exact_image = False; base64_img = False; url_img = False
if type(img).__module__ == np.__name__:
exact_image = True
elif len(img) > 11 and img[0:11] == "data:image/":
base64_img = True
elif len(img) > 11 and img.startswith("http"):
url_img = True
#---------------------------
if base64_img == True:
img = loadBase64Img(img)
elif url_img:
img = np.array(Image.open(requests.get(img, stream=True).raw))
elif exact_image != True: #image path passed as input
if os.path.isfile(img) != True:
raise ValueError("Confirm that ",img," exists")
img = cv2.imread(img)
return img
# mask-detect
maskDetect_path = './app/face_mask/mobilenetv2_pt_2.pth'
maskDetectModel = torch.load(maskDetect_path)
maskDetectModel.eval()
def log_execution_time(file_path):
def decorator(func):
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
execution_time = end_time - start_time
# Ghi thời gian vào file
with open(file_path, "a") as log_file:
log_file.write(f"Function '{func.__name__}' executed in {execution_time:.6f} seconds.\n")
return result
return wrapper
return decorator
def mask_detect(img):
predictions = []
# image_transforms = transforms.Compose([transforms.Resize(size=(244,244)), transforms.ToTensor(),transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])])
image_transforms = transforms.Compose([
transforms.Resize(size=256),
transforms.CenterCrop(size=224),
transforms.ToTensor(),
transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])
img = Image.fromarray(img)
img = image_transforms(img)
img = img.unsqueeze(0)
img = img.to(device)
prediction = maskDetectModel(img)
prediction = prediction.argmax()
predictions.append(prediction.data)
prediction_data = prediction.item()
# print('Mask Detect')
# print(predictions)
if (prediction == 0):
print("Mask")
else:
print("No Mask")
return prediction_data
###############################################################################
# #deep3d
# #set up model
# opt = TestOptions().parse()
# device = torch.device(0)
# torch.cuda.set_device(device)
# model_deep3d = create_model(opt)
# model_deep3d.setup(opt)
# model_deep3d.device = device
# model_deep3d.parallelize()
# model_deep3d.eval()
# lm3d_std = load_lm3d('BFM')
# visualizer = MyVisualizer(opt)
# #setup detector
# # detector = MTCNN()
# fa = face_alignment.FaceAlignment(face_alignment.LandmarksType.TWO_D, device='cpu', face_detector='blazeface')
# #detect landmark function
# def detect(img):
# # global detector_mtcnn
# # detector_mtcnn = MTCNN()
# image = np.array(img)
# # print(detector)
# #detect keypoints
# preds = fa.get_landmarks_from_image(image)[0]
# #extract landmarks from keypoints
# left_eye_x = (preds[37][0] + preds[40][0])/2
# left_eye_y = (preds[37][1] + preds[40][1])/2
# right_eye_x = (preds[43][0] + preds[46][0])/2
# right_eye_y = (preds[43][1] + preds[46][1])/2
# nose_x = (preds[30][0] + preds[33][0])/2
# nose_y = (preds[30][1] + preds[33][1])/2
# mouth_left_x = preds[48][0]
# mouth_left_y = preds[48][1]
# mouth_right_x = preds[54][0]
# mouth_right_y = preds[54][1]
# #create numpy ndarray
# landmark = np.array([[left_eye_x, left_eye_y],
# [right_eye_x, right_eye_y],
# [nose_x, nose_y],
# [mouth_left_x, mouth_left_y],
# [mouth_right_x, mouth_right_y]],
# dtype='f')
# # return landmark, img
# return landmark
# #reconstruct 3d face function
# def reconstruct(im, lm):
# W,H = im.size
# lm = lm.reshape([-1, 2])
# lm[:, -1] = H - 1 - lm[:, -1]
# _, im, lm, _ = align_img(im, lm, lm3d_std)
# im_tensor = torch.tensor(np.array(im)/255., dtype=torch.float32).permute(2, 0, 1).unsqueeze(0)
# lm_tensor = torch.tensor(lm).unsqueeze(0)
# data = {
# 'imgs': im_tensor,
# 'lms': lm_tensor
# }
# model_deep3d.set_input(data) # unpack data from data loader
# model_deep3d.test() # run inference
# recon_shape, tri, recon_tex = model.export_mesh()
# return recon_shape, tri, recon_tex
# #rasterize 3d face
# def rasterize():
# visuals = model.get_current_visuals() # get image results
# result = visualizer.save_img(visuals)
# b,g,red = cv2.split(result)
# result = cv2.merge((red,g,b))
# return result
# #main api function
# def unmask(input_img):
# lm = detect(input_img)
# shape, tri, texture = reconstruct(input_img, lm)
# recon_img = rasterize()
# return recon_img
###############################################################################
### Trung addd
fa = face_alignment.FaceAlignment(face_alignment.LandmarksType.TWO_D, device='cpu', face_detector='blazeface')
opt = TestOptions().parse()
model_deep3d = create_model(opt)
model_deep3d.setup(opt)
model_deep3d.device = device
model_deep3d.parallelize()
model_deep3d.eval()
visualizer = MyVisualizer(opt)
lm3d_std = load_lm3d('./deep3d/BFM')
def detect_landmark(img):
# global detector_mtcnn
# detector_mtcnn = MTCNN()
image = np.array(img)
# print(detector)
#detect keypoints
preds = fa.get_landmarks_from_image(image)[0]
#extract landmarks from keypoints
left_eye_x = (preds[37][0] + preds[40][0])/2
left_eye_y = (preds[37][1] + preds[40][1])/2
right_eye_x = (preds[43][0] + preds[46][0])/2
right_eye_y = (preds[43][1] + preds[46][1])/2
nose_x = (preds[30][0] + preds[33][0])/2
nose_y = (preds[30][1] + preds[33][1])/2
mouth_left_x = preds[48][0]
mouth_left_y = preds[48][1]
mouth_right_x = preds[54][0]
mouth_right_y = preds[54][1]
#create numpy ndarray
landmark = np.array([[left_eye_x, left_eye_y],
[right_eye_x, right_eye_y],
[nose_x, nose_y],
[mouth_left_x, mouth_left_y],
[mouth_right_x, mouth_right_y]],
dtype='f')
# return landmark, img
return landmark
#reconstruct 3d face function
@log_execution_time("execution_times.txt")
def reconstruct(im, lm):
print ('Img size:', im.size)
W,H = im.size
lm = lm.reshape([-1, 2])
lm[:, -1] = H - 1 - lm[:, -1]
_, im, lm, _ = align_img(im, lm, lm3d_std)
im_tensor = torch.tensor(np.array(im)/255., dtype=torch.float32).permute(2, 0, 1).unsqueeze(0)
lm_tensor = torch.tensor(lm).unsqueeze(0)
data = {
'imgs': im_tensor,
'lms': lm_tensor
}
model_deep3d.set_input(data) # unpack data from data loader
model_deep3d.test() # run inference
recon_shape, tri, recon_tex = model_deep3d.export_mesh()
return recon_shape, tri, recon_tex
#rasterize 3d face
def rasterize():
visuals = model_deep3d.get_current_visuals() # get image results
result = visualizer.save_img(visuals)
b,g,red = cv2.split(result)
result = cv2.merge((red,g,b))
return result
#main api function
def unmask(input_img):
lm = detect_landmark(input_img)
shape, tri, texture = reconstruct(input_img, lm)
recon_img = rasterize()
return recon_img
### End Trung Add
### Viet-Bac Nguyen
def estimatePose(frame, landmarks):
"""Calculate poses"""
size = frame.shape #(height, width, color_channel)
image_points = np.array([
(landmarks[30][0], landmarks[30][1]), # Nose tip
(landmarks[8][0], landmarks[8][1]), # Chin
(landmarks[36][0], landmarks[36][1]), # Left eye left corner
(landmarks[45][0], landmarks[45][1]), # Right eye right corne
(landmarks[48][0], landmarks[48][1]), # Left Mouth corner
(landmarks[54][0], landmarks[54][1]) # Right mouth corner
], dtype="double")
model_points = np.array([
(0.0, 0.0, 0.0), # Nose tip
(0.0, -330.0, -65.0), # Chin
(-165.0, 170.0, -135.0), # Left eye left corner
(165.0, 170.0, -135.0), # Right eye right corne
(-150.0, -150.0, -125.0), # Left Mouth corner
(150.0, -150.0, -125.0) # Right mouth corner
])
# Camera internals
center = (size[1]/2, size[0]/2)
focal_length = center[0] / np.tan(60/2 * np.pi / 180)
camera_matrix = np.array(
[[focal_length, 0, center[0]],
[0, focal_length, center[1]],
[0, 0, 1]], dtype = "double"
)
dist_coeffs = np.zeros((4,1)) # Assuming no lens distortion
(success, rotation_vector, translation_vector) = cv2.solvePnP(model_points, image_points, camera_matrix, dist_coeffs)#, flags=cv2.CV_ITERATIVE)
axis = np.float32([[500,0,0],
[0,500,0],
[0,0,500]])
imgpts, jac = cv2.projectPoints(axis, rotation_vector, translation_vector, camera_matrix, dist_coeffs)
modelpts, jac2 = cv2.projectPoints(model_points, rotation_vector, translation_vector, camera_matrix, dist_coeffs)
rvec_matrix = cv2.Rodrigues(rotation_vector)[0]
proj_matrix = np.hstack((rvec_matrix, translation_vector))
eulerAngles = cv2.decomposeProjectionMatrix(proj_matrix)[6]
pitch, yaw, roll = [math.radians(_) for _ in eulerAngles]
pitch = math.degrees(math.asin(math.sin(pitch)))
roll = -math.degrees(math.asin(math.sin(roll)))
yaw = math.degrees(math.asin(math.sin(yaw)))
# return imgpts, modelpts, (str(int(roll)), str(int(pitch)), str(int(yaw))), (image_points[0][0], image_points[0][1]), image_points
return str(int(roll)), str(int(pitch)), str(int(yaw)), image_points
### End
class LoginApp():
def __init__(self):
pass
def login_required(self, f):
@wraps(f)
def decorated(*args, **kwargs):
user = None
try:
access_key = args[0].cookies['user_face_key']
except:
access_key = None
if not access_key:
user = None
else:
user = Users.query.filter_by(access_key=access_key).first()
if user:
user = user.__dict__
return f(*((user,) + args), **kwargs)
return decorated
def login_user(self, user, request):
access_key = uuid.uuid4()
user.access_key = str(access_key)
db_session.commit()
response = web.HTTPSeeOther(location='./')
response.cookies['user_face_key'] = user.access_key
return response
def logout_user(self, request):
response = web.HTTPSeeOther(location='./')
response.cookies['user_face_key'] = ''
return response
login_app = LoginApp()
def create_error_middleware(overrides):
@web.middleware
async def error_middleware(request, handler):
try:
Base.metadata.create_all(bind=engine)
resp = await handler(request)
db_session.remove()
return resp
except web.HTTPException as ex:
override = overrides.get(ex.status)
if override:
resp = await override(request, ex)
resp.set_status(ex.status)
return resp
except Exception as e:
print(traceback.format_exc())
resp = await overrides[500](request, web.HTTPError(text=traceback.format_exc()))
resp.set_status(500)
return resp
return error_middleware
async def handle_403(request, ex):
# return web.json_response({"result": {'message': ex.text}}, status=403)
raise web.HTTPFound(location='./')
async def handle_404(request, ex):
# return web.json_response({"result": {'message': ex.text}}, status=404)
raise web.HTTPFound(location='./')
async def handle_500(request, ex):
# return web.json_response({"result": {'message': ex.text}}, status=500)
raise web.HTTPFound(location='./')
def setup_middlewares(app):
error_middleware = create_error_middleware({
403: handle_403,
404: handle_404,
500: handle_500,
})
app.middlewares.append(error_middleware)
setup_middlewares(app)
@aiohttp_jinja2.template('index.html')
@login_app.login_required
async def index(current_user, request):
data = await request.post()
if 'login' in data and request.method == 'POST':
# read form data
username = data.get('username')
password = data.get('password')
# Locate user
user = Users.query.filter_by(username=username).first()
# Check the password
if user and verify_pass(password, user.password):
return login_app.login_user(user, request)
raise web.HTTPFound(location='./')
if 'register' in data and request.method == 'POST':
username = data.get('username')
password = data.get('password')
if username.strip() == "" or password.strip() == "":
raise web.HTTPFound(location='./')
# Check usename exists
user = Users.query.filter_by(username=username).first()
if user:
raise web.HTTPFound(location='./')
# else we can create the user
user = Users(username=username, password=password, secret_key="")
db_session.add(user)
db_session.commit()
p = hnswlib.Index(space = 'cosine', dim = 512)
p.init_index(max_elements = 1000, ef_construction = 200, M = 16)
p.set_ef(10)
p.set_num_threads(4)
p.save_index("indexes/index_" + str(user.secret_key) + ".bin")
return login_app.login_user(user, request)
if not current_user:
return {'is_login': False, 'current_user': None}
return {'is_login': True, 'current_user': current_user}
@aiohttp_jinja2.template('client.html')
@login_app.login_required
async def client_a(current_user, request):
return {'current_user':current_user}
async def logout(request):
return login_app.logout_user(request)
### Trung add dream here
model_dream = Branch(feat_dim=512)
# model.cuda()
checkpoint = torch.load('./face_dream/checkpoint_512.pth')
model_dream.load_state_dict(checkpoint['state_dict'])
model_dream.eval()
def dream_embedding(embedding_I, input_yaw):
yaw = np.zeros([1, 1])
yaw[0,0] = norm_angle(float(input_yaw))
original_embedding_tensor = np.expand_dims(embedding_I.detach().cpu().numpy(), axis=0)
# feat = torch.autograd.Variable(torch.from_numpy(feat.astype(np.float32)), volatile=True).cuda()
# yaw = torch.autograd.Variable(torch.from_numpy(yaw.astype(np.float32)), volatile=True).cuda()
feature_original = torch.autograd.Variable(torch.from_numpy(original_embedding_tensor.astype(np.float32)))
yaw = torch.autograd.Variable(torch.from_numpy(yaw.astype(np.float32)))
new_embedding = model_dream(feature_original, yaw)
new_embedding = new_embedding.cpu().data.numpy()
#new_embedding = new_embedding.to(device).data.numpy()
embedding_I = new_embedding[0, :]
return embedding_I
#### End of Trung add dream here
def get_embeddings(img_input, local_register = False):
# img = []
img = load_image(img_input)
print ('img size', img.size, type(img))
image = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
print ('image size', image.size, type(image))
## Trung Add mask phrase here
# mask_prediction = mask_detect(image)
# if not mask_prediction:
# tmp_image = Image.fromarray(np.uint8(image)).convert('RGB')
# image = unmask(tmp_image)
#cv2.imwrite("infer_img.jpg", image)
## End of Trung Add mask phrase here
### 8/8/2023 Viet-Bac Nguyen
### Estimate pose
# preds = fa.get_landmarks_from_image(image)[0]
# roll, pitch, yaw, _ = estimatePose(image, preds)
### End
if not local_register:
boxes, labels, probs = predictor.predict(image, candidate_size / 2, threshold)
boxes = boxes.detach().cpu().numpy()
else:
boxes = np.array([[0, 0, image.shape[1], image.shape[0]]])
feats = []
images = []
bboxes = []
masks = []
yaws = []
for i in range(boxes.shape[0]):
box = boxes[i, :]
xmin, ymin, xmax, ymax = box
xmin -= (xmax-xmin)/18
xmax += (xmax-xmin)/18
ymin -= (ymax-ymin)/18
ymax += (ymax-ymin)/18
xmin = 0 if xmin < 0 else xmin
ymin = 0 if ymin < 0 else ymin
xmax = image.shape[1] if xmax >= image.shape[1] else xmax
ymax = image.shape[0] if ymax >= image.shape[0] else ymax
boxes[i,:] = [xmin, ymin, xmax, ymax]
infer_img = image[int(ymin): int(ymax), int(xmin): int(xmax), :]
if infer_img is not None and infer_img.shape[0] != 0 and infer_img.shape[1] != 0:
with torch.no_grad():
mask_prediction = mask_detect(infer_img)
if not mask_prediction:
PIL_image = Image.fromarray(np.uint8(infer_img)).convert('RGB')
infer_img = unmask(PIL_image)
cv2.imwrite("infer_img.jpg", infer_img)
feat = F.normalize(backbone(transform(Image.fromarray(infer_img)).unsqueeze(0).to(device))).cpu()
preds = fa.get_landmarks_from_image(infer_img)[0]
roll, pitch, yaw, _ = estimatePose(infer_img, preds)
#Trung add dream here
ang = 3
if int(yaw) >= -ang and int(yaw) <= ang:
#Hieu chinh mat khi goc nghieng lon hon 5 do
feat = dream_embedding(feat, yaw)
##Trung add dream here
#feats.append(feat.detach().cpu().numpy())
yaws.append(yaw)
masks.append(not mask_prediction)
feats.append(feat)
images.append(infer_img.copy())
bboxes.append("{} {} {} {}".format(xmin, ymin, xmax, ymax))
return feats, images, bboxes, masks, yaws
#Duong-add
# Thêm hàm đo thời gian xử lý và lưu vào file
def measure_processing_time(func):
@wraps(func)
async def wrapper(*args, **kwargs):
start_time = time.time() # Bắt đầu đo thời gian
result = await func(*args, **kwargs) # Gọi hàm xử lý
end_time = time.time() # Kết thúc đo thời gian
processing_time = end_time - start_time
# Ghi thời gian xử lý vào file processing_time_log.txt
with open("processing_time_log.txt", "a") as log_file:
log_file.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - Processing time: {processing_time:.4f} seconds\n")
return result
return wrapper
# Áp dụng hàm đo thời gian vào một endpoint cần kiểm tra, ví dụ với /facerec
@measure_processing_time
async def facerec(request):
"""
---
description: This end-point allow to recognize face identity.
tags:
- Face Recognition
produces:
- text/json
responses:
"200":
description: successful operation
"400":
description: Vui lòng truyền secret key
"400":
description: Vui lòng truyền ảnh dưới dạng Base64
"403":
description: Secret key không hợp lệ
"""
req = await request.json()
if 'secret_key' not in req:
return web.json_response({"result": {'message': 'Vui lòng truyền secret key'}}, status=400)
user = Users.query.filter_by(secret_key=req['secret_key']).first()
if not user:
return web.json_response({"result": {'message': 'Secret key không hợp lệ'}}, status=403)
img_input = ""
if "img" in list(req.keys()):
img_input = req["img"]
validate_img = False
if len(img_input) > 11 and img_input[0:11] == "data:image/":
validate_img = True
if validate_img != True:
return web.json_response({"result": {'message': 'Vui lòng truyền ảnh dưới dạng Base64'}}, status=400)
feats, images, bboxes, masks, yaws = get_embeddings(img_input)
generated_face_ids = []
profile_face_ids = []
p = hnswlib.Index(space = 'cosine', dim = 512)
p.load_index("indexes/index_" + str(user.secret_key) + '.bin')
person_access_keys = []
identities = []
timelines = []
now = 0
for feat, image, mask, yaw in zip(feats, images, masks, yaws):
person_access_key = -1
try:
neighbors, distances = p.knn_query(feat, k=1)
if distances[0][0] <= 0.45:
person_access_key = db_session.query(DefineImages.person_access_key, func.count(DefineImages.person_access_key).label('total'))\
.filter(DefineImages.id.in_(neighbors.tolist()[0]))\
.filter(DefineImages.person_access_key==People.access_key)\
.filter(People.user_id==user.id)\
.group_by(DefineImages.person_access_key)\
.order_by(text('total DESC')).first().person_access_key
except:
person_access_key = -1
person = People.query.filter_by(access_key=person_access_key).first()
identities.append('Người lạ' if not person else person.name)
person_access_keys.append(person_access_key)
profile_image_id = DefineImages.query.filter_by(person_access_key=person_access_key).first()
# profile_image = cv2.imread("images/" + req['secret_key'] + '/' + profile_image_id.image_id + ".jpg")
profile_face_ids.append(profile_image_id.image_id if profile_image_id is not None else None)
generated_face_ids.append(None)
now = round(datetime.datetime.now().timestamp() * 1000)
extra = str(uuid.uuid4())
if not os.path.isdir("images/" + req['secret_key'] ):
os.mkdir("images/" + req['secret_key'] )
cv2.imwrite("images/" + req['secret_key'] + "/face_" + str(now) + '_' + extra + ".jpg", cv2.cvtColor(image, cv2.COLOR_RGB2BGR))
image = Timeline(user_id=user.id, person_access_key=person_access_key, image_id="face_" + str(now) + '_' + extra, embedding=np.array2string(feat, separator=','), timestamp=now, mask=mask, yaw=yaw)
db_session.add(image)
db_session.commit()
timelines.append(now)
pub.sendMessage('face_vkist', uid=req['secret_key'], message='facerec ' + str(now))
return web.json_response({'result': {"bboxes": bboxes, "identities": identities, "id": person_access_keys, "timelines": timelines, "profilefaceIDs": profile_face_ids, "3DFace": generated_face_ids, "masks": masks}}, status=200)
def validate_request(req, keys, values):
new_req = {}
for key in keys:
if not key in req:
new_req[key] = values[key]
else:
new_req[key] = req[key]
return new_req
async def facereg(request):
"""
---
description: This end-point allow to enroll face identity.
tags:
- Face Registation
produces:
- text/json
responses:
"200":
description: successful operation
"400":
description: Vui lòng truyền secret key
"400":
description: Vui lòng truyền ảnh dưới dạng Base64
"403":
description: Secret key không hợp lệ
"""
req = await request.json()
if 'secret_key' not in req:
return web.json_response({"result": {'message': 'Vui lòng truyền secret key'}}, status=400)
user = Users.query.filter_by(secret_key=req['secret_key']).first()
if not user:
return web.json_response({"result": {'message': 'Secret key không hợp lệ'}}, status=400)
feats, images, boxes = ([], [], [])
if "img" in list(req.keys()):
for img_input in req["img"]:
if not (len(img_input) > 11 and img_input[0:11] == "data:image/"):
return web.json_response({"result": {'message': 'Vui lòng truyền ảnh dưới dạng Base64'}}, status=400)
if 'local_register' in list(req.keys()):
feats_, images_, boxes_, masks_, _ = get_embeddings(img_input, True)
else:
feats_, images_, boxes_, masks_, _ = get_embeddings(img_input)
feats += feats_
images += images_
boxes += boxes_
if len(boxes) < 1 and 'access_key' not in req:
return web.json_response({"result": {'message': 'Không xác định khuôn mặt'}}, status=400)
if 'access_key' not in req or req['access_key']=="":
if 'img' not in req or 'type_role' not in req or 'name' not in req:
return web.json_response({"result": {'message': 'Vui lòng tryền đầy đủ dữ liệu'}}, status=400)
access_key = str(uuid.uuid4())
# NVB 13-1-2023
req = validate_request(req, ['name', 'age', 'type_role', 'class_access_key', 'gender', 'phone', 'secret_key'], {'name': None, 'age': None, 'type_role': None, 'class_access_key': None, 'gender': None, 'phone': None, 'secret_key': None})
person = People(user_id=user.id, name=req['name'], age=req['age'], type_role=req['type_role'], gender=req['gender'], phone=req['phone'], access_key=access_key)
db_session.add(person)
db_session.commit()
pc = PeopleClasses.query.filter_by(person_access_key=person.access_key).first()
if not pc:
if not req['class_access_key'] is None and req['type_role'] != 'parent':
pc = PeopleClasses(class_access_key = req['class_access_key'], person_access_key = person.access_key)
db_session.add(pc)
db_session.commit()
else:
access_key = req['access_key']
person = People.query.filter_by(access_key=access_key, user_id=user.id).first()
# req = validate_request(req, ['name', 'age', 'type_role', 'gender', 'phone'], person._asdict())
req = validate_request(req, ['name', 'age', 'type_role', 'class_access_key', 'gender', 'phone', 'secret_key'], person.__dict__)
person.name=req['name']
person.age=req['age']
person.type_role=req['type_role']
person.gender=req['gender']
person.phone=req['phone']
db_session.commit()
pc = PeopleClasses.query.filter_by(person_access_key=person.access_key).first()
if not pc:
if not req['class_access_key'] is None and req['type_role'] != 'parent':
pc = PeopleClasses(class_access_key = req['class_access_key'], person_access_key = person.access_key)
db_session.add(pc)
db_session.commit()
else:
pc.class_access_key = req['class_access_key']
db_session.commit()
person = People.query.filter_by(access_key=access_key).first()
if 'applicant_access_keys' in req and not req['applicant_access_keys'] is None and req['type_role'] !="teacher":
for ak in req['applicant_access_keys']:
if req['type_role'] == 'student':
parent = People.query.filter_by(access_key=ak, type_role="parent", user_id=user.id).first()
if parent:
pc = ChildrenPicker.query.filter_by(child_access_key=person.access_key, picker_access_key=parent.id).first()
if not pc:
pc = ChildrenPicker(child_access_key=person.access_key, picker_access_key=parent.id)
db_session.add(pc)
db_session.commit()
else:
child = People.query.filter_by(access_key=ak, type_role="student", user_id=user.id).first()
pc = ChildrenPicker.query.filter_by(child_access_key=child.id, picker_access_key=person.access_key).first()
if child:
if not pc:
pc = ChildrenPicker(child_access_key=child.id, picker_access_key=person.access_key)
db_session.add(pc)
db_session.commit()
p = hnswlib.Index(space = 'cosine', dim = 512)
p.load_index("indexes/index_" + str(user.secret_key) + '.bin', max_elements=1000)
now = 0
for feat, image in zip(feats, images):
now = round(datetime.datetime.now().timestamp() * 1000)
if not os.path.isdir("images/" + req['secret_key'] ):
os.mkdir("images/" + req['secret_key'] )
extra = str(uuid.uuid4())
cv2.imwrite("images/" + req['secret_key'] + "/face_" + str(now) + '_' + extra + ".jpg", cv2.cvtColor(image, cv2.COLOR_RGB2BGR))
image = Timeline(user_id=user.id, person_access_key=person.access_key, image_id="face_" + str(now) + '_' + extra, embedding=np.array2string(feat, separator=','), timestamp=now)
db_session.add(image)
db_session.commit()
define_image = DefineImages(person_access_key=person.access_key, image_id=image.image_id)
db_session.add(define_image)
db_session.commit()
define_image = DefineImages.query.filter_by(image_id=image.image_id).first()
p.add_items(feat, np.array([define_image.id]))
p.save_index("indexes/index_" + str(user.secret_key) + '.bin')
now = datetime.datetime.now().timestamp() * 1000
pub.sendMessage('face_vkist', uid=user.secret_key, message='facereg ' + str(now))
return web.json_response({"result": {'message': 'success'}}, status=200)
@login_app.login_required
async def delete_image(current_user, request):
"""
---
description: This end-point allow to delete face image.
tags:
- Face Image Deleting
produces:
- text/json
responses:
"200":
description: successfull operation
"400":
description: Vui lòng truyền access key
"403":
description: Bạn không có quyền xóa
"""
req = await request.json()
if 'access_key' not in req:
return web.json_response({"result": {'message': 'Vui lòng truyền access key'}}, status=400)
access_key = req['access_key']
person = People.query.filter_by(access_key=access_key, user_id=current_user['id']).first()
if not person:
return web.json_response({"result": {'message': 'Bạn không có quyền xóa'}}, status=403)
sql1 = delete(DefineImages).where(DefineImages.person_access_key == person.access_key)
db_session.execute(sql1)
db_session.commit()
sql2 = delete(ChildrenPicker).where(ChildrenPicker.child_access_key == person.access_key)
db_session.execute(sql2)
db_session.commit()
sql3 = delete(ChildrenPicker).where(ChildrenPicker.picker_access_key == person.access_key)
db_session.execute(sql3)
db_session.commit()
sql4 = delete(PeopleClasses).where(PeopleClasses.person_access_key == person.access_key)
db_session.execute(sql4)
db_session.commit()
sql5 = delete(PickUp).where(PickUp.child_access_key == person.access_key)
db_session.execute(sql5)
db_session.commit()
sql6 = delete(PickUp).where(PickUp.picker_access_key == person.access_key)
db_session.execute(sql6)
db_session.commit()
db_session.delete(person)
db_session.commit()
p = hnswlib.Index(space = 'cosine', dim = 512)
p.init_index(max_elements = 1000, ef_construction = 200, M = 16)
p.set_ef(10)
p.set_num_threads(4)
print('\n\n\nRebuild hnswlib\n\n\n')
# Rebuild index
remainImages = db_session.query(People.id, DefineImages.id, Timeline.embedding) \
.filter(People.user_id==current_user['id']) \
.filter(DefineImages.person_access_key==People.access_key) \
.filter(Timeline.image_id==DefineImages.image_id) \
.all()
for imageI in remainImages:
embedding = imageI[2]
embedding = embedding[2:-2]
embedding = np.expand_dims(np.fromstring(embedding, dtype='float32', sep=','), axis=0)
p.add_items(embedding, np.array([imageI[1]]))
p.save_index("indexes/index_" + str(current_user['secret_key']) + ".bin")
return web.json_response({"result" : {'mesage' : "Ok"}}, status=200)
@login_app.login_required
async def add_class(current_user, request):
"""
---
description: This end-point allow to add class.
tags:
- Class Adding
produces:
- text/json
responses:
"200":
description: successfull operation
"""
req = await request.json()
if 'class_access_key' not in req or req['class_access_key'] == "":
access_key = str(uuid.uuid4())
new_class = Classes(name=req['class_name'], user_id=current_user['id'], access_key=access_key)
db_session.add(new_class)
db_session.commit()
else:
new_class = Classes.query.filter_by(access_key=req['class_access_key'], user_id=current_user['id']).first()
if 'class_name' in req and req['class_name'] != "":
new_class.name = req['class_name']
db_session.commit()
if 'teacher_access_keys' in req and not req['teacher_access_keys'] is None:
for ak in req['teacher_access_keys']:
person = People.query.filter_by(access_key=ak, user_id=current_user['id'], type_role="teacher").first()
if person:
pc = PeopleClasses.query.filter_by(person_access_key=person.access_key).first()
if not pc:
pc = PeopleClasses(class_access_key = new_class.access_key, person_access_key = person.access_key)
db_session.add(pc)
db_session.commit()
return web.json_response({"result": {'message': 'success'}}, status=200)
@login_app.login_required
async def delete_class(current_user, request):
"""
---
description: This end-point allow to delete class.
tags:
- Class Deleting
produces:
- text/json
responses:
"200":
description: successfull operation
"403":
description: Bạn không có quyền xóa
"""
req = await request.json()
class_access_key = req['class_access_key']
target_class = Classes.query.filter_by(access_key=class_access_key, user_id=current_user['id']).first()
if not target_class:
return web.json_response({"result": {'message': 'Bạn không có quyền xóa lớp này'}}, status=403)
db_session.delete(target_class)
db_session.commit()
sql1 = delete(PeopleClasses).where(PeopleClasses.class_access_key == class_access_key)
db_session.execute(sql1)
db_session.commit()
return web.json_response({"result": {'message': 'success'}}, status=200)
async def check_pickup(request):
"""
---
description: This end-point allow to check pickup between parent and children.
tags:
- Pickup Checking
produces:
- text/json
responses:
"200":
description: successfull operation
"400":
description: Vui lòng truyền secret key
"403":
description: Secret key không hợp lệ
"""
req = await request.json()
if 'secret_key' not in req:
return web.json_response({"result": {'message': 'Vui lòng truyền secret key'}}, status=400)
user = Users.query.filter_by(secret_key=req['secret_key']).first()
if not user:
return web.json_response({"result": {'message': 'Secret key không hợp lệ'}}, status=403)
if 'id1' not in req or 'id2' not in req or 'timeline_id1' not in req or 'timeline_id2' not in req:
return web.json_response({"result": {'message': 'Vui lòng tryền đầy đủ dữ liệu'}}, status=400)
person1 = None
person2 = None
if int(req['id1']) != -1:
person1 = People.query.filter_by(access_key=req['id1'], user_id=user.id).first()
if not person1:
return web.json_response({"result": {'message': 'Bạn không có quyền thay đổi'}}, status=403)
if int(req['id2']) != -1:
person2 = People.query.filter_by(access_key=req['id2'], user_id=user.id).first()
if not person2:
return web.json_response({"result": {'message': 'Bạn không có quyền thay đổi'}}, status=403)
if not person1 or person1.type_role == "parent" or person1.type_role == "teacher":
pk = PickUp(child_access_key=int(req['id2']), picker_access_key=int(req['id1']), child_timeline=int(req['timeline_id2']), parent_timeline=int(req['timeline_id1']))
db_session.add(pk)
db_session.commit()
else:
pk = PickUp(child_access_key=int(req['id1']), picker_access_key=int(req['id2']), child_timeline=int(req['timeline_id1']), parent_timeline=int(req['timeline_id2']))
db_session.add(pk)
db_session.commit()
return web.json_response({"result": {'message': 'success'}}, status=200)
@login_app.login_required
async def get_pickup(current_user, request): #them page
"""
---
description: This end-point allow to get all pickup between parent and children.
tags:
- Pickup Listing
produces:
- text/json
responses:
"200":
description: successfull operation
"""
args = request.rel_url.query
page = request.match_info.get('page','1')
page = int(page)
if page <= 0:
page = 1
page_size = 10
args = validate_request(args, ['name'], {'name': ''})
People_alias = aliased(People)
Timeline_alias = aliased(Timeline)
all_pickups_available = db_session.query(PickUp.id, Timeline.timestamp, People.name, People_alias.name, Timeline.image_id, Timeline_alias.image_id)\
.filter(Timeline.user_id==current_user['id'])\
.filter(Timeline_alias.user_id==current_user['id'])\
.filter(PickUp.child_access_key==People.access_key)\
.filter(PickUp.picker_access_key==People_alias.id)\
.filter(PickUp.child_timeline==Timeline.id)\
.filter(PickUp.picker_timeline==Timeline_alias.id)\
.filter(func.lower(People.name).like('%' + args['name'].lower() + '%') | func.lower(People_alias.name).like('%' + args['name'].lower() +'%'))\
.order_by(Timeline.timestamp.desc())\
all_pickups = all_pickups_available.limit(page_size).offset((page - 1) * page_size).all()
safe_pickups = all_pickups_available.filter(PickUp.child_access_key==ChildrenPicker.child_access_key & PickUp.picker_access_key==ChildrenPicker.picker_access_key).all()
pickup_array = {}
for u in all_pickups:
pickup_array[str(u[0])] = {'timestamp': u[1], 'child_name': u[2], 'parent_name': u[3], 'child_image_id': u[4], 'parent_image_id': u[5], 'is_acceptable': False}
for u in safe_pickups:
if str(u[0]) in pickup_array:
pickup_array[str(u[0])]['is_acceptable'] = True
pickup_array_list = [pickup_array[u] for u in pickup_array.keys()]
number_of_pickup = len(all_pickups_available.all())
return web.json_response({
"result": {
"number_of_pickup": number_of_pickup,
"pickup_list": pickup_array_list,
}
}, status=200)
@login_app.login_required
async def get_class(current_user, request): #them page
"""
---
description: This end-point allow to check get all class of yours.
tags:
- Class Listing
produces:
- text/json
responses:
"200":
description: successfull operation
"""
args = request.rel_url.query
page = request.match_info.get('page','1')
page = int(page)
if page <= 0:
page = 1
page_size = 10
args = validate_request(args, ['name', 'class_access_key'], {'name': '', 'class_access_key': '%%'})
all_classes_count = db_session.query(Classes.access_key, Classes.name)\
.filter(Classes.user_id==current_user['id'])\
.filter(func.lower(Classes.name).like('%'+args['name'].lower()+'%') & Classes.access_key.like(args['class_access_key']))
all_classes_available = all_classes_count.limit(page_size).offset((page - 1) * page_size).all()
all_classes = db_session.query(Classes.access_key, Classes.name, func.count(Classes.access_key))\
.filter(Classes.user_id==current_user['id'])\
.filter(Classes.access_key==PeopleClasses.class_access_key)\
.filter(People.access_key==PeopleClasses.person_access_key)\
.filter(People.type_role=='student')\
.group_by(Classes.access_key)\
.all()
all_classes_teacher = db_session.query(Classes.access_key, Classes.name, People.name, DefineImages.image_id)\
.filter(Classes.user_id==current_user['id'])\
.filter(Classes.access_key==PeopleClasses.class_access_key)\
.filter(People.access_key==PeopleClasses.person_access_key)\
.filter(People.type_role=='teacher')\
.filter(DefineImages.person_access_key==People.access_key)\
.group_by(Classes.access_key)\
.all()
class_array = {}
for u in all_classes_available:
class_array[str(u[0])] = {'access_key': u[0], 'classname': u[1], 'number_of_student': 0}
for u in all_classes:
if str(u[0]) in class_array:
class_array[str(u[0])] = {'access_key': u[0], 'classname': u[1], 'number_of_student': u[2]}
for u in all_classes_teacher:
if str(u[0]) in class_array:
class_array[str(u[0])]['teachers'] = {'name': u[2], 'image_id': u[3]}
class_array_list = [class_array[u] for u in class_array.keys()]
number_of_class = len(all_classes_count.all())
return web.json_response({
"result": {
"number_of_class": number_of_class,
"class_list": class_array_list,
}
}, status=200)
@login_app.login_required
async def people_list(current_user, request): #tham page
"""
---
description: This end-point allow to check get all people of yours.
tags:
- People Listing
produces:
- text/json
responses:
"200":
description: successfull operation
"""
args = request.rel_url.query
page = request.match_info.get('page','1')
page = int(page)
if page <= 0:
page = 1
page_size = 10
today = datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0).timestamp() * 1000
args = validate_request(args, ['name', 'type_role', 'class_name', 'access_key', 'begin', 'end'], {'name': "%{}%".format(''), 'type_role': "%{}%".format(''), 'access_key': "%{}%".format(''), 'class_name': "%{}%".format(''), 'begin': today, 'end': today + 86400000})
sub = db_session.query(DefineImages.image_id, DefineImages.id.label('id')).subquery()
all_classes = Classes.query.filter_by(user_id = current_user['id']).all()
number_class = len(all_classes)
all_people_count = db_session.query(DefineImages.person_access_key, DefineImages.image_id, People.name, People.access_key, People.type_role, People.age, People.gender, People.phone)\
.filter(People.user_id==current_user['id'])\
.filter(People.access_key==DefineImages.person_access_key)\
.filter(func.lower(People.name).like("%" + args['name'].lower() + "%") & People.type_role.like(args['type_role']) & People.access_key.like(args['access_key']))
if number_class > 0:
all_people_count = all_people_count.filter(Classes.name.like(args['class_name']))\
all_people_count = all_people_count.group_by(DefineImages.person_access_key, People.name, People.access_key)\
.filter(DefineImages.id==sub.c.id)\
# .filter(func.lower(People.name).like('%' + args['name'].lower() + '%') & People.type_role.like('%' + args['type_role'] + '%') & Classes.name.like('%' + args['class_name'] + '%'))\
all_people_have_class = db_session.query(DefineImages.person_access_key, DefineImages.image_id, People.name, People.access_key, People.type_role, Classes.name, Classes.access_key)\
.filter(People.user_id==current_user['id'])\
.filter(People.access_key==DefineImages.person_access_key)\
.filter(People.access_key==PeopleClasses.person_access_key)\
.filter(Classes.access_key==PeopleClasses.class_access_key)\
.all()
# .filter(People.access_key==PeopleClasses.person_access_key)\
# .filter(func.lower(People.name).like(args['name'].lower()) & People.type_role.like(args['type_role']) & Classes.name.like(args['class_name']))\
# .group_by(DefineImages.person_access_key, People.name, People.access_key)\
# .filter(DefineImages.id==sub.c.id)\
current_checkin = db_session.query(Timeline.person_access_key, func.min(Timeline.timestamp), func.max(Timeline.timestamp), Timeline.image_id, People.name)\
.filter(Timeline.user_id==current_user['id'])\
.filter(People.access_key==Timeline.person_access_key)\
.filter(Timeline.timestamp >= int(args['begin']))\
.filter(Timeline.timestamp <= int(args['end']))\
.group_by(Timeline.person_access_key, People.name)\
.all()
all_people = all_people_count.limit(page_size).offset((page - 1) * page_size).all()
if not current_checkin:
current_checkin = []
people_array = {}
for u in all_people:
people_array[str(u[0])] = {
'name': u[2],
'image_ids': u[1],
'begin': '--',
'end': '--',
'checkin': False,
'access_key': u[3],
'type_role': u[4],
'class_name': None,
'class_access_key': None,
'age': u[5],
'gender': u[6],
'phone': u[7]
}
for u in current_checkin:
if str(u[0]) in people_array:
people_array[str(u[0])] = {
'name': u[4],
'image_ids': u[3],
'begin': str(u[1]),
'end': str(u[2]),
'checkin': True,
'access_key': people_array[str(u[0])]['access_key'],
'type_role': people_array[str(u[0])]['type_role'],
'class_name': people_array[str(u[0])]['class_name'],
'class_access_key': people_array[str(u[0])]['class_access_key'],
'age': people_array[str(u[0])]['age'],
'gender': people_array[str(u[0])]['gender'],
'phone': people_array[str(u[0])]['phone'],
}
for u in all_people_have_class:
if str(u[0]) in people_array:
people_array[str(u[0])]['class_name'] = u[5]
people_array[str(u[0])]['class_access_key'] = u[6]
number_of_current_checkin = len(current_checkin)
number_of_people = len(all_people_count.all())
current_checkin = [people_array[u] for u in people_array.keys()]
return web.json_response({
"result": {
"people_list": current_checkin,
'number_of_people': number_of_people,
'number_of_current_checkin': number_of_current_checkin,
}
}, status = 200)
@login_app.login_required
async def data_a(current_user, request):
"""
---
description: This end-point allow to check get all data of yours.
tags:
- Data Profiling
produces:
- text/json
responses:
"200":
description: successfull operation
"""
today = datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0).timestamp() * 1000
all_people = db_session.query(DefineImages.person_access_key, DefineImages.image_id, People.name, People.access_key)\
.filter(People.user_id==current_user['id'])\
.filter(People.access_key==DefineImages.person_access_key)\
.group_by(DefineImages.person_access_key, People.name, People.access_key)\
.all()
current_checkin = db_session.query(Timeline.person_access_key, Timeline.timestamp, Timeline.image_id, Timeline.mask, Timeline.yaw, People.name)\
.filter(Timeline.user_id==current_user['id'])\
.filter(People.access_key==Timeline.person_access_key)\
.filter(Timeline.timestamp >= today)\
.group_by(Timeline.person_access_key, People.name)\
.all()
current_timeline = db_session.query(Timeline.person_access_key, Timeline.timestamp, Timeline.image_id, Timeline.mask, Timeline.yaw, People.name)\
.filter(Timeline.user_id==current_user['id'])\
.filter(People.access_key==Timeline.person_access_key)\
.order_by(Timeline.timestamp.desc())\
.limit(10)\
.all()
strangers = db_session.query(Timeline.person_access_key, Timeline.timestamp, Timeline.image_id, Timeline.mask, Timeline.yaw)\
.filter(Timeline.user_id==current_user['id'])\
.filter(Timeline.person_access_key==-1)\
.order_by(Timeline.timestamp.desc())\
.limit(10)\
.all()
if not current_checkin:
current_checkin = []
number_of_current_checkin = len(current_checkin)
number_of_people = len(all_people)
current_timeline_ = [{'name': u[5], 'image_id': u[2], 'timestamp': str(u[1]), 'mask': u[3], 'yaw': u[4]} for u in current_timeline]
strangers = [{'image_id': u[2], 'timestamp': str(u[1]), 'mask': u[3], 'yaw': u[4]} for u in strangers]
t = 0
r = 0
a = 0
if torch.cuda.is_available():
t = torch.cuda.get_device_properties(0).total_memory
r = torch.cuda.memory_reserved(0)
a = torch.cuda.memory_allocated(0)
return web.json_response({
"result": {
'secret_key': current_user['secret_key'],
'number_of_people': number_of_people,
'number_of_current_checkin': number_of_current_checkin,
'current_timeline': current_timeline_,
'strangers': strangers,
'gpu': {
't': t,
'r': r,
'a': a
}
}
}, status = 200)
async def images(request): #them secretkey, image_id
"""
---
description: This end-point allow to check get image.
tags:
- Image
produces:
- text/json
responses:
"200":
description: successfull operation
"""
secret_key = request.match_info.get('secret_key','')
image_id = request.match_info.get('image_id', '')
return web.FileResponse('images/' + secret_key + "/" + image_id + '.jpg')
app.router.add_route('*','/', index, name="index")
app.router.add_route('GET','/client', client_a)
app.router.add_route('GET','/logout', logout)
app.router.add_route('POST',"/facerec", facerec)
app.router.add_route('POST','/facereg', facereg)
app.router.add_route('POST','/delete_image', delete_image)
app.router.add_route('POST',"/add_class", add_class)
app.router.add_route('POST',"/delete_class", delete_class)
app.router.add_route('POST',"/check_pickup", check_pickup)
app.router.add_route('GET',"/pickup_list/{page}", get_pickup)
app.router.add_route('GET',"/class_list/{page}", get_class)
app.router.add_route('GET',"/people_list/{page}", people_list)
app.router.add_route('GET',"/data", data_a)
app.router.add_route('GET',"/images/{secret_key}/{image_id}", images)
app.add_routes([web.static('/static', 'static')])
setup_swagger(app, swagger_url="./api/v1/doc", ui_version=3)
if __name__ == "__main__":
# img_input='/home/coder/face_mask/mask.jpg'
# input_img = Image.open(img_input)
# print ('input_img size', input_img.size, type(input_img))
# feats, images, bboxes, masks = get_embeddings(img_input, local_register = False)
# print(feats)
#output_img = unmask(input_img)
#cv2.imwrite("output.jpg", output_img)
# web.run_app(app, port=5002)
web.run_app(app, host='0.0.0.0', port=5002)
# image = np.array(input_img)
# # print(detector)
# #detect keypoints
# preds = fa.get_landmarks_from_image(image)[0]
# roll, pitch, yaw, _ = estimatePose(image, preds)
# print('yaw:', yaw)Editor is loading...
Leave a Comment