diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,1240 +1,1240 @@ -""" ---- -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", "587")) - 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. -
    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. -
    3. Set the number of frames to skip - higher values process fewer frames
      4. -
    4. Choose an analysis depth: -
        -
      • Granular - Analyzes each frame individually
        • -
      • Cumulative - Provides an overall summary with key frames
        • -
      -
      5. -
    5. Enter a prompt describing what anomaly to look for
      6. -
    6. Enter your OpenAI API key with access to the selected model (not needed for Phi-4)
      7. -
    7. Click "Analyze Video" to start processing
      8. -
    - -

    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) - - - - - +""" +--- +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. +
    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. +
    3. Set the number of frames to skip - higher values process fewer frames
      4. +
    4. Choose an analysis depth: +
        +
      • Granular - Analyzes each frame individually
        • +
      • Cumulative - Provides an overall summary with key frames
        • +
      +
      5. +
    5. Enter a prompt describing what anomaly to look for
      6. +
    6. Enter your OpenAI API key with access to the selected model (not needed for Phi-4)
      7. +
    7. Click "Analyze Video" to start processing
      8. +
    + +

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