import requests import time import threading import cv2 as cv import numpy as np class DoorWorker(): def __init__(self): self.door_status = False def open_door(self): if not self.door_status: self.door_status = True print("OPEN DOOR") #GPIO.setmode(GPIO.BOARD) #GPIO.setup(16, GPIO.OUT) #GPIO.output(16, 1) time.sleep(5) #GPIO.output(16, 0) white_list.clear() permit_list.clear() self.door_status = False class HttpWorker(): def __init__(self): self.last_answer = None self.search_filter = threading.Lock() def search_face(self, position, img_data): return if not self.search_filter.locked(): with self.search_filter: URL = ADDRESS + "/find_by_img" headers = { 'accept': 'application/json' } files = { 'img_data': ('photo.jpg', img_data["image"], 'image/jpeg') } answer = requests.post(URL, headers=headers, files=files).json() if answer["result"]: if answer["result"] in white_list: white_list[answer["result"]] = \ {"position": position, "counter": white_list[answer["result"]]["counter"] + 1} else: white_list[answer["result"]] = \ {"position": position, "counter": 1} print(white_list) self.last_answer = answer time.sleep(0.1) class RecognitionWorker(): def __init__(self, cap): self.face_cascade = cv.CascadeClassifier(cv.data.haarcascades + 'haarcascade_frontalface_default.xml') def check_frame(self, frame): gray = cv.cvtColor(frame, cv.COLOR_BGR2GRAY) faces = self.face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5) return faces class VisualizeWorker(): def drawer(self, image, faces, door_status, permit_list, fps=None): output = image.copy() if fps: cv.putText(output, 'FPS: {:.2f}'.format(fps), (0, 15), cv.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0)) white_position = list() if permit_list: for person in permit_list: try: white_position.append(white_list[person]["position"]) except: pass crop_img = list() for position, face in enumerate(faces): coords = faces[position].astype(np.int32) for i in range(len(coords)): if coords[i] < 0: coords[i] = 0 if door_status and position in white_position: color = (0, 255, 0) else: color = (0, 0, 255) cv.rectangle(output, (coords[0], coords[1]), (coords[0] + coords[2], coords[1] + coords[3]), color, 2) cv.rectangle(output, (coords[0], coords[1]), (coords[0] + coords[2] + 10, coords[1] + coords[3] + 40), (0, 255, 0), 2) crop_img.append(output[coords[1]:coords[1] + coords[3]+40, coords[0]:coords[0] + coords[2]+10]) return output, crop_img def resize(img): height_size = 640 print(height_size, img.shape[0]) scale = height_size / img.shape[0] width = int(img.shape[1] * scale) height = int(img.shape[0] * scale) return cv.resize(img, (width, height)) def permit(check_list): permit_ids = list() if len(check_list) != 0: for person in check_list: if check_list[person]["counter"] >= 3: permit_ids.append(person) return permit_ids else: return if __name__ == '__main__': ADDRESS = "http://127.0.0.1:5000" device_id = 0 white_list = dict() door_close = True cap = cv.VideoCapture(device_id) tm = cv.TickMeter() cv.namedWindow('libfacedetection demo', cv.WINDOW_NORMAL) cv.setWindowProperty('libfacedetection demo', cv.WND_PROP_FULLSCREEN, cv.WINDOW_FULLSCREEN) visualize = VisualizeWorker() recognition = RecognitionWorker(cap=cap) request = HttpWorker() door = DoorWorker() while cv.waitKey(1) < 0: has_frame, frame = cap.read() if not has_frame: print('No frames grabbed!') permit_list = permit(white_list) tm.start() faces = recognition.check_frame(frame) tm.stop() if faces is not None: frame, crop_imgs = visualize.drawer(frame, faces, door.door_status, permit_list, fps=tm.getFPS()) for position, img in enumerate(crop_imgs): resize_frame = resize(img) _, im_buf_arr = cv.imencode(".jpg", resize_frame) byte_im = im_buf_arr.tobytes() threading.Thread(target=request.search_face, args=(position, {'image': byte_im},)).start() if permit_list: threading.Thread(target=door.open_door).start() cv.imshow('libfacedetection demo', frame) tm.reset() print("END")