#14 StudentCapture & TrainClassifier added to capture & save student encodings for recognition

Author: mohamedmujtabaraza

backend/src/app.py

@@ -2,11 +2,13 @@
 from flask_restful import Api
 from flask_jwt_extended import JWTManager
 from flask_cors import CORS
+from flask_uploads import configure_uploads, patch_request_class
 from marshmallow import ValidationError
 
+from src.libs.image_helper import IMAGE_SET
 from src.resources.dashboard import Dashboard
 from src.resources.teacher import Teacher, TeacherRegister, TeacherLogin
-from src.resources.student import StudentList, StudentAdd, StudentDelete
+from src.resources.student import StudentList, StudentAdd, StudentCapture, StudentDelete, TrainClassifier
 from src.resources.attendance import AttendanceList
 from src.resources.video_feed import (
     VideoFeedList, VideoFeedAdd, VideoFeed, VideoFeedPreview, VideoFeedStop, VideoFeedStart, VideoFeedDelete
@@ -16,6 +18,8 @@
 app = Flask(__name__)
 app.config.from_object("src.settings.FlaskAppConfiguration")
 api = Api(app)
+patch_request_class(app, 2 * 1024 * 1024)  # 2 MB max size upload
+configure_uploads(app, IMAGE_SET)
 jwt = JWTManager(app)
 cors = CORS(app)
 
@@ -45,7 +49,9 @@ def handle_marshmallow_validation(err):
 # /students
 api.add_resource(StudentList, "/students")
 api.add_resource(StudentAdd, "/students/add")
+api.add_resource(StudentCapture, "/students/capture/<int:student_id>")
 api.add_resource(StudentDelete, "/students/delete/<int:student_id>")
+api.add_resource(TrainClassifier, "/students/train")
 
 # /attendance
 api.add_resource(AttendanceList, "/attendance")

backend/src/libs/cli_utils.py

@@ -153,136 +153,6 @@ def train_classifier(cls):
         f.write(pickle.dumps(data))
         f.close()
 
-    # def recognize_n_attendance_dnn(self):
-    #     print("[INFO] loading encodings...")
-    #     data = pickle.loads(open(ENCODINGS_FILE, "rb").read())
-    #     # print(len(data['encodings']) == len(data['ids']))
-    #
-    #     print("[INFO] starting video stream...")
-    #     # store input video stream in cap variable
-    #     cap = cv2.VideoCapture(self.input_video)
-    #     # load our serialized model from disk
-    #     net = cv2.dnn.readNetFromCaffe(prototxt=PROTOTXT_PATH, caffeModel=CAFFEMODEL_PATH)
-    #
-    #     # find if today's attendance exists in the database
-    #     attendance = AttendanceModel.find_by_date(date=dt.today())
-    #     # if not
-    #     if attendance is None:
-    #         # create new instance for today's attendance
-    #         attendance = AttendanceModel()
-    #
-    #     # loop over the frames from the video stream
-    #     while True:
-    #         # grab the frame from the video stream
-    #         ret, img = cap.read()
-    #
-    #         # convert the input frame from BGR to RGB then resize it to have
-    #         # a width of 750px (to speedup processing)
-    #         rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
-    #         # rgb = imutils.resize(img, width=750)
-    #         r = img.shape[1] / float(rgb.shape[1])
-    #
-    #         # grab the image frame dimensions and convert it to a blob
-    #         (h, w) = img.shape[:2]
-    #         blob = cv2.dnn.blobFromImage(
-    #             cv2.resize(img, (300, 300)), 1.0, (300, 300), (104.0, 177.0, 123.0)
-    #         )
-    #
-    #         # pass the blob through the network and obtain the detections and
-    #         # predictions
-    #         net.setInput(blob)
-    #         detections = net.forward()
-    #
-    #         # loop over the detections
-    #         for i in range(0, detections.shape[2]):
-    #             # extract the confidence (i.e., probability) associated with the
-    #             # prediction
-    #             confidence = detections[0, 0, i, 2]
-    #
-    #             # filter out weak detections by ensuring the `confidence` is
-    #             # greater than the minimum confidence
-    #             if confidence < 0.5:
-    #                 continue
-    #
-    #             # compute the (x, y)-coordinates of the bounding box for the
-    #             # object
-    #             box = detections[0, 0, i, 3:7] * np.array([w, h, w, h])
-    #             (startX, startY, endX, endY) = box.astype("int")
-    #
-    #             # detect the (x, y)-coordinates of the bounding boxes
-    #             # corresponding to each face in the input frame, then compute
-    #             # the facial embeddings for each face
-    #             boxes = [(startY, endX, endY, startX)]
-    #
-    #             encodings = face_recognition.face_encodings(rgb, boxes)
-    #             names = []
-    #
-    #             # loop over the facial embeddings
-    #             for encoding in encodings:
-    #                 # attempt to match each face in the input image to our known encodings
-    #                 matches = face_recognition.compare_faces(data["encodings"], encoding, DLIB_TOLERANCE)
-    #                 # name to be displayed on video
-    #                 display_name = "Unknown"
-    #
-    #                 # check to see if we have found a match
-    #                 if True in matches:
-    #                     # find the indexes of all matched faces then initialize a
-    #                     # dictionary to count the total number of times each face
-    #                     # was matched
-    #                     matched_indexes = [i for (i, b) in enumerate(matches) if b]
-    #                     counts = {}
-    #
-    #                     # loop over the matched indexes and maintain a count for
-    #                     # each recognized face
-    #                     for matched_index in matched_indexes:
-    #                         _id = data["ids"][matched_index]
-    #                         counts[_id] = counts.get(_id, 0) + 1
-    #
-    #                     # determine the recognized face with the largest number
-    #                     # of votes (note: in the event of an unlikely tie Python
-    #                     # will select first entry in the dictionary)
-    #                     _id = max(counts, key=counts.get)
-    #                     if _id:
-    #                         # find matched student in the database by id
-    #                         student = StudentModel.find_by_id(_id)
-    #                         # if student's attendance is not marked
-    #                         if not attendance.is_marked(student):
-    #                             # then mark student's attendance
-    #                             attendance.students.append(student)
-    #                             # commit changes to database
-    #                             attendance.save_to_db()
-    #                         # update displayed name to student's name
-    #                         display_name = student.name
-    #                 # append the name to be displayed in names list
-    #                 names.append(display_name)
-    #             # loop over the recognized faces
-    #             for ((top, right, bottom, left), display_name) in zip(boxes, names):
-    #                 if display_name == "Unknown":
-    #                     continue
-    #                 # rescale the face coordinates
-    #                 top = int(top * r)
-    #                 right = int(right * r)
-    #                 bottom = int(bottom * r)
-    #                 left = int(left * r)
-    #                 top_left = (left, top)
-    #                 bottom_right = (right, bottom)
-    #
-    #                 # draw the predicted face name on the image
-    #                 cv2.rectangle(img, top_left, bottom_right, (0, 255, 0), 2)
-    #                 y = top - 15 if top - 15 > 15 else top + 15
-    #                 cv2.putText(img, display_name, (left, y), cv2.FONT_HERSHEY_SIMPLEX, 0.75, (0, 255, 0), 2)
-    #
-    #         # display the output frames to the screen
-    #         cv2.imshow(f"Recognizing Faces - {self.app_title}", img)
-    #         k = cv2.waitKey(100) & 0xff  # Press 'ESC' for exiting from the loop
-    #         if k == 27:
-    #             break
-    #
-    #     # do a bit of cleanup
-    #     cap.release()
-    #     cv2.destroyAllWindows()
-    #     print("Attendance Successful!")
-
     def recognize_n_attendance(self):
         print("[INFO] loading encodings...")
         data = pickle.loads(open(ENCODINGS_FILE, "rb").read())

backend/src/libs/image_helper.py

@@ -0,0 +1,74 @@
+import os
+import re
+from typing import Union
+from werkzeug.datastructures import FileStorage
+
+from flask_uploads import UploadSet, IMAGES
+
+IMAGE_SET = UploadSet("images", IMAGES)  # set name and allowed extensions
+
+
+def save_image(image: FileStorage, folder: str = None, name: str = None) -> str:
+    return IMAGE_SET.save(image, folder, name)
+
+
+def get_path(filename: str = None, folder: str = None) -> str:
+    return IMAGE_SET.path(filename, folder)
+
+
+def find_image_any_format(filename: str, folder: str) -> Union[str, None]:
+    """
+    Given a format-less filename, try to find the file by appending each of the allowed formats to the given
+    filename and check if the file exists
+    :param filename: formatless filename
+    :param folder: the relative folder in which to search
+    :return: the path of the image if exists, otherwise None
+    """
+    for _format in IMAGES:  # look for existing avatar and delete it
+        avatar = f"{filename}.{_format}"
+        avatar_path = IMAGE_SET.path(filename=avatar, folder=folder)
+        if os.path.isfile(avatar_path):
+            return avatar_path
+    return None
+
+
+def _retrieve_filename(file: Union[str, FileStorage]) -> str:
+    """
+    Make our filename related functions generic, able to deal with FileStorage object as well as filename str.
+    """
+    if isinstance(file, FileStorage):
+        return file.filename
+    return file
+
+
+def is_filename_safe(file: Union[str, FileStorage]) -> bool:
+    """
+    Check if a filename is secure according to our definition
+    - starts with a-z A-Z 0-9 at least one time
+    - only contains a-z A-Z 0-9 and _().-
+    - followed by a dot (.) and a allowed_format at the end
+    """
+    filename = _retrieve_filename(file)
+
+    allowed_format = "|".join(IMAGES)
+    # format IMAGES into regex, eg: ('jpeg','png') --> 'jpeg|png'
+    regex = f"^[a-zA-Z0-9][a-zA-Z0-9_()-\.]*\.({allowed_format})$"
+    return re.match(regex, filename) is not None
+
+
+def get_basename(file: Union[str, FileStorage]) -> str:
+    """
+    Return file's basename, for example
+    get_basename('some/folder/image.jpg') returns 'image.jpg'
+    """
+    filename = _retrieve_filename(file)
+    return os.path.split(filename)[1]
+
+
+def get_extension(file: Union[str, FileStorage]) -> str:
+    """
+    Return file's extension, for example
+    get_extension('image.jpg') returns '.jpg'
+    """
+    filename = _retrieve_filename(file)
+    return os.path.splitext(filename)[1]

backend/src/libs/web_utils.py

@@ -8,53 +8,18 @@
 import face_recognition
 
 from src.settings import (
-    VIDEO_SOURCE,
-    PROTOTXT_PATH, CAFFEMODEL_PATH,
     DLIB_MODEL, DLIB_TOLERANCE,
     ENCODINGS_FILE
 )
 from src.libs.base_camera import BaseCamera
-from src.models import StudentModel, AttendanceModel, VideoFeedModel
-
-
-class DetectionCamera(BaseCamera):
-    video_source = 0
-
-    def __init__(self):
-        if VIDEO_SOURCE:
-            DetectionCamera.set_video_source(VIDEO_SOURCE)
-        super(DetectionCamera, self).__init__()
-
-    @classmethod
-    def set_video_source(cls, source):
-        cls.video_source = source
-
-    @classmethod
-    def frames(cls):
-        camera = cv2.VideoCapture(cls.video_source)
-        if not camera.isOpened():
-            raise RuntimeError('Could not start camera.')
-
-        while True:
-            # read current frame
-            _, img = camera.read()
-
-            # TODO: logic of face detection
-
-            # encode as a jpeg image and return it
-            yield cv2.imencode('.jpg', img)[1].tobytes()
+from src.models import StudentModel, AttendanceModel
 
 
 class RecognitionCamera(BaseCamera):
     video_source = 0
     # this class variable will help to process every other frame of video to save time
     process_this_frame = True
 
-    # def __init__(self, unique_id=None):
-    #     if VIDEO_SOURCE:
-    #         RecognitionCamera.set_video_source(VIDEO_SOURCE)
-    #     super(RecognitionCamera, self).__init__(unique_id)
-
     @classmethod
     def set_video_source(cls, source):
         cls.video_source = source