app.py

import gradio as gr
import time
from video_processing import process_video
from PIL import Image
import matplotlib
import threading

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

def process_and_show_completion(video_input_path, anomaly_threshold_input, fps, progress=gr.Progress()):
    global processing
    processing = True
    start_time = time.time()
    
    try:
        print("Starting video processing...")
        results = process_video(video_input_path, anomaly_threshold_input, fps, progress=progress)
        print("Video processing completed.")

        if isinstance(results[0], str) and results[0].startswith("Error"):
            print(f"Error occurred: {results[0]}")
            processing = False
            return [results[0]] + [None] * 27

        exec_time, results_summary, df, mse_embeddings, mse_posture, mse_voice, \
            mse_plot_embeddings, mse_plot_posture, mse_plot_voice, \
            mse_histogram_embeddings, mse_histogram_posture, mse_histogram_voice, \
            mse_heatmap_embeddings, mse_heatmap_posture, mse_heatmap_voice, \
            face_samples_frequent, \
            anomaly_faces_embeddings, anomaly_frames_posture_images, \
            aligned_faces_folder, frames_folder, \
            heatmap_video_path, combined_mse_plot, correlation_heatmap = results

        anomaly_faces_embeddings_pil = [Image.fromarray(face) for face in anomaly_faces_embeddings] if anomaly_faces_embeddings is not None else []
        anomaly_frames_posture_pil = [Image.fromarray(frame) for frame in anomaly_frames_posture_images] if anomaly_frames_posture_images is not None else []

        face_samples_frequent = [Image.open(path) for path in face_samples_frequent] if face_samples_frequent is not None else []

        end_time = time.time()
        total_exec_time = end_time - start_time

        output = [
            f"Total execution time: {total_exec_time:.2f} seconds", results_summary,
            df, mse_embeddings, mse_posture, mse_voice,
            mse_plot_embeddings, mse_plot_posture, mse_plot_voice,
            mse_histogram_embeddings, mse_histogram_posture, mse_histogram_voice,
            mse_heatmap_embeddings, mse_heatmap_posture, mse_heatmap_voice,
            anomaly_faces_embeddings_pil, anomaly_frames_posture_pil,
            face_samples_frequent,
            aligned_faces_folder, frames_folder,
            mse_embeddings, mse_posture, mse_voice,
            heatmap_video_path, combined_mse_plot, correlation_heatmap
        ]

        processing = False
        return output

    except Exception as e:
        error_message = f"An error occurred: {str(e)}"
        print(error_message)
        import traceback
        traceback.print_exc()
        processing = False
        return [error_message] + [None] * 27

def show_results():
    return [gr.update(visible=True) for _ in range(4)]

def start_execution_timer():
    return gr.update(visible=True), gr.update(visible=False)

def update_execution_time():
    current_time = time.time() - start_time
    return f"Execution time: {current_time:.2f} seconds"

processing = False
start_time = 0

with gr.Blocks() as iface:
    with gr.Row():
        video_input = gr.Video(label="Input Video")

    anomaly_threshold = gr.Slider(minimum=1, maximum=5, step=0.1, value=3, label="Anomaly Detection Threshold (Standard deviation)")
    fps_slider = gr.Slider(minimum=5, maximum=20, step=1, value=10, label="Frames Per Second (FPS)")
    process_btn = gr.Button("Detect Anomalies")
    progress_bar = gr.Progress()

    with gr.Tabs() as all_tabs:
        description_tab = gr.Tab("Description")
        with description_tab:
            description_md = gr.Markdown("""
            # Multimodal Behavioral Anomalies Detection

            This tool detects anomalies in facial expressions, body language, and voice over the timeline of a video.
            It extracts faces, postures, and voice from video frames, and analyzes them to identify anomalies using time series analysis and a variational autoencoder (VAE) approach.
            """)
            execution_time_md = gr.Markdown(visible=False)

        facial_features_tab = gr.Tab("Facial Features", visible=False)
        with facial_features_tab:
            results_text = gr.TextArea(label="Faces Breakdown", lines=5)
            mse_features_plot = gr.Plot(label="MSE: Facial Features")
            mse_features_hist = gr.Plot(label="MSE Distribution: Facial Features")
            mse_features_heatmap = gr.Plot(label="MSE Heatmap: Facial Features")
            anomaly_frames_features = gr.Gallery(label="Anomaly Frames (Facial Features)", columns=6, rows=2, height="auto")
            face_samples_most_frequent = gr.Gallery(label="Most Frequent Person Samples", columns=10, rows=2, height="auto")

        body_posture_tab = gr.Tab("Body Posture", visible=False)
        with body_posture_tab:
            mse_posture_plot = gr.Plot(label="MSE: Body Posture")
            mse_posture_hist = gr.Plot(label="MSE Distribution: Body Posture")
            mse_posture_heatmap = gr.Plot(label="MSE Heatmap: Body Posture")
            anomaly_frames_posture = gr.Gallery(label="Anomaly Frames (Body Posture)", columns=6, rows=2, height="auto")

        voice_tab = gr.Tab("Voice", visible=False)
        with voice_tab:
            mse_voice_plot = gr.Plot(label="MSE: Voice")
            mse_voice_hist = gr.Plot(label="MSE Distribution: Voice")
            mse_voice_heatmap = gr.Plot(label="MSE Heatmap: Voice")

        combined_tab = gr.Tab("Combined", visible=False)
        with combined_tab:
            heatmap_video = gr.Video(label="Video with Anomaly Heatmap")
            combined_mse_plot = gr.Plot(label="Combined MSE Plot")
            correlation_heatmap_plot = gr.Plot(label="Correlation Heatmap")
                
    df_store = gr.State()
    mse_features_store = gr.State()
    mse_posture_store = gr.State()
    mse_voice_store = gr.State()
    aligned_faces_folder_store = gr.State()
    frames_folder_store = gr.State()
    mse_heatmap_embeddings_store = gr.State()
    mse_heatmap_posture_store = gr.State()
    mse_heatmap_voice_store = gr.State()
    
    def start_processing():
        global start_time, processing
        start_time = time.time()
        processing = True

    process_btn.click(
        start_processing,
        inputs=None,
        outputs=None
    ).then(
        start_execution_timer,
        inputs=None,
        outputs=[execution_time_md, description_md]
    ).then(
        process_and_show_completion,
        inputs=[video_input, anomaly_threshold, fps_slider],
        outputs=[
            execution_time_md, results_text, df_store,
            mse_features_store, mse_posture_store, mse_voice_store,
            mse_features_plot, mse_posture_plot, mse_voice_plot,
            mse_features_hist, mse_posture_hist, mse_voice_hist,
            mse_features_heatmap, mse_posture_heatmap, mse_voice_heatmap,
            anomaly_frames_features, anomaly_frames_posture,
            face_samples_most_frequent,
            aligned_faces_folder_store, frames_folder_store,
            mse_heatmap_embeddings_store, mse_heatmap_posture_store, mse_heatmap_voice_store,
            heatmap_video, combined_mse_plot, correlation_heatmap_plot
        ]
    ).then(
        show_results,
        inputs=None,
        outputs=all_tabs.children[1:]
    )

    execution_time_md.change(
        update_execution_time,
        inputs=None,
        outputs=execution_time_md,
        every=0.1
    )

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