# Project export: NeuroDrive This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2025 - Tagline: NeuroDrive helps you drive safer and longer by tracking your brain waves and surroundings, gently alerting you when you're distracted or at risk, so you and your loved ones enjoy the ride with peace. - Devpost: https://devpost.com/software/neurodrive-xl3iur - GitHub: https://github.com/deanzahci/inner-eye-drive - Team: 4 GitHub contributor(s) — Koichi Nakayamada (18 commits), AlanTMa (8 commits), TheGrandBraker (5 commits), shahxsheel (1 commits) ## Devpost submission (written by the team) ### Inspiration Randomly, we encountered an unexpected statistic: 1 in 25 adults has dozed off while driving in the previous month. This made us discover that fatigue-related driving is responsible for more than 100,000 accidents annually in the U.S., with a significant number involving long-haul operators. We were amazed at how frequently fatigue and distraction remain overlooked until it's too late — particularly on lengthy journeys. That’s when we started considering if we could create a system that not only comprehends the road but also focuses on the driver. That concept formed the basis for NeuroGuardian. ### What it does Data Entries An EEG headset was utilized to track the driver’s cognitive condition in real time, recording signals indicative of attention and fatigue. In addition, we installed a dual-camera system: one directed at the driver to monitor behavior, and another aimed at the road to collect traffic and lane information. Detection Workflow We incorporated YOLOv8 for identifying objects and utilized UltraFast Lane Detection to analyze the drivable region. These outputs were merged to categorize objects according to the lane zone where they were detected. At the same time, EEG signals were examined to identify states of distraction or fatigue. Results & Notifications If the system identifies distraction or potential risk in the driver's lane, it activates a voice alert through a text-to-speech system (using Gemini). The alert aims to be simple yet impactful, assisting in redirecting the driver’s attention without causing them to feel overloaded ### How we built it NeuroGuardian is an assistant for driver awareness that integrates brainwave signals with computer vision to minimize distraction and fatigue. An EEG headset tracks the driver's concentration in real time, supplemented by a dual-camera system—one directed at the road and the other observing the driver. It identifies lane boundaries and detects objects with YOLOv8, subsequently classifying those objects according to their location in relation to the driving lane. If it senses a lapse in attention or a possible danger in a crucial area, it provides a brief voice alert via a text-to-speech system to redirect the driver’s focus to the road. ### Challenges we ran into We faced issues with training and fine-tuning models due to limited processing power. Additionally, finding data sets for our EEG model was difficult. ### Accomplishments we're proud of We are proud of combining both EEG technology and CV to protect drivers, which hasn't been done before. ### What we learned We learned how to reduce the noise in data. The data and dataset for all our AI models were noisy, so we've learned the importance and a lot of techniques in it while tackling the data analysis. ### What's next We would like to use a smaller BCI device such as Muse2, but the data quality is low, so we also need to improve the machine learning model. We will also need feedback from the users so that we can find further improvements. ## README (from the GitHub repository) # NeuroDrive **UC Berkeley AI Hackathon 2025 Project** A real-time driver monitoring system that integrates computer vision and EEG analysis to detect distraction, drowsiness, and potential hazards for enhanced driving safety. ## 🏆 Project Vision We win these - Building the future of intelligent driver assistance systems. ## 🏗️ Architecture ### Microservices Design ``` ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ Distraction CV │ │ Dizziness EEG │ │ Object Detection│ │ Port 8001 │ │ Port 8002 │ │ Port 8003 │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ └───────────────────────┼───────────────────────┘ │ ┌─────────────────┐ │ Main GUI │ │ (Tkinter) │ └─────────────────┘ ``` ### Core Services 1. **Distraction CV Service** (`services/distraction_cv.py`) - Real-time eye tracking and attention detection - Head pose estimation - WebSocket streaming on port 8001 2. **Dizziness EEG Service** (`services/dizziness_eeg.py`) - EEG signal processing for drowsiness detection - Brain wave analysis (Alpha, Beta, Theta) - WebSocket streaming on port 8002 3. **Object Detection Service** (`services/object_detection_cv.py`) - Real-time road hazard detection - Vehicle and pedestrian identification - WebSocket streaming on port 8003 4. **Main Integration** (`main.py`) - Tkinter-based GUI for unified monitoring - Real-time data aggregation from all services - Central control and visualization ## 🚀 Quick Start ### Prerequisites - Python 3.8+ - pip package manager ### Installation 1. **Clone the repository** ```bash git clone cd inner-eye-drive ``` 2. **Install dependencies** ```bash pip install -r requirements.txt ``` ### Running the System 1. **Start all services** (in separate terminals): ```bash # Terminal 1 - Distraction CV python3 services/distraction_cv.py # Terminal 2 - Dizziness EEG python3 services/dizziness_eeg.py # Terminal 3 - Object Detection python3 services/object_detection_cv.py ``` 2. **Launch the main application**: ```bash # Terminal 4 - Main GUI python3 main.py ``` 3. **Test with multi-client** (optional): ```bash # View all service outputs simultaneously python3 multi_client.py ``` ## 🔧 API Endpoints ### Service Health Checks - **Distraction CV**: `GET http://127.0.0.1:8001/` - **Dizziness EEG**: `GET http://127.0.0.1:8002/` - **Object Detection**: `GET http://127.0.0.1:8003/` ### WebSocket Endpoints - **Distraction CV**: `ws://127.0.0.1:8001/ws` - **Dizziness EEG**: `ws://127.0.0.1:8002/ws` - **Object Detection**: `ws://127.0.0.1:8003/ws` ## 📁 Project Structure ``` inner-eye-drive/ ├── main.py # Main GUI application ├── multi_client.py # Multi-service test client ├── client.py # Single service test client ├── requirements.txt # Python dependencies ├── README.md # Project documentation ├── services/ │ ├── distraction_cv.py # Eye tracking & attention detection │ ├── dizziness_eeg.py # EEG drowsiness detection │ └── object_detection_cv.py # Road hazard detection └── models/ # ML models (future implementation) ``` ## 🛠️ Development Status ### ✅ Completed - [x] Microservices architecture setup - [x] FastAPI + WebSocket infrastructure - [x] Real-time communication between services - [x] Basic Tkinter GUI framework - [x] Multi-service client testing ### 🚧 In Progress - [ ] Computer vision algorithms implementation - [ ] EEG signal processing integration - [ ] Machine learning model integration - [ ] Advanced GUI with real-time visualizations ### 📋 Planned Features - [ ] Real-time driver alerting system - [ ] Data logging and analytics - [ ] Mobile app integration - [ ] Cloud-based monitoring dashboard ## 🤝 Contributing This project was developed during the UC Berkeley AI Hackathon 2025. ## 📄 License [Add your license information here] --- **Built with ❤️ at UC Berkeley AI Hackathon 2025** ## Detected evidence (automated analysis) Indexed codebase: 28 recognized source files, 167 KB. - FastAPI (technology) — detected in the code - Python (language) — detected in the code - PyTorch (technology) — detected in the code - TensorFlow (technology) — detected in the code ## Codebase structure (from repository index) ### Files (33 of 33) ``` .gitignore check_dependencies.py client.py main.py models/distraction_cv/config_vgg16.py models/distraction_cv/download_model.py models/distraction_cv/README.md models/distraction_cv/requirements.txt models/distraction_cv/run_vgg16_detection.py models/dizziness_nn/acquisition.py models/dizziness_nn/config.py models/dizziness_nn/dizziness_eeg.py models/dizziness_nn/model_weights.pth models/drowsiness_cv/config.py models/drowsiness_cv/drowsinessCV.py models/drowsiness_cv/install.py models/drowsiness_cv/run_drowsiness_detector.py models/object_cv/yolov8n.pt not-used/distraction_cv/config.py not-used/distraction_cv/distraction_detector.py not-used/distraction_cv/run_distraction_detection.py not-used/distraction_cv/test_system.py not-used/MongoDB/database.py README.md requirements.txt services/distraction_cv_minimal.py services/distraction_cv.py services/dizziness_eeg_minimal.py services/dizziness_eeg.py services/object_detection_cv_minimal.py services/object_detection_cv.py test_drowsiness_api.py test_eeg_api.py ``` ### Dependencies - models/distraction_cv/requirements.txt: numpy@>=1.21.0, opencv-python@>=4.5.0, tensorflow@>=2.10.0 - requirements.txt: addict, fastapi@>=0.100.0, matplotlib@>=3.5.0, mediapipe@>=0.10.0, numpy, opencv-python, pathspec, pydantic@>=2.0.0, pyOpenBCI, pyserial, requests, scikit-learn, scipy, shapely, tensorboard, torch, torchvision, tqdm, ultralytics, uvicorn@>=0.20.0, websockets@>=11.0 ### Recent commits (newest first) - commit again - Merge branch 'main' of https://github.com/deanzahci/inner-eye-drive - Add EEG dizziness detection service with simulation and WebSocket support - Update requirements.txt to include additional dependencies for EEG acquisition - Update distraction detection: confidence threshold, flexible model loading, and improved overlay for high-confidence predictions only - Add Driver Distraction Detection and Drowsiness Detection Services - Update requirements.txt - Add Driver Distraction Detection system with real-time processing and configuration - Add EEG data acquisition system with OpenBCI integration and random data fallback - Merge branch 'main' of https://github.com/deanzahci/inner-eye-drive - upd - Create acquisition.py - Merge branch 'main' of https://github.com/deanzahci/inner-eye-drive - Implement EEG monitoring with WebSocket integration and real-time data visualization - a - misspelling - dizziness model - test - mine - update ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` # Core dependencies for the current system fastapi>=0.100.0 uvicorn>=0.20.0 websockets>=11.0 pydantic>=2.0.0 matplotlib>=3.5.0 # Computer Vision dependencies opencv-python mediapipe>=0.10.0 tqdm tensorboard addict scikit-learn pathspec shapely ultralytics requests # Deep Learning dependencies torch torchvision numpy scipy # EEG acquisition dependencies pyOpenBCI pyserial ``` ### models/distraction_cv/requirements.txt ``` tensorflow>=2.10.0 opencv-python>=4.5.0 numpy>=1.21.0 ``` ### main.py ```python import tkinter as tk from tkinter import ttk import webbrowser import threading import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg from matplotlib.figure import Figure import asyncio import websockets import json import time from collections import deque import client is_started = False client_thread = None eeg_thread = None # EEG data storage eeg_data = deque(maxlen=100) # Store last 100 EEG readings time_data = deque(maxlen=100) current_eeg_state = 0 class EEGClient: def __init__(self, update_callback): self.update_callback = update_callback self.running = False self.loop = None async def connect_to_eeg_service(self): uri = "ws://127.0.0.1:8002/ws" try: async with websockets.connect(uri, ping_timeout=20, ping_interval=20) as websocket: print("[EEG] Connected to EEG service") await websocket.send("hello") while self.running: try: message = await asyncio.wait_for(websocket.recv(), timeout=1.0) data = json.loads(message) if 'state' in data: self.update_callback(data['state']) except asyncio.TimeoutError: continue except websockets.exceptions.ConnectionClosed: print("[EEG] Connection closed") break except json.JSONDecodeError: print(f"[EEG] Invalid JSON: {message}") except Exception as e: print(f"[EEG] Connection failed: {e}") def start(self): self.running = True self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) try: self.loop.run_until_complete(self.connect_to_eeg_service()) except Exception as e: print(f"[EEG] Error: {e}") finally: self.loop.close() def stop(self): self.running = False if self.loop and not self.loop.is_closed(): self.loop.call_soon_threadsafe(self.loop.stop) eeg_client = None def update_eeg_data(state_value): """Callback function to update EEG data from WebSocket""" global eeg_data, time_data, current_eeg_state current_time = time.time() current_eeg_state = state_value eeg_data.append(state_value) time_data.append(current_time) # Update the plot (this will be called from a different thread) root.after(0, update_plot) def update_plot(): """Update the matplotlib plot with new EEG data""" if len(eeg_data) > 0: ax.clear() # Convert time data to relative seconds for better visualization if len(time_data) > 0: start_time = time_data[0] relative_times = [(t - start_time) for t in time_data] else: relative_times = list(range(len(eeg_data))) # Plot EEG data ax.plot(relative_times, list(eeg_data), 'b-', linewidth=2, marker='o', markersize=3) ax.set_title('Real-time EEG State Monitoring', fontsize=14, fontweight='bold') ax.set_xlabel('Time (seconds)') ax.set_ylabel('EEG State') ax.set_ylim(-0.5, 3.5) ax.grid(True, alpha=0.3) # Add colored zones for different states ax.axhspan(-0.5, 0.5, alpha=0.1, color='green', label='Alert') ax.axhspan(0.5, 1.5, alpha=0.1, color='yellow', label='Normal') ax.axhspan(1.5, 2.5, alpha=0.1, color='orange', label='Tired') ax.axhspan(2.5, 3.5, alpha=0.1, color='red', label='Drowsy') # Add current state indicator if len(eeg_data) > 0: latest_value = eeg_data[-1] ax.axhline(y=latest_value, color='black', linestyle='--', alpha=0.8, linewidth=1) ax.text(0.02, 0.98, f'Current State: {latest_value:.1f}', transform=ax.transAxes, verticalalignment='top', bbox=dict(boxstyle='round', facecolor='white', alpha=0.8)) canvas.draw() def get_state_description(state): """Get human-readable description of EEG state""" if state <= 0.5: return "Alert & Focused" elif state <= 1.5: return "Normal Attention" elif state <= 2.5: return "Getting Tired" else: return "Drowsy - Alert Needed!" def update_button(): if is_started: start_stop_button.config(text="Stop", command=stop) else: start_stop_button.config(text="Start", command=start) def start(): global is_started, client_thread, eeg_thread, eeg_client print("Starting services...") is_started = True # Run main client in a separate thread client_thread = threading.Thread(target=client.start, daemon=True) client_thread.start() # Run EEG client in a separate thread for the plot eeg_client = EEGClient(update_eeg_data) eeg_thread = threading.Thread(target=eeg_client.start, daemon=True) eeg_thread.start() update_button() def stop(): global is_started, client_thread, eeg_client, eeg_thread print("Stopping services...") is_started = False # Stop main client client.stop() if client_thread and client_thread.is_alive(): client_thread.join(timeout=2) # Stop EEG client if eeg_client: eeg_client.stop() if eeg_thread and eeg_thread.is_alive(): eeg_thread.join(timeout=2) update_button() root = tk.Tk() root.title("Inner Eye Drive - EEG Monitor") root.geometry("900x700") # Configure grid weights so the plot can expand root.grid_columnconfigure(0, weight=1) root.grid_columnconfigure(1, weight=1) root.grid_rowconfigure(1, weight=1) # Create control frame control_frame = ttk.Frame(root) control_frame.grid(row=0, column=0, columnspan=2, padx=10, pady [truncated — 1984 more characters] ``` ### client.py ```python import asyncio import websockets import json import threading # Global variable to control the running state running = False loop = None async def connect_to_service(service_name, port): global running uri = f"ws://127.0.0.1:{port}/ws" print(f"[{service_name}] Attempting to connect to {uri}") try: async with websockets.connect(uri, ping_timeout=20, ping_interval=20) as websocket: print(f"[{service_name}] Successfully connected to service on port {port}") # Send initial message to establish connection (needed for some services) await websocket.send("hello") while running: try: message = await asyncio.wait_for(websocket.recv(), timeout=1.0) data = json.loads(message) print(f"[{service_name}] Received: {data}") except asyncio.TimeoutError: # Timeout allows us to check the running flag periodically continue except websockets.exceptions.ConnectionClosed: print(f"[{service_name}] Connection closed") break except json.JSONDecodeError: print(f"[{service_name}] Invalid JSON: {message}") except ConnectionRefusedError: print(f"[{service_name}] Connection refused - service not running on port {port}") except Exception as e: print(f"[{service_name}] Connection failed: {e}") print(f"[{service_name}] Service connection ended") async def main(): global running services = [ ("DISTRACTION_CV", 8001), ("DIZZINESS_EEG", 8002), ("OBJECT_DETECTION", 8003) ] tasks = [] for service_name, port in services: task = asyncio.create_task(connect_to_service(service_name, port)) tasks.append(task) try: await asyncio.gather(*tasks) except asyncio.CancelledError: print("Client tasks cancelled") def start(): global running, loop running = True loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(main()) except KeyboardInterrupt: print("Client interrupted") finally: loop.close() def stop(): global running, loop running = False if loop and not loop.is_closed(): # Schedule the loop to stop loop.call_soon_threadsafe(loop.stop) print("Client stopped") ``` ### test_drowsiness_api.py ```python #!/usr/bin/env python3 """ Test script for Drowsiness Detection API Integration """ import asyncio import websockets import json import requests import time async def test_websocket(): """Test WebSocket connection to drowsiness detection service""" uri = "ws://127.0.0.1:8001/ws" print("🔌 Testing WebSocket connection...") try: async with websockets.connect(uri) as websocket: print("✅ Connected to Drowsiness Detection WebSocket") # Send hello message await websocket.send("hello") # Listen for a few messages for i in range(10): try: message = await asyncio.wait_for(websocket.recv(), timeout=5.0) data = json.loads(message) print(f"📊 Detection Data {i+1}:") print(f" Face Detected: {data.get('face_detected', 'N/A')}") print(f" Drowsy: {data.get('drowsy', 'N/A')}") print(f" Alert Level: {data.get('alert_level', 'N/A')}") print(f" Avg EAR: {data.get('avg_ear', 0):.3f}") print(f" Total Blinks: {data.get('total_blinks', 'N/A')}") print() except asyncio.TimeoutError: print("⏰ Timeout waiting for message") break except json.JSONDecodeError: print(f"❌ Invalid JSON: {message}") except Exception as e: print(f"❌ WebSocket connection failed: {e}") def test_http_endpoints(): """Test HTTP endpoints""" base_url = "http://127.0.0.1:8001" print("🌐 Testing HTTP endpoints...") # Test root endpoint try: response = requests.get(f"{base_url}/") print(f"✅ Root endpoint: {response.json()}") except Exception as e: print(f"❌ Root endpoint failed: {e}") # Test health endpoint try: response = requests.get(f"{base_url}/health") print(f"✅ Health endpoint: {response.json()}") except Exception as e: print(f"❌ Health endpoint failed: {e}") # Test status endpoint try: response = requests.get(f"{base_url}/status") print(f"✅ Status endpoint: {response.json()}") except Exception as e: print(f"❌ Status endpoint failed: {e}") def main(): print("🧪 Drowsiness Detection API Test") print("=" * 50) print("Make sure the service is running: python services/distraction_cv.py") print() # Test HTTP endpoints first test_http_endpoints() print("\n" + "=" * 50) # Test WebSocket asyncio.run(test_websocket()) print("🏁 Test complete!") if __name__ == "__main__": main() ``` ### test_eeg_api.py ```python #!/usr/bin/env python3 """ Test script for Dizziness EEG API Integration """ import asyncio import websockets import json import requests import time async def test_websocket(): """Test WebSocket connection to dizziness EEG service""" uri = "ws://127.0.0.1:8002/ws" print("🔌 Testing WebSocket connection...") try: async with websockets.connect(uri) as websocket: print("✅ Connected to Dizziness EEG WebSocket") # Send hello message await websocket.send("hello") # Listen for a few messages for i in range(10): try: message = await asyncio.wait_for(websocket.recv(), timeout=10.0) data = json.loads(message) print(f"🧠 EEG Analysis {i+1}:") print(f" Dominant State: {data.get('state_name', 'N/A')}") print(f" Confidence: {data.get('confidence', 0):.3f}") print(f" Probabilities:") probs = data.get('dizziness_probabilities', {}) for state, prob in probs.items(): print(f" {state.capitalize()}: {prob:.3f}") print(f" Sample Count: {data.get('sample_count', 'N/A')}") print(f" Window Duration: {data.get('window_duration', 0):.3f}s") print() except asyncio.TimeoutError: print("⏰ Timeout waiting for message") break except json.JSONDecodeError: print(f"❌ Invalid JSON: {message}") except Exception as e: print(f"❌ WebSocket connection failed: {e}") def test_http_endpoints(): """Test HTTP endpoints""" base_url = "http://127.0.0.1:8002" print("🌐 Testing HTTP endpoints...") # Test root endpoint try: response = requests.get(f"{base_url}/") print(f"✅ Root endpoint: {response.json()}") except Exception as e: print(f"❌ Root endpoint failed: {e}") # Test health endpoint try: response = requests.get(f"{base_url}/health") print(f"✅ Health endpoint: {response.json()}") except Exception as e: print(f"❌ Health endpoint failed: {e}") # Test status endpoint try: response = requests.get(f"{base_url}/status") print(f"✅ Status endpoint: {response.json()}") except Exception as e: print(f"❌ Status endpoint failed: {e}") def main(): print("🧪 Dizziness EEG API Test") print("=" * 50) print("Make sure the service is running: python services/dizziness_eeg.py") print() # Test HTTP endpoints first test_http_endpoints() print("\n" + "=" * 50) # Test WebSocket asyncio.run(test_websocket()) print("🏁 Test complete!") if __name__ == "__main__": main() ``` ### check_dependencies.py ```python #!/usr/bin/env python3 """ Dependency checker for Inner Eye Drive Object Detection Service Run this script to verify all required dependencies are available. """ import sys import os import importlib.util def check_import(module_name, package_name=None): """Check if a module can be imported""" try: __import__(module_name) print(f"✅ {package_name or module_name}") return True except ImportError as e: print(f"❌ {package_name or module_name}: {e}") return False def check_file_exists(file_path, description): """Check if a file exists""" if os.path.exists(file_path): print(f"✅ {description}: {file_path}") return True else: print(f"❌ {description}: {file_path} (NOT FOUND)") return False def main(): print("🔍 Checking Inner Eye Drive Dependencies...") print("=" * 50) # Core Python packages print("\n📦 Core Python Packages:") core_packages = [ ("fastapi", "FastAPI"), ("uvicorn", "Uvicorn"), ("asyncio", "AsyncIO"), ("json", "JSON"), ("websockets", "WebSockets"), ] core_ok = all(check_import(pkg, name) for pkg, name in core_packages) # Computer Vision packages print("\n🖼️ Computer Vision Packages:") cv_packages = [ ("cv2", "OpenCV"), ("mediapipe", "MediaPipe"), ("torch", "PyTorch"), ("torchvision", "TorchVision"), ("numpy", "NumPy"), ("PIL", "Pillow"), ("scipy", "SciPy"), ("ultralytics", "YOLOv8"), ("shapely", "Shapely"), ] cv_ok = all(check_import(pkg, name) for pkg, name in cv_packages) # Ultra-Fast-Lane-Detection specific print("\n🛣️ Lane Detection Model:") ultra_fast_path = os.path.join(os.path.dirname(__file__), 'models', 'object_cv', 'Ultra-Fast-Lane-Detection') sys.path.append(ultra_fast_path) lane_packages = [] try: from model.model import parsingNet print("✅ parsingNet model") lane_packages.append(True) except ImportError as e: print(f"❌ parsingNet model: {e}") lane_packages.append(False) try: from utils.config import Config print("✅ Config utils") lane_packages.append(True) except ImportError as e: print(f"❌ Config utils: {e}") lane_packages.append(False) lane_ok = all(lane_packages) # Required files print("\n📁 Required Files:") config_path = os.path.join(ultra_fast_path, 'configs', 'tusimple.py') model_path = os.path.join(ultra_fast_path, 'weights', 'tusimple_18.pth') yolo_path = os.path.join(ultra_fast_path, 'yolov8n.pt') files_ok = all([ check_file_exists(config_path, "TuSimple Config"), check_file_exists(model_path, "Lane Detection Weights"), check_file_exists(yolo_path, "YOLO Model") or check_file_exists('yolov8n.pt', "YOLO Model (fallback)") ]) # Camera check print("\n📹 Camera Check:") try: import cv2 cap = cv2.VideoCapture(0) if cap.isOpened(): print("✅ Camera is available") cap.release() camera_ok = True else: print("⚠️ Camera not available (service will use fallback mode)") camera_ok = False except Exception as e: print(f"❌ Camera check failed: {e}") camera_ok = False # Summary print("\n" + "=" * 50) print("📋 DEPENDENCY SUMMARY:") print(f" Core packages: {'✅ OK' if core_ok else '❌ MISSING'}") print(f" CV packages: {'✅ OK' if cv_ok else '❌ MISSING'}") print(f" Lane detection: {'✅ OK' if lane_ok else '❌ MISSING'}") print(f" Required files: {'✅ OK' if files_ok else '❌ MISSING'}") print(f" Camera: {'✅ OK' if camera_ok else '⚠️ UNAVAILABLE'}") if core_ok and cv_ok: if lane_ok and files_ok: print("\n🎉 ALL SYSTEMS GO! You can run the full service.") else: print("\n⚠️ PARTIAL FUNCTIONALITY: Service will run with YOLO-only mode.") else: print("\n❌ MISSING CRITICAL DEPENDENCIES: Please install missing packages.") print("\n💡 Installation suggestions:") if not core_ok: print(" pip install fastapi uvicorn websockets") if not cv_ok: print(" pip install opencv-python torch torchvision ultralytics shapely scipy pillow numpy") print("\n🚀 To start the service: python services/object_detection_cv.py") if __name__ == "__main__": main() ``` ### services/object_detection_cv_minimal.py ```python #!/usr/bin/env python3 """ Minimal test version of object detection service This version only uses YOLO and basic dependencies """ from fastapi import FastAPI, WebSocket import asyncio import json import uvicorn app = FastAPI() class ConnectionManager: def __init__(self): self.active_connections = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) manager = ConnectionManager() @app.get("/") async def root(): return {"message": "Object Detection CV Service - Minimal Mode"} @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except: manager.disconnect(websocket) async def minimal_cv_processing(): """Minimal computer vision with just basic object detection""" try: import cv2 from ultralytics import YOLO # Try to load YOLO yolo_model = YOLO('yolov8n.pt') # This will download if not available cap = cv2.VideoCapture(0) print("✅ Minimal CV mode: YOLO + Camera ready") while True: if cap.isOpened(): ret, frame = cap.read() if ret: # Basic YOLO detection results = yolo_model(frame, conf=0.25) objects = [] if len(results[0].boxes) > 0: boxes = results[0].boxes.xyxy.cpu().numpy() classes = results[0].boxes.cls.cpu().numpy() confs = results[0].boxes.conf.cpu().numpy() for box, cls, conf in zip(boxes, classes, confs): objects.append({ "object": yolo_model.names[int(cls)], "confidence": float(conf), "bbox": box.tolist() }) data = { "timestamp": asyncio.get_event_loop().time(), "objects": objects, "total_objects": len(objects), "mode": "minimal_cv" } await manager.broadcast(json.dumps(data)) await asyncio.sleep(0.1) # 10 FPS except Exception as e: print(f"Minimal CV failed: {e}, using simulation") await simulation_mode() async def simulation_mode(): """Pure simulation mode""" counter = 0 while True: counter += 1 data = { "timestamp": asyncio.get_event_loop().time(), "objects": [ {"object": "car", "confidence": 0.85, "bbox": [100, 100, 200, 200]}, {"object": "person", "confidence": 0.76, "bbox": [300, 150, 350, 300]} ], "total_objects": 2, "mode": "simulation" } await manager.broadcast(json.dumps(data)) await asyncio.sleep(1) @app.on_event("startup") async def startup_event(): print("🚀 Starting minimal object detection service...") asyncio.create_task(minimal_cv_processing()) if __name__ == "__main__": print("🔧 Minimal Object Detection CV Service") uvicorn.run(app, host="127.0.0.1", port=8003) ``` ### services/dizziness_eeg_minimal.py ```python #!/usr/bin/env python3 """ Minimal EEG dizziness detection service This version provides simulation when OpenBCI/PyTorch is not available """ from fastapi import FastAPI, WebSocket import asyncio import json import uvicorn import random import math import time import numpy as np app = FastAPI() class ConnectionManager: def __init__(self): self.active_connections = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) manager = ConnectionManager() DIZZINESS_LEVELS = { 0: "High", 1: "Moderate", 2: "Low", 3: "None" } @app.get("/") async def root(): return {"message": "Dizziness EEG Service - Minimal Simulation"} @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except: manager.disconnect(websocket) async def simulate_eeg_analysis(): """Simulate EEG analysis with realistic dizziness patterns""" print("🧠 Starting EEG dizziness simulation...") counter = 0 while True: counter += 1 # Simulate natural patterns time_factor = time.time() / 20 # Slow changes # Base probabilities base_none = 0.6 + 0.3 * math.sin(time_factor) base_low = 0.2 + 0.1 * math.sin(time_factor + 1) base_moderate = 0.15 + 0.1 * math.sin(time_factor + 2) base_high = 0.05 + 0.05 * math.sin(time_factor + 3) # Normalize probabilities total = base_none + base_low + base_moderate + base_high probs = [base_high/total, base_moderate/total, base_low/total, base_none/total] # Add some random variation for i in range(len(probs)): probs[i] += random.uniform(-0.05, 0.05) probs[i] = max(0, min(1, probs[i])) # Renormalize total = sum(probs) probs = [p/total for p in probs] # Determine dominant state dominant_state = np.argmax(probs) confidence = probs[dominant_state] # Generate analysis data analysis_data = { "timestamp": asyncio.get_event_loop().time(), "dizziness_probabilities": { "none": float(probs[3]), "low": float(probs[2]), "moderate": float(probs[1]), "high": float(probs[0]) }, "dominant_state": int(dominant_state), "state_name": DIZZINESS_LEVELS[dominant_state], "confidence": float(confidence), "window_duration": 1.024, # Simulated window duration "sample_count": 256, "channels": 8, "service": "dizziness_eeg", "mode": "simulation" } # Broadcast to all connected clients await manager.broadcast(json.dumps(analysis_data)) # Update every 2 seconds (simulating processing time) await asyncio.sleep(2) @app.get("/health") async def health_check(): return { "status": "healthy", "service": "Dizziness EEG Service - Simulation Mode", "model_loaded": True, "eeg_acquisition": "Simulation", "mode": "simulation" } @app.get("/status") async def get_status(): return { "service": "Dizziness EEG - Neural Network Analysis", "model_loaded": True, "eeg_source": "Simulation", "connections": len(manager.active_connections), "channels": 8, "sample_rate": 250, "window_size": 256, "mode": "simulation" } @app.on_event("startup") async def startup_event(): print("🧠 Starting minimal EEG dizziness analysis simulation...") asyncio.create_task(simulate_eeg_analysis()) if __name__ == "__main__": print("🚀 Starting Dizziness EEG Service (Simulation Mode) on port 8002") print("🧠 This version provides realistic EEG dizziness detection simulation") uvicorn.run(app, host="127.0.0.1", port=8002) ``` ### services/distraction_cv_minimal.py ```python #!/usr/bin/env python3 """ Minimal drowsiness detection service This version provides basic simulation when full computer vision is not available """ from fastapi import FastAPI, WebSocket import asyncio import json import uvicorn import random import math import time app = FastAPI() class ConnectionManager: def __init__(self): self.active_connections = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) manager = ConnectionManager() @app.get("/") async def root(): return {"message": "Distraction CV Service - Minimal Drowsiness Simulation"} @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except: manager.disconnect(websocket) async def simulate_drowsiness_detection(): """Simulate drowsiness detection with realistic patterns""" print("🎭 Starting drowsiness simulation mode...") counter = 0 total_blinks = 0 drowsiness_alerts = 0 while True: counter += 1 # Simulate natural blinking and occasional drowsiness base_ear = 0.3 # Normal open eye EAR # Add natural variation time_factor = time.time() / 10 # Slow variation ear_variation = 0.05 * math.sin(time_factor) # Simulate blinks (quick drops in EAR) if counter % 60 == 0: # Blink every ~2 seconds at 30fps total_blinks += 1 avg_ear = 0.15 # Closed eye EAR consecutive_frames = 2 is_drowsy = False alert_level = "normal" # Simulate drowsiness episodes (longer periods of low EAR) elif counter % 300 < 50: # Drowsy for 50 frames every 300 frames (~10 seconds) avg_ear = 0.18 + ear_variation consecutive_frames = (counter % 300) + 20 is_drowsy = consecutive_frames > 20 if consecutive_frames > 40: alert_level = "critical" if counter % 300 == 25: # Count alert once per episode drowsiness_alerts += 1 else: alert_level = "warning" else: # Normal alert state avg_ear = base_ear + ear_variation consecutive_frames = 0 is_drowsy = False alert_level = "normal" # Generate detection data detection_data = { "timestamp": asyncio.get_event_loop().time(), "face_detected": True, "drowsy": is_drowsy, "alert_level": alert_level, "left_ear": avg_ear + random.uniform(-0.02, 0.02), "right_ear": avg_ear + random.uniform(-0.02, 0.02), "avg_ear": avg_ear, "consecutive_closed_frames": consecutive_frames, "total_blinks": total_blinks, "drowsiness_alerts": drowsiness_alerts, "service": "drowsiness_detection", "mode": "simulation" } # Broadcast to all connected clients await manager.broadcast(json.dumps(detection_data)) # Run at ~30 FPS await asyncio.sleep(0.033) @app.get("/health") async def health_check(): return { "status": "healthy", "service": "Distraction CV Service - Simulation Mode", "mode": "simulation" } @app.get("/status") async def get_status(): return { "service": "Distraction CV - Drowsiness Detection", "detector_loaded": True, "camera_available": False, "connections": len(manager.active_connections), "mode": "simulation" } @app.on_event("startup") async def startup_event(): print("🎭 Starting minimal drowsiness detection simulation...") asyncio.create_task(simulate_drowsiness_detection()) if __name__ == "__main__": print("🚀 Starting Distraction CV Service (Simulation Mode) on port 8001") print("🎭 This version provides realistic drowsiness detection simulation") uvicorn.run(app, host="127.0.0.1", port=8001) ``` [18 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]