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()