college-work/Дипломная работа/app/client/client.py

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")