# Project export: CrashCourse 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: Cal Hacks 12.0 - Tagline: AI-powered VR driving simulator that analyzes your driving in real time using YOLOv8 and gives live coaching feedback through an AI voice agent. Build confidence in driving before you hit the road! - Devpost: https://devpost.com/software/crashcourse-edybia - GitHub: https://github.com/LucasStevenson/CrashCourse - Video: https://www.youtube.com/embed/LPhXPLBcztA?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — Lucas Stevenson (9 commits), Shashwat (4 commits) ## Devpost submission (written by the team) ### Inspiration Learning to drive can be stressful, expensive, and often inaccessible. Traditional driving lessons and instructors cost hundreds of dollars, and many beginners struggle to build confidence behind the wheel. We wanted to create a safe, affordable, and AI-powered virtual alternative, a system that helps people learn and improve their driving skills from anywhere. By combining VR immersion, AI feedback, and custom hardware, we aimed to build a simulator that feels realistic, builds confidence, and provides the same (or better) learning experience as a real instructor, without the cost or risk of being on the road. ### What it does The VR driving simulator has different experiences or tracks for drivers to try and test their skills. Each experience is built with a specific goal in mind such as merging lanes on a highway, turning at intersections, and much more. The game feed is sent to a YOLOv8 computer vision model, which performs real-time object detection (cars, signs, pedestrians, etc.) and extracts telemetry data (speed, steering angle, lane deviation, collisions). This data is analyzed by the AI to generate performance cues and feedback. These cues are passed to our custom Toolhouse AI agent that acts as a virtual driving coach, providing real-time commentary and post-session advice. The coach’s responses are converted into natural speech using Fish AI and played inside the VR environment, creating the effect of a live instructor talking to you while you drive. The coach also gives the driver a final score based on their overall performance along with tips to improve their driving! ### How we built it Hardware: Built a custom steering wheel system using an ESP32 microcontroller and rotary encoder for steering input. Established serial communication between the ESP32 and Unity for real-time input mapping. (The encoder was later discovered to be faulty, preventing final calibration — but hardware communication and Unity integration were functional.) Software & AI Pipeline: The driving simulator environment was created in Unity using the XR SDK for VR support. The gameplay feed was captured and sent to a Python FastAPI server running YOLOv8 for real-time object detection. Telemetry data (steering angle, acceleration, collisions) was streamed to the same endpoint for analysis. The AI model evaluated performance and generated driver cues such as “Slow down,” “Maintain lane,” or “Pedestrian ahead.” These cues were sent to a Toolhouse AI agent, which structured them into natural coaching dialogue and feedback. The final text was passed through Fish AI for text-to-speech conversion, and the resulting audio was played directly inside Unity. Tech Stack: Hardware: ESP32, Rotary Encoder, Serial over USB Software: Unity (C#), Python, FastAPI, YOLOv8, OpenCV AI Tools: Toolhouse AI (driving coach), Fish AI (text-to-speech) VR Platform: Oculus/Meta headset ### Challenges we ran into While building our custom steering wheel, we ran into multiple hardware related challenges such as our rotary encoder malfunctioning or being defective and not reading the pulses correctly. This prevented us from completing the steering input mapping. However, we pushed through and worked with what we had to connect the VR headset to the Unity driving simulation. Another challenge we had was integrating real-time object detection with Unity while maintaining performance in VR. This required significant optimization of the models. Managing synchronization between multiple systems (Unity, ESP32, YOLOv8, Toolhouse AI, Fish AI) in real time was complex. Building natural-sounding, contextual AI feedback that felt like a real driving coach required prompt engineering and tuning. ### Accomplishments we're proud of We are really proud that even through times of adversity when our hardware was not functioning properly, we did not give up and worked through creating this project to build something that integrated hardware, VR, and AI! We achieved real time analysis of the driving footage using the YOLOv8 model We also created a driving coach AI agent that talks to the user making the experience more human like for the driver. Finally we were able to almost wrap up creating a custom steering wheel for the VR headset driving. ### What we learned We learned a lot about choosing various AI models pretrained on hugging face and fine tuning them to our use case. Moreover, we learned how to use custom hardware such as ESP32 boards, rotary encoder, and connecting embedded systems with VR. ### What's next We plan on building more experiences in the driving simulation for people to use. We want to create all kinds of driving scenarios for our users to practice so that they feel prepared and confident before they hit the roads! ## README (from the GitHub repository) # CrashCourse - Driving Evaluation System A real-time driving evaluation and coaching system that analyzes driving video footage to assess driver safety and behavior. ## Features - Real-time object detection using YOLOv8 - Lane detection and departure warnings - Time-to-collision (TTC) estimation - Driving performance scoring with multiple dimensions: - Speeding violations - Lane keeping - Headway management - Smooth driving (harsh braking detection) - Traffic compliance (red lights, stop signs) - Real-time coaching cues - Multiple integration options (FastAPI, WebSocket, LiveKit) ## Project Structure ``` CrashCourse/ ├── ai/src/ # AI inference engine │ ├── api.py # FastAPI endpoints │ ├── detector.py # YOLOv8 object detection │ ├── rules.py # Scoring and cuing logic │ ├── lane_simple.py # Lane detection │ └── video_only.py # Vision-based utilities ├── backend/ # WebSocket backend │ └── app.py # WebSocket server └── livekit_backend/ # LiveKit integration └── livekit_backend.py ``` ## Setup ### 1. Install Dependencies ```bash pip install -r ai/requirements.txt ``` ### 2. Download YOLOv8 Model The YOLOv8n model will be automatically downloaded on first run, or you can place `yolov8n.pt` in the `ai/src/` directory. ### 3. Configure Environment (for LiveKit only) If using LiveKit integration: ```bash cd livekit_backend cp .env.example .env # Edit .env with your LiveKit credentials ``` ## Usage ### Option 1: FastAPI Server (Recommended) Start the inference API server: ```bash cd ai/src uvicorn api:app --host 0.0.0.0 --port 8000 ``` API Endpoints: - `POST /infer_frame` - Send frame + telemetry for inference - Parameters: - `image`: multipart image file - `telemetry`: JSON string with driving data - Returns: `{"cues": [...], "ttc": float, "detections": int}` - `POST /end_session` - Get final driving score - Returns: `{"subscores": {...}, "final": float, "violations": {...}}` ### Option 2: WebSocket Server Start the WebSocket server (automatically connects to FastAPI): ```bash # Terminal 1: Start FastAPI server cd ai/src uvicorn api:app --host 0.0.0.0 --port 8000 # Terminal 2: Start WebSocket server cd backend python app.py ``` The WebSocket server listens on `ws://localhost:8765` **Protocol:** 1. Send binary frame data (JPEG encoded) 2. Send JSON telemetry data 3. Receive real-time inference results 4. Send "DONE" message to get final score ### Option 3: LiveKit Integration For Unity/WebRTC integration: ```bash # Terminal 1: Start FastAPI server cd ai/src uvicorn api:app --host 0.0.0.0 --port 8000 # Terminal 2: Start LiveKit backend cd livekit_backend python livekit_backend.py ``` ## Telemetry Data Format ```json { "t": 1.5, // timestamp in seconds "speed_mps": 15.0, // current speed in m/s "speed_limit_mps": 13.4, // speed limit in m/s "throttle": 0.5, // throttle position (0-1) "brake": 0.0, // brake position (0-1) "steer_deg": -5.0, // steering angle in degrees "lane_offset_m": 0.2, // lane offset in meters (optional) "tl_state": "green", // traffic light state (optional) "in_stop_zone": false, // in stop zone (optional) "collision": false // collision detected (boolean) } ``` ## Coaching Cues The system generates the following real-time cues: - `SLOW_DOWN` - Speed exceeds limit - `KEEP_LANE` - Lane departure detected - `INCREASE_HEADWAY` - Following too closely (low TTC) - `SMOOTHER_BRAKE` - Harsh braking detected - `BRAKE_NOW` - Red light/stop sign violation imminent ## Scoring Final scores are calculated across 5 dimensions: - **Speeding** (25% weight): Time spent over speed limit - **Lane Keeping** (25% weight): Time spent out of lane - **Headway** (20% weight): Time with inadequate TTC - **Smoothness** (15% weight): Number of harsh braking events - **Compliance** (15% weight): Red light violations and collisions Each subscore ranges from 0-100, with the final score being a weighted average. ## Troubleshooting **Model not found error:** - Ensure `yolov8n.pt` is in `ai/src/` or let it auto-download - Check internet connection for first-time model download **Connection refused errors:** - Verify FastAPI server is running on port 8000 - Check firewall settings **No cues generated:** - Verify telemetry data format matches specification - Check that speed limits and thresholds are realistic ## Development To run tests with sample video: ```bash cd ai/src python replay_test.py # With synthetic telemetry python replay_video_only.py # Vision-only mode ``` ## License [Add your license here] ## Detected evidence (automated analysis) Indexed codebase: 14 recognized source files, 54 KB. - FastAPI (technology) — detected in the code - Python (language) — detected in the code - C (language) — claimed on Devpost, not found in the code - C# (language) — claimed on Devpost, not found in the code - PyTorch (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (24 of 24) ``` .gitignore ai/.env.example ai/requirements.txt ai/src/__init__.py ai/src/api.py ai/src/detector.py ai/src/lane_simple.py ai/src/replay_test.py ai/src/replay_video_only.py ai/src/rules.py ai/src/video_only.py ai/src/yolov8n.pt backend/.DS_Store backend/.env.example backend/app.py backend/requirements.txt backend/test.py backend/verbal_audio.py backend/yolov8n.pt livekit_backend/.env livekit_backend/.env.example livekit_backend/livekit_backend.py README.md send_mp4_ws.py ``` ### Dependencies - ai/requirements.txt: aiohttp@>=3.8, fastapi@>=0.111, livekit@>=0.10.0, numpy@>=1.26, opencv-python@>=4.9.0.80, pydantic@>=2.7, python-dotenv@>=1.0.0, python-multipart@>=0.0.9, ultralytics@==8.3.20, uvicorn[standard]@>=0.30, websockets@>=11.0 - backend/requirements.txt: PyAudio@==0.2.11 ### Recent commits (newest first) - voice stuff mostly works - tts connects to toolhouse output - deleted misc files - merged audio tts code - audio working - added toolhouse ai agent - merge conflicts resolved - updated model and testing - tryna test yolo cues - made it an actual websocket now - websocket stuff - yolov8 detection for driving evaluation - Initial commit ## Key source files (fetched from GitHub, selected and truncated for size) ### backend/requirements.txt ``` PyAudio==0.2.11 ``` ### ai/requirements.txt ``` ultralytics==8.3.20 opencv-python>=4.9.0.80 numpy>=1.26 fastapi>=0.111 uvicorn[standard]>=0.30 pydantic>=2.7 aiohttp>=3.8 websockets>=11.0 livekit>=0.10.0 python-dotenv>=1.0.0 python-multipart>=0.0.9 ``` ### backend/app.py ```python import asyncio import websockets import json import numpy as np import cv2 import aiohttp import os import time import uuid from dotenv import load_dotenv from verbal_audio import FishTTSStreamer from websockets.exceptions import ConnectionClosed, ConnectionClosedOK # Store frames and telemetry for each connection connections = {} # Load environment variables from a .env file (e.g., backend/.env) load_dotenv() FISHAUDIO_API_KEY = os.getenv("FISHAUDIO_API_KEY", "") VOICE_MODEL_ID = None # Toolhouse agent config (set via environment) TOOLHOUSE_URL = os.getenv("TOOLHOUSE_URL", "") TOOLHOUSE_API_KEY = os.getenv("TOOLHOUSE_API_KEY", "") # To conserve runs: throttle forwards and only send when cue changes FORWARD_MIN_INTERVAL_S = float(os.getenv("TOOLHOUSE_MIN_INTERVAL_S", "1.0")) # Max seconds to wait for final coach response before falling back locally TOOLHOUSE_FINAL_TIMEOUT_S = float(os.getenv("TOOLHOUSE_FINAL_TIMEOUT_S", "20.0")) # How to send payload to Toolhouse: 'wrapped' (default, uses 'message'), 'wrapped_input' (uses 'input'), or 'raw' PAYLOAD_STYLE = os.getenv("TOOLHOUSE_PAYLOAD_STYLE", "wrapped").lower() tts_streamer = FishTTSStreamer(FISHAUDIO_API_KEY, VOICE_MODEL_ID) def _bucket(val, step): try: return None if val is None else round(float(val) / step) * step except Exception: return None def _cue_fingerprint(obs): cue = obs.get("cue") or "" lvl = _bucket(obs.get("cue_level"), 0.2) return f"{cue}|{lvl}" async def forward_to_toolhouse(session: aiohttp.ClientSession, payload: dict, timeout_s: float = 5.0) -> dict | None: """Example output {"cue": "INCREASE_HEADWAY", "cue_level": 0.99, "message": "Increase following distance—too close to vehicle ahead.", "notes": null} """ if not TOOLHOUSE_URL: return None headers = {"Content-Type": "application/json"} if TOOLHOUSE_API_KEY: headers["Authorization"] = f"Bearer {TOOLHOUSE_API_KEY}" # Build a wrapped text prompt if requested, improves deterministic JSON replies body = payload if PAYLOAD_STYLE != "raw": obs = json.dumps(payload, ensure_ascii=False) if payload.get("event") == "session_end": # Session summary prompt prompt = ( "You are VR Driving Coach AI. Summarize the driver's overall session based on this final score JSON and respond with a SINGLE compact JSON object only, no prose. " "Final: " + obs + ". " "Output schema: {\"summary\": string, \"tips\": [string,string,string], \"drills\": [string,string,string], \"priority\": \"speeding|lane|headway|smooth|compliance\"}. " "Rules: (1) Keep summary <= 160 chars. (2) Tips must be short, actionable, and specific. (3) Choose priority based on weakest weighted dimension." ) else: # Realtime observation prompt prompt = ( "You are VR Driving Coach AI. Evaluate this driving observation JSON and respond with a SINGLE compact JSON object only, no prose. " "Observation: " + obs + ". " "Output schema: {\"cue\": string|null, \"cue_level\": number|null, \"message\": string, \"notes\": string|null}. " "Rules: (1) Choose at most one cue unless imminent danger. (2) If no issue, set cue=null and write a short positive message. (3) Keep message under 140 chars." ) if PAYLOAD_STYLE == "wrapped_input": body = {"input": prompt} else: # default 'wrapped' uses the 'message' field (matches your working curl) body = {"message": prompt} try: async with session.post(TOOLHOUSE_URL, json=body, headers=headers, timeout=timeout_s) as resp: status = resp.status try: body = await resp.json() except Exception: body = {"text": await resp.text()} print(f"[coach] forwarded event={payload.get('event')} status={status}") body["status"] = status return body except Exception as e: print(f"Coach forward error: {e}") return None async def safe_send(ws, obj: dict) -> bool: if getattr(ws, "closed", False): return False try: await ws.send(json.dumps(obj)) return True except (ConnectionClosed, ConnectionClosedOK): return False except Exception as e: print(f"Send error: {e}") return False def _fallback_final_coach(final_result: dict) -> dict: """Example Output { "summary": "Excellent lane keeping and smoothness. Some improvement needed in speed control, headway, and compliance. No red light violations, but collisions occurred.", "tips": [ "Maintain safe following distance.", "Monitor speed near limits.", "Anticipate hazards to avoid collisions." ], "drills": [ "Practice safe headway in traffic.", "Run speed control exercises.", "Complete collision avoidance scenarios." ], "priority": "headway" } """ subs = final_result.get("subscores", {}) final = final_result.get("final", 0) pri = min(subs, key=subs.get) if subs else "headway" tips_map = { "speeding": [ "Match speed to posted limit", "Lift early when approaching slower traffic", "Use cruise control to avoid creep", ], "lane": [ "Center the car between lines", "Look farther ahead to stabilize steering", "Ease steering inputs—avoid ping‑pong", ], "headway": [ "Open following gap to 2–3s", "Brake earlier, lighter when closing", "Avoid tailgating after lane changes", ], "smooth": [ "Feather brake before stopping", "Plan ahead—no hard stabs", "Keep throttle steady [truncated — 8916 more characters] ``` ### send_mp4_ws.py ```python import asyncio import json import time import cv2 import websockets import argparse import math import sounddevice as sd import numpy as np from pydub import AudioSegment from io import BytesIO def gen_telemetry(t: float, speed_limit_mps: float = 13.4) -> dict: # Simple synthetic telemetry resembling README format speed = 13.0 + 5.0 * math.sin(t * 0.4) return { "t": float(t), "speed_mps": float(max(0.0, speed)), "speed_limit_mps": float(speed_limit_mps), "throttle": 0.3, "brake": 0.0, "steer_deg": 0.0, "lane_offset_m": 0.0, "tl_state": "green", "in_stop_zone": False, "collision": False, } def play_audio(audio_bytes): # Decode bytes (assumed MP3 or other compressed format) to PCM audio audio_segment = AudioSegment.from_file(BytesIO(audio_bytes), format="mp3") # adjust format if needed raw_data = audio_segment.raw_data sample_rate = audio_segment.frame_rate channels = audio_segment.channels sample_width = audio_segment.sample_width # Map sample width in bytes to numpy dtype dtype_map = {1: np.int8, 2: np.int16, 4: np.int32} dtype = dtype_map.get(sample_width, np.int16) # Convert raw PCM bytes to numpy array audio_array = np.frombuffer(raw_data, dtype=dtype) # Reshape for multi-channel audio if channels > 1: audio_array = audio_array.reshape(-1, channels) # Normalize to float32 between -1.0 and 1.0 for sounddevice max_val = float(2 ** (8 * sample_width - 1)) audio_float = audio_array.astype(np.float32) / max_val # Play audio and wait for completion sd.play(audio_float, samplerate=sample_rate) sd.wait() async def stream_video(video_path: str, ws_url: str = "ws://localhost:8765", fps: float = 12.0): cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise SystemExit(f"Cannot open {video_path}") frame_period = 1.0 / max(1e-3, fps) t0 = time.time() async with websockets.connect(ws_url) as ws: audio_file_idx = 0 audio_bytes = b'' try: while True: ok, frame = cap.read() if not ok: break ok_jpg, enc = cv2.imencode('.jpg', frame) if not ok_jpg: print("Warn: JPEG encode failed; skipping frame") await asyncio.sleep(frame_period) continue # Send binary frame await ws.send(enc.tobytes()) # Send matching telemetry t = time.time() - t0 tel = gen_telemetry(t) await ws.send(json.dumps(tel)) # Optionally read back inference result (non-blocking) try: while (resp := await asyncio.wait_for(ws.recv(), timeout=2.5)) and isinstance(resp, bytes): # then we are receiving audio bytes audio_bytes += resp if audio_bytes: play_audio(audio_bytes) audio_bytes = b'' audio_file_idx += 1 continue try: data = json.loads(resp) collided = data.get("collision") ttc = data.get("ttc") dist = data.get("lead_distance_m") cues = data.get("cues") kind = data.get("type", "inference") print(f"Result[{kind}]: ttc={ttc} dist_m={dist} collided={collided} cues={cues}") coach = data.get("coach") if coach is not None: print("Coach:", coach) except Exception: print("Result:", resp) except asyncio.TimeoutError: pass await asyncio.sleep(frame_period) # Signal end of session and read final await ws.send("DONE") # Drain messages until we receive the one marked as final received_final = False deadline = time.time() + 25.0 while time.time() < deadline and not received_final: try: msg = await asyncio.wait_for(ws.recv(), timeout=max(0.1, deadline - time.time())) except asyncio.TimeoutError: break try: data = json.loads(msg) except Exception: print("Final (raw):", msg) continue if data.get("type") == "final": print("Final:", data) if data.get("coach") is not None: print("Coach (final):", data["coach"]) received_final = True else: # late in-flight inference; print briefly and continue waiting print("Late inference after DONE:", data) if not received_final: print("Final score response timed out or missing") finally: cap.release() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--video", default="ai/src/sample_drive.mp4", help="Path to MP4 file") parser.add_argument("--ws", default="ws://localhost:8765", help="WebSocket URL of backend") parser.add_argument("--fps", type=float, default=12.0, help="Send rate (frames per second)") args = parser.parse_args() asyncio.run(stream_video(args.video, args.ws, args.fps)) ``` ### backend/verbal_audio.py ```python # Using fish audio AI for the verbal cues sent back to the user from fish_audio_sdk import WebSocketSession, TTSRequest class FishTTSStreamer: def __init__(self, api_key, voice_model_id=None): self.api_key = api_key self.voice_model_id = voice_model_id async def stream_tts(self, text, send_audio): # Generator splits text for lower latency streaming def text_chunks(): for word in text.split(): yield word + " " # Setup Fish Audio session with WebSocketSession(self.api_key) as session: request = TTSRequest(text="", reference_id=self.voice_model_id) # Receive audio chunks and forward to send_audio callback for chunk in session.tts(request, text_chunks()): await send_audio(chunk) ``` ### backend/test.py ```python import asyncio import websockets import json TELEM_DATA = { "t": 5.3, "speed_mps": 20.5, "speed_limit_mps": 15.1, "throttle": 0.0, # unknown from video; keep 0 "brake": 0.0, # unknown; rules still handle TTC/lanes "steer_deg": 0.0, # unknown "lane_offset_m": 2.9, "tl_state": None, # None if unknown "in_stop_zone": False, # we don't know stop zone without a map "collision": False } async def test(): uri = "ws://localhost:8765" image_path = "./Screen Shot 2025-10-26 at 3.47.09 AM.png" with open(image_path, "rb") as f: image = f.read() async with websockets.connect(uri) as websocket: # Send image and some telemetry first await websocket.send(image) await websocket.send(json.dumps(TELEM_DATA)) await websocket.send("DONE") # Initialize a list to store audio chunks audio_data = b'' idx = 0 while True: try: message = await websocket.recv() if isinstance(message, bytes): audio_data += message else: print(f"Received JSON: {message}") if audio_data: with open(f"output{idx}.mp3", 'wb') as f: f.write(audio_data) print(f"Audio saved to output{idx}.mp3") audio_data = b'' idx += 1 except websockets.exceptions.ConnectionClosedOK: print("Connection closed") break # Save the received audio data as a WAV file if audio_data: with open(f"output{idx}.mp3", 'wb') as f: f.write(audio_data) print(f"Audio saved to output{idx}.mp3") asyncio.run(test()) ``` ### livekit_backend/livekit_backend.py ```python # The idea is that the unity game sends about 10-15 frames + some textual telemetry data to the livekit python backend, which then sends the data to appropriate AI models. It sends the AI data back to the unity game, which verbally says the output from livekit import api, rtc import logging import asyncio import aiohttp import requests import os from dotenv import load_dotenv import cv2 import numpy as np load_dotenv() # take environment variables # Code of your application, which uses environment variables (e.g. from `os.environ` or # `os.getenv`) as if they came from the actual environment. URL = os.getenv('LIVEKIT_URL') TOKEN = api.AccessToken() \ .with_identity("crashcourse") \ .with_name("Python Bot") \ .with_grants(api.VideoGrants( room_join=True, room="my-room", )).to_jwt() async def main(): room = rtc.Room() @room.on("participant_connected") def on_participant_connected(participant: rtc.RemoteParticipant): logging.info( "participant connected: %s %s", participant.sid, participant.identity) telemetry_cache = {} async def receive_video_frames(stream: rtc.VideoStream): async with aiohttp.ClientSession() as session: async for frame in stream: # Convert frame to JPEG bytes properly # Convert the frame to numpy array arr = frame.to_ndarray(format="bgr24") # Encode as JPEG _, image_bytes = cv2.imencode('.jpg', arr) image_bytes = image_bytes.tobytes() telemetry_str = telemetry_cache.get("latest", "{}") data = aiohttp.FormData() data.add_field('image', image_bytes, filename='frame.jpg', content_type='image/jpeg') data.add_field('telemetry', telemetry_str) try: async with session.post('http://localhost:8000/infer_frame', data=data) as resp: result = await resp.json() print("Inference result:", result) except Exception as e: logging.error(f"Error during inference: {e}") async def receive_telemetry_data(track: rtc.Track): while True: data = await track.read() # reads the next data packet (bytes) if data is None: break # track ended telemetry_cache["latest"] = data.decode('utf-8') # track_subscribed is emitted whenever the local participant is subscribed to a new track @room.on("track_subscribed") async def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant): logging.info("track subscribed: %s", publication.sid) if track.kind == rtc.TrackKind.KIND_VIDEO: video_stream = rtc.VideoStream(track) await receive_video_frames(video_stream) # if track.kind == rtc.TrackKind.TELEMETRY_DATA: if track.kind == rtc.TrackKind.KIND_UNKNOWN: await receive_telemetry_data(track) # By default, autosubscribe is enabled. The participant will be subscribed to # all published tracks in the room await room.connect(URL, TOKEN) logging.info("connected to room %s", room.name) # participants and tracks that are already available in the room # participant_connected and track_published events will *not* be emitted for them for identity, participant in room.remote_participants.items(): print(f"identity: {identity}") print(f"participant: {participant}") for tid, publication in participant.track_publications.items(): print(f"\ttrack id: {publication}") # Keep the connection alive await asyncio.Future() # runs forever if __name__ == "__main__": logging.basicConfig(level=logging.INFO) asyncio.run(main()) ``` ### ai/src/detector.py ```python from typing import List, Dict, Any, Optional, Tuple import numpy as np from ultralytics import YOLO COCO = {"person":0,"bicycle":1,"car":2,"motorcycle":3,"bus":5,"truck":7,"traffic light":10,"stop sign":13} INTEREST = {COCO["person"],COCO["car"],COCO["bicycle"],COCO["motorcycle"],COCO["bus"],COCO["truck"],COCO["traffic light"],COCO["stop sign"]} class YoloDetector: def __init__(self, model_name: str = "yolov8n.pt", conf: float = 0.25, imgsz: int = 640): self.model = YOLO(model_name) self.conf = conf self.imgsz = imgsz def infer(self, bgr_frame: np.ndarray) -> List[Dict[str, Any]]: res = self.model.predict(bgr_frame, imgsz=self.imgsz, conf=self.conf, verbose=False)[0] out: List[Dict[str, Any]] = [] if res.boxes is None or res.boxes.xyxy is None: return out boxes = res.boxes.xyxy.cpu().numpy() clss = res.boxes.cls.cpu().numpy().astype(int) confs = res.boxes.conf.cpu().numpy() names = self.model.names for (x1,y1,x2,y2), c, p in zip(boxes, clss, confs): if c not in INTEREST: continue out.append({"cls_id":int(c),"cls_name":names[int(c)],"conf":float(p),"xyxy":[float(x1),float(y1),float(x2),float(y2)], "center":[float((x1+x2)/2), float((y1+y2)/2)]}) return out def estimate_lead_distance_px(det: List[Dict[str, Any]], frame_shape: Tuple[int,int,int]) -> Optional[float]: h, w = frame_shape[:2]; cx = w/2; best=None for d in det: if d["cls_id"] not in [COCO["car"],COCO["bus"],COCO["truck"],COCO["motorcycle"],COCO["bicycle"]]: continue x1,y1,x2,y2 = d["xyxy"]; box_h = (y2-y1); center_x = (x1+x2)/2 if abs(center_x-cx) < w*0.22: if best is None or box_h>best[0]: best=(box_h,d) if best is None: return None return 1.0/max(best[0],1.0) # bigger box -> closer ``` ### ai/src/lane_simple.py ```python # lane_simple.py import cv2, numpy as np from typing import Optional, Tuple def _roi_mask(img: np.ndarray) -> np.ndarray: h, w = img.shape[:2] mask = np.zeros_like(img) pts = np.array([[ (int(0.10*w), int(0.95*h)), (int(0.45*w), int(0.62*h)), (int(0.55*w), int(0.62*h)), (int(0.90*w), int(0.95*h)) ]], dtype=np.int32) cv2.fillPoly(mask, pts, 255) return cv2.bitwise_and(img, mask) def _fit_line(points): if len(points) < 2: return None vx, vy, x0, y0 = cv2.fitLine(np.array(points, np.float32), cv2.DIST_L2,0,0.01,0.01) return float(vx), float(vy), float(x0), float(y0) def _x_at_y(line, y): if line is None: return None vx, vy, x0, y0 = line if abs(vy) < 1e-6: return None t = (y - y0) / vy return x0 + vx * t def estimate_lane_offset_m(bgr: np.ndarray, lane_width_m: float = 3.7): """Return (offset_m, dbg) where + is right of center; None if cannot estimate.""" h, w = bgr.shape[:2] gray = cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY) blur = cv2.GaussianBlur(gray, (5,5), 0) edges = cv2.Canny(blur, 60, 150) edges = _roi_mask(edges) lines = cv2.HoughLinesP(edges, 1, np.pi/180, threshold=60, minLineLength=40, maxLineGap=50) left_pts, right_pts = [], [] cx = w/2 if lines is not None: for x1,y1,x2,y2 in lines[:,0,:]: if y2==y1: continue slope = (x2-x1)/(y2-y1) if abs(slope) < 0.2: # reject near-horizontal continue x_bottom = x1 if y1>y2 else x2 (left_pts if x_bottom < cx else right_pts).extend([(x1,y1),(x2,y2)]) L = _fit_line(left_pts) if left_pts else None R = _fit_line(right_pts) if right_pts else None y_eval = int(h*0.90) xl = _x_at_y(L, y_eval) if L else None xr = _x_at_y(R, y_eval) if R else None dbg = {"xl": xl, "xr": xr, "y_eval": y_eval} if xl is None or xr is None or xr <= xl: return (None, dbg) lane_center_x = 0.5*(xl + xr) lane_width_px = xr - xl meters_per_px = lane_width_m / lane_width_px offset_m = (lane_center_x - cx) * meters_per_px return (float(offset_m), dbg) ``` ### ai/src/replay_test.py ```python import cv2, time, math, numpy as np from .detector import YoloDetector, estimate_lead_distance_px from .rules import ScoringState, Telemetry VIDEO_PATH = "src/sample_drive.mp4" IMG_SIZE = 640 FPS_INFER = 12 SPEED_LIMIT_MPS = 13.4 # ~30 mph def synthetic_telemetry(t: float) -> Telemetry: speed = 13 + 5*math.sin(t*0.2) brake = max(0.0, 0.4*math.sin(t*1.3)) throttle = max(0.0, 0.6*math.cos(t*0.7)) steer = 5*math.sin(t*0.5) lane = 0.25*math.sin(t*0.15) tl = "red" if 20.0 <= t%60.0 <= 25.0 else "green" in_stop = 19.0 <= t%60.0 <= 26.0 return Telemetry(t=t, speed_mps=max(0.0,speed), speed_limit_mps=SPEED_LIMIT_MPS, throttle=throttle, brake=brake, steer_deg=steer, lane_offset_m=lane, tl_state=tl, in_stop_zone=in_stop, collision=False) def px_to_ttc(px_proxy: float|None, speed_mps: float) -> float|None: if px_proxy is None or speed_mps<0.1: return None k=40.0; dist=k*px_proxy; return dist/max(speed_mps,0.1) def main(): cap=cv2.VideoCapture(VIDEO_PATH); if not cap.isOpened(): raise SystemExit(f"Cannot open {VIDEO_PATH}") det=YoloDetector("yolov8n.pt", conf=0.25, imgsz=IMG_SIZE) scorer=ScoringState() frame_period=1.0/FPS_INFER; next_tick=time.time(); t0=time.time() while True: now=time.time() if now1.0: frame=cv2.resize(frame,(int(w/scale),int(h/scale))) dets=det.infer(frame) lead_proxy=estimate_lead_distance_px(dets, frame.shape) t=time.time()-t0 tel=synthetic_telemetry(t) ttc=px_to_ttc(lead_proxy, tel.speed_mps) cues=scorer.step(tel, ttc) for d in dets: x1,y1,x2,y2=map(int,d["xyxy"]) cv2.rectangle(frame,(x1,y1),(x2,y2),(0,255,0),2) cv2.putText(frame,f"{d['cls_name']} {d['conf']:.2f}",(x1,max(12,y1-6)),cv2.FONT_HERSHEY_SIMPLEX,0.45,(0,255,0),1) hud=f"spd={tel.speed_mps*2.236:.1f}mph lim={tel.speed_limit_mps*2.236:.0f} lane={tel.lane_offset_m:+.2f}m TTC={'{:.2f}s'.format(ttc) if ttc else 'NA'}" cv2.putText(frame,hud,(10,22),cv2.FONT_HERSHEY_SIMPLEX,0.55,(50,200,255),2) y=45 for cue in cues[:2]: cv2.putText(frame,f"CUE: {cue['cue']} {cue['level']:.2f}",(10,y),cv2.FONT_HERSHEY_SIMPLEX,0.6,(0,0,255),2); y+=22 cv2.imshow("Scoring Replay (q to quit)", frame) if cv2.waitKey(1)&0xFF==ord('q'): break cap.release(); cv2.destroyAllWindows() print("\n=== SCORECARD ===") print(scorer.finalize()) if __name__=="__main__": main() ``` [4 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]