Upload 4 files

app.py

@@ -0,0 +1,51 @@
+import os
+from concurrent.futures import ThreadPoolExecutor
+import gradio as gr
+from bpm_app.heartBPM_modified_copy import heart
+from stress_detection.eyebrow_detection_modified_copy import stress
+from age_estimator.mivolo.demo_copy import main as age_estimation_main
+
+def process_video(video_file):
+    # Validate the input file path
+    if not video_file or not os.path.isfile(video_file):
+        return {'error': 'Invalid video path'}
+
+    # Run functions in parallel
+    with ThreadPoolExecutor() as executor:
+        heart_future = executor.submit(heart, video_file)
+        stress_future = executor.submit(stress, video_file, duration=30)
+        
+        # Define parameters for age estimation
+        output_folder = 'output'
+        detector_weights = 'age_estimator/mivolo/models/yolov8x_person_face.pt'
+        checkpoint = 'age_estimator/mivolo/models/model_imdb_cross_person_4.22_99.46.pth.tar'
+        device = 'cpu'
+        with_persons = True
+        disable_faces = False
+        draw = True
+        
+        age_future = executor.submit(
+            age_estimation_main, video_file, output_folder, detector_weights, checkpoint, device, with_persons, disable_faces, draw
+        )
+        
+        # Retrieve results
+        avg_bpm, frames_processed = heart_future.result()
+        stressed_count, not_stressed_count, most_frequent_label = stress_future.result()
+        absolute_age, lower_bound, upper_bound = age_future.result()
+
+    # Compile results
+    results = {
+        'Average BPM': avg_bpm,
+        'Most Frequent State': most_frequent_label,
+        'Age Range': f"{lower_bound} - {upper_bound}"
+    }
+
+    return results
+
+# Define Gradio interface
+gr.Interface(
+    fn=process_video,
+    inputs=gr.Video(label="Upload a video file"),
+    outputs="json",
+    title="Parallel Video Processing for Heart Rate, Stress, and Age Estimation"
+).launch()

eyebrow_detection_modified_copy.py

@@ -0,0 +1,125 @@
+import argparse
+from scipy.spatial import distance as dist
+from imutils import face_utils
+import numpy as np
+import imutils
+import time
+import dlib
+import cv2
+import matplotlib.pyplot as plt
+from keras.preprocessing.image import img_to_array
+from keras.models import load_model
+
+def eye_brow_distance(leye, reye):
+    global points
+    distq = dist.euclidean(leye, reye)
+    points.append(int(distq))
+    return distq
+
+def emotion_finder(faces, frame):
+    global emotion_classifier
+    EMOTIONS = ["angry", "disgust", "scared", "happy", "sad", "surprised", "neutral"]
+    x, y, w, h = face_utils.rect_to_bb(faces)
+    frame = frame[y:y + h, x:x + w]
+    roi = cv2.resize(frame, (64, 64))
+    roi = roi.astype("float") / 255.0
+    roi = img_to_array(roi)
+    roi = np.expand_dims(roi, axis=0)
+    preds = emotion_classifier.predict(roi)[0]
+    emotion_probability = np.max(preds)
+    label = EMOTIONS[preds.argmax()]
+    return label
+
+def normalize_values(points, disp):
+    normalized_value = abs(disp - np.min(points)) / abs(np.max(points) - np.min(points))
+    stress_value = np.exp(-(normalized_value))
+    return stress_value
+
+def stress(video_path, duration):
+    global points, emotion_classifier
+    detector = dlib.get_frontal_face_detector()
+    predictor = dlib.shape_predictor("stress_detection/models/data")
+    emotion_classifier = load_model("stress_detection/models/_mini_XCEPTION.102-0.66.hdf5", compile=False)
+    
+    cap = cv2.VideoCapture(video_path)
+    points = []
+    stress_labels = []
+    start_time = time.time()
+
+    while True:
+        current_time = time.time()
+        if current_time - start_time >= duration:
+            break
+
+        ret, frame = cap.read()
+        if not ret:
+            break
+        
+        frame = cv2.flip(frame, 1)
+        frame = imutils.resize(frame, width=500, height=500)
+
+        (lBegin, lEnd) = face_utils.FACIAL_LANDMARKS_IDXS["right_eyebrow"]
+        (rBegin, rEnd) = face_utils.FACIAL_LANDMARKS_IDXS["left_eyebrow"]
+
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+
+        try:
+            detections = detector(gray, 0)
+            for detection in detections:
+                emotion = emotion_finder(detection, gray)
+                shape = predictor(gray, detection)
+                shape = face_utils.shape_to_np(shape)
+
+                leyebrow = shape[lBegin:lEnd]
+                reyebrow = shape[rBegin:rEnd]
+
+                distq = eye_brow_distance(leyebrow[-1], reyebrow[0])
+                stress_value = normalize_values(points, distq)
+
+                # Determine stress label for this frame
+                if emotion in ['scared', 'sad', 'angry'] and stress_value >= 0.75:
+                    stress_label = 'stressed'
+                else:
+                    stress_label = 'not stressed'
+
+                # Store stress label in list
+                stress_labels.append(stress_label)
+
+        except Exception as e:
+            print(f'Error: {e}')
+
+        key = cv2.waitKey(1) & 0xFF
+        if key == ord('q'):
+            break
+
+    cap.release()
+
+    # Count occurrences of 'stressed' and 'not stressed'
+    stressed_count = stress_labels.count('stressed')
+    not_stressed_count = stress_labels.count('not stressed')
+
+    # Determine which label occurred more frequently
+    if stressed_count > not_stressed_count:
+        most_frequent_label = 'stressed'
+    else:
+        most_frequent_label = 'not stressed'
+
+    return stressed_count, not_stressed_count, most_frequent_label
+
+def main():
+    # Argument parsing
+    parser = argparse.ArgumentParser(description='Stress Detection from Video')
+    parser.add_argument('--video', type=str, required=True, default='output.mp4', help='Path to the input video file')
+    parser.add_argument('--duration', type=int, default=30, help='Duration for analysis in seconds')
+    args = parser.parse_args()
+
+    # Call the stress function and get the results
+    stressed_count, not_stressed_count, most_frequent_label = stress(args.video, args.duration)
+
+    # Display the result
+    print(f"Stressed frames: {stressed_count}")
+    print(f"Not stressed frames: {not_stressed_count}")
+    print(f"Most frequent state: {most_frequent_label}")
+
+if __name__ == '__main__':
+    main()

heartBPM_modified_copy.py

@@ -0,0 +1,101 @@
+import numpy as np
+import cv2
+import time
+from cvzone.FaceDetectionModule import FaceDetector
+
+# Initialization
+videoWidth = 160
+videoHeight = 120
+videoChannels = 3
+videoFrameRate = 15
+
+# Helper Methods
+def buildGauss(frame, levels):
+    pyramid = [frame]
+    for level in range(levels):
+        frame = cv2.pyrDown(frame)
+        pyramid.append(frame)
+    return pyramid
+
+def reconstructFrame(pyramid, index, levels):
+    filteredFrame = pyramid[index]
+    for level in range(levels):
+        filteredFrame = cv2.pyrUp(filteredFrame)
+    filteredFrame = filteredFrame[:videoHeight, :videoWidth]
+    return filteredFrame
+
+# Main heart rate function
+def heart(video_file_path):
+    levels = 3
+    alpha = 170
+    minFrequency = 1.0
+    maxFrequency = 2.0
+    bufferSize = 150
+    bufferIndex = 0
+
+    detector = FaceDetector()
+
+    video = cv2.VideoCapture(video_file_path)
+
+    firstFrame = np.zeros((videoHeight, videoWidth, videoChannels))
+    firstGauss = buildGauss(firstFrame, levels + 1)[levels]
+    videoGauss = np.zeros((bufferSize, firstGauss.shape[0], firstGauss.shape[1], videoChannels))
+    fourierTransformAvg = np.zeros((bufferSize))
+
+    frequencies = (1.0 * videoFrameRate) * np.arange(bufferSize) / (1.0 * bufferSize)
+    mask = (frequencies >= minFrequency) & (frequencies <= maxFrequency)
+
+    bpmCalculationFrequency = 10
+    bpmBufferIndex = 0
+    bpmBufferSize = 10
+    bpmBuffer = np.zeros((bpmBufferSize))
+
+    bpmList = []
+    startTime = time.time()
+    frameCount = 0
+
+    while True:
+        ret, frame = video.read()
+        if not ret:
+            break
+
+        elapsedTime = time.time() - startTime
+        if elapsedTime >= 30:
+            break
+
+        frame, bboxs = detector.findFaces(frame, draw=False)
+        frameCount += 1
+
+        if bboxs:
+            x1, y1, w1, h1 = bboxs[0]['bbox']
+            
+            # Check if the bounding box is valid
+            if x1 >= 0 and y1 >= 0 and w1 > 0 and h1 > 0:
+                detectionFrame = frame[y1:y1 + h1, x1:x1 + w1]
+                
+                # Check if detectionFrame is valid and not empty before resizing
+                if detectionFrame.size != 0:
+                    detectionFrame = cv2.resize(detectionFrame, (videoWidth, videoHeight))
+
+                    videoGauss[bufferIndex] = buildGauss(detectionFrame, levels + 1)[levels]
+                    fourierTransform = np.fft.fft(videoGauss, axis=0)
+                    fourierTransform[mask == False] = 0
+
+                    if bufferIndex % bpmCalculationFrequency == 0:
+                        for buf in range(bufferSize):
+                            fourierTransformAvg[buf] = np.real(fourierTransform[buf]).mean()
+                        hz = frequencies[np.argmax(fourierTransformAvg)]
+                        bpm = 60.0 * hz
+                        bpmBuffer[bpmBufferIndex] = bpm
+                        bpmBufferIndex = (bpmBufferIndex + 1) % bpmBufferSize
+                        bpmList.append(bpmBuffer.mean())
+
+                    bufferIndex = (bufferIndex + 1) % bufferSize
+        else:
+            # If no face is detected, skip to the next frame
+            continue
+
+    avgBPM = np.mean(bpmList) if bpmList else 0
+    video.release()
+
+    return avgBPM, frameCount

requirements.txt

@@ -0,0 +1,14 @@
+huggingface_hub
+tensorflow
+ultralytics==8.1.0
+timm==0.8.13.dev0
+yt_dlp
+lapx>=0.5.2
+typing-extensions
+cvzone
+keras
+cmake
+dlib
+imutils
+opencv-python
+