1

app.py

@@ -4,13 +4,6 @@ from tabs.FACS_analysis import create_facs_analysis_tab
 from ui_components import CUSTOM_CSS, HEADER_HTML, DISCLAIMER_HTML
 import spaces  # Importing spaces to utilize Zero GPU
 
-# Initialize Zero GPU
-if torch.cuda.is_available():
-    zero = torch.Tensor([0]).cuda()
-    print(f"Initial device: {zero.device}")
-else:
-    zero = torch.Tensor([0])
-    print("CUDA is not available. Using CPU.")
 
 # Define the tab structure
 TAB_STRUCTURE = [
@@ -22,10 +15,6 @@ TAB_STRUCTURE = [
 # Decorate GPU-dependent function with Zero GPU
 @spaces.GPU(duration=120)  # Allocates GPU for 120 seconds when needed
 def create_demo():
-    if torch.cuda.is_available():
-        print(f"Device inside create_demo: {zero.device}")
-    else:
-        print("CUDA is not available inside create_demo.")
     
     # Gradio blocks to create the interface
     with gr.Blocks(css=CUSTOM_CSS) as demo:

app/app_utils.py

@@ -1,333 +0,0 @@
-import torch
-import numpy as np
-import mediapipe as mp
-from PIL import Image
-import cv2
-from pytorch_grad_cam.utils.image import show_cam_on_image
-import matplotlib.pyplot as plt
-
-# Importing necessary components for the Gradio app
-from app.model import pth_model_static, pth_model_dynamic, cam, pth_processing
-from app.face_utils import get_box, display_info
-from app.config import DICT_EMO, config_data
-from app.plot import statistics_plot
-
-mp_face_mesh = mp.solutions.face_mesh
-
-def get_device():
-    if torch.backends.mps.is_available():
-        return torch.device("mps")
-    elif torch.cuda.is_available():
-        return torch.device("cuda")
-    else:
-        return torch.device("cpu")
-
-device = get_device()
-print(f"Using device: {device}")
-
-# Move models to the selected device
-pth_model_static = pth_model_static.to(device)
-pth_model_dynamic = pth_model_dynamic.to(device)
-
-def preprocess_image_and_predict(inp):
-    inp = np.array(inp)
-
-    if inp is None:
-        return None, None, None
-
-    try:
-        h, w = inp.shape[:2]
-    except Exception:
-        return None, None, None
-
-    with mp_face_mesh.FaceMesh(
-        max_num_faces=1,
-        refine_landmarks=False,
-        min_detection_confidence=0.5,
-        min_tracking_confidence=0.5,
-    ) as face_mesh:
-        results = face_mesh.process(inp)
-        if results.multi_face_landmarks:
-            for fl in results.multi_face_landmarks:
-                startX, startY, endX, endY = get_box(fl, w, h)
-                cur_face = inp[startY:endY, startX:endX]
-                cur_face_n = pth_processing(Image.fromarray(cur_face)).to(device)
-                with torch.no_grad():
-                    prediction = (
-                        torch.nn.functional.softmax(pth_model_static(cur_face_n), dim=1)
-                        .detach()
-                        .cpu()
-                        .numpy()[0]
-                    )
-                confidences = {DICT_EMO[i]: float(prediction[i]) for i in range(7)}
-                grayscale_cam = cam(input_tensor=cur_face_n)
-                grayscale_cam = grayscale_cam[0, :]
-                cur_face_hm = cv2.resize(cur_face,(224,224))
-                cur_face_hm = np.float32(cur_face_hm) / 255
-                heatmap = show_cam_on_image(cur_face_hm, grayscale_cam, use_rgb=True)
-
-    return cur_face, heatmap, confidences
-
-def preprocess_frame_and_predict_aus(frame):
-    if len(frame.shape) == 2:
-        frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
-    elif frame.shape[2] == 4:
-        frame = cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
-
-    with mp_face_mesh.FaceMesh(
-        max_num_faces=1,
-        refine_landmarks=False,
-        min_detection_confidence=0.5,
-        min_tracking_confidence=0.5
-    ) as face_mesh:
-        results = face_mesh.process(frame)
-        
-        if results.multi_face_landmarks:
-            h, w = frame.shape[:2]
-            for fl in results.multi_face_landmarks:
-                startX, startY, endX, endY = get_box(fl, w, h)
-                cur_face = frame[startY:endY, startX:endX]
-                cur_face_n = pth_processing(Image.fromarray(cur_face)).to(device)
-                
-                with torch.no_grad():
-                    features = pth_model_static(cur_face_n)
-                    au_intensities = features_to_au_intensities(features)
-                
-                grayscale_cam = cam(input_tensor=cur_face_n)
-                grayscale_cam = grayscale_cam[0, :]
-                cur_face_hm = cv2.resize(cur_face, (224, 224))
-                cur_face_hm = np.float32(cur_face_hm) / 255
-                heatmap = show_cam_on_image(cur_face_hm, grayscale_cam, use_rgb=True)
-                
-                return cur_face, au_intensities, heatmap
-    
-    return None, None, None
-
-def features_to_au_intensities(features):
-    features_np = features.detach().cpu().numpy()[0]
-    au_intensities = (features_np - features_np.min()) / (features_np.max() - features_np.min())
-    return au_intensities[:24]  # Assuming we want 24 AUs
-
-def preprocess_video_and_predict(video):
-    cap = cv2.VideoCapture(video)
-    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-    fps = np.round(cap.get(cv2.CAP_PROP_FPS))
-
-    path_save_video_face = 'result_face.mp4'
-    vid_writer_face = cv2.VideoWriter(path_save_video_face, cv2.VideoWriter_fourcc(*'mp4v'), fps, (224, 224))
-
-    path_save_video_hm = 'result_hm.mp4'
-    vid_writer_hm = cv2.VideoWriter(path_save_video_hm, cv2.VideoWriter_fourcc(*'mp4v'), fps, (224, 224))
-
-    lstm_features = []
-    count_frame = 1
-    count_face = 0
-    probs = []
-    frames = []
-    au_intensities_list = []
-    last_output = None
-    last_heatmap = None 
-    last_au_intensities = None
-    cur_face = None
-
-    with mp_face_mesh.FaceMesh(
-    max_num_faces=1,
-    refine_landmarks=False,
-    min_detection_confidence=0.5,
-    min_tracking_confidence=0.5) as face_mesh:
-
-        while cap.isOpened():
-            _, frame = cap.read()
-            if frame is None: break
-
-            frame_copy = frame.copy()
-            frame_copy.flags.writeable = False
-            frame_copy = cv2.cvtColor(frame_copy, cv2.COLOR_BGR2RGB)
-            results = face_mesh.process(frame_copy)
-            frame_copy.flags.writeable = True
-
-            if results.multi_face_landmarks:
-                for fl in results.multi_face_landmarks:
-                    startX, startY, endX, endY  = get_box(fl, w, h)
-                    cur_face = frame_copy[startY:endY, startX: endX]
-
-                    if count_face%config_data.FRAME_DOWNSAMPLING == 0:
-                        cur_face_copy = pth_processing(Image.fromarray(cur_face)).to(device)
-                        with torch.no_grad():
-                            features = torch.nn.functional.relu(pth_model_static.extract_features(cur_face_copy)).detach().cpu().numpy()
-                            au_intensities = features_to_au_intensities(pth_model_static(cur_face_copy))
-
-                        grayscale_cam = cam(input_tensor=cur_face_copy)
-                        grayscale_cam = grayscale_cam[0, :]
-                        cur_face_hm = cv2.resize(cur_face,(224,224), interpolation = cv2.INTER_AREA)
-                        cur_face_hm = np.float32(cur_face_hm) / 255
-                        heatmap = show_cam_on_image(cur_face_hm, grayscale_cam, use_rgb=False)
-                        last_heatmap = heatmap
-                        last_au_intensities = au_intensities
-        
-                        if len(lstm_features) == 0:
-                            lstm_features = [features]*10
-                        else:
-                            lstm_features = lstm_features[1:] + [features]
-
-                        lstm_f = torch.from_numpy(np.vstack(lstm_features)).to(device)
-                        lstm_f = torch.unsqueeze(lstm_f, 0)
-                        with torch.no_grad():
-                            output = pth_model_dynamic(lstm_f).detach().cpu().numpy()
-                        last_output = output
-
-                        if count_face == 0:
-                            count_face += 1
-
-                    else:
-                        if last_output is not None:
-                            output = last_output
-                            heatmap = last_heatmap
-                            au_intensities = last_au_intensities
-
-                        elif last_output is None:
-                            output = np.empty((1, 7))
-                            output[:] = np.nan
-                            au_intensities = np.empty(24)
-                            au_intensities[:] = np.nan
-                            
-                    probs.append(output[0])
-                    frames.append(count_frame)
-                    au_intensities_list.append(au_intensities)
-            else:
-                if last_output is not None:
-                    lstm_features = []
-                    empty = np.empty((7))
-                    empty[:] = np.nan
-                    probs.append(empty)
-                    frames.append(count_frame)
-                    au_intensities_list.append(np.full(24, np.nan))
-
-            if cur_face is not None:
-                heatmap_f = display_info(heatmap, 'Frame: {}'.format(count_frame), box_scale=.3)
-
-                cur_face = cv2.cvtColor(cur_face, cv2.COLOR_RGB2BGR)
-                cur_face = cv2.resize(cur_face, (224,224), interpolation = cv2.INTER_AREA)
-                cur_face = display_info(cur_face, 'Frame: {}'.format(count_frame), box_scale=.3)
-                vid_writer_face.write(cur_face)
-                vid_writer_hm.write(heatmap_f)
-
-            count_frame += 1
-            if count_face != 0:
-                count_face += 1
-
-        vid_writer_face.release()
-        vid_writer_hm.release()
-
-        stat = statistics_plot(frames, probs)
-        au_stat = au_statistics_plot(frames, au_intensities_list)
-
-        if not stat or not au_stat:
-            return None, None, None, None, None
-        
-    return video, path_save_video_face, path_save_video_hm, stat, au_stat
-
-# The rest of the functions remain the same
-# ...
-
-def au_statistics_plot(frames, au_intensities_list):
-    fig, ax = plt.subplots(figsize=(12, 6))
-    au_intensities_array = np.array(au_intensities_list)
-    
-    for i in range(au_intensities_array.shape[1]):
-        ax.plot(frames, au_intensities_array[:, i], label=f'AU{i+1}')
-    
-    ax.set_xlabel('Frame')
-    ax.set_ylabel('AU Intensity')
-    ax.set_title('Action Unit Intensities Over Time')
-    ax.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
-    plt.tight_layout()
-    return fig
-
-def preprocess_video_and_predict_sleep_quality(video):
-    cap = cv2.VideoCapture(video)
-    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-    fps = np.round(cap.get(cv2.CAP_PROP_FPS))
-
-    path_save_video_original = 'result_original.mp4'
-    path_save_video_face = 'result_face.mp4'
-    path_save_video_sleep = 'result_sleep.mp4'
-    
-    vid_writer_original = cv2.VideoWriter(path_save_video_original, cv2.VideoWriter_fourcc(*'mp4v'), fps, (w, h))
-    vid_writer_face = cv2.VideoWriter(path_save_video_face, cv2.VideoWriter_fourcc(*'mp4v'), fps, (224, 224))
-    vid_writer_sleep = cv2.VideoWriter(path_save_video_sleep, cv2.VideoWriter_fourcc(*'mp4v'), fps, (224, 224))
-
-    frames = []
-    sleep_quality_scores = []
-    eye_bags_images = []
-
-    with mp_face_mesh.FaceMesh(
-    max_num_faces=1,
-    refine_landmarks=False,
-    min_detection_confidence=0.5,
-    min_tracking_confidence=0.5) as face_mesh:
-
-        while cap.isOpened():
-            ret, frame = cap.read()
-            if not ret:
-                break
-
-            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-            results = face_mesh.process(frame_rgb)
-
-            if results.multi_face_landmarks:
-                for fl in results.multi_face_landmarks:
-                    startX, startY, endX, endY = get_box(fl, w, h)
-                    cur_face = frame_rgb[startY:endY, startX:endX]
-                    
-                    sleep_quality_score, eye_bags_image = analyze_sleep_quality(cur_face)
-                    sleep_quality_scores.append(sleep_quality_score)
-                    eye_bags_images.append(cv2.resize(eye_bags_image, (224, 224)))
-
-                    sleep_quality_viz = create_sleep_quality_visualization(cur_face, sleep_quality_score)
-                    
-                    cur_face = cv2.resize(cur_face, (224, 224))
-                    
-                    vid_writer_face.write(cv2.cvtColor(cur_face, cv2.COLOR_RGB2BGR))
-                    vid_writer_sleep.write(sleep_quality_viz)
-
-            vid_writer_original.write(frame)
-            frames.append(len(frames) + 1)
-
-    cap.release()
-    vid_writer_original.release()
-    vid_writer_face.release()
-    vid_writer_sleep.release()
-
-    sleep_stat = sleep_quality_statistics_plot(frames, sleep_quality_scores)
-    
-    if eye_bags_images:
-        average_eye_bags_image = np.mean(np.array(eye_bags_images), axis=0).astype(np.uint8)
-    else:
-        average_eye_bags_image = np.zeros((224, 224, 3), dtype=np.uint8)
-
-    return (path_save_video_original, path_save_video_face, path_save_video_sleep, 
-            average_eye_bags_image, sleep_stat)
-
-def analyze_sleep_quality(face_image):
-    # Placeholder function - implement your sleep quality analysis here
-    sleep_quality_score = np.random.random()
-    eye_bags_image = cv2.resize(face_image, (224, 224))
-    return sleep_quality_score, eye_bags_image
-
-def create_sleep_quality_visualization(face_image, sleep_quality_score):
-    viz = face_image.copy()
-    cv2.putText(viz, f"Sleep Quality: {sleep_quality_score:.2f}", (10, 30), 
-                cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
-    return cv2.cvtColor(viz, cv2.COLOR_RGB2BGR)
-
-def sleep_quality_statistics_plot(frames, sleep_quality_scores):
-    # Placeholder function - implement your statistics plotting here
-    fig, ax = plt.subplots()
-    ax.plot(frames, sleep_quality_scores)
-    ax.set_xlabel('Frame')
-    ax.set_ylabel('Sleep Quality Score')
-    ax.set_title('Sleep Quality Over Time')
-    return fig

app/face_utils.py

@@ -1,68 +0,0 @@
-"""
-File: face_utils.py
-Author: Elena Ryumina and Dmitry Ryumin
-Description: This module contains utility functions related to facial landmarks and image processing.
-License: MIT License
-"""
-
-import numpy as np
-import math
-import cv2
-
-
-def norm_coordinates(normalized_x, normalized_y, image_width, image_height):
-    x_px = min(math.floor(normalized_x * image_width), image_width - 1)
-    y_px = min(math.floor(normalized_y * image_height), image_height - 1)
-    return x_px, y_px
-
-
-def get_box(fl, w, h):
-    idx_to_coors = {}
-    for idx, landmark in enumerate(fl.landmark):
-        landmark_px = norm_coordinates(landmark.x, landmark.y, w, h)
-        if landmark_px:
-            idx_to_coors[idx] = landmark_px
-
-    x_min = np.min(np.asarray(list(idx_to_coors.values()))[:, 0])
-    y_min = np.min(np.asarray(list(idx_to_coors.values()))[:, 1])
-    endX = np.max(np.asarray(list(idx_to_coors.values()))[:, 0])
-    endY = np.max(np.asarray(list(idx_to_coors.values()))[:, 1])
-
-    (startX, startY) = (max(0, x_min), max(0, y_min))
-    (endX, endY) = (min(w - 1, endX), min(h - 1, endY))
-
-    return startX, startY, endX, endY
-
-def display_info(img, text, margin=1.0, box_scale=1.0):
-    img_copy = img.copy()
-    img_h, img_w, _ = img_copy.shape
-    line_width = int(min(img_h, img_w) * 0.001)
-    thickness = max(int(line_width / 3), 1)
-
-    font_face = cv2.FONT_HERSHEY_SIMPLEX
-    font_color = (0, 0, 0)
-    font_scale = thickness / 1.5
-
-    t_w, t_h = cv2.getTextSize(text, font_face, font_scale, None)[0]
-
-    margin_n = int(t_h * margin)
-    sub_img = img_copy[0 + margin_n: 0 + margin_n + t_h + int(2 * t_h * box_scale),
-              img_w - t_w - margin_n - int(2 * t_h * box_scale): img_w - margin_n]
-
-    white_rect = np.ones(sub_img.shape, dtype=np.uint8) * 255
-
-    img_copy[0 + margin_n: 0 + margin_n + t_h + int(2 * t_h * box_scale),
-    img_w - t_w - margin_n - int(2 * t_h * box_scale):img_w - margin_n] = cv2.addWeighted(sub_img, 0.5, white_rect, .5, 1.0)
-
-    cv2.putText(img=img_copy,
-                text=text,
-                org=(img_w - t_w - margin_n - int(2 * t_h * box_scale) // 2,
-                     0 + margin_n + t_h + int(2 * t_h * box_scale) // 2),
-                fontFace=font_face,
-                fontScale=font_scale,
-                color=font_color,
-                thickness=thickness,
-                lineType=cv2.LINE_AA,
-                bottomLeftOrigin=False)
-
-    return img_copy

app/model.py

@@ -1,78 +0,0 @@
-import os
-import torch
-import torch.nn as nn
-import torchvision.transforms as transforms
-from pytorch_grad_cam import GradCAM
-from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
-import logging
-from app.model_architectures import ResNet50, LSTMPyTorch
-
-# Set up logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-# Determine the device
-device = torch.device('mps' if torch.backends.mps.is_available() else 'cuda' if torch.cuda.is_available() else 'cpu')
-logger.info(f"Using device: {device}")
-
-# Define paths
-STATIC_MODEL_PATH = 'assets/models/FER_static_ResNet50_AffectNet.pt'
-DYNAMIC_MODEL_PATH = 'assets/models/FER_dynamic_LSTM.pt'
-
-def load_model(model_class, model_path, *args, **kwargs):
-    model = model_class(*args, **kwargs).to(device)
-    if os.path.exists(model_path):
-        try:
-            model.load_state_dict(torch.load(model_path, map_location=device))
-            model.eval()
-            logger.info(f"Model loaded successfully from {model_path}")
-        except Exception as e:
-            logger.error(f"Error loading model from {model_path}: {str(e)}")
-            logger.info("Initializing with random weights.")
-    else:
-        logger.warning(f"Model file not found at {model_path}. Initializing with random weights.")
-    return model
-
-# Load the static model
-pth_model_static = load_model(ResNet50, STATIC_MODEL_PATH, num_classes=7, channels=3)
-
-# Load the dynamic model
-pth_model_dynamic = load_model(LSTMPyTorch, DYNAMIC_MODEL_PATH, input_size=2048, hidden_size=256, num_layers=2, num_classes=7)
-
-# Set up GradCAM
-target_layers = [pth_model_static.resnet.layer4[-1]]
-cam = GradCAM(model=pth_model_static, target_layers=target_layers)
-
-# Define image preprocessing
-pth_transform = transforms.Compose([
-    transforms.Resize((224, 224)),
-    transforms.ToTensor(),
-    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-])
-
-def pth_processing(img):
-    img = pth_transform(img).unsqueeze(0).to(device)
-    return img
-
-def predict_emotion(img):
-    with torch.no_grad():
-        output = pth_model_static(pth_processing(img))
-        _, predicted = torch.max(output, 1)
-    return predicted.item()
-
-def get_emotion_probabilities(img):
-    with torch.no_grad():
-        output = nn.functional.softmax(pth_model_static(pth_processing(img)), dim=1)
-    return output.squeeze().cpu().numpy()
-
-def generate_cam(img):
-    input_tensor = pth_processing(img)
-    targets = [ClassifierOutputTarget(predict_emotion(img))]
-    grayscale_cam = cam(input_tensor=input_tensor, targets=targets)
-    return grayscale_cam[0, :]
-
-# Add any other necessary functions or variables here
-
-if __name__ == "__main__":
-    logger.info("Model initialization complete.")
-    # You can add some test code here to verify everything is working correctly

app/model_architectures.py

@@ -1,46 +0,0 @@
-import torch
-import torch.nn as nn
-import torchvision.models as models
-
-class ResNet50(nn.Module):
-    def __init__(self, num_classes=7, channels=3):
-        super(ResNet50, self).__init__()
-        self.resnet = models.resnet50(pretrained=True)
-        # Modify the first convolutional layer if channels != 3
-        if channels != 3:
-            self.resnet.conv1 = nn.Conv2d(channels, 64, kernel_size=7, stride=2, padding=3, bias=False)
-        num_features = self.resnet.fc.in_features
-        self.resnet.fc = nn.Linear(num_features, num_classes)
-
-    def forward(self, x):
-        return self.resnet(x)
-
-    def extract_features(self, x):
-        x = self.resnet.conv1(x)
-        x = self.resnet.bn1(x)
-        x = self.resnet.relu(x)
-        x = self.resnet.maxpool(x)
-
-        x = self.resnet.layer1(x)
-        x = self.resnet.layer2(x)
-        x = self.resnet.layer3(x)
-        x = self.resnet.layer4(x)
-
-        x = self.resnet.avgpool(x)
-        x = torch.flatten(x, 1)
-        return x
-
-class LSTMPyTorch(nn.Module):
-    def __init__(self, input_size, hidden_size, num_layers, num_classes):
-        super(LSTMPyTorch, self).__init__()
-        self.hidden_size = hidden_size
-        self.num_layers = num_layers
-        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
-        self.fc = nn.Linear(hidden_size, num_classes)
-
-    def forward(self, x):
-        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
-        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
-        out, _ = self.lstm(x, (h0, c0))
-        out = self.fc(out[:, -1, :])
-        return out