import threading import socket import re import time import datetime import os import numpy as np import pandas as pd import faiss import cv2 from flask import Flask from flask import request from flask import render_template # live stream for channel 0, change to right url first live_stream_0 = "http://192.168.77.60/flv?app=live&stream=face_recognize_0" # user info, belongs to business database user_infos_df = pd.DataFrame(columns=["user_id", "name", "sex", "age", "birthplace", "register_photo"]) # user embeddings, belong to vector database user_embeddings_index = None # recognize history, belongs to business database recognize_history_df = pd.DataFrame(columns=["time", "rec_status", "rec_photo", "user_id", "name", "sex", "age", "birthplace", "register_photo", "top1_similiarity"]) recognize_history_df['time'] = pd.to_datetime(recognize_history_df['time']) # threshold for success of recognize recognize_similiarity_threshold = 18 ''' generate embedding for face image ''' # face embedding model keep as same as face_encoding_pipeline.cpp face_embedding_model_path = "../../../vp_data/models/face/face_recognition_sface_2021dec.onnx" net = cv2.dnn.readNetFromONNX(face_embedding_model_path) output_layer_names = net.getUnconnectedOutLayersNames() def generate_embedding(face_image): input_image = cv2.resize(face_image, (112, 112)) blob = cv2.dnn.blobFromImage(input_image) net.setInput(blob) out = net.forward(output_layer_names) embedding = np.array(out[0]) return embedding ''' simulate the action of user register, add record into vector database AND business database ''' def read_user_infos_from_disk(): user_infos_path = "./static/user_register_infos" for dirpath, dirnames, filenames in os.walk(user_infos_path): for filename in filenames: if filename.endswith('.png'): filename_no_ext = filename.split('.')[0] items = filename_no_ext.split('-') ''' item0: user_id item1: name item2: sex item3: age item4 birthplace ''' if len(items) == 5: file_path = os.path.join(dirpath, filename) face_image = cv2.imread(file_path) embedding = generate_embedding(face_image) # initialize vector database using specific dims of embedding global user_embeddings_index if user_embeddings_index == None: user_embeddings_index = faiss.IndexFlatIP(embedding.shape[1]) # 128 dims # add to vector database user_embeddings_index.add(embedding) # add to business database user_infos_df.loc[len(user_infos_df)] = [items[0], items[1], items[2], items[3], items[4], os.path.join(user_infos_path, filename)] print(f"new user register record: {items[0]} {items[1]} {items[2]} {items[3]} {items[4]} {os.path.join(user_infos_path, filename)}") ''' step0, search the most similar embedding from vecotr database, step1, search the corresponding user info from business database, step2, insert recognize history record into business database which would be rendered on web. ''' def recognize(embedding, path): embedding_list = [[e for e in embedding.split(',')]] embedding_np = np.array(embedding_list, dtype=np.float32) # step0, search top1 from vector database D, I = user_embeddings_index.search(embedding_np, 1) # step1, search user info from business database shoot_user_info_df = user_infos_df.iloc[I[0]] assert(len(shoot_user_info_df) == 1) # only 1 user returned # step2, attempt to insert history record into business database recognize_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())) user_id = shoot_user_info_df['user_id'].iloc[0] name = shoot_user_info_df['name'].iloc[0] sex = shoot_user_info_df['sex'].iloc[0] age = shoot_user_info_df['age'].iloc[0] birthplace = shoot_user_info_df['birthplace'].iloc[0] register_photo = shoot_user_info_df['register_photo'].iloc[0] recognize_photo = path recognize_similiarity = D[0][0] rec_status = "" # check if top1 similiarity satisfy threshold(define by your experience) if recognize_similiarity >= recognize_similiarity_threshold : rec_status = "passed" else: rec_status = "failed" recognize_history_df.loc[len(recognize_history_df)] = [recognize_time, rec_status, recognize_photo, user_id, name, sex, age, birthplace, register_photo, recognize_similiarity] ''' receive face embedding from socket (sent from VideoPipe) ''' def face_embedding_receiver(): udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) bind_ip = '192.168.77.68' bind_port = 5555 udp_socket.bind((bind_ip, bind_port)) while True: data, addr = udp_socket.recvfrom(1024 * 64) s = data.decode() print('############################################') print(f'Received data: \n{s}') print(f'From address: {addr}') li = re.findall("\<\-\-.*?\-\-\>", s, re.DOTALL) print(f"total {len(li)} embeddings parsed! going to recognize...") for l in li: items = str(l).split('\n') ''' items[0] <-- items[1] path items[2] embedding items[3] --> ''' assert(len(items) == 4) # recognize recognize(items[2], items[1]) # face embedding and path of cropped face image print(recognize_history_df) print('############################################') ''' create flask app ''' app = Flask(__name__) # home page @app.route("/", methods=["GET", "POST"]) def home(): ''' filter properties ''' select_rec_status = "all" if 'select_rec_status' in request.form: select_rec_status = request.form['select_rec_status'] # 1 hour by default now = time.time() past_1_hour = now - 3600 select_start_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(past_1_hour)) select_end_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(now)) if 'select_start_time' in request.form and request.form['select_start_time'] != '': select_start_time = datetime.datetime.strptime(request.form['select_start_time'], '%Y-%m-%dT%H:%M:%S') if 'select_end_time' in request.form and request.form['select_end_time'] != '': select_end_time = datetime.datetime.strptime(request.form['select_end_time'], '%Y-%m-%dT%H:%M:%S') sql = "rec_status != ''" if select_rec_status != 'all': sql += f" & rec_status == '{select_rec_status}'" sql += f" & time >= '{select_start_time}'" sql += f" & time < '{select_end_time}'" print(sql) filter_df = recognize_history_df.query(sql) recognize_results = [] for index, row in filter_df.iterrows(): recognize_results.append({'uid':index, \ 'time':row['time'], \ 'rec_status':row['rec_status'], \ 'rec_photo':row['rec_photo'], \ 'user_id':row['user_id'], \ 'name': row['name'], \ 'sex': row['sex'], \ 'age': row['age'], \ 'birthplace': row['birthplace'], \ 'register_photo': row['register_photo'], \ 'top1_similiarity': row['top1_similiarity']}) return render_template("home.html", \ rec_results = recognize_results, \ total_records = len(filter_df), \ select_rec_status = select_rec_status, \ select_start_time = select_start_time, \ select_end_time = select_end_time, \ live_stream_0 = live_stream_0) if __name__ == "__main__": # simulate the action of user register read_user_infos_from_disk() # start face embedding receiving thread receiver = threading.Thread(target = face_embedding_receiver) receiver.start() # run server app.run(host="192.168.77.68", port=5656)