Create main.py

main.py

@@ -0,0 +1,609 @@
+import streamlit as st
+import cv2
+import numpy as np
+import os
+import tempfile
+import time
+from PIL import Image
+import matplotlib.pyplot as plt
+import io
+import base64
+
+# Set page configuration
+st.set_page_config(
+    page_title="Motion Detection and Object Tracking",
+    page_icon="🎥",
+    layout="wide"
+)
+
+# Custom CSS for styling
+st.markdown("""
+<style>
+    .main-header {
+        font-size: 2.5rem;
+        font-weight: bold;
+        margin-bottom: 1rem;
+        color: #2c3e50;
+    }
+    .sub-header {
+        font-size: 1.5rem;
+        margin-bottom: 1rem;
+        color: #34495e;
+    }
+    .stAlert {
+        background-color: #f8f9fa;
+        border: 1px solid #e9ecef;
+        border-radius: 0.25rem;
+    }
+    .thumbnail-container {
+        display: flex;
+        overflow-x: auto;
+        padding: 10px 0;
+    }
+    .thumbnail {
+        width: 120px;
+        height: 80px;
+        margin-right: 10px;
+        border: 2px solid transparent;
+    }
+</style>
+""", unsafe_allow_html=True)
+
+# Utility functions from backend
+def f_keepLargeComponents(I, th):
+    """Keep only large connected components in a binary image"""
+    R = np.zeros(I.shape) < 0
+    unique_labels = np.unique(I.flatten())
+    for label in unique_labels:
+        if label == 0:
+            pass
+        else:
+            I2 = I == label
+            if np.sum(I2) > th:
+                R = R | I2
+    return np.float32(255 * R)
+
+def convert_to_image(cv2_img):
+    """Convert OpenCV image to PIL Image for Streamlit display"""
+    img_rgb = cv2.cvtColor(cv2_img, cv2.COLOR_BGR2RGB)
+    return Image.fromarray(img_rgb)
+
+def get_image_base64(img):
+    """Convert image to base64 for HTML display"""
+    buffered = io.BytesIO()
+    img.save(buffered, format="JPEG")
+    img_str = base64.b64encode(buffered.getvalue()).decode()
+    return img_str
+
+def process_frame(frame, bg_subtractor, min_area):
+    """Process a frame with background subtraction and component filtering"""
+    # Resize for consistent processing
+    frame = cv2.resize(frame, dsize=(600, 400))
+    
+    # Apply background subtraction
+    fgmask = bg_subtractor.apply(frame)
+    
+    # Apply morphological operations
+    K_r = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 3))
+    fgmask = cv2.morphologyEx(np.float32(fgmask), cv2.MORPH_OPEN, K_r)
+    
+    # Connected components analysis
+    num_labels, labels_im = cv2.connectedComponents(np.array(fgmask > 0, np.uint8))
+    
+    # Keep only large components
+    fgmask = f_keepLargeComponents(labels_im, min_area)
+    
+    # Create visualization mask
+    F = np.zeros(frame.shape, np.uint8)
+    F[:, :, 0], F[:, :, 1], F[:, :, 2] = fgmask, fgmask, fgmask
+    
+    # Combine original and mask
+    combined = np.hstack((frame, F))
+    
+    has_motion = np.sum(fgmask) > 0
+    
+    return frame, fgmask, combined, has_motion
+
+# App title and description
+st.markdown("<div class='main-header'>Motion Detection and Object Tracking</div>", unsafe_allow_html=True)
+st.markdown("Track objects and detect motion in videos or image sequences. Upload your own video or use your webcam.")
+
+# Sidebar for settings
+st.sidebar.markdown("<div class='sub-header'>Settings</div>", unsafe_allow_html=True)
+
+# Input source selection
+input_source = st.sidebar.radio("Select Input Source", ["Upload Video", "Upload Image Sequence", "Sample Video"])
+
+# Motion detection parameters
+min_area = st.sidebar.slider("Minimum Component Area", 100, 5000, 1000, 100)
+history = st.sidebar.slider("Background History", 100, 1000, 500, 50)
+var_threshold = st.sidebar.slider("Variance Threshold", 5, 100, 16, 1)
+detect_shadows = st.sidebar.checkbox("Detect Shadows", value=True)
+
+# Output settings
+save_output = st.sidebar.checkbox("Save Processed Frames", value=False)
+min_sequence_frames = st.sidebar.number_input("Minimum Frames in Sequence", 1, 20, 5)
+
+# Object detection option
+use_object_detection = st.sidebar.checkbox("Enable Object Detection", value=False)
+if use_object_detection:
+    st.sidebar.warning("Note: Object detection requires the cvlib package which must be installed on your Streamlit Cloud deployment.")
+
+# Create tabs for different functionalities
+tab1, tab2, tab3 = st.tabs(["Process Video", "Results", "About"])
+
+with tab1:
+    # Process input based on selection
+    if input_source == "Upload Video":
+        uploaded_file = st.file_uploader("Upload a video file", type=["mp4", "avi", "mov", "mkv"])
+        
+        if uploaded_file is not None:
+            # Save uploaded file to a temporary file
+            temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4')
+            temp_file.write(uploaded_file.read())
+            
+            # Create background subtractor
+            fgModel = cv2.createBackgroundSubtractorMOG2(history=history, varThreshold=var_threshold, detectShadows=detect_shadows)
+            
+            # Process video
+            if st.button("Process Video"):
+                # Create output directory if saving results
+                if save_output:
+                    output_dir = tempfile.mkdtemp()
+                    st.session_state['output_dir'] = output_dir
+                    st.session_state['saved_frames'] = []
+                
+                # Open video capture
+                cap = cv2.VideoCapture(temp_file.name)
+                
+                # Process frames
+                frame_idx = 0
+                motion_sequence = []
+                sequence_counter = 0
+                
+                progress_bar = st.progress(0)
+                frame_display = st.empty()
+                status_text = st.empty()
+                
+                # Get total frames for progress calculation
+                total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+                
+                while cap.isOpened():
+                    ret, frame = cap.read()
+                    if not ret:
+                        break
+                    
+                    frame_idx += 1
+                    
+                    # Process frame
+                    original, mask, combined, has_motion = process_frame(frame, fgModel, min_area)
+                    
+                    # Update motion sequence
+                    if has_motion:
+                        motion_sequence.append(original)
+                    elif len(motion_sequence) > 0:
+                        # Save sequence if it meets minimum length
+                        if len(motion_sequence) >= min_sequence_frames and save_output:
+                            sequence_counter += 1
+                            for i, seq_frame in enumerate(motion_sequence):
+                                frame_name = f"{sequence_counter}_{i+1}.jpg"
+                                frame_path = os.path.join(output_dir, frame_name)
+                                
+                                # Apply object detection if enabled
+                                if use_object_detection:
+                                    try:
+                                        import cvlib as cv
+                                        from cvlib.object_detection import draw_bbox
+                                        bbox, labels, conf = cv.detect_common_objects(seq_frame)
+                                        seq_frame = draw_bbox(seq_frame, bbox, labels, conf)
+                                    except ImportError:
+                                        st.warning("cvlib not available. Skipping object detection.")
+                                
+                                cv2.imwrite(frame_path, seq_frame)
+                                st.session_state['saved_frames'].append(frame_path)
+                        
+                        # Reset sequence
+                        motion_sequence = []
+                    
+                    # Display current frame
+                    frame_display.image(convert_to_image(combined), caption="Processing: Original | Foreground Mask", use_column_width=True)
+                    
+                    # Update progress
+                    progress = min(frame_idx / total_frames, 1.0)
+                    progress_bar.progress(progress)
+                    
+                    # Display status
+                    status_text.text(f"Processing frame {frame_idx}/{total_frames} | Motion sequences: {sequence_counter}")
+                    
+                    # Slow down processing slightly for display
+                    time.sleep(0.01)
+                
+                # Save final sequence if any
+                if len(motion_sequence) >= min_sequence_frames and save_output:
+                    sequence_counter += 1
+                    for i, seq_frame in enumerate(motion_sequence):
+                        frame_name = f"{sequence_counter}_{i+1}.jpg"
+                        frame_path = os.path.join(output_dir, frame_name)
+                        
+                        # Apply object detection if enabled
+                        if use_object_detection:
+                            try:
+                                import cvlib as cv
+                                from cvlib.object_detection import draw_bbox
+                                bbox, labels, conf = cv.detect_common_objects(seq_frame)
+                                seq_frame = draw_bbox(seq_frame, bbox, labels, conf)
+                            except ImportError:
+                                pass
+                        
+                        cv2.imwrite(frame_path, seq_frame)
+                        st.session_state['saved_frames'].append(frame_path)
+                
+                cap.release()
+                
+                # Complete
+                progress_bar.progress(1.0)
+                status_text.success(f"Processing complete! {sequence_counter} motion sequences detected.")
+                
+                # Clean up
+                os.unlink(temp_file.name)
+                
+                # Set results flag
+                if save_output and sequence_counter > 0:
+                    st.session_state['has_results'] = True
+                    st.info("Results are available in the Results tab.")
+    
+    elif input_source == "Upload Image Sequence":
+        uploaded_files = st.file_uploader("Upload image sequence", type=["jpg", "jpeg", "png"], accept_multiple_files=True)
+        
+        if uploaded_files:
+            # Create background subtractor
+            fgModel = cv2.createBackgroundSubtractorMOG2(history=history, varThreshold=var_threshold, detectShadows=detect_shadows)
+            
+            # Process images
+            if st.button("Process Images"):
+                # Create output directory if saving results
+                if save_output:
+                    output_dir = tempfile.mkdtemp()
+                    st.session_state['output_dir'] = output_dir
+                    st.session_state['saved_frames'] = []
+                
+                # Process frames
+                frame_idx = 0
+                motion_sequence = []
+                sequence_counter = 0
+                
+                progress_bar = st.progress(0)
+                frame_display = st.empty()
+                status_text = st.empty()
+                
+                total_frames = len(uploaded_files)
+                
+                for file in uploaded_files:
+                    frame_idx += 1
+                    
+                    # Read image
+                    file_bytes = np.asarray(bytearray(file.read()), dtype=np.uint8)
+                    frame = cv2.imdecode(file_bytes, cv2.IMREAD_COLOR)
+                    
+                    # Reset file pointer
+                    file.seek(0)
+                    
+                    # Process frame
+                    original, mask, combined, has_motion = process_frame(frame, fgModel, min_area)
+                    
+                    # Update motion sequence
+                    if has_motion:
+                        motion_sequence.append(original)
+                    elif len(motion_sequence) > 0:
+                        # Save sequence if it meets minimum length
+                        if len(motion_sequence) >= min_sequence_frames and save_output:
+                            sequence_counter += 1
+                            for i, seq_frame in enumerate(motion_sequence):
+                                frame_name = f"{sequence_counter}_{i+1}.jpg"
+                                frame_path = os.path.join(output_dir, frame_name)
+                                
+                                # Apply object detection if enabled
+                                if use_object_detection:
+                                    try:
+                                        import cvlib as cv
+                                        from cvlib.object_detection import draw_bbox
+                                        bbox, labels, conf = cv.detect_common_objects(seq_frame)
+                                        seq_frame = draw_bbox(seq_frame, bbox, labels, conf)
+                                    except ImportError:
+                                        st.warning("cvlib not available. Skipping object detection.")
+                                
+                                cv2.imwrite(frame_path, seq_frame)
+                                st.session_state['saved_frames'].append(frame_path)
+                        
+                        # Reset sequence
+                        motion_sequence = []
+                    
+                    # Display current frame
+                    frame_display.image(convert_to_image(combined), caption="Processing: Original | Foreground Mask", use_column_width=True)
+                    
+                    # Update progress
+                    progress = min(frame_idx / total_frames, 1.0)
+                    progress_bar.progress(progress)
+                    
+                    # Display status
+                    status_text.text(f"Processing frame {frame_idx}/{total_frames} | Motion sequences: {sequence_counter}")
+                    
+                    # Slow down processing slightly for display
+                    time.sleep(0.01)
+                
+                # Save final sequence if any
+                if len(motion_sequence) >= min_sequence_frames and save_output:
+                    sequence_counter += 1
+                    for i, seq_frame in enumerate(motion_sequence):
+                        frame_name = f"{sequence_counter}_{i+1}.jpg"
+                        frame_path = os.path.join(output_dir, frame_name)
+                        
+                        # Apply object detection if enabled
+                        if use_object_detection:
+                            try:
+                                import cvlib as cv
+                                from cvlib.object_detection import draw_bbox
+                                bbox, labels, conf = cv.detect_common_objects(seq_frame)
+                                seq_frame = draw_bbox(seq_frame, bbox, labels, conf)
+                            except ImportError:
+                                pass
+                        
+                        cv2.imwrite(frame_path, seq_frame)
+                        st.session_state['saved_frames'].append(frame_path)
+                
+                # Complete
+                progress_bar.progress(1.0)
+                status_text.success(f"Processing complete! {sequence_counter} motion sequences detected.")
+                
+                # Set results flag
+                if save_output and sequence_counter > 0:
+                    st.session_state['has_results'] = True
+                    st.info("Results are available in the Results tab.")
+    
+    else:  # Sample video
+        st.info("Using a sample video for demonstration")
+        
+        # Create sample video data - in a real app, you'd use a sample video file
+        sample_video = st.selectbox("Select sample video", ["Campus", "Shopping Mall", "Office Room"])
+        
+        # Process sample video
+        if st.button("Process Sample Video"):
+            # Create background subtractor
+            fgModel = cv2.createBackgroundSubtractorMOG2(history=history, varThreshold=var_threshold, detectShadows=detect_shadows)
+            
+            # Create output directory if saving results
+            if save_output:
+                output_dir = tempfile.mkdtemp()
+                st.session_state['output_dir'] = output_dir
+                st.session_state['saved_frames'] = []
+            
+            # Generate some dummy frames for demonstration
+            total_frames = 100
+            progress_bar = st.progress(0)
+            frame_display = st.empty()
+            status_text = st.empty()
+            
+            sequence_counter = 0
+            motion_sequence = []
+            
+            # Simulate processing frames
+            for i in range(total_frames):
+                # Create a dummy frame with some motion
+                frame = np.zeros((400, 600, 3), dtype=np.uint8)
+                
+                # Add some moving objects
+                if i % 10 < 5:  # Motion every other 5 frames
+                    cv2.circle(frame, (300 + i % 100, 200), 50, (0, 0, 255), -1)
+                    has_motion = True
+                else:
+                    has_motion = False
+                
+                # Apply background subtraction (simplified for demo)
+                if i == 0:
+                    fgmask = np.zeros((400, 600), dtype=np.float32)
+                else:
+                    # Process frame - simplified for demo
+                    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+                    fgmask = fgModel.apply(frame)
+                    fgmask = np.float32(fgmask)
+                
+                # Create visualization mask
+                F = np.zeros(frame.shape, np.uint8)
+                F[:, :, 0], F[:, :, 1], F[:, :, 2] = fgmask, fgmask, fgmask
+                
+                # Combine original and mask
+                combined = np.hstack((frame, F))
+                
+                # Update motion sequence
+                if has_motion:
+                    motion_sequence.append(frame)
+                elif len(motion_sequence) > 0:
+                    # Save sequence if it meets minimum length
+                    if len(motion_sequence) >= min_sequence_frames and save_output:
+                        sequence_counter += 1
+                        for j, seq_frame in enumerate(motion_sequence):
+                            frame_name = f"{sequence_counter}_{j+1}.jpg"
+                            frame_path = os.path.join(output_dir, frame_name)
+                            cv2.imwrite(frame_path, seq_frame)
+                            st.session_state['saved_frames'].append(frame_path)
+                    
+                    # Reset sequence
+                    motion_sequence = []
+                
+                # Display current frame
+                frame_display.image(convert_to_image(combined), caption="Processing: Original | Foreground Mask", use_column_width=True)
+                
+                # Update progress
+                progress = (i + 1) / total_frames
+                progress_bar.progress(progress)
+                
+                # Display status
+                status_text.text(f"Processing frame {i+1}/{total_frames} | Motion sequences: {sequence_counter}")
+                
+                # Slow down processing slightly for display
+                time.sleep(0.05)
+            
+            # Save final sequence if any
+            if len(motion_sequence) >= min_sequence_frames and save_output:
+                sequence_counter += 1
+                for j, seq_frame in enumerate(motion_sequence):
+                    frame_name = f"{sequence_counter}_{j+1}.jpg"
+                    frame_path = os.path.join(output_dir, frame_name)
+                    cv2.imwrite(frame_path, seq_frame)
+                    st.session_state['saved_frames'].append(frame_path)
+            
+            # Complete
+            progress_bar.progress(1.0)
+            status_text.success(f"Processing complete! {sequence_counter} motion sequences detected.")
+            
+            # Set results flag
+            if save_output and sequence_counter > 0:
+                st.session_state['has_results'] = True
+                st.info("Results are available in the Results tab.")
+
+with tab2:
+    st.markdown("<div class='sub-header'>Detected Motion Sequences</div>", unsafe_allow_html=True)
+    
+    if 'has_results' in st.session_state and st.session_state['has_results']:
+        if 'saved_frames' in st.session_state and st.session_state['saved_frames']:
+            # Group frames by sequence
+            sequences = {}
+            for frame_path in st.session_state['saved_frames']:
+                frame_name = os.path.basename(frame_path)
+                seq_id = frame_name.split('_')[0]
+                
+                if seq_id not in sequences:
+                    sequences[seq_id] = []
+                
+                sequences[seq_id].append(frame_path)
+            
+            # Display sequences
+            selected_sequence = st.selectbox("Select motion sequence", list(sequences.keys()))
+            
+            if selected_sequence:
+                st.write(f"Sequence {selected_sequence} - {len(sequences[selected_sequence])} frames")
+                
+                # Display thumbnails
+                cols = st.columns(min(5, len(sequences[selected_sequence])))
+                for i, (col, frame_path) in enumerate(zip(cols, sequences[selected_sequence])):
+                    img = Image.open(frame_path)
+                    col.image(img, caption=f"Frame {i+1}", use_column_width=True)
+                
+                # Display full sequence
+                st.write("Full sequence:")
+                sequence_frames = []
+                for frame_path in sequences[selected_sequence]:
+                    img = Image.open(frame_path)
+                    sequence_frames.append(np.array(img))
+                
+                # Create animated GIF option
+                if st.button("Create GIF from Sequence"):
+                    with st.spinner("Creating GIF..."):
+                        # Create a temporary file for the GIF
+                        with tempfile.NamedTemporaryFile(suffix='.gif', delete=False) as temp_gif:
+                            gif_path = temp_gif.name
+                        
+                        # Convert frames to GIF
+                        images = [Image.fromarray(frame) for frame in sequence_frames]
+                        images[0].save(
+                            gif_path,
+                            save_all=True,
+                            append_images=images[1:],
+                            duration=200,
+                            loop=0
+                        )
+                        
+                        # Display the GIF
+                        with open(gif_path, 'rb') as gif_file:
+                            gif_data = gif_file.read()
+                        
+                        st.image(gif_data, caption="Motion Sequence GIF")
+                        
+                        # Provide download link
+                        st.download_button(
+                            label="Download GIF",
+                            data=gif_data,
+                            file_name=f"sequence_{selected_sequence}.gif",
+                            mime="image/gif"
+                        )
+                        
+                        # Clean up
+                        os.unlink(gif_path)
+                
+                # Display stacked view
+                st.write("All frames in sequence:")
+                for i, frame in enumerate(sequence_frames):
+                    st.image(frame, caption=f"Frame {i+1}", use_column_width=True)
+    else:
+        st.info("No results available yet. Process a video or image sequence with 'Save Processed Frames' enabled.")
+
+with tab3:
+    st.markdown("<div class='sub-header'>About this application</div>", unsafe_allow_html=True)
+    
+    st.markdown("""
+    This application implements motion detection and object tracking using background subtraction and connected component analysis.
+    
+    ### Features:
+    - Process videos or image sequences to detect motion
+    - Background subtraction using the MOG2 algorithm
+    - Connected component filtering to remove noise
+    - Optional object detection (requires cvlib)
+    - Save and display motion sequences
+    - Export sequences as GIFs
+    
+    ### How to use:
+    1. Select an input source (upload video, image sequence, or use sample)
+    2. Adjust parameters in the sidebar
+    3. Process the input
+    4. View results in the Results tab
+    
+    ### Requirements for deployment:
+    ```
+    streamlit
+    opencv-python-headless
+    numpy
+    pillow
+    matplotlib
+    ```
+    
+    For object detection functionality, also install:
+    ```
+    cvlib
+    tensorflow
+    ```
+    
+    ### How it works:
+    The application uses background subtraction to identify moving objects in a video sequence. It then applies connected component analysis to filter out small noise components and track larger moving objects. Optionally, it can apply object detection to identify the types of objects detected.
+    """)
+    
+    # Show parameters explanation
+    st.markdown("### Parameter Explanation:")
+    
+    st.markdown("""
+    - **Minimum Component Area**: The minimum size (in pixels) of connected components to keep. Smaller components are filtered out as noise.
+    - **Background History**: Number of frames used to build the background model in MOG2.
+    - **Variance Threshold**: Threshold on the squared Mahalanobis distance to decide whether a pixel is foreground or background.
+    - **Detect Shadows**: Enable shadow detection in MOG2 algorithm.
+    - **Minimum Frames in Sequence**: The minimum number of consecutive frames with motion to save as a sequence.
+    """)
+    
+    # Add deployment instructions
+    st.markdown("### Deployment to Streamlit Cloud:")
+    
+    st.markdown("""
+    To deploy this application to Streamlit Cloud:
+    
+    1. Create a GitHub repository with this code
+    2. Include a `requirements.txt` file with the necessary dependencies
+    3. Connect your GitHub repository to Streamlit Cloud
+    4. Deploy the application
+    
+    Example `requirements.txt`:
+    ```
+    streamlit==1.24.0
+    opencv-python-headless==4.7.0.72
+    numpy==1.24.3
+    pillow==9.5.0
+    matplotlib==3.7.1
+    ```
+    """)