app.py

import os
import cv2
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import seaborn as sns
from facenet_pytorch import InceptionResnetV1, MTCNN
import mediapipe as mp
from fer import FER
from scipy import interpolate
from sklearn.cluster import DBSCAN, KMeans
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.metrics import silhouette_score
import umap
import pandas as pd
import matplotlib
import matplotlib.pyplot as plt
from moviepy.editor import VideoFileClip
from PIL import Image
import gradio as gr
import tempfile
import shutil
import io

# Suppress TensorFlow warnings
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
import tensorflow as tf

tf.get_logger().setLevel('ERROR')

matplotlib.rcParams['figure.dpi'] = 400
matplotlib.rcParams['savefig.dpi'] = 400

# Initialize models and other global variables
device = 'cuda' if torch.cuda.is_available() else 'cpu'

mtcnn = MTCNN(keep_all=False, device=device, thresholds=[0.98, 0.98, 0.98], min_face_size=50,
              selection_method='largest')
model = InceptionResnetV1(pretrained='vggface2').eval().to(device)
mp_face_mesh = mp.solutions.face_mesh
face_mesh = mp_face_mesh.FaceMesh(static_image_mode=False, max_num_faces=1, min_detection_confidence=0.5)
emotion_detector = FER(mtcnn=False)

def frame_to_timecode(frame_num, total_frames, duration):
    total_seconds = (frame_num / total_frames) * duration
    hours = int(total_seconds // 3600)
    minutes = int((total_seconds % 3600) // 60)
    seconds = int(total_seconds % 60)
    milliseconds = int((total_seconds - int(total_seconds)) * 1000)
    return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}"

def get_face_embedding_and_emotion(face_img):
    face_tensor = torch.tensor(face_img).permute(2, 0, 1).unsqueeze(0).float() / 255
    face_tensor = (face_tensor - 0.5) / 0.5
    face_tensor = face_tensor.to(device)
    with torch.no_grad():
        embedding = model(face_tensor)

    emotions = emotion_detector.detect_emotions(face_img)
    if emotions:
        emotion_dict = emotions[0]['emotions']
    else:
        emotion_dict = {e: 0 for e in ['angry', 'disgust', 'fear', 'happy', 'sad', 'surprise', 'neutral']}

    return embedding.cpu().numpy().flatten(), emotion_dict

def alignFace(img):
    img_raw = img.copy()
    results = face_mesh.process(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
    if not results.multi_face_landmarks:
        return None
    landmarks = results.multi_face_landmarks[0].landmark
    left_eye = np.array([[landmarks[33].x, landmarks[33].y], [landmarks[160].x, landmarks[160].y],
                         [landmarks[158].x, landmarks[158].y], [landmarks[144].x, landmarks[144].y],
                         [landmarks[153].x, landmarks[153].y], [landmarks[145].x, landmarks[145].y]])
    right_eye = np.array([[landmarks[362].x, landmarks[362].y], [landmarks[385].x, landmarks[385].y],
                          [landmarks[387].x, landmarks[387].y], [landmarks[263].x, landmarks[263].y],
                          [landmarks[373].x, landmarks[373].y], [landmarks[380].x, landmarks[380].y]])
    left_eye_center = left_eye.mean(axis=0).astype(np.int32)
    right_eye_center = right_eye.mean(axis=0).astype(np.int32)
    dY = right_eye_center[1] - left_eye_center[1]
    dX = right_eye_center[0] - left_eye_center[0]
    angle = np.degrees(np.arctan2(dY, dX))
    desired_angle = 0
    angle_diff = desired_angle - angle
    height, width = img_raw.shape[:2]
    center = (width // 2, height // 2)
    rotation_matrix = cv2.getRotationMatrix2D(center, angle_diff, 1)
    new_img = cv2.warpAffine(img_raw, rotation_matrix, (width, height))
    return new_img

def extract_frames(video_path, output_folder, desired_fps, progress_callback=None):
    os.makedirs(output_folder, exist_ok=True)
    clip = VideoFileClip(video_path)
    original_fps = clip.fps
    duration = clip.duration
    total_frames = int(duration * original_fps)
    step = max(1, original_fps / desired_fps)
    total_frames_to_extract = int(total_frames / step)

    frame_count = 0
    for t in np.arange(0, duration, step / original_fps):
        frame = clip.get_frame(t)
        img = Image.fromarray(frame)
        img.save(os.path.join(output_folder, f"frame_{frame_count:04d}.jpg"))
        frame_count += 1
        if progress_callback:
            progress = min(100, (frame_count / total_frames_to_extract) * 100)
            progress_callback(progress, f"Extracting frame")
        if frame_count >= total_frames_to_extract:
            break
    clip.close()
    return frame_count, original_fps

def process_frames(frames_folder, aligned_faces_folder, frame_count, progress, batch_size):
    embeddings_by_frame = {}
    emotions_by_frame = {}
    aligned_face_paths = []
    frame_files = sorted([f for f in os.listdir(frames_folder) if f.endswith('.jpg')])

    for i in range(0, len(frame_files), batch_size):
        batch_files = frame_files[i:i + batch_size]
        batch_frames = []
        batch_nums = []

        for frame_file in batch_files:
            frame_num = int(frame_file.split('_')[1].split('.')[0])
            frame_path = os.path.join(frames_folder, frame_file)
            frame = cv2.imread(frame_path)
            if frame is not None:
                batch_frames.append(frame)
                batch_nums.append(frame_num)

        if batch_frames:
            batch_boxes, batch_probs = mtcnn.detect(batch_frames)

            for j, (frame, frame_num, boxes, probs) in enumerate(
                    zip(batch_frames, batch_nums, batch_boxes, batch_probs)):
                if boxes is not None and len(boxes) > 0 and probs[0] >= 0.99:
                    x1, y1, x2, y2 = [int(b) for b in boxes[0]]
                    face = frame[y1:y2, x1:x2]
                    if face.size > 0:
                        aligned_face = alignFace(face)
                        if aligned_face is not None:
                            aligned_face_resized = cv2.resize(aligned_face, (160, 160))
                            output_path = os.path.join(aligned_faces_folder, f"frame_{frame_num}_face.jpg")
                            cv2.imwrite(output_path, aligned_face_resized)
                            aligned_face_paths.append(output_path)
                            embedding, emotion = get_face_embedding_and_emotion(aligned_face_resized)
                            embeddings_by_frame[frame_num] = embedding
                            emotions_by_frame[frame_num] = emotion

        progress((i + len(batch_files)) / frame_count,
                 f"Processing frames {i + 1} to {min(i + len(batch_files), frame_count)} of {frame_count}")

    return embeddings_by_frame, emotions_by_frame, aligned_face_paths


def cluster_faces(embeddings):
    if len(embeddings) < 2:
        print("Not enough faces for clustering. Assigning all to one cluster.")
        return np.zeros(len(embeddings), dtype=int)

    X = np.stack(embeddings)

    dbscan = DBSCAN(eps=0.5, min_samples=5, metric='cosine')
    clusters = dbscan.fit_predict(X)

    if np.all(clusters == -1):
        print("DBSCAN assigned all to noise. Considering as one cluster.")
        return np.zeros(len(embeddings), dtype=int)

    return clusters

def organize_faces_by_person(embeddings_by_frame, clusters, aligned_faces_folder, organized_faces_folder):
    for (frame_num, embedding), cluster in zip(embeddings_by_frame.items(), clusters):
        person_folder = os.path.join(organized_faces_folder, f"person_{cluster}")
        os.makedirs(person_folder, exist_ok=True)
        src = os.path.join(aligned_faces_folder, f"frame_{frame_num}_face.jpg")
        dst = os.path.join(person_folder, f"frame_{frame_num}_face.jpg")
        shutil.copy(src, dst)

def save_person_data_to_csv(embeddings_by_frame, emotions_by_frame, clusters, desired_fps, original_fps, output_folder,
                            num_components, video_duration):
    emotions = ['angry', 'disgust', 'fear', 'happy', 'sad', 'surprise', 'neutral']
    person_data = {}

    for (frame_num, embedding), (_, emotion_dict), cluster in zip(embeddings_by_frame.items(),
                                                                  emotions_by_frame.items(), clusters):
        if cluster not in person_data:
            person_data[cluster] = []
        person_data[cluster].append((frame_num, embedding, {e: emotion_dict[e] for e in emotions}))

    largest_cluster = max(person_data, key=lambda k: len(person_data[k]))

    data = person_data[largest_cluster]
    data.sort(key=lambda x: x[0])
    frames, embeddings, emotions_data = zip(*data)

    embeddings_array = np.array(embeddings)
    np.save(os.path.join(output_folder, 'face_embeddings.npy'), embeddings_array)

    reducer = umap.UMAP(n_components=num_components, random_state=1)
    embeddings_reduced = reducer.fit_transform(embeddings)

    scaler = MinMaxScaler(feature_range=(0, 1))
    embeddings_reduced_normalized = scaler.fit_transform(embeddings_reduced)

    total_frames = max(frames)
    timecodes = [frame_to_timecode(frame, total_frames, video_duration) for frame in frames]
    times_in_minutes = [frame / total_frames * video_duration / 60 for frame in frames]

    df_data = {
        'Frame': frames,
        'Timecode': timecodes,
        'Time (Minutes)': times_in_minutes,
        'Embedding_Index': range(len(embeddings))
    }

    for i in range(num_components):
        df_data[f'Comp {i + 1}'] = embeddings_reduced_normalized[:, i]

    for emotion in emotions:
        df_data[emotion] = [e[emotion] for e in emotions_data]

    df = pd.DataFrame(df_data)

    return df, largest_cluster

def determine_optimal_anomalies(anomaly_scores, z_threshold=3):
    mean = np.mean(anomaly_scores)
    std = np.std(anomaly_scores)
    threshold = mean + z_threshold * std
    anomalies = anomaly_scores > threshold
    return anomalies, np.where(anomalies)[0]

def timecode_to_seconds(timecode):
    h, m, s = map(float, timecode.split(':'))
    return h * 3600 + m * 60 + s

def group_similar_timecodes(timecodes, scores, threshold_seconds=5):
    grouped = []
    current_group = []

    for i, (timecode, score) in enumerate(zip(timecodes, scores)):
        if not current_group or abs(
                timecode_to_seconds(timecode) - timecode_to_seconds(current_group[0][0])) <= threshold_seconds:
            current_group.append((timecode, score, i))
        else:
            grouped.append(current_group)
            current_group = [(timecode, score, i)]

    if current_group:
        grouped.append(current_group)

    return grouped

class LSTMAutoencoder(nn.Module):
    def __init__(self, input_size, hidden_size=64, num_layers=2):
        super(LSTMAutoencoder, self).__init__()
        self.input_size = input_size
        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, input_size)

    def forward(self, x):
        outputs, (hidden, _) = self.lstm(x)
        out = self.fc(outputs)
        return out

def lstm_anomaly_detection(X, feature_columns, epochs=100, batch_size=64):
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    X = torch.FloatTensor(X).to(device)
    if X.dim() == 2:
        X = X.unsqueeze(0)
    elif X.dim() == 1:
        X = X.unsqueeze(0).unsqueeze(2)
    elif X.dim() > 3:
        raise ValueError(f"Input X should be 1D, 2D or 3D, but got {X.dim()} dimensions")

    print(f"X shape after reshaping: {X.shape}")

    train_size = int(0.85 * X.shape[1])
    X_train, X_val = X[:, :train_size, :], X[:, train_size:, :]

    model = LSTMAutoencoder(input_size=X.shape[2]).to(device)
    criterion = nn.MSELoss()
    optimizer = optim.Adam(model.parameters())

    for epoch in range(epochs):
        model.train()
        optimizer.zero_grad()
        output_train = model(X_train)
        loss_train = criterion(output_train, X_train.squeeze(0))
        loss_train.backward()
        optimizer.step()

        model.eval()
        with torch.no_grad():
            output_val = model(X_val)
            loss_val = criterion(output_val, X_val.squeeze(0))

    model.eval()
    with torch.no_grad():
        reconstructed = model(X).squeeze(0).cpu().numpy()

    mse_all = np.mean(np.power(X.squeeze(0).cpu().numpy() - reconstructed, 2), axis=1)
    anomalies_all, top_indices_all = determine_optimal_anomalies(mse_all)

    component_columns = [col for col in feature_columns if col.startswith('Comp')]
    component_indices = [feature_columns.index(col) for col in component_columns]

    if len(component_indices) > 0:
        mse_comp = np.mean(
            np.power(X.squeeze(0).cpu().numpy()[:, component_indices] - reconstructed[:, component_indices], 2), axis=1)
    else:
        mse_comp = mse_all

    anomalies_comp, top_indices_comp = determine_optimal_anomalies(mse_comp)

    return (anomalies_all, mse_all, top_indices_all,
            anomalies_comp, mse_comp, top_indices_comp,
            model)

def emotion_anomaly_detection(emotion_data, epochs=100, batch_size=64):
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    X = torch.FloatTensor(emotion_data.values).to(device)
    if X.dim() == 1:
        X = X.unsqueeze(0).unsqueeze(2)  # Add batch and feature dimensions
    elif X.dim() == 2:
        X = X.unsqueeze(0)  # Add batch dimension

    model = LSTMAutoencoder(input_size=1).to(device)
    criterion = nn.MSELoss()
    optimizer = optim.Adam(model.parameters())

    for epoch in range(epochs):
        model.train()
        optimizer.zero_grad()
        output = model(X)
        loss = criterion(output, X)
        loss.backward()
        optimizer.step()

    model.eval()
    with torch.no_grad():
        reconstructed = model(X).squeeze(0).cpu().numpy()

    mse = np.mean(np.power(X.squeeze(0).cpu().numpy() - reconstructed, 2), axis=1)
    anomalies, top_indices = determine_optimal_anomalies(mse)

    return anomalies, mse, top_indices

def normalize_scores(scores):
    min_score = np.min(scores)
    max_score = np.max(scores)
    if max_score == min_score:
        return np.full_like(scores, 100)
    return ((scores - min_score) / (max_score - min_score)) * 100

def plot_to_image(fig):
    buf = io.BytesIO()
    fig.savefig(buf, format='png', dpi=300, bbox_inches='tight')
    buf.seek(0)
    return buf

def plot_anomaly_scores(df, anomaly_scores, top_indices, title, timecodes):
    plt.figure(figsize=(16, 8), dpi=300)
    fig, ax = plt.subplots(figsize=(16, 8))

    df['Seconds'] = df['Timecode'].apply(
        lambda x: sum(float(t) * 60 ** i for i, t in enumerate(reversed(x.split(':')))))

    normalized_scores = normalize_scores(anomaly_scores)

    seconds = df['Seconds'].values[1:]
    scores = normalized_scores[1:]

    ax.scatter(seconds, scores, color='blue', alpha=0.7, s=10)

    top_indices = [idx for idx in top_indices if idx > 0]
    ax.scatter(df['Seconds'].iloc[top_indices], normalized_scores[top_indices], color='red', s=50, zorder=5)

    # Calculate and plot baseline
    non_anomalous_scores = np.delete(normalized_scores, top_indices)
    baseline = np.mean(non_anomalous_scores)
    ax.axhline(y=baseline, color='black', linestyle='--', linewidth=2.5)
    ax.text(df['Seconds'].max(), baseline, f'Baseline ({baseline:.2f})',
            verticalalignment='bottom', horizontalalignment='right', color='black')

    grouped_timecodes = group_similar_timecodes([df['Timecode'].iloc[idx] for idx in top_indices],
                                                normalized_scores[top_indices])

    for group in grouped_timecodes:
        max_score_idx = max(range(len(group)), key=lambda i: group[i][1])
        timecode, score, idx = group[max_score_idx]
        ax.annotate(timecode,
                    (df['Seconds'].iloc[top_indices[idx]], score),
                    xytext=(5, 5), textcoords='offset points',
                    fontsize=6, color='red')

    max_seconds = df['Seconds'].max()
    ax.set_xlim(0, max_seconds)
    num_ticks = 100
    ax.set_xticks(np.linspace(0, max_seconds, num_ticks))
    ax.set_xticklabels([f"{int(x // 60):02d}:{int(x % 60):02d}" for x in ax.get_xticks()],
                       rotation=90, ha='center', va='top')

    ax.set_xlabel('Time')
    ax.set_ylabel('Anomaly Score')
    ax.set_title(title)

    ax.grid(True, linestyle='--', alpha=0.7)
    plt.tight_layout()
    plt.close()
    return fig

def plot_emotion(df, emotion, anomaly_scores, top_indices, color, timecodes):
    plt.figure(figsize=(16, 8), dpi=300)
    fig, ax = plt.subplots(figsize=(16, 8))

    df['Seconds'] = df['Timecode'].apply(
        lambda x: sum(float(t) * 60 ** i for i, t in enumerate(reversed(x.split(':')))))

    seconds = df['Seconds'].values[1:]
    scores = anomaly_scores[1:]

    ax.scatter(seconds, scores, color=color, alpha=0.7, s=10)

    top_indices = [idx for idx in top_indices if idx > 0]
    ax.scatter(df['Seconds'].iloc[top_indices], anomaly_scores[top_indices], color='red', s=50, zorder=5)

    # Calculate and plot baseline
    non_anomalous_scores = np.delete(anomaly_scores, top_indices)
    baseline = np.mean(non_anomalous_scores)
    ax.axhline(y=baseline, color='black', linestyle='--', linewidth=2.5)
    ax.text(df['Seconds'].max(), baseline, f'Baseline ({baseline:.2f})',
            verticalalignment='bottom', horizontalalignment='right', color='black')


    grouped_timecodes = group_similar_timecodes([df['Timecode'].iloc[idx] for idx in top_indices],
                                                anomaly_scores[top_indices])

    for group in grouped_timecodes:
        max_score_idx = max(range(len(group)), key=lambda i: group[i][1])
        timecode, score, idx = group[max_score_idx]
        ax.annotate(timecode,
                    (df['Seconds'].iloc[top_indices[idx]], score),
                    xytext=(5, 5), textcoords='offset points',
                    fontsize=6, color='red')

    max_seconds = df['Seconds'].max()
    ax.set_xlim(0, max_seconds)
    num_ticks = 100
    ax.set_xticks(np.linspace(0, max_seconds, num_ticks))
    ax.set_xticklabels([f"{int(x // 60):02d}:{int(x % 60):02d}" for x in ax.get_xticks()],
                       rotation=90, ha='center', va='top')

    ax.set_xlabel('Time')
    ax.set_ylabel(f'{emotion.capitalize()} Anomaly Score')
    ax.set_title(f'{emotion.capitalize()} Anomaly Scores')

    ax.grid(True, linestyle='--', alpha=0.7)
    plt.tight_layout()
    plt.close()
    return fig

def get_random_face_samples(organized_faces_folder, output_folder, largest_cluster, num_samples=100):
    face_samples = []
    for cluster_folder in os.listdir(organized_faces_folder):
        if cluster_folder.startswith("person_"):
            person_folder = os.path.join(organized_faces_folder, cluster_folder)
            face_files = [f for f in os.listdir(person_folder) if f.endswith('.jpg')]
            if face_files:
                if int(cluster_folder.split('_')[1]) == largest_cluster:
                    samples = np.random.choice(face_files, min(num_samples, len(face_files)), replace=False)
                else:
                    samples = [np.random.choice(face_files)]

                for i, sample in enumerate(samples):
                    face_path = os.path.join(person_folder, sample)
                    output_path = os.path.join(output_folder, f"face_sample_{cluster_folder}_{i}.jpg")
                    face_img = cv2.imread(face_path)
                    if face_img is not None:
                        small_face = cv2.resize(face_img, (160, 160))
                        cv2.imwrite(output_path, small_face)
                        face_samples.append(output_path)
    return face_samples

def process_video(video_path, num_components, desired_fps, batch_size, progress=gr.Progress()):
    output_folder = "output"
    os.makedirs(output_folder, exist_ok=True)

    with tempfile.TemporaryDirectory() as temp_dir:
        aligned_faces_folder = os.path.join(temp_dir, 'aligned_faces')
        organized_faces_folder = os.path.join(temp_dir, 'organized_faces')
        os.makedirs(aligned_faces_folder, exist_ok=True)
        os.makedirs(organized_faces_folder, exist_ok=True)

        clip = VideoFileClip(video_path)
        video_duration = clip.duration
        clip.close()

        progress(0, "Starting frame extraction")
        frames_folder = os.path.join(temp_dir, 'extracted_frames')

        def extraction_progress(percent, message):
            progress(percent / 100, f"Extracting frames")

        frame_count, original_fps = extract_frames(video_path, frames_folder, desired_fps, extraction_progress)

        progress(1, "Frame extraction complete")
        progress(0.3, "Processing frames")
        embeddings_by_frame, emotions_by_frame, aligned_face_paths = process_frames(frames_folder, aligned_faces_folder,
                                                                                    frame_count,
                                                                                    progress, batch_size)

        if not aligned_face_paths:
            return ("No faces were extracted from the video.",
                    None, None, None, None, None, None, None, None, None)

        progress(0.6, "Clustering faces")
        embeddings = [embedding for _, embedding in embeddings_by_frame.items()]
        clusters = cluster_faces(embeddings)
        num_clusters = len(set(clusters))  # Get the number of unique clusters

        progress(0.7, "Organizing faces")
        organize_faces_by_person(embeddings_by_frame, clusters, aligned_faces_folder, organized_faces_folder)

        progress(0.8, "Saving person data")
        df, largest_cluster = save_person_data_to_csv(embeddings_by_frame, emotions_by_frame, clusters, desired_fps,
                                                      original_fps, temp_dir, num_components, video_duration)

        progress(0.85, "Getting face samples")
        face_samples = get_random_face_samples(organized_faces_folder, output_folder, largest_cluster)

        progress(0.9, "Performing anomaly detection")
        feature_columns = [col for col in df.columns if
                           col not in ['Frame', 'Timecode', 'Time (Minutes)', 'Embedding_Index']]
        X = df[feature_columns].values

        try:
            anomalies_all, anomaly_scores_all, top_indices_all, anomalies_comp, anomaly_scores_comp, top_indices_comp, _ = lstm_anomaly_detection(
                X, feature_columns, batch_size=batch_size)

            anomaly_scores_all = normalize_scores(anomaly_scores_all)
            anomaly_scores_comp = normalize_scores(anomaly_scores_comp)

            emotion_anomalies = {}
            for emotion in ['fear', 'sad', 'angry', 'happy', 'surprise', 'neutral']:
                anomalies, scores, indices = emotion_anomaly_detection(df[emotion])
                emotion_anomalies[emotion] = {
                    'anomalies': anomalies,
                    'scores': normalize_scores(scores),
                    'indices': indices
                }

        except Exception as e:
            print(f"Error details: {str(e)}")
            return f"Error in anomaly detection: {str(e)}", None, None, None, None, None, None, None, None, None

        progress(0.95, "Generating plots")
        try:
            anomaly_plot_all = plot_anomaly_scores(df, anomaly_scores_all, top_indices_all,
                                                   "Facial Features + Emotions",
                                                   df['Timecode'].iloc[top_indices_all].values)
            anomaly_plot_comp = plot_anomaly_scores(df, anomaly_scores_comp, top_indices_comp, "Facial Features",
                                                    df['Timecode'].iloc[top_indices_comp].values)
            emotion_plots = [
                plot_emotion(df, emotion,
                             emotion_anomalies[emotion]['scores'],
                             emotion_anomalies[emotion]['indices'],
                             color,
                             df['Timecode'].iloc[emotion_anomalies[emotion]['indices']].values)
                for emotion, color in zip(['fear', 'sad', 'angry', 'happy', 'surprise', 'neutral'],
                                          ['purple', 'green', 'orange', 'darkblue', 'gold', 'grey'])
            ]
        except Exception as e:
            return f"Error generating plots: {str(e)}", None, None, None, None, None, None, None, None, None

        progress(1.0, "Preparing results")
        results = f"Number of persons/clusters detected: {num_clusters}\n\n"
        results += f"Breakdown of persons/clusters:\n"
        for cluster_id in range(num_clusters):
            results += f"Person/Cluster {cluster_id + 1}: {len([c for c in clusters if c == cluster_id])} frames\n"
        results += f"\nAnomalies (Facial Features + Emotions):\n"
        results += "\n".join([f"{score:.2f} at {timecode}" for score, timecode in
                              zip(anomaly_scores_all[top_indices_all[1:]],
                                  df['Timecode'].iloc[top_indices_all[1:]].values)])
        results += f"\n\nAnomalies (Facial Features):\n"
        results += "\n".join([f"{score:.2f} at {timecode}" for score, timecode in
                              zip(anomaly_scores_comp[top_indices_comp[1:]],
                                  df['Timecode'].iloc[top_indices_comp[1:]].values)])

        for emotion in ['fear', 'sad', 'angry', 'happy', 'surprise', 'neutral']:
            results += f"\n\n{emotion.capitalize()} Anomalies:\n"
            results += "\n".join([f"{emotion_anomalies[emotion]['scores'][i]:.2f} at {df['Timecode'].iloc[i]}"
                                  for i in emotion_anomalies[emotion]['indices'] if i > 0])

        return (
            results,
            anomaly_plot_all,
            anomaly_plot_comp,
            *emotion_plots,
            face_samples
        )

iface = gr.Interface(
    fn=process_video,
    inputs=[
        gr.Video(),
        gr.Slider(minimum=1, maximum=20, step=1, value=10, label="Number of Components"),
        gr.Slider(minimum=1, maximum=20, step=1, value=15, label="Desired FPS"),
        gr.Slider(minimum=1, maximum=32, step=1, value=8, label="Batch Size")
    ],
    outputs=[
        gr.Textbox(label="Anomaly Detection Results"),
        gr.Plot(label="Anomaly Scores (Facial Features + Emotions)"),
        gr.Plot(label="Anomaly Scores (Facial Features)"),
        gr.Plot(label="Fear Anomalies"),
        gr.Plot(label="Sad Anomalies"),
        gr.Plot(label="Angry Anomalies"),
        gr.Plot(label="Happy Anomalies"),
        gr.Plot(label="Surprise Anomalies"),
        gr.Plot(label="Neutral Anomalies"),
        gr.Gallery(label="Random Samples of Detected Persons", columns=[5], rows=[2], height="auto")
    ],
    title="Facial Expressions Anomaly Detection",
    description="""
        This application detects anomalies in facial expressions and emotions from a video input. 
        It identifies distinct persons in the video and provides sample faces for each, with 10 samples for the most frequent person.

        Adjust the parameters as needed:
        - Number of Components: Complexity of the facial expression model
        - Desired FPS: Frames per second to analyze (lower for faster processing)
        - Batch Size: Affects processing speed and memory usage

        Click on any graph to enlarge it.
        """,
    allow_flagging="never"
)

if __name__ == "__main__":
    iface.launch()