Spaces:
Running
Running
| import gradio as gr | |
| import cv2 | |
| import numpy as np | |
| from queue_monitor import QueueMonitor | |
| from llm_analyzer import LogAnalyzer | |
| import json | |
| # Initialize components | |
| monitor = QueueMonitor() | |
| # Define a default zone for demonstration | |
| default_polygon = np.array([[100, 100], [1100, 100], [1100, 600], [100, 600]]) | |
| monitor.setup_zones([default_polygon]) | |
| # Lazy load LLM to save resources until needed | |
| analyzer = None | |
| def get_analyzer(): | |
| global analyzer | |
| if analyzer is None: | |
| analyzer = LogAnalyzer() | |
| return analyzer | |
| def process_video(video_path): | |
| cap = cv2.VideoCapture(video_path) | |
| frames = [] | |
| total_stats = [] | |
| # Process first 30 frames for demo purposes | |
| for _ in range(30): | |
| ret, frame = cap.read() | |
| if not ret: | |
| break | |
| annotated, stats = monitor.process_frame(frame) | |
| frames.append(cv2.cvtColor(annotated, cv2.COLOR_BGR2RGB)) | |
| total_stats.append(stats) | |
| cap.release() | |
| return frames[0] if frames else None, json.dumps(total_stats[0] if total_stats else {}, indent=2) | |
| def analyze_logs(log_json): | |
| try: | |
| log_data = json.loads(log_json) | |
| llm = get_analyzer() | |
| analysis = llm.analyze_logs(log_data) | |
| return analysis | |
| except Exception as e: | |
| return f"Error: {str(e)}" | |
| with gr.Blocks(title="AI Queue Management for CCTV and YOLO") as demo: | |
| gr.Markdown("# AI Queue Management for CCTV and YOLO") | |
| with gr.Tab("Real-time Monitoring"): | |
| with gr.Row(): | |
| video_input = gr.Video(label="Upload CCTV Footage") | |
| image_output = gr.Image(label="Processed Frame") | |
| with gr.Row(): | |
| stats_output = gr.Code(label="Zone Statistics (JSON)", language="json") | |
| process_btn = gr.Button("Process Video") | |
| process_btn.click(process_video, inputs=video_input, outputs=[image_output, stats_output]) | |
| with gr.Tab("AI Log Analysis"): | |
| gr.Markdown("### Analyze Queue Logs with Qwen-2.5") | |
| log_input = gr.Textbox( | |
| label="Input Logs (JSON)", | |
| value=json.dumps({ | |
| "date": "2026-01-24", | |
| "branch": "SBI Jabalpur", | |
| "avg_wait_time_sec": 420, | |
| "max_wait_time_sec": 980, | |
| "customers_served": 134, | |
| "counter_1_avg_service": 180, | |
| "counter_2_avg_service": 310, | |
| "peak_hour": "12:00-13:00", | |
| "queue_overflow_events": 5 | |
| }, indent=2), | |
| lines=10 | |
| ) | |
| analyze_btn = gr.Button("Generate AI Insights") | |
| analysis_output = gr.Markdown(label="AI Recommendations") | |
| analyze_btn.click(analyze_logs, inputs=log_input, outputs=analysis_output) | |
| with gr.Tab("Use Cases"): | |
| gr.Markdown(""" | |
| ## Expanded Use Cases | |
| - **Retail Heatmap & Dwell Time**: Identify which product sections attract the most customers. | |
| - **Bank Branch Efficiency**: Optimize staffing based on counter service times. | |
| - **Airport Security**: Predict wait times and manage lane openings. | |
| - **Hospital Triage**: Monitor ER waiting areas for timely care. | |
| - **Smart Parking**: Manage vehicle turnover in specific zones. | |
| - **Safety Monitoring**: Detect unauthorized presence in restricted zones. | |
| """) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860) | |