""" Smart Public Queue Traffic Analyzer & Decision Assistant Final Year Engineering Project Uses Computer Vision (OpenCV HOG+SVM) for Queue Analysis """ import gradio as gr import cv2 import numpy as np import matplotlib.pyplot as plt from matplotlib.figure import Figure from PIL import Image import io import tempfile import os class QueueAnalyzer: """Core queue analysis engine using OpenCV HOG + SVM""" def __init__(self): # Initialize HOG descriptor with default people detector self.hog = cv2.HOGDescriptor() self.hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector()) # Queue type service time rules (minutes per person) self.service_times = { "College Office": 2, "Hospital": 5, "Railway Counter": 3, "Supermarket": 1 } def detect_people(self, image): """ Detect people in image using HOG+SVM Returns: list of bounding boxes and count """ # Resize for better performance on CPU height, width = image.shape[:2] scale = 1.0 if width > 800: scale = 800 / width image = cv2.resize(image, None, fx=scale, fy=scale) # Detect people # Parameters tuned for CPU performance boxes, weights = self.hog.detectMultiScale( image, winStride=(8, 8), padding=(4, 4), scale=1.05, useMeanshiftGrouping=False ) # Scale boxes back if image was resized if scale != 1.0: boxes = [[int(x/scale), int(y/scale), int(w/scale), int(h/scale)] for x, y, w, h in boxes] return boxes, len(boxes) def annotate_image(self, image, boxes, count, wait_time, decision): """Draw bounding boxes and overlay information""" annotated = image.copy() # Draw bounding boxes for (x, y, w, h) in boxes: cv2.rectangle(annotated, (x, y), (x + w, y + h), (0, 255, 0), 2) # Prepare overlay text overlay_height = 120 overlay = annotated.copy() cv2.rectangle(overlay, (0, 0), (annotated.shape[1], overlay_height), (0, 0, 0), -1) cv2.addWeighted(overlay, 0.7, annotated, 0.3, 0, annotated) # Add text information font = cv2.FONT_HERSHEY_SIMPLEX cv2.putText(annotated, f"People Count: {count}", (10, 30), font, 1, (255, 255, 255), 2) cv2.putText(annotated, f"Wait Time: {wait_time:.0f} min", (10, 70), font, 1, (255, 255, 255), 2) # Color-coded decision decision_color = self._get_decision_color(wait_time) cv2.putText(annotated, f"Decision: {decision}", (10, 110), font, 1, decision_color, 2) return annotated def _get_decision_color(self, wait_time): """Get color for decision based on wait time""" if wait_time < 10: return (0, 255, 0) # Green elif wait_time <= 20: return (0, 255, 255) # Yellow else: return (0, 0, 255) # Red def calculate_wait_time(self, queue_size, queue_type): """Calculate estimated waiting time""" service_time = self.service_times.get(queue_type, 2) return queue_size * service_time def make_decision(self, wait_time): """Generate decision recommendation""" if wait_time < 10: return "🟢 Go Now", "success" elif wait_time <= 20: return "🟡 Moderate Wait", "warning" else: return "🔴 Come Later", "error" def process_image(self, image_path, queue_type, show_analytics): """Process single image""" # Read image image = cv2.imread(image_path) if image is None: return None, "Error: Could not read image", None image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) # Detect people boxes, count = self.detect_people(image) # Calculate metrics wait_time = self.calculate_wait_time(count, queue_type) decision_text, decision_type = self.make_decision(wait_time) # Annotate image annotated = self.annotate_image(image, boxes, count, wait_time, decision_text) # Prepare metrics metrics = f""" ### 📊 Queue Analysis Results **👥 People Count:** {count} **⏱ Estimated Waiting Time:** {wait_time:.0f} minutes **🎯 Decision:** {decision_text} """ # Generate analytics chart if requested chart = None if show_analytics and count > 0: chart = self._create_simple_bar_chart(count) return annotated, metrics, chart def process_video(self, video_path, queue_type, show_analytics): """Process video by sampling frames""" cap = cv2.VideoCapture(video_path) if not cap.isOpened(): return None, "Error: Could not read video", None frame_counts = [] frame_indices = [] last_annotated = None frame_idx = 0 sample_interval = 10 # Process every 10th frame for CPU efficiency while True: ret, frame = cap.read() if not ret: break # Sample frames if frame_idx % sample_interval == 0: frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) boxes, count = self.detect_people(frame_rgb) frame_counts.append(count) frame_indices.append(frame_idx) # Keep last frame for annotation if len(frame_counts) > 0: wait_time = self.calculate_wait_time(count, queue_type) decision_text, _ = self.make_decision(wait_time) last_annotated = self.annotate_image(frame_rgb, boxes, count, wait_time, decision_text) frame_idx += 1 cap.release() if len(frame_counts) == 0: return None, "Error: No frames could be processed", None # Calculate statistics avg_count = np.mean(frame_counts) max_count = np.max(frame_counts) # Use average count for decision wait_time = self.calculate_wait_time(avg_count, queue_type) decision_text, decision_type = self.make_decision(wait_time) # Prepare metrics metrics = f""" ### 📊 Queue Analysis Results (Video) **👥 Average People Count:** {avg_count:.1f} **👥 Maximum People Count:** {max_count} **📹 Frames Analyzed:** {len(frame_counts)} **⏱ Estimated Waiting Time:** {wait_time:.0f} minutes **🎯 Decision:** {decision_text} """ # Generate analytics charts chart = None if show_analytics: chart = self._create_video_analytics(frame_indices, frame_counts, avg_count, max_count) return last_annotated, metrics, chart def _create_simple_bar_chart(self, count): """Create simple bar chart for image analysis""" fig = Figure(figsize=(8, 4)) ax = fig.add_subplot(111) ax.bar(['Detected People'], [count], color='#2196F3', width=0.4) ax.set_ylabel('Count', fontsize=12) ax.set_title('People Detection Result', fontsize=14, fontweight='bold') ax.grid(axis='y', alpha=0.3) # Convert to image buf = io.BytesIO() fig.savefig(buf, format='png', bbox_inches='tight', dpi=100) buf.seek(0) img = Image.open(buf) plt.close(fig) return img def _create_video_analytics(self, frame_indices, frame_counts, avg_count, max_count): """Create analytics charts for video analysis""" fig = Figure(figsize=(14, 5)) # Line chart: People count over frames ax1 = fig.add_subplot(121) ax1.plot(frame_indices, frame_counts, marker='o', linewidth=2, markersize=4, color='#2196F3', label='Detected People') ax1.axhline(y=avg_count, color='#FF9800', linestyle='--', linewidth=2, label=f'Average: {avg_count:.1f}') ax1.set_xlabel('Frame Index', fontsize=11) ax1.set_ylabel('People Count', fontsize=11) ax1.set_title('People Count Over Time', fontsize=13, fontweight='bold') ax1.legend() ax1.grid(True, alpha=0.3) # Bar chart: Statistics ax2 = fig.add_subplot(122) metrics = ['Average', 'Maximum'] values = [avg_count, max_count] colors = ['#4CAF50', '#F44336'] bars = ax2.bar(metrics, values, color=colors, width=0.5) ax2.set_ylabel('People Count', fontsize=11) ax2.set_title('Queue Statistics', fontsize=13, fontweight='bold') ax2.grid(axis='y', alpha=0.3) # Add value labels on bars for bar, value in zip(bars, values): height = bar.get_height() ax2.text(bar.get_x() + bar.get_width()/2., height, f'{value:.1f}', ha='center', va='bottom', fontsize=10, fontweight='bold') fig.tight_layout() # Convert to image buf = io.BytesIO() fig.savefig(buf, format='png', bbox_inches='tight', dpi=100) buf.seek(0) img = Image.open(buf) plt.close(fig) return img def analyze_queue(file, queue_type, show_analytics): """Main analysis function called by Gradio""" if file is None: return None, "⚠️ Please upload an image or video file.", None analyzer = QueueAnalyzer() # Determine file type file_ext = os.path.splitext(file.name)[1].lower() try: if file_ext in ['.jpg', '.jpeg', '.png', '.bmp', '.webp']: # Process as image return analyzer.process_image(file.name, queue_type, show_analytics) elif file_ext in ['.mp4', '.avi', '.mov', '.mkv', '.webm']: # Process as video return analyzer.process_video(file.name, queue_type, show_analytics) else: return None, "❌ Unsupported file format. Please upload an image or video.", None except Exception as e: return None, f"❌ Error processing file: {str(e)}", None # Build Gradio Interface def create_interface(): """Create professional Gradio Blocks interface""" with gr.Blocks(theme=gr.themes.Soft(), title="Smart Queue Analyzer") as app: # Header gr.Markdown(""" # 🎯 Smart Public Queue Traffic Analyzer ### AI-Powered Decision Assistant Using Computer Vision Upload an image or video of a public queue to get instant analysis and recommendations. """) with gr.Row(): # Input Section with gr.Column(scale=1): gr.Markdown("### 📤 Input") file_input = gr.File( label="Upload Image or Video", file_types=["image", "video"], type="filepath" ) queue_type = gr.Dropdown( choices=["College Office", "Hospital", "Railway Counter", "Supermarket"], value="College Office", label="Queue Type", info="Select the type of queue for accurate wait time estimation" ) show_analytics = gr.Checkbox( label="Show Analytics Charts", value=True, info="Display detailed visualization (for video: trend analysis)" ) analyze_btn = gr.Button( "🔍 Analyze Queue", variant="primary", size="lg" ) gr.Markdown(""" --- **Supported Formats:** - Images: JPG, PNG, BMP, WEBP - Videos: MP4, AVI, MOV, MKV, WEBM **Decision Guide:** - 🟢 **Go Now**: < 10 min wait - 🟡 **Moderate Wait**: 10-20 min - 🔴 **Come Later**: > 20 min """) # Output Section with gr.Column(scale=2): gr.Markdown("### 📊 Analysis Results") output_image = gr.Image( label="Annotated Output", type="numpy", height=400 ) output_metrics = gr.Markdown( value="*Analysis results will appear here*" ) output_chart = gr.Image( label="Analytics Visualization", type="pil", visible=True ) # Footer gr.Markdown(""" --- **Technology Stack:** OpenCV HOG+SVM | Gradio | Python **Project Type:** Final Year Engineering Project **Detection Method:** Histogram of Oriented Gradients (HOG) with SVM Classifier **Deployment:** Optimized for CPU-only environments (Hugging Face Spaces compatible) """) # Event Handler analyze_btn.click( fn=analyze_queue, inputs=[file_input, queue_type, show_analytics], outputs=[output_image, output_metrics, output_chart] ) return app # Launch Application if __name__ == "__main__": app = create_interface() app.launch()