Spaces:
Runtime error
Runtime error
| import streamlit as st | |
| import torch | |
| import cv2 | |
| import numpy as np | |
| from PIL import Image | |
| import io | |
| import time | |
| import plotly.graph_objects as go | |
| import plotly.express as px | |
| import pandas as pd | |
| from pathlib import Path | |
| import tempfile | |
| import os | |
| import logging | |
| import warnings | |
| import requests | |
| import threading | |
| import av | |
| import imageio.v2 as imageio | |
| from streamlit_webrtc import webrtc_streamer, VideoProcessorBase | |
| # Suppress WebRTC/asyncio warnings and errors | |
| logging.getLogger('aioice').setLevel(logging.CRITICAL) | |
| logging.getLogger('asyncio').setLevel(logging.CRITICAL) | |
| logging.getLogger('streamlit_webrtc').setLevel(logging.ERROR) | |
| warnings.filterwarnings('ignore', category=DeprecationWarning) | |
| warnings.filterwarnings('ignore', category=FutureWarning) | |
| # Page config | |
| st.set_page_config( | |
| page_title="𦽠AI Wheelchair Navigation System", | |
| page_icon="π¦½", | |
| layout="wide", | |
| initial_sidebar_state="expanded", | |
| menu_items={ | |
| 'Get Help': 'https://huggingface.co/spaces/fouadmahmoud281/processing-image', | |
| 'About': 'AI Wheelchair Navigation System - Graduation Project' | |
| } | |
| ) | |
| # Custom CSS for wheelchair theme | |
| st.markdown(""" | |
| <style> | |
| .main-header { | |
| background: linear-gradient(90deg, #2E86AB 0%, #A23B72 50%, #F18F01 100%); | |
| color: white; | |
| padding: 2rem; | |
| border-radius: 10px; | |
| margin-bottom: 2rem; | |
| text-align: center; | |
| } | |
| .metric-card { | |
| background: #f8f9fa; | |
| padding: 1rem; | |
| border-radius: 10px; | |
| border-left: 4px solid #2E86AB; | |
| margin: 0.5rem 0; | |
| } | |
| .detection-box { | |
| background: #e8f4f8; | |
| padding: 1rem; | |
| border-radius: 8px; | |
| border: 1px solid #2E86AB; | |
| margin: 0.5rem 0; | |
| } | |
| .safety-alert { | |
| background: #fff3cd; | |
| color: #856404; | |
| padding: 1rem; | |
| border-radius: 8px; | |
| border-left: 4px solid #ffc107; | |
| margin: 1rem 0; | |
| } | |
| .success-alert { | |
| background: #d4edda; | |
| color: #155724; | |
| padding: 1rem; | |
| border-radius: 8px; | |
| border-left: 4px solid #28a745; | |
| margin: 1rem 0; | |
| } | |
| .wheelchair-icon { | |
| font-size: 2rem; | |
| color: #2E86AB; | |
| } | |
| .stButton > button { | |
| background: linear-gradient(90deg, #2E86AB, #A23B72); | |
| color: white; | |
| border-radius: 20px; | |
| border: none; | |
| padding: 0.5rem 2rem; | |
| font-weight: bold; | |
| } | |
| .stSelectbox > div > div { | |
| border-radius: 10px; | |
| } | |
| </style> | |
| """, unsafe_allow_html=True) | |
| # Initialize session state | |
| if 'model' not in st.session_state: | |
| st.session_state.model = None | |
| if 'detection_history' not in st.session_state: | |
| st.session_state.detection_history = [] | |
| if 'current_image' not in st.session_state: | |
| st.session_state.current_image = None | |
| if 'demo_selected' not in st.session_state: | |
| st.session_state.demo_selected = False | |
| if 'processed_video_path' not in st.session_state: | |
| st.session_state.processed_video_path = None | |
| if 'video_stats' not in st.session_state: | |
| st.session_state.video_stats = None | |
| if 'processed_video_mime' not in st.session_state: | |
| st.session_state.processed_video_mime = None | |
| # Demo images URLs (publicly accessible images) | |
| DEMO_IMAGES = { | |
| "Indoor Scene": "https://plus.unsplash.com/premium_photo-1661346079168-5152a94fee62?q=80&w=869&auto=format&fit=crop&ixlib=rb-4.1.0&ixid=M3wxMjA3fDB8MHxwaG90by1wYWdlfHx8fGVufDB8fHx8fA%3D%3D", # Office/indoor | |
| "Outdoor Scene": "https://images.unsplash.com/photo-1767034241658-0319e450c0eb?q=80&w=870&auto=format&fit=crop&ixlib=rb-4.1.0&ixid=M3wxMjA3fDB8MHxwaG90by1wYWdlfHx8fGVufDB8fHx8fA%3D%3D", # Street scene | |
| "Crowded Area": "https://images.unsplash.com/photo-1723930298143-48843627a478?q=80&w=387&auto=format&fit=crop&ixlib=rb-4.1.0&ixid=M3wxMjA3fDB8MHxwaG90by1wYWdlfHx8fGVufDB8fHx8fA%3D%3D" # Crowded place | |
| } | |
| def load_demo_image(demo_name): | |
| """Load demo image from URL""" | |
| try: | |
| url = DEMO_IMAGES[demo_name] | |
| response = requests.get(url, timeout=10) | |
| response.raise_for_status() | |
| image = Image.open(io.BytesIO(response.content)) | |
| return image | |
| except Exception as e: | |
| st.error(f"Error loading demo image: {e}") | |
| return None | |
| # Wheelchair-relevant classes from COCO | |
| WHEELCHAIR_CLASSES = { | |
| 0: 'person', # People to avoid/navigate around | |
| 1: 'bicycle', # Other mobility devices | |
| 2: 'car', # Vehicles to avoid | |
| 3: 'motorcycle', # Vehicles to avoid | |
| 5: 'bus', # Large vehicles | |
| 7: 'truck', # Large vehicles | |
| 9: 'traffic light', # Navigation signals | |
| 11: 'stop sign', # Navigation signals | |
| 24: 'backpack', # Personal items/obstacles | |
| 26: 'handbag', # Personal items/obstacles | |
| 56: 'chair', # Furniture/obstacles | |
| 58: 'potted plant' # Environmental obstacles | |
| } | |
| CLASS_COLORS = { | |
| 'person': '#FF6B6B', | |
| 'bicycle': '#4ECDC4', | |
| 'car': '#45B7D1', | |
| 'motorcycle': '#96CEB4', | |
| 'bus': '#FECA57', | |
| 'truck': '#FF9F43', | |
| 'traffic light': '#6C5CE7', | |
| 'stop sign': '#FD79A8', | |
| 'chair': '#A0E7E5', | |
| 'backpack': '#DDA0DD', | |
| 'handbag': '#F7DC6F', | |
| 'potted plant': '#82E0AA' | |
| } | |
| def load_model(confidence_threshold=0.5): | |
| """Load YOLOv5 model with caching""" | |
| try: | |
| # Try to load from local path first (for development) | |
| model_paths = [ | |
| "wheelchair_runs/wheelchair_exp5/weights/best.pt", | |
| "wheelchair_runs/wheelchair_exp/weights/best.pt", | |
| "best.pt" # Fallback | |
| ] | |
| model = None | |
| for path in model_paths: | |
| if os.path.exists(path): | |
| model = torch.hub.load('ultralytics/yolov5', 'custom', path=path, trust_repo=True) | |
| break | |
| # If no local model, load pretrained YOLOv5s | |
| if model is None: | |
| model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True, trust_repo=True) | |
| model.conf = confidence_threshold | |
| model.iou = 0.45 | |
| return model | |
| except Exception as e: | |
| st.error(f"Error loading model: {e}") | |
| return None | |
| def get_navigation_advice(detections, image_width=640): | |
| """Generate navigation advice based on detections""" | |
| advice = [] | |
| safety_level = "π’ SAFE" | |
| if detections is None or len(detections) == 0: | |
| return ["β Clear path ahead"], safety_level | |
| critical_detections = 0 | |
| for _, detection in detections.iterrows(): | |
| class_id = int(detection['class']) | |
| class_name = detection['name'] | |
| confidence = detection['confidence'] | |
| x_center = (detection['xmin'] + detection['xmax']) / 2 | |
| y_center = (detection['ymin'] + detection['ymax']) / 2 | |
| # Determine position relative to wheelchair | |
| if x_center < image_width * 0.33: | |
| position = "left" | |
| elif x_center > image_width * 0.67: | |
| position = "right" | |
| else: | |
| position = "ahead" | |
| critical_detections += 1 | |
| # Generate specific advice based on object type and position | |
| if class_name == 'person': | |
| if position == "ahead": | |
| advice.append(f"β οΈ PERSON DETECTED AHEAD - STOP and wait for clear path") | |
| safety_level = "π΄ CRITICAL" | |
| else: | |
| advice.append(f"π€ Person on {position} (conf: {confidence:.0%})") | |
| if safety_level == "π’ SAFE": | |
| safety_level = "π‘ CAUTION" | |
| elif class_name in ['car', 'truck', 'bus', 'motorcycle']: | |
| advice.append(f"π {class_name.title()} {position} - proceed with extreme caution") | |
| safety_level = "π΄ CRITICAL" if position == "ahead" else "π‘ CAUTION" | |
| elif class_name in ['traffic light', 'stop sign']: | |
| advice.append(f"π¦ {class_name.replace('_', ' ').title()} detected - follow traffic rules") | |
| elif class_name == 'chair': | |
| advice.append(f"πͺ Chair detected on {position} - navigate around") | |
| if position == "ahead": | |
| safety_level = "π‘ CAUTION" | |
| elif class_name in ['backpack', 'handbag']: | |
| advice.append(f"π Personal item on {position} - person nearby") | |
| elif class_name == 'potted plant': | |
| advice.append(f"πͺ΄ Obstacle on {position} - adjust path") | |
| if critical_detections > 2: | |
| safety_level = "π΄ CRITICAL" | |
| advice.insert(0, "β οΈ MULTIPLE OBSTACLES AHEAD - STOP AND REASSESS") | |
| return advice[:5], safety_level # Limit to top 5 pieces of advice | |
| def process_image(image, model, conf_threshold): | |
| """Process image and return results""" | |
| if model is None: | |
| return None, None, [] | |
| # Run inference | |
| results = model(image) | |
| # Get detections | |
| detections = results.pandas().xyxy[0] | |
| # Filter to wheelchair-relevant classes | |
| relevant_detections = detections[detections['name'].isin(WHEELCHAIR_CLASSES.values())] | |
| # Get rendered image | |
| rendered_img = results.render()[0] | |
| rendered_img = cv2.cvtColor(rendered_img, cv2.COLOR_BGR2RGB) | |
| return rendered_img, relevant_detections, results | |
| def process_video(video_path, model, conf_threshold, frame_skip=1): | |
| """Process video and return output path + stats.""" | |
| if model is None: | |
| return None, None | |
| cap = cv2.VideoCapture(video_path) | |
| if not cap.isOpened(): | |
| return None, None | |
| fps = cap.get(cv2.CAP_PROP_FPS) or 24 | |
| width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) | |
| height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) | |
| output_fd, output_path = tempfile.mkstemp(suffix=".mp4") | |
| os.close(output_fd) | |
| total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) or 0 | |
| processed_frames = 0 | |
| detection_counts = {} | |
| progress = st.progress(0, text="Processing video frames...") | |
| container = None | |
| stream = None | |
| codec_name = None | |
| error_message = None | |
| output_mime = "video/mp4" | |
| def _process_frame(frame_bgr): | |
| nonlocal processed_frames | |
| img_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) | |
| model.conf = conf_threshold | |
| results = model(img_rgb) | |
| detections = results.pandas().xyxy[0] | |
| relevant = detections[detections['name'].isin(WHEELCHAIR_CLASSES.values())] | |
| for name in relevant['name'].tolist(): | |
| detection_counts[name] = detection_counts.get(name, 0) + 1 | |
| rendered_frame = results.render()[0] | |
| if rendered_frame.shape[1] != width or rendered_frame.shape[0] != height: | |
| rendered_frame = cv2.resize(rendered_frame, (width, height)) | |
| processed_frames += 1 | |
| if total_frames > 0: | |
| progress.progress(min(processed_frames / total_frames, 1.0)) | |
| return rendered_frame | |
| def _reset_stats(): | |
| nonlocal processed_frames, detection_counts | |
| processed_frames = 0 | |
| detection_counts = {} | |
| progress.progress(0) | |
| def _run_loop(write_frame): | |
| cap.set(cv2.CAP_PROP_POS_FRAMES, 0) | |
| frame_index = 0 | |
| while True: | |
| ret, frame = cap.read() | |
| if not ret: | |
| break | |
| frame_index += 1 | |
| if frame_skip > 1 and frame_index % frame_skip != 0: | |
| continue | |
| rendered_frame = _process_frame(frame) | |
| write_frame(rendered_frame) | |
| try: | |
| container = av.open(output_path, mode="w") | |
| try: | |
| stream = container.add_stream("libx264", rate=fps) | |
| codec_name = "libx264" | |
| output_mime = "video/mp4" | |
| except av.AVError: | |
| container.close() | |
| output_path = output_path.replace(".mp4", ".webm") | |
| container = av.open(output_path, mode="w") | |
| stream = container.add_stream("libvpx", rate=fps) | |
| codec_name = "libvpx" | |
| output_mime = "video/webm" | |
| stream.width = width | |
| stream.height = height | |
| stream.pix_fmt = "yuv420p" | |
| _run_loop(lambda rendered_frame: [container.mux(p) for p in stream.encode(av.VideoFrame.from_ndarray(rendered_frame, format="bgr24"))]) | |
| for packet in stream.encode(): | |
| container.mux(packet) | |
| except Exception as e: | |
| error_message = f"PyAV encode failed: {e}" | |
| finally: | |
| if container is not None: | |
| container.close() | |
| if error_message: | |
| _reset_stats() | |
| try: | |
| imageio_path = output_path.replace(".webm", ".mp4") | |
| writer = imageio.get_writer( | |
| imageio_path, | |
| fps=fps, | |
| codec="libx264", | |
| ffmpeg_params=["-pix_fmt", "yuv420p"], | |
| format="FFMPEG" | |
| ) | |
| _run_loop(lambda rendered_frame: writer.append_data(cv2.cvtColor(rendered_frame, cv2.COLOR_BGR2RGB))) | |
| writer.close() | |
| output_path = imageio_path | |
| codec_name = "libx264" | |
| output_mime = "video/mp4" | |
| error_message = None | |
| except Exception as e: | |
| error_message = f"ImageIO H.264 encode failed: {e}" | |
| if error_message: | |
| _reset_stats() | |
| try: | |
| imageio_path = output_path.replace(".mp4", ".webm") | |
| writer = imageio.get_writer( | |
| imageio_path, | |
| fps=fps, | |
| codec="libvpx", | |
| ffmpeg_params=["-pix_fmt", "yuv420p"], | |
| format="FFMPEG" | |
| ) | |
| _run_loop(lambda rendered_frame: writer.append_data(cv2.cvtColor(rendered_frame, cv2.COLOR_BGR2RGB))) | |
| writer.close() | |
| output_path = imageio_path | |
| codec_name = "libvpx" | |
| output_mime = "video/webm" | |
| error_message = None | |
| except Exception as e: | |
| error_message = f"ImageIO WebM encode failed: {e}" | |
| if error_message: | |
| _reset_stats() | |
| try: | |
| fourcc = cv2.VideoWriter_fourcc(*"mp4v") | |
| writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height)) | |
| _run_loop(lambda rendered_frame: writer.write(rendered_frame)) | |
| writer.release() | |
| error_message = None | |
| codec_name = "mp4v" | |
| output_mime = "video/mp4" | |
| except Exception as e: | |
| error_message = f"OpenCV encode failed: {e}" | |
| output_path = None | |
| progress.empty() | |
| cap.release() | |
| stats = { | |
| "total_frames": total_frames, | |
| "processed_frames": processed_frames, | |
| "detection_counts": detection_counts, | |
| "codec": codec_name, | |
| "error": error_message, | |
| "mime": output_mime | |
| } | |
| return output_path, stats | |
| def create_detection_chart(detections): | |
| """Create a bar chart of detections""" | |
| if detections is None or len(detections) == 0: | |
| return None | |
| detection_counts = detections['name'].value_counts() | |
| fig = px.bar( | |
| x=detection_counts.index, | |
| y=detection_counts.values, | |
| color=detection_counts.index, | |
| color_discrete_map=CLASS_COLORS, | |
| title="π Detected Objects Count", | |
| labels={'x': 'Object Type', 'y': 'Count'} | |
| ) | |
| fig.update_layout( | |
| showlegend=False, | |
| plot_bgcolor='rgba(0,0,0,0)', | |
| paper_bgcolor='rgba(0,0,0,0)', | |
| font=dict(size=12), | |
| title_font=dict(size=16, color='#2E86AB') | |
| ) | |
| return fig | |
| def create_confidence_chart(detections): | |
| """Create a confidence score visualization""" | |
| if detections is None or len(detections) == 0: | |
| return None | |
| fig = px.scatter( | |
| detections, | |
| x='name', | |
| y='confidence', | |
| size='confidence', | |
| color='name', | |
| color_discrete_map=CLASS_COLORS, | |
| title="π― Detection Confidence Scores", | |
| labels={'confidence': 'Confidence Score', 'name': 'Object Type'} | |
| ) | |
| fig.update_layout( | |
| showlegend=False, | |
| plot_bgcolor='rgba(0,0,0,0)', | |
| paper_bgcolor='rgba(0,0,0,0)', | |
| font=dict(size=12), | |
| title_font=dict(size=16, color='#2E86AB') | |
| ) | |
| return fig | |
| def get_rtc_configuration(): | |
| """Build RTC config from env vars (TURN optional).""" | |
| turn_url = os.getenv("TURN_URL") | |
| turn_username = os.getenv("TURN_USERNAME") | |
| turn_password = os.getenv("TURN_PASSWORD") | |
| ice_servers = [ | |
| {"urls": ["stun:stun.l.google.com:19302", "stun:stun1.l.google.com:19302"]} | |
| ] | |
| if turn_url and turn_username and turn_password: | |
| ice_servers.append({ | |
| "urls": [turn_url], | |
| "username": turn_username, | |
| "credential": turn_password | |
| }) | |
| return {"iceServers": ice_servers} | |
| class RealtimeVideoProcessor(VideoProcessorBase): | |
| def __init__(self, model, conf_threshold): | |
| self.model = model | |
| self.conf_threshold = conf_threshold | |
| self.last_detections = None | |
| self.lock = threading.Lock() | |
| def recv(self, frame): | |
| img_bgr = frame.to_ndarray(format="bgr24") | |
| img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB) | |
| if self.model is None: | |
| return frame | |
| self.model.conf = self.conf_threshold | |
| results = self.model(img_rgb) | |
| detections = results.pandas().xyxy[0] | |
| relevant_detections = detections[detections['name'].isin(WHEELCHAIR_CLASSES.values())] | |
| with self.lock: | |
| self.last_detections = relevant_detections | |
| rendered_img = results.render()[0] | |
| return av.VideoFrame.from_ndarray(rendered_img, format="bgr24") | |
| # Main app | |
| def main(): | |
| # Header | |
| st.markdown(""" | |
| <div class="main-header"> | |
| <h1>𦽠AI Wheelchair Navigation System</h1> | |
| <p>Intelligent Computer Vision for Safe Wheelchair Navigation</p> | |
| <p><strong>Graduation Project</strong> | YOLOv5 Object Detection | Raspberry Pi Optimized</p> | |
| </div> | |
| """, unsafe_allow_html=True) | |
| # Sidebar | |
| with st.sidebar: | |
| st.markdown("### π§ Model Configuration") | |
| # Model settings | |
| conf_threshold = st.slider( | |
| "Confidence Threshold", | |
| min_value=0.1, | |
| max_value=1.0, | |
| value=0.5, | |
| step=0.1, | |
| help="Minimum confidence for object detection" | |
| ) | |
| st.markdown("### π Model Performance") | |
| st.markdown(""" | |
| <div class="metric-card"> | |
| <strong>π― Overall Accuracy:</strong> 87.5% mAP@0.5<br> | |
| <strong>π€ Person Detection:</strong> 84.4%<br> | |
| <strong>πͺ Chair Detection:</strong> 93.9%<br> | |
| <strong>π Vehicle Detection:</strong> 56.3%<br> | |
| <strong>βοΈ Model Size:</strong> 14.7 MB | |
| </div> | |
| """, unsafe_allow_html=True) | |
| st.markdown("### π― Wheelchair-Relevant Objects") | |
| for class_id, class_name in WHEELCHAIR_CLASSES.items(): | |
| color = CLASS_COLORS.get(class_name, '#808080') | |
| st.markdown(f'<span style="color: {color}; font-weight: bold;">β</span> {class_name.title()}', unsafe_allow_html=True) | |
| # Main content tabs | |
| tab1, tab2, tab3, tab4, tab5 = st.tabs(["πΈ Live Detection", "πΉ Realtime Camera", "π Analytics", "βΉοΈ About", "π Deployment"]) | |
| with tab1: | |
| col1, col2 = st.columns([2, 1]) | |
| with col1: | |
| st.markdown("### π· Upload Image for Detection") | |
| uploaded_file = st.file_uploader( | |
| "Choose an image...", | |
| type=['jpg', 'jpeg', 'png', 'bmp'], | |
| help="Upload an image to test wheelchair navigation detection", | |
| key="image_uploader" | |
| ) | |
| st.markdown("### π₯ Upload Video for Detection") | |
| uploaded_video = st.file_uploader( | |
| "Choose a video...", | |
| type=['mp4', 'avi', 'mov', 'mkv'], | |
| help="Upload a video to run obstacle detection", | |
| key="video_uploader" | |
| ) | |
| # Demo images buttons | |
| st.markdown("### π¬ Or Try Demo Images") | |
| demo_col1, demo_col2, demo_col3 = st.columns(3) | |
| demo_image = None | |
| with demo_col1: | |
| if st.button("πͺ Indoor Scene", use_container_width=True): | |
| st.session_state.demo_selected = True | |
| demo_image = load_demo_image("Indoor Scene") | |
| if demo_image: | |
| st.session_state.current_image = demo_image | |
| with demo_col2: | |
| if st.button("ποΈ Outdoor Scene", use_container_width=True): | |
| st.session_state.demo_selected = True | |
| demo_image = load_demo_image("Outdoor Scene") | |
| if demo_image: | |
| st.session_state.current_image = demo_image | |
| with demo_col3: | |
| if st.button("πΆ Crowded Area", use_container_width=True): | |
| st.session_state.demo_selected = True | |
| demo_image = load_demo_image("Crowded Area") | |
| if demo_image: | |
| st.session_state.current_image = demo_image | |
| with col2: | |
| st.markdown("### π‘οΈ Safety Status") | |
| safety_placeholder = st.empty() | |
| st.markdown("### π§ Navigation Advice") | |
| advice_placeholder = st.empty() | |
| # Process uploaded or demo image | |
| image = None | |
| if uploaded_file is not None: | |
| st.session_state.demo_selected = False | |
| image = Image.open(uploaded_file) | |
| st.session_state.current_image = image | |
| elif st.session_state.demo_selected and st.session_state.current_image is not None: | |
| image = st.session_state.current_image | |
| # Process uploaded video | |
| if uploaded_video is not None: | |
| st.session_state.demo_selected = False | |
| if st.session_state.model is None: | |
| with st.spinner("π€ Loading AI model..."): | |
| st.session_state.model = load_model(conf_threshold) | |
| if st.button("βΆοΈ Run Video Detection", use_container_width=True): | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=Path(uploaded_video.name).suffix) as tmp: | |
| tmp.write(uploaded_video.getbuffer()) | |
| tmp_path = tmp.name | |
| with st.spinner("π Processing video for obstacles..."): | |
| output_path, stats = process_video(tmp_path, st.session_state.model, conf_threshold, frame_skip=1) | |
| st.session_state.processed_video_path = output_path | |
| st.session_state.video_stats = stats | |
| st.session_state.processed_video_mime = stats.get("mime") if stats else None | |
| try: | |
| os.remove(tmp_path) | |
| except Exception: | |
| pass | |
| if st.session_state.processed_video_path: | |
| st.markdown("### π¬ Processed Video") | |
| try: | |
| if os.path.exists(st.session_state.processed_video_path) and os.path.getsize(st.session_state.processed_video_path) > 0: | |
| with open(st.session_state.processed_video_path, "rb") as f: | |
| st.video(f.read(), format=st.session_state.processed_video_mime or "video/mp4") | |
| else: | |
| st.error("Processed video file is empty or missing.") | |
| except Exception as e: | |
| st.error(f"Unable to load processed video: {e}") | |
| if st.session_state.video_stats: | |
| st.markdown("### π Video Detection Summary") | |
| stats = st.session_state.video_stats | |
| st.metric("ποΈ Total Frames", stats.get("total_frames", 0)) | |
| st.metric("β Processed Frames", stats.get("processed_frames", 0)) | |
| if stats.get("codec"): | |
| st.caption(f"Codec: {stats.get('codec')}") | |
| if stats.get("mime"): | |
| st.caption(f"MIME: {stats.get('mime')}") | |
| if stats.get("error"): | |
| st.error(stats.get("error")) | |
| if stats.get("detection_counts"): | |
| counts_df = pd.DataFrame( | |
| sorted(stats["detection_counts"].items(), key=lambda x: x[1], reverse=True), | |
| columns=["Object", "Count"] | |
| ) | |
| st.dataframe(counts_df, use_container_width=True) | |
| if image is not None: | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| st.markdown("#### πΈ Original Image") | |
| st.image(image, width=500) | |
| # Load model | |
| if st.session_state.model is None: | |
| with st.spinner("π€ Loading AI model..."): | |
| st.session_state.model = load_model(conf_threshold) | |
| if st.session_state.model is not None: | |
| # Process image | |
| with st.spinner("π Analyzing image for obstacles..."): | |
| start_time = time.time() | |
| rendered_img, detections, results = process_image(image, st.session_state.model, conf_threshold) | |
| processing_time = time.time() - start_time | |
| with col2: | |
| st.markdown("#### π― Detection Results") | |
| if rendered_img is not None: | |
| st.image(rendered_img, width=500) | |
| # Generate navigation advice | |
| advice, safety_level = get_navigation_advice(detections, image.width) | |
| # Update safety status | |
| with safety_placeholder.container(): | |
| if "CRITICAL" in safety_level: | |
| st.markdown(f'<div class="safety-alert"><strong>{safety_level}</strong></div>', unsafe_allow_html=True) | |
| elif "CAUTION" in safety_level: | |
| st.markdown(f'<div style="background: #fff3cd; color: #856404; padding: 1rem; border-radius: 8px; border-left: 4px solid #ffc107;"><strong>{safety_level}</strong></div>', unsafe_allow_html=True) | |
| else: | |
| st.markdown(f'<div class="success-alert"><strong>{safety_level}</strong></div>', unsafe_allow_html=True) | |
| # Display navigation advice | |
| with advice_placeholder.container(): | |
| for advice_text in advice: | |
| st.markdown(f'<div class="detection-box">{advice_text}</div>', unsafe_allow_html=True) | |
| # Performance metrics | |
| st.markdown("### β‘ Performance Metrics") | |
| perf_col1, perf_col2, perf_col3, perf_col4 = st.columns(4) | |
| with perf_col1: | |
| st.metric("β±οΈ Processing Time", f"{processing_time:.2f}s") | |
| with perf_col2: | |
| fps = 1 / processing_time if processing_time > 0 else 0 | |
| st.metric("π¬ Estimated FPS", f"{fps:.1f}") | |
| with perf_col3: | |
| total_detections = len(detections) if detections is not None else 0 | |
| st.metric("π Objects Detected", total_detections) | |
| with perf_col4: | |
| relevant_count = len(detections) if detections is not None else 0 | |
| st.metric("π― Relevant Objects", relevant_count) | |
| # Detailed detection results | |
| if detections is not None and len(detections) > 0: | |
| st.markdown("### π Detailed Detection Results") | |
| # Create a formatted dataframe | |
| display_df = detections[['name', 'confidence', 'xmin', 'ymin', 'xmax', 'ymax']].copy() | |
| display_df['confidence'] = display_df['confidence'].apply(lambda x: f"{x:.1%}") | |
| display_df.columns = ['Object', 'Confidence', 'X Min', 'Y Min', 'X Max', 'Y Max'] | |
| st.dataframe(display_df, use_container_width=True) | |
| with tab2: | |
| st.markdown("### πΉ Realtime Camera Detection") | |
| st.caption("Start the camera to run live detection on each frame.") | |
| if 'model' not in st.session_state: | |
| st.session_state.model = None | |
| if st.session_state.model is None: | |
| with st.spinner("π€ Loading AI model..."): | |
| st.session_state.model = load_model(conf_threshold) | |
| model = st.session_state.model | |
| rtc_config = get_rtc_configuration() | |
| if not os.getenv("TURN_URL"): | |
| st.info("Live camera may fail on some networks without TURN. Set TURN_URL, TURN_USERNAME, TURN_PASSWORD to improve connectivity.") | |
| webrtc_ctx = webrtc_streamer( | |
| key="realtime_camera", | |
| video_processor_factory=lambda: RealtimeVideoProcessor(model, conf_threshold), | |
| media_stream_constraints={"video": True, "audio": False}, | |
| async_processing=True, | |
| rtc_configuration=rtc_config | |
| ) | |
| realtime_detections = None | |
| if webrtc_ctx.video_processor: | |
| with webrtc_ctx.video_processor.lock: | |
| realtime_detections = webrtc_ctx.video_processor.last_detections | |
| if realtime_detections is not None: | |
| st.markdown("### π‘οΈ Safety Status") | |
| advice, safety_level = get_navigation_advice(realtime_detections, image_width=640) | |
| if "CRITICAL" in safety_level: | |
| st.markdown(f'<div class="safety-alert"><strong>{safety_level}</strong></div>', unsafe_allow_html=True) | |
| elif "CAUTION" in safety_level: | |
| st.markdown(f'<div style="background: #fff3cd; color: #856404; padding: 1rem; border-radius: 8px; border-left: 4px solid #ffc107;"><strong>{safety_level}</strong></div>', unsafe_allow_html=True) | |
| else: | |
| st.markdown(f'<div class="success-alert"><strong>{safety_level}</strong></div>', unsafe_allow_html=True) | |
| st.markdown("### π§ Navigation Advice") | |
| for advice_text in advice: | |
| st.markdown(f'<div class="detection-box">{advice_text}</div>', unsafe_allow_html=True) | |
| if len(realtime_detections) > 0: | |
| st.markdown("### π Latest Detections") | |
| display_df = realtime_detections[['name', 'confidence', 'xmin', 'ymin', 'xmax', 'ymax']].copy() | |
| display_df['confidence'] = display_df['confidence'].apply(lambda x: f"{x:.1%}") | |
| display_df.columns = ['Object', 'Confidence', 'X Min', 'Y Min', 'X Max', 'Y Max'] | |
| st.dataframe(display_df, use_container_width=True) | |
| with tab3: | |
| st.markdown("### π Detection Analytics") | |
| if uploaded_file is not None and 'detections' in locals() and detections is not None: | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| chart1 = create_detection_chart(detections) | |
| if chart1: | |
| st.plotly_chart(chart1, use_container_width=True) | |
| with col2: | |
| chart2 = create_confidence_chart(detections) | |
| if chart2: | |
| st.plotly_chart(chart2, use_container_width=True) | |
| # Detection statistics | |
| st.markdown("### π Statistics") | |
| if len(detections) > 0: | |
| stats_col1, stats_col2, stats_col3 = st.columns(3) | |
| with stats_col1: | |
| avg_conf = detections['confidence'].mean() | |
| st.metric("π Average Confidence", f"{avg_conf:.1%}") | |
| with stats_col2: | |
| max_conf = detections['confidence'].max() | |
| st.metric("π― Highest Confidence", f"{max_conf:.1%}") | |
| with stats_col3: | |
| unique_classes = detections['name'].nunique() | |
| st.metric("π·οΈ Unique Object Types", unique_classes) | |
| else: | |
| st.info("πΈ Upload an image in the 'Live Detection' tab to see analytics") | |
| with tab4: | |
| st.markdown("### βΉοΈ About This System") | |
| col1, col2 = st.columns([2, 1]) | |
| with col1: | |
| st.markdown(""" | |
| #### 𦽠AI Wheelchair Navigation System | |
| This intelligent computer vision system is designed to assist wheelchair users with safe navigation by detecting and identifying potential obstacles, people, vehicles, and navigation signals in real-time. | |
| **π― Key Features:** | |
| - **Real-time Object Detection**: Identifies 12 wheelchair-relevant object types | |
| - **Safety Warnings**: Provides immediate alerts for potential hazards | |
| - **Navigation Guidance**: Offers contextual advice for safe path planning | |
| - **Raspberry Pi Optimized**: Lightweight model for edge deployment | |
| - **High Accuracy**: 87.5% mAP@0.5 overall accuracy | |
| **π§ Technical Specifications:** | |
| - **Model**: YOLOv5s (Small) - optimized for speed and accuracy | |
| - **Input Size**: 416x416 pixels | |
| - **Model Size**: 14.7 MB (perfect for embedded systems) | |
| - **Target Platform**: Raspberry Pi 4 | |
| - **Processing Speed**: 5-10 FPS on Raspberry Pi | |
| **π Graduation Project Context:** | |
| This system represents a comprehensive computer vision solution for assistive technology, demonstrating: | |
| - Advanced deep learning implementation | |
| - Edge computing optimization | |
| - Real-world application development | |
| - Safety-critical system design | |
| """) | |
| with col2: | |
| st.markdown(""" | |
| #### π Model Performance | |
| **Overall Metrics:** | |
| - mAP@0.5: 87.5% | |
| - mAP@0.5:0.95: 63.6% | |
| - Precision: 88.6% | |
| - Recall: 80.8% | |
| **Class-Specific Performance:** | |
| - Person: 84.4% mAP | |
| - Chair: 93.9% mAP | |
| - Vehicle: 56.3% mAP | |
| - Bicycle: 80.9% mAP | |
| **π Safety Features:** | |
| - Emergency obstacle detection | |
| - Multi-level alert system | |
| - Contextual navigation advice | |
| - Real-time processing | |
| """) | |
| st.markdown("### π οΈ Technology Stack") | |
| tech_col1, tech_col2, tech_col3, tech_col4 = st.columns(4) | |
| with tech_col1: | |
| st.markdown(""" | |
| **π§ AI/ML** | |
| - YOLOv5 | |
| - PyTorch | |
| - OpenCV | |
| - NumPy | |
| """) | |
| with tech_col2: | |
| st.markdown(""" | |
| **π Web App** | |
| - Streamlit | |
| - Plotly | |
| - PIL/Pillow | |
| - Pandas | |
| """) | |
| with tech_col3: | |
| st.markdown(""" | |
| **β‘ Deployment** | |
| - Hugging Face Spaces | |
| - Docker | |
| - Git LFS | |
| - ONNX (optional) | |
| """) | |
| with tech_col4: | |
| st.markdown(""" | |
| **π§ Hardware** | |
| - Raspberry Pi 4 | |
| - USB/Pi Camera | |
| - MicroSD Storage | |
| - Power Supply | |
| """) | |
| with tab5: | |
| st.markdown("### π Deployment Information") | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| st.markdown(""" | |
| #### π¦ Hugging Face Spaces Deployment | |
| This application is deployed on Hugging Face Spaces, providing: | |
| - **Free hosting** for demonstration purposes | |
| - **Easy sharing** with project evaluators | |
| - **Scalable infrastructure** for multiple users | |
| - **Integrated CI/CD** for automatic updates | |
| **π Deployment Features:** | |
| - Real-time inference on uploaded images | |
| - Interactive web interface | |
| - Performance analytics and visualization | |
| - Mobile-responsive design | |
| """) | |
| with col2: | |
| st.markdown(""" | |
| #### π Local/Raspberry Pi Deployment | |
| For real wheelchair deployment: | |
| 1. **Download the deployment package** | |
| 2. **Transfer to Raspberry Pi** | |
| 3. **Install dependencies** | |
| 4. **Connect camera** | |
| 5. **Run inference script** | |
| **π Requirements:** | |
| - Raspberry Pi 4 (4GB RAM recommended) | |
| - Python 3.7+ | |
| - PyTorch (CPU version) | |
| - USB Camera or Pi Camera | |
| """) | |
| st.markdown("### π» Code Repository") | |
| st.markdown(""" | |
| **π Project Structure:** | |
| ``` | |
| wheelchair_deployment/ | |
| βββ wheelchair_model.pt # Trained model weights | |
| βββ wheelchair_inference.py # Raspberry Pi inference script | |
| βββ requirements_rpi.txt # Dependencies | |
| βββ README_deployment.md # Setup instructions | |
| βββ wheelchair_config.yaml # Model configuration | |
| ``` | |
| """) | |
| st.markdown("### π€ Integration Guidelines") | |
| st.markdown(""" | |
| **For Wheelchair Integration:** | |
| 1. **Motor Control Interface**: Connect detection results to wheelchair motor control system | |
| 2. **Safety Protocols**: Implement emergency stop and collision avoidance | |
| 3. **User Interface**: Add audio/visual feedback for navigation guidance | |
| 4. **Sensor Fusion**: Combine with ultrasonic/LiDAR sensors for enhanced safety | |
| 5. **Custom Training**: Collect and label wheelchair-specific navigation data | |
| """) | |
| if __name__ == "__main__": | |
| main() |