Дипломная работа

Дипломная работа/app/.env

@@ -0,0 +1,4 @@
+MONGO_HOST=127.0.0.1
+MONGO_PORT=27017
+MONGO_USER=user
+MONGO_PASS=pass

Дипломная работа/app/README.md

@@ -0,0 +1,19 @@
+# Система распознавания лиц для электронных замков
+
+## Запуск:
+
+### Установка зависимостей Python:
+```commandline
+python3.9 -m venv venv
+source venv/bin/activate
+pip3 install -r requirements.txt
+```
+### Запуск MongoDB
+```commandline
+docker run --name mongodb -d -p 27017:27017 /
+    -e MONGO_INITDB_ROOT_USERNAME=user /
+    -e MONGO_INITDB_ROOT_PASSWORD=pass /
+    mongodb/mongodb-community-server
+```
+
+developed by FabLab for Igor's diplom rabota

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

@@ -0,0 +1,166 @@
+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")
+  

Дипломная работа/app/schemas/api_schemas.py

@@ -0,0 +1,13 @@
+from typing import Any
+
+from pydantic import BaseModel
+
+
+ALLOWED_MIME_TYPES = {"image/png", "image/jpeg"}
+
+class AddPersonJSON(BaseModel):
+    username: str
+
+class BaseResponse(BaseModel):
+    result: Any
+    error: bool = False

Дипломная работа/app/schemas/base_schemas.py

@@ -0,0 +1,20 @@
+from pydantic import BaseModel, NonNegativeInt, ConfigDict
+from typing import Any
+
+from datetime import datetime
+
+from numpy import ndarray, dtype
+
+class Person(BaseModel):
+    person_id: NonNegativeInt = 0
+    person_name: str
+
+    encodings: list[float]
+    image_name: str
+
+    adding_date: str = datetime.now().isoformat()
+    active_status: bool = True
+
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+    )

Дипломная работа/app/server.py

@@ -0,0 +1,61 @@
+#!/usr/bin/python3
+import json
+
+import cv2
+import uvicorn, asyncio
+import numpy as np
+
+from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Depends
+
+from schemas.api_schemas import *
+
+from workers.mongo_worker import MongoWorker
+from workers.face_rec_worker import FaceRecognition
+
+app = FastAPI()
+mongo_worker = MongoWorker()
+face_recognition_worker = FaceRecognition()
+
+@app.post('/add_person')
+async def add_person(json_data: str = Form(...),
+                     img_data: UploadFile = File(...),
+                     face_recognition: FaceRecognition = Depends(lambda: face_recognition_worker),
+                     mongo: MongoWorker = Depends(lambda: mongo_worker)) -> BaseResponse:
+    json_validate = AddPersonJSON(**json.loads(json_data))
+    if img_data.content_type not in ALLOWED_MIME_TYPES:
+        raise HTTPException(400, detail="Invalid file type")
+
+    file_data = await img_data.read()
+    face_img = cv2.imdecode(np.fromstring(file_data, np.uint8), cv2.IMREAD_COLOR)
+    encodings = face_recognition.get_encodings(face_img)
+    if not encodings:
+        raise HTTPException(422, detail="Can't find human face")
+
+    result = mongo.add_person(json_validate.username, encodings[0], img_data.filename, face_img)
+    if result.error:
+        raise HTTPException(422, detail=str(result))
+    return BaseResponse(result="Success")
+
+@app.post('/find_by_img')
+async def find_by_img(img_data: UploadFile = File(...),
+                     face_recognition: FaceRecognition = Depends(lambda: face_recognition_worker),
+                     mongo: MongoWorker = Depends(lambda: mongo_worker)) -> BaseResponse:
+    if img_data.content_type not in ALLOWED_MIME_TYPES:
+        raise HTTPException(400, detail="Invalid file type")
+
+    file_data = await img_data.read()
+    face_img = cv2.imdecode(np.fromstring(file_data, np.uint8), cv2.IMREAD_COLOR)
+    encodings = face_recognition.get_encodings(face_img)
+    if not encodings:
+        raise HTTPException(422, detail="Can't find human face")
+    res = face_recognition_worker.face_distance(encodings)
+    print(res)
+    return BaseResponse(result=res)
+
+async def main():
+    config = uvicorn.Config("server:app", host="0.0.0.0", port=5000, log_level="info")
+    server = uvicorn.Server(config)
+    await server.serve()
+
+if __name__ == "__main__":
+    asyncio.run(main())

Дипломная работа/app/workers/face_rec_worker.py

@@ -0,0 +1,65 @@
+import face_recognition
+import numpy as np
+import cv2
+
+from workers.mongo_worker import MongoWorker
+
+class FaceRecognition():
+    def __init__(self):
+        self.mongo_worker = MongoWorker()
+        self.face_encodings_in_cache = self.mongo_worker.get_all_encodings()
+
+    def get_encodings(self, face_image):
+        return face_recognition.face_encodings(face_image)
+
+    def softmax(self, values):
+        exp_values = np.exp(values)
+        exp_values_sum = np.sum(exp_values)
+
+        return exp_values / exp_values_sum
+
+    def kl_divergence(self, face_encodings, face_to_compare):
+        if len(face_encodings) == 0:
+            return np.empty((0))
+
+        face_encodings = np.asarray(self.softmax(face_encodings))
+        face_to_compare = np.asarray(self.softmax(face_to_compare))
+
+        face_encodings = face_encodings / face_encodings.sum(axis=1, keepdims=True)
+        face_to_compare = face_to_compare / face_to_compare.sum()
+
+        epsilon = 1e-10
+        face_encodings = np.clip(face_encodings, epsilon, 1)
+        face_to_compare = np.clip(face_to_compare, epsilon, 1)
+
+        return np.sum(face_encodings * np.log(face_encodings / face_to_compare), axis=1)
+
+    def face_distance(self, face_encodings:list[float]):
+        result = {}
+        if self.mongo_worker.count_persons() > len(self.face_encodings_in_cache):
+            self.face_encodings_in_cache = self.mongo_worker.get_all_encodings()
+
+        for person_id in self.face_encodings_in_cache:
+            result.update({person_id: self.kl_divergence(face_encodings, self.face_encodings_in_cache[person_id])})
+
+        print(result.values())
+        try:
+            minimal = min(result.values())
+            print(minimal)
+            if minimal > 0.001:
+                return None
+            return [key for key, val in result.items() if val == minimal][0]
+        except:
+            best_in_frame = dict()
+            for key, elements in result.items():
+                best_in_frame.update({key:min(elements)})
+            minimal = min(best_in_frame.values())
+            print(minimal)
+            if minimal > 0.001:
+                return None
+            return [key for key, val in best_in_frame.items() if val == minimal][0]
+
+if __name__ == "__main__":
+    face_recognition_worker = FaceRecognition()
+    img = cv2.imread("obama.jpg")
+    print(face_recognition_worker.get_encodings(img))

Дипломная работа/app/workers/mongo_worker.py

@@ -0,0 +1,79 @@
+import os
+
+import gridfs
+import numpy as np
+import cv2
+
+from pymongo import MongoClient
+from dotenv import load_dotenv
+
+from schemas.base_schemas import *
+from schemas.api_schemas import *
+
+class MongoWorker():
+    def __init__(self):
+        load_dotenv()
+        self.client = MongoClient(host=os.getenv('MONGO_HOST'),
+                                  port=int(os.getenv('MONGO_PORT')),
+                                  username=os.getenv('MONGO_USER'),
+                                  password=os.getenv('MONGO_PASS'))
+        self.db = self.client["Face_recognition"]
+        self.persons = self.db["persons"]
+        self.counters = self.db["counters"]
+        self.gfs = gridfs.GridFS(self.db)
+
+    def count_persons(self):
+        docs_count = self.persons.count_documents({})
+        return docs_count
+
+    def get_and_update_counter(self, counter_name: str) -> int:
+        counter = self.counters.find_one_and_update(
+            {"counter_name": counter_name},
+            {"$inc": {"counter": 1}},
+            upsert=True,
+            return_document=True)
+        return counter["counter"]
+
+    def add_person(self, person_name:str, encodings:ndarray[Any, dtype], image_name:str, face_img:np.ndarray) -> BaseResponse:
+        try:
+            new_person_id = self.get_and_update_counter("persons")
+            new_person = Person(
+                person_id = new_person_id,
+                person_name = person_name,
+                encodings = encodings.tolist(),
+                image_name = f"{new_person_id}_{image_name}")
+
+            file_id = self.save_image_to_gridfs(face_img, image_name)
+            mongo_result = self.persons.insert_one(new_person.model_dump())
+            return BaseResponse(result={"file_id": file_id, "mongo_result": mongo_result})
+        except Exception as exception:
+            print(exception)
+            return BaseResponse(result=exception, error=True)
+
+    def save_image_to_gridfs(self, image_array, filename):
+        success, encoded_image = cv2.imencode('.png', image_array)
+        if not success:
+            raise ValueError("Не удалось закодировать изображение")
+        file_id = self.gfs.put(encoded_image.tobytes(), filename=filename)
+        return file_id
+
+    def get_all_encodings(self):
+        result = {}
+        projection = {"person_id": 1, "encodings": 1, "_id": 0}
+
+        for document in self.persons.find({}, projection):
+            entry = {document["person_id"]: document["encodings"]}
+            result.update(entry)
+
+        return result
+
+    def retrieve_image_from_gridfs(self, file_id):
+        try:
+            gridfs_file = self.gfs.get(file_id)
+            image_bytes = gridfs_file.read()
+            image_array = cv2.imdecode(np.frombuffer(image_bytes, np.uint8), cv2.IMREAD_COLOR)
+            if image_array is None:
+                raise ValueError("Не удалось декодировать изображение")
+            return image_array
+        except gridfs.errors.NoFile:
+            raise FileNotFoundError(f"Файл с ID {file_id} не найден в GridFS")