""" --- title: Video Anomaly Detector emoji: 🎥 colorFrom: blue colorTo: green sdk: streamlit sdk_version: 1.31.0 app_file: app.py pinned: false license: mit --- """ import streamlit as st import os import tempfile import time from detector import VideoAnomalyDetector import cv2 from PIL import Image import numpy as np from dotenv import load_dotenv import streamlit.components.v1 as components import json import base64 from io import BytesIO import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.image import MIMEImage import requests import re # Custom JSON encoder to handle numpy arrays and other non-serializable types class NumpyEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, np.ndarray): # Convert numpy arrays to base64 encoded strings pil_img = Image.fromarray(obj) buffered = BytesIO() pil_img.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode("utf-8") return {"__ndarray__": img_str} return super(NumpyEncoder, self).default(obj) def send_email_notification(to_email, subject, body, image=None): """Send email notification with optional image attachment""" try: # Get email credentials from environment variables smtp_server = os.getenv("SMTP_SERVER", "smtp.gmail.com") smtp_port = int(os.getenv("SMTP_PORT", "1587")) smtp_username = os.getenv("SMTP_USERNAME") smtp_password = os.getenv("SMTP_PASSWORD") if not smtp_username or not smtp_password: st.warning("Email notification failed: SMTP credentials not configured. Please set SMTP_USERNAME and SMTP_PASSWORD environment variables.") return False # Create message msg = MIMEMultipart() msg['From'] = smtp_username msg['To'] = to_email msg['Subject'] = subject # Attach text msg.attach(MIMEText(body, 'plain')) # Attach image if provided if image is not None: # Convert numpy array to image if isinstance(image, np.ndarray): pil_img = Image.fromarray(image) img_byte_arr = BytesIO() pil_img.save(img_byte_arr, format='PNG') img_data = img_byte_arr.getvalue() else: # Assume it's already bytes img_data = image img_attachment = MIMEImage(img_data) img_attachment.add_header('Content-Disposition', 'attachment', filename='anomaly.png') msg.attach(img_attachment) # Connect to server and send server = smtplib.SMTP(smtp_server, smtp_port) server.starttls() server.login(smtp_username, smtp_password) server.send_message(msg) server.quit() return True except Exception as e: st.warning(f"Email notification failed: {str(e)}") return False def send_whatsapp_notification(to_number, message): """Send WhatsApp notification using WhatsApp Business API""" try: # Get WhatsApp API credentials from environment variables whatsapp_api_key = os.getenv("WHATSAPP_API_KEY") whatsapp_phone_id = os.getenv("WHATSAPP_PHONE_ID") if not whatsapp_api_key or not whatsapp_phone_id: st.warning("WhatsApp notification failed: API credentials not configured. Please set WHATSAPP_API_KEY and WHATSAPP_PHONE_ID environment variables.") return False # For demonstration purposes, we'll show how to use the WhatsApp Business API # In a real implementation, you would need to set up a WhatsApp Business account # and use their official API # Example using WhatsApp Business API url = f"https://graph.facebook.com/v17.0/{whatsapp_phone_id}/messages" headers = { "Authorization": f"Bearer {whatsapp_api_key}", "Content-Type": "application/json" } data = { "messaging_product": "whatsapp", "to": to_number, "type": "text", "text": { "body": message } } # For demonstration, we'll just log the request instead of actually sending it print(f"Would send WhatsApp message to {to_number}: {message}") # In a real implementation, you would uncomment this: # response = requests.post(url, headers=headers, json=data) # return response.status_code == 200 return True except Exception as e: st.warning(f"WhatsApp notification failed: {str(e)}") return False # Helper functions for notifications def validate_email(email): """Validate email format""" pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$' return re.match(pattern, email) is not None def validate_phone(phone): """Validate phone number format (should include country code)""" pattern = r'^\+\d{1,3}\d{6,14}$' return re.match(pattern, phone) is not None def send_notification(notification_type, contact, message, image=None): """Send notification based on type""" if notification_type == "email": if validate_email(contact): return send_email_notification( contact, "Anomaly Detected - Video Anomaly Detector", message, image ) else: st.warning("Invalid email format. Notification not sent.") return False elif notification_type == "whatsapp": if validate_phone(contact): return send_whatsapp_notification(contact, message) else: st.warning("Invalid phone number format. Please include country code (e.g., +1234567890). Notification not sent.") return False return False # Helper functions for displaying results def display_single_result(result): """Display a single analysis result""" if isinstance(result, dict): # This is a single frame result or cumulative result if "anomaly_detected" in result: # Create columns for image and text if "frame" in result: col1, col2 = st.columns([1, 2]) with col1: st.image(result["frame"], caption="Captured Frame", use_column_width=True) with col2: anomaly_detected = result["anomaly_detected"] # Start building the HTML content html_content = f"""
""" # Add confidence if available if "confidence" in result: html_content += f"

Confidence: {result['confidence']}%

" # Add analysis/text if available (check multiple possible keys) analysis_text = None for key in ["analysis", "text", "description"]: if key in result and result[key]: analysis_text = result[key] break if analysis_text: html_content += f"

Analysis: {analysis_text}

" # Add anomaly type if available if "anomaly_type" in result and result["anomaly_type"]: html_content += f"

Anomaly Type: {result['anomaly_type']}

" # Close the div html_content += "
" # Display the HTML content st.markdown(html_content, unsafe_allow_html=True) else: # No frame available, just show the text # Start building the HTML content html_content = "
" # Add confidence if available if "confidence" in result: html_content += f"

Confidence: {result['confidence']}%

" # Add analysis/text if available (check multiple possible keys) analysis_text = None for key in ["analysis", "text", "description"]: if key in result and result[key]: analysis_text = result[key] break if analysis_text: html_content += f"

Analysis: {analysis_text}

" # Add anomaly type if available if "anomaly_type" in result and result["anomaly_type"]: html_content += f"

Anomaly Type: {result['anomaly_type']}

" # Close the div html_content += "
" # Display the HTML content st.markdown(html_content, unsafe_allow_html=True) else: # Display other types of results st.json(result) else: # Unknown result type st.write(result) def display_results(results, analysis_depth): """Display analysis results based on analysis depth""" if not results: st.warning("No results to display") return # Add a main results header st.markdown("

📊 Analysis Results

", unsafe_allow_html=True) # Add high-level summary at the top if analysis_depth == "granular": # For granular analysis, check if any frame has an anomaly anomaly_frames = sum(1 for r in results if r.get("anomaly_detected", False)) total_frames = len(results) if anomaly_frames > 0: # Get the anomaly types from frames with anomalies anomaly_types = set(r.get("anomaly_type", "Unknown") for r in results if r.get("anomaly_detected", False)) anomaly_types_str = ", ".join(anomaly_types) st.markdown( f"""

⚠️ ANOMALY DETECTED

Frames with anomalies: {anomaly_frames} out of {total_frames}

Anomaly types: {anomaly_types_str}

""", unsafe_allow_html=True ) else: st.markdown( """

✅ No Anomalies Detected

No anomalies were detected in any of the analyzed frames.

""", unsafe_allow_html=True ) else: # cumulative # For cumulative analysis, check the overall result if results.get("anomaly_detected", False): anomaly_type = results.get("anomaly_type", "Unknown") st.markdown( f"""

⚠️ ANOMALY DETECTED

Anomaly type: {anomaly_type}

""", unsafe_allow_html=True ) else: st.markdown( """

✅ No Anomalies Detected

No anomalies were detected in the video.

""", unsafe_allow_html=True ) # Display detailed results if analysis_depth == "granular": # For granular analysis, results is a list of frame analyses st.markdown("

🔍 Frame-by-Frame Analysis

", unsafe_allow_html=True) # Display detailed view directly without tabs for i, result in enumerate(results): with st.expander(f"Frame {i+1} - {'⚠️ ANOMALY' if result.get('anomaly_detected', False) else '✅ Normal'}"): display_single_result(result) else: # cumulative st.markdown("

🔍 Overall Video Analysis

", unsafe_allow_html=True) display_single_result(results) # Display key frames if available if "frames" in results and results["frames"]: st.markdown("

🖼️ Key Frames

", unsafe_allow_html=True) # Create a row of columns for the frames num_frames = len(results["frames"]) cols = st.columns(min(3, num_frames)) # Display each frame in a column for i, (col, frame) in enumerate(zip(cols, results["frames"])): with col: st.image(frame, caption=f"Key Frame {i+1}", use_column_width=True) # Initialize session state for stop button if 'stop_requested' not in st.session_state: st.session_state.stop_requested = False def request_stop(): st.session_state.stop_requested = True # Conditionally import Phi-4 detector try: from phi4_detector import Phi4AnomalyDetector PHI4_AVAILABLE = True except ImportError: PHI4_AVAILABLE = False # Load environment variables from .env file load_dotenv() # Set page configuration st.set_page_config( page_title="Video Anomaly Detector", page_icon="🔍", layout="wide" ) # Custom CSS for better UI st.markdown(""" """, unsafe_allow_html=True) # Header with icon st.markdown("

🔍 Video Anomaly Detector

", unsafe_allow_html=True) st.markdown("

Analyze video frames for anomalies using advanced AI models

", unsafe_allow_html=True) # Sidebar for inputs with st.sidebar: st.markdown("

⚙️ Settings

", unsafe_allow_html=True) # Input source selection st.markdown("
📹Input Source
", unsafe_allow_html=True) input_source = st.radio( "", ["Video File", "Live Stream"], index=0, help="Select the input source for analysis" ) # File uploader or stream URL based on input source if input_source == "Video File": st.markdown("
📁Upload Video
", unsafe_allow_html=True) # Find sample .mp4 files in the current directory sample_files = [] for file in os.listdir(): if file.endswith('.mp4'): sample_files.append(file) # Show sample files if available if sample_files: st.info(f"Sample videos available: {', '.join(sample_files)}") use_sample = st.checkbox("Use a sample video instead of uploading") if use_sample: selected_sample = st.selectbox("Select a sample video", sample_files) uploaded_file = selected_sample # We'll handle this specially later # Add video preview section st.markdown("

🎬 Video Preview

", unsafe_allow_html=True) # Create a container for the video preview with custom styling st.markdown("
", unsafe_allow_html=True) # Get the full path to the selected sample video video_path = os.path.join(os.getcwd(), selected_sample) # Display the video player st.video(video_path) # Display video information try: cap = cv2.VideoCapture(video_path) if cap.isOpened(): # Get video properties fps = cap.get(cv2.CAP_PROP_FPS) frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) # Calculate duration duration = frame_count / fps if fps > 0 else 0 # Format duration as minutes:seconds minutes = int(duration // 60) seconds = int(duration % 60) duration_str = f"{minutes}:{seconds:02d}" cap.release() except Exception as e: st.warning(f"Could not read video properties: {str(e)}") st.markdown("
", unsafe_allow_html=True) else: uploaded_file = st.file_uploader("", type=["mp4", "avi", "mov", "mkv"]) else: uploaded_file = st.file_uploader("", type=["mp4", "avi", "mov", "mkv"]) stream_source = None else: # Live Stream st.markdown("
🔗Stream Source
", unsafe_allow_html=True) stream_options = ["Webcam", "IP Camera / RTSP Stream"] stream_type = st.selectbox("", stream_options, index=0) if stream_type == "Webcam": stream_source = 0 # Default webcam else: stream_source = st.text_input("Stream URL", placeholder="rtsp://username:password@ip_address:port/path") # Max frames to process for live stream st.markdown("
🔢Frame Capture Settings
", unsafe_allow_html=True) capture_mode = st.radio( "Capture Mode", ["Frame Count Limit", "Time Interval (Continuous)"], index=0, help="Choose how to capture frames from the live stream" ) if capture_mode == "Frame Count Limit": max_frames = st.number_input( "Maximum Frames", min_value=1, max_value=100, value=30, help="Maximum number of frames to process from the live stream" ) time_interval = None else: # Time Interval mode max_frames = None # No frame limit in time interval mode time_interval = st.number_input( "Seconds Between Captures", min_value=1, max_value=60, value=5, help="Capture one frame every X seconds indefinitely" ) st.info("⚠️ In time interval mode, processing will continue indefinitely. Use the Stop button to end capture.") uploaded_file = None # Model selection st.markdown("
🧠AI Model
", unsafe_allow_html=True) # Add Phi-4 to the model options if available model_options = ["GPT-4o", "GPT-4o-mini"] if PHI4_AVAILABLE: model_options.append("Phi-4") model_options.append("Phi-3 (Coming Soon)") model = st.selectbox( "", model_options, index=0, help="Select the AI model to use for analysis" ) # Display model info based on selection if model == "GPT-4o": st.markdown("
Most powerful model with highest accuracy
", unsafe_allow_html=True) model_value = "gpt-4o" use_phi4 = False elif model == "GPT-4o-mini": st.markdown("
Faster and more cost-effective
", unsafe_allow_html=True) model_value = "gpt-4o-mini" use_phi4 = False elif model == "Phi-4": st.markdown("
Microsoft's multimodal model, runs locally
", unsafe_allow_html=True) model_value = "phi-4" use_phi4 = True else: # Phi-3 st.markdown("
Not yet implemented
", unsafe_allow_html=True) model_value = "gpt-4o" # Default to GPT-4o if Phi-3 is selected use_phi4 = False st.warning("Phi-3 support is coming soon. Using GPT-4o instead.") # Skip frames input with icon st.markdown("
⏭️Frame Skip Rate
", unsafe_allow_html=True) skip_frames = st.number_input( "", min_value=0, max_value=100, value=5, help="Higher values process fewer frames, making analysis faster but potentially less accurate" ) # Analysis depth selection st.markdown("
🔬Analysis Depth
", unsafe_allow_html=True) analysis_depth = st.radio( "", ["Granular (Frame by Frame)", "Cumulative (Overall)"], index=0, help="Granular provides analysis for each frame, Cumulative gives an overall assessment" ) # Map the radio button value to the actual value analysis_depth_value = "granular" if analysis_depth == "Granular (Frame by Frame)" else "cumulative" # Notification options st.markdown("
🔔Notifications
", unsafe_allow_html=True) enable_notifications = st.checkbox("Enable notifications for anomaly detection", value=False) if enable_notifications: notification_type = st.radio( "Notification Method", ["Email", "WhatsApp"], index=0, help="Select how you want to be notified when anomalies are detected" ) if notification_type == "Email": notification_email = st.text_input( "Email Address", placeholder="your.email@example.com", help="Enter the email address to receive notifications" ) st.session_state.notification_contact = notification_email if notification_email else None st.session_state.notification_type = "email" if notification_email else None else: # WhatsApp notification_phone = st.text_input( "WhatsApp Number", placeholder="+1234567890 (include country code)", help="Enter your WhatsApp number with country code" ) st.session_state.notification_contact = notification_phone if notification_phone else None st.session_state.notification_type = "whatsapp" if notification_phone else None else: st.session_state.notification_type = None st.session_state.notification_contact = None # Prompt input with icon st.markdown("
💬Anomaly Description
", unsafe_allow_html=True) prompt = st.text_area( "", value="Analyze this frame and describe if there are any unusual or anomalous activities or objects. If you detect anything unusual, explain what it is and why it might be considered an anomaly.", height=150, help="Describe what kind of anomaly to look for" ) # API key input with default from environment variable and icon (only show for OpenAI models) if not use_phi4: st.markdown("
🔑OpenAI API Key
", unsafe_allow_html=True) default_api_key = os.getenv("OPENAI_API_KEY", "") api_key = st.text_input( "", value=default_api_key, type="password", help="Your OpenAI API key with access to the selected model" ) else: # For Phi-4, we don't need an API key api_key = "not-needed-for-phi4" # Submit button with icon submit_button = st.button("🚀 Analyze Video") # Main content area for video file if input_source == "Video File" and uploaded_file is not None: # Display video info st.markdown("

📊 Video Information

", unsafe_allow_html=True) # Check if we're using a sample file or an uploaded file if isinstance(uploaded_file, str) and os.path.exists(uploaded_file): # This is a sample file from the directory video_path = uploaded_file st.success(f"Using sample video: {os.path.basename(video_path)}") else: # This is an uploaded file # Save uploaded file to a temporary file with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as tmp_file: tmp_file.write(uploaded_file.getvalue()) video_path = tmp_file.name # Get video metadata # For video files, use the default backend instead of DirectShow cap = cv2.VideoCapture(video_path) # Don't set MJPG format for video files as it can interfere with proper decoding # cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc('M','J','P','G')) # Try to get video properties fps = cap.get(cv2.CAP_PROP_FPS) frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # Prevent division by zero, but only show warning for live streams # For video files, this is likely an actual error if fps <= 0: # Check if this is a video file (not a webcam/stream) if isinstance(video_path, str) and os.path.exists(video_path): # This is a file that exists but has FPS issues fps = 30.0 # Use a default value st.warning(f"Could not determine frame rate for video file: {os.path.basename(video_path)}. Using default value of 30 FPS.") else: # This is likely a webcam or stream fps = 30.0 st.info("Using default frame rate of 30 FPS for live stream.") duration = frame_count / fps width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) cap.release() # Display video metadata in a nicer format col1, col2, col3 = st.columns(3) with col1: st.markdown("
⏱️
", unsafe_allow_html=True) st.metric("Duration", f"{duration:.2f} seconds") with col2: st.markdown("
🎞️
", unsafe_allow_html=True) st.metric("Total Frames", frame_count) with col3: st.markdown("
📐
", unsafe_allow_html=True) st.metric("Resolution", f"{width}x{height}") # Display estimated frames to process estimated_frames = frame_count // (skip_frames + 1) + 1 st.info(f"With current settings, approximately {estimated_frames} frames will be processed.") # Main content area for live stream elif input_source == "Live Stream" and stream_source is not None: # Display live stream info st.markdown("

📊 Live Stream Information

", unsafe_allow_html=True) # Display stream source info if stream_source == 0: st.info("Using default webcam as the stream source.") else: st.info(f"Using stream URL: {stream_source}") # Display estimated frames to process st.info(f"Will process up to {max_frames} frames with a skip rate of {skip_frames}.") # Show a placeholder for the live stream st.markdown("

Live stream preview will appear here during processing

", unsafe_allow_html=True) # Process video or stream when submit button is clicked if submit_button: if not api_key and not use_phi4: st.error("⚠️ Please enter your OpenAI API key") elif input_source == "Video File" and uploaded_file is None: st.error("⚠️ Please upload a video file") elif input_source == "Live Stream" and stream_source is None: st.error("⚠️ Please provide a valid stream source") else: try: # Initialize detector based on selected model if use_phi4: with st.spinner("Loading Phi-4 model... This may take a while if downloading for the first time."): detector = Phi4AnomalyDetector() st.success("Phi-4 model loaded successfully!") else: detector = VideoAnomalyDetector(api_key, model_value) # Progress bar and status st.markdown("

⏳ Processing Video

", unsafe_allow_html=True) progress_bar = st.progress(0) status_text = st.empty() # Create a callback function to update progress def update_progress(current, total): if total == -1: # Continuous mode status_text.text(f"Processed {current} frames (continuous mode)...") else: # Normal mode with a known total if total > 0: progress = current / total progress_bar.progress(progress) else: # Handle case where total is zero progress_bar.progress(0) status_text.text(f"Processing frame {current+1} of {total if total > 0 else '?'}...") # Process the video or stream start_time = time.time() if input_source == "Video File": results = detector.process_video(video_path, skip_frames, prompt, analysis_depth_value, update_progress) print(f"Results: {results}") # Results will be displayed after processing else: # Live Stream if capture_mode == "Frame Count Limit": # Process with frame count limit (original behavior) results = detector.process_live_stream(stream_source, skip_frames, prompt, analysis_depth_value, max_frames, update_progress) # Results will be displayed after processing else: # Time Interval mode # Create a placeholder for continuous results results_container = st.empty() # Reset stop request flag at the beginning of processing st.session_state.stop_requested = False # Create a stop button outside the loop st.button("Stop Capture", key="stop_continuous_main", on_click=request_stop) # Process with time interval (generator mode) results_generator = detector.process_live_stream( stream_source, skip_frames, prompt, analysis_depth_value, None, update_progress, time_interval ) # Collect results for cumulative analysis if needed all_results = [] frame_counter = 0 try: # Process results as they come in for result in results_generator: # Check if stop button was pressed if st.session_state.stop_requested: st.success("Capture stopped by user") break frame_counter += 1 all_results.append(result) # Display the latest result with results_container.container(): if analysis_depth_value == "granular": # For granular analysis, show the latest frame result st.markdown(f"### Frame {frame_counter}") display_single_result(result) # Send notification if anomaly detected and notifications are enabled if result.get("anomaly_detected", False) and st.session_state.notification_type and st.session_state.notification_contact: # Create notification message anomaly_type = result.get("anomaly_type", "Unknown") anomaly_message = f"Anomaly detected in live stream (Frame {frame_counter}).\n" anomaly_message += f"Anomaly type: {anomaly_type}\n\n" # Add analysis details analysis_text = None for key in ["analysis", "text", "description"]: if key in result and result[key]: analysis_text = result[key] break if analysis_text: anomaly_message += f"Analysis: {analysis_text[:500]}..." # Send notification with st.spinner("Sending notification about detected anomaly..."): notification_sent = send_notification( st.session_state.notification_type, st.session_state.notification_contact, anomaly_message, result.get("frame") ) if notification_sent: st.success(f"Notification sent to {st.session_state.notification_contact} via {st.session_state.notification_type.capitalize()}") else: st.error(f"Failed to send notification. Please check your {st.session_state.notification_type} settings.") else: # For cumulative analysis, we get periodic updates st.markdown(f"### Cumulative Analysis (Updated)") display_single_result(result) # Send notification if anomaly detected and notifications are enabled if result.get("anomaly_detected", False) and st.session_state.notification_type and st.session_state.notification_contact: # Create notification message anomaly_type = result.get("anomaly_type", "Unknown") anomaly_message = f"Anomaly detected in live stream (Cumulative Analysis).\n" anomaly_message += f"Anomaly type: {anomaly_type}\n\n" # Add analysis details analysis_text = None for key in ["analysis", "text", "description"]: if key in result and result[key]: analysis_text = result[key] break if analysis_text: anomaly_message += f"Analysis: {analysis_text[:500]}..." # Get a frame for the notification if available anomaly_image = None if "frames" in result and result["frames"]: anomaly_image = result["frames"][0] # Send notification with st.spinner("Sending notification about detected anomaly..."): notification_sent = send_notification( st.session_state.notification_type, st.session_state.notification_contact, anomaly_message, anomaly_image ) if notification_sent: st.success(f"Notification sent to {st.session_state.notification_contact} via {st.session_state.notification_type.capitalize()}") else: st.error(f"Failed to send notification. Please check your {st.session_state.notification_type} settings.") # Sleep briefly to allow UI updates time.sleep(0.1) except StopIteration: if not st.session_state.stop_requested: st.info("Stream ended") # Final results if analysis_depth_value == "granular": results = all_results else: results = all_results[-1] if all_results else None end_time = time.time() # Calculate processing time processing_time = end_time - start_time st.success(f"Processing completed in {processing_time:.2f} seconds") # Check if notifications are enabled and if anomalies were detected if st.session_state.notification_type and st.session_state.notification_contact: # Check if anomalies were detected anomalies_detected = False anomaly_image = None anomaly_message = "" if analysis_depth_value == "granular": # For granular analysis, check if any frame has an anomaly anomaly_frames = [r for r in results if r.get("anomaly_detected", False)] if anomaly_frames: anomalies_detected = True # Get the first anomaly frame for the notification first_anomaly = anomaly_frames[0] anomaly_image = first_anomaly.get("frame") # Create notification message anomaly_types = set(r.get("anomaly_type", "Unknown") for r in anomaly_frames) anomaly_message = f"Anomaly detected in {len(anomaly_frames)} out of {len(results)} frames.\n" anomaly_message += f"Anomaly types: {', '.join(anomaly_types)}\n\n" # Add details of the first anomaly analysis_text = None for key in ["analysis", "text", "description"]: if key in first_anomaly and first_anomaly[key]: analysis_text = first_anomaly[key] break if analysis_text: anomaly_message += f"Analysis of first anomaly: {analysis_text[:500]}..." else: # For cumulative analysis, check the overall result if results.get("anomaly_detected", False): anomalies_detected = True # Get a frame for the notification if available if "frames" in results and results["frames"]: anomaly_image = results["frames"][0] # Create notification message anomaly_type = results.get("anomaly_type", "Unknown") anomaly_message = f"Anomaly detected in video analysis.\n" anomaly_message += f"Anomaly type: {anomaly_type}\n\n" # Add analysis details analysis_text = None for key in ["analysis", "text", "description"]: if key in results and results[key]: analysis_text = results[key] break if analysis_text: anomaly_message += f"Analysis: {analysis_text[:500]}..." # Send notification if anomalies were detected if anomalies_detected: with st.spinner("Sending notification about detected anomalies..."): notification_sent = send_notification( st.session_state.notification_type, st.session_state.notification_contact, anomaly_message, anomaly_image ) if notification_sent: st.success(f"Notification sent to {st.session_state.notification_contact} via {st.session_state.notification_type.capitalize()}") else: st.error(f"Failed to send notification. Please check your {st.session_state.notification_type} settings.") # Only display results here if we're not in time interval mode # (time interval mode displays results as they come in) if not (input_source == "Live Stream" and capture_mode == "Time Interval (Continuous)"): # Display the results without an additional header display_results(results, analysis_depth_value) # Download results button if results: try: # Convert results to JSON using our custom encoder results_json = json.dumps(results, indent=2, cls=NumpyEncoder) # Create a download button st.download_button( label="Download Results as JSON", data=results_json, file_name="anomaly_detection_results.json", mime="application/json" ) except Exception as e: st.warning(f"Could not create downloadable results: {str(e)}") st.info("This is usually due to large image data in the results. The analysis is still valid.") # Clean up the temporary file if using a video file if input_source == "Video File" and 'video_path' in locals(): # Only delete the file if it's a temporary file, not a sample file if not isinstance(uploaded_file, str): os.unlink(video_path) except Exception as e: st.error(f"⚠️ An error occurred: {str(e)}") if input_source == "Video File" and 'video_path' in locals(): # Only delete the file if it's a temporary file, not a sample file if not isinstance(uploaded_file, str): os.unlink(video_path) # Instructions when no file is uploaded or stream is selected if (input_source == "Video File" and uploaded_file is None) or (input_source == "Live Stream" and stream_source is None) or not submit_button: # Using HTML component to properly render the HTML model_options_html = "" if PHI4_AVAILABLE: model_options_html += "
  • Phi-4 - Microsoft's multimodal model, runs locally
    • " instructions_html = f"""

    📝 How to use this application

    1. Select an input source:
      • Video File - Upload a video file for analysis
      • Live Stream - Connect to a webcam or IP camera stream
    2. Select an AI model for analysis:
      • GPT-4o-mini - Faster and more cost-effective
      • GPT-4o - Most powerful model with highest accuracy
        • {model_options_html}
    3. Set the number of frames to skip - higher values process fewer frames
    4. Choose an analysis depth:
      • Granular - Analyzes each frame individually
      • Cumulative - Provides an overall summary with key frames
    5. Enter a prompt describing what anomaly to look for
    6. Enter your OpenAI API key with access to the selected model (not needed for Phi-4)
    7. Click "Analyze Video" to start processing

    The application will extract frames from your video or stream, analyze them using the selected AI model, and display the results with clear indicators for detected anomalies.

    """ components.html(instructions_html, height=500) # Footer st.markdown("---") st.markdown("
    Powered by OpenAI's GPT-4o, GPT-4o-mini, and Microsoft's Phi-4 models | © 2023 Video Anomaly Detector
    ", unsafe_allow_html=True)