# Project export: Bob 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: AR glasses for physical work. It sees what you see, hears what you hear, and gives real-time responses to help you build anything. - Devpost: https://devpost.com/software/bob-vj43mq - GitHub: https://github.com/raghavrajsah/tethyr/ - Video: https://www.youtube.com/embed/1mViSjfp9T0?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Result: winner (Cal Hacks: Most Creative Hack) - Team: 3 GitHub contributor(s) — Brian Zhao (41 commits), Yubo-Cao (39 commits), raghavrajsah (13 commits) ## Devpost submission (written by the team) ### Inspiration Last year, the US had a shortage of 70,000 electricians and 642,000 mechanics. With one in five of these tradespeople over the age of 55 and growing demand in data centers and electric vehicles, this gap is only getting bigger. But what if everyone could become a skilled technician in under 5 minutes? This is why we built Bob. By using this AI+AR assistant, homeowners could handle simple repairs, vocational schools could train workers more efficiently, and professionals could work faster, safer, and more collaboratively. ### What it does Bob is a pair of agentic AR glasses that automatically watches over your actions, listens to your questions, and responds to your needs in real time. This could be instructions for your next steps, object-specific details (like a resistor's resistance), or warnings before you do something dangerous. For complex, collaborative tasks, it can also contact your teammates or highlight selected objects (in case you don't know what a “Kellum grip” is). While existing smart glasses focus on specific workflows like trivia or design, we built Bob to be a generalist from the outset. It can help you build electrical circuits, repair cars, and assemble furniture. With more tools, the variety of tasks Bob can do would be even greater. ### How we built it We’re using Snapchat Spectacles as our hardware and Gemini Live as our base model: the Spectacles stream video and audio data to a Python-built WebSocket server. On initial connection, we establish a new WebSocket connection to Gemini Live and store all subsequent audio and video frames in the session. Asynchronous workers handle buffering uploads, processing responses, executing tool calls, and resuming Gemini Live sessions efficiently. For tools, we used SMTP to integrate Gmail, YOLOE-11 for object detection via text prompts, the Python Slack SDK to integrate Slack, and Google Generative AI SDK for Google Search and Google Map. When voice activation detects that the user has stopped speaking, Gemini Live returns text and tool calls. These get executed and sent back to the Spectacles to update the overlay and bounding box highlights, guiding users through their project. Finally, we would like to note that all WebSockets connections are reused, minimizing the latency between Spectacles and the server. ### Challenges we ran into Messaging with Gemini Live over WebSocket turned out to be particularly challenging, with bugs in the asynchronous context manager and a demanding manual implementation of retry and bidirectional socket management. In addition, projecting pixel coordinates from the camera frame to the Snap Spectacle for object detection required debugging complex coordinate transformations. We solved these issues through test-driven development, A/B testing, and binary search. ### Accomplishments we're proud of As far as we know, we made the first pair of AR glasses with a multimodal AI agent that can talk back and forth with the user and instruct them in completing physical tasks. Although smart glasses exist, they are incapable of maintaining coherence over a physical task while accepting real-time input, often relying on obtrusive UI like buttons. By integrating live, multimodal agent and object detection into Snap Spectacles, we turned AR glasses into an agent with memory that helps anyone build whatever they want. We’re especially proud of getting the Spectacles to work since none of us had touched AR glasses before this project. ### What we learned Developing AR applications with Lens Studio Working with live instead of turn-based agents State management for WebSocket ### What's next When we interviewed our users about what else they would like to do with Bob, they gave really creative answers: cooking, first aid, martial arts…While these tasks are far from our original goal, Bob can quickly adapt to them because of its agentic framework. Every new tool can unlock a new field for Bob. For example, if we had added Composio’s toolset, Bob would be able to manage your calendar, send Slack messages, and read Notion pages. We could even link Bob to a humanoid robot that collaborates with the user on physical tasks. The future path for Bob is to become the orchestrator directing tens, hundreds or even thousands of humans at a time concurrently on large projects. Managing and monitoring all of them towards common goals while maintaining a common state across workers which would allow for effective collaboration. In addition, Bob is limited by its base models. If we had the hardware, we would run Qwen 2.5-Omni locally to reduce latency and use GroundingDINO to detect objects with greater accuracy. ## README (from the GitHub repository) # Tethyr Labs - AI-Powered AR Glasses An AI-powered AR glasses development platform with real-time computer vision, object detection, and AI agent capabilities. ## Architecture ### AR Glasses (Snap Lens Studio) **Location**: `snap/snap/Assets/CoordinateFetcher.ts` - **File**: TypeScript component for Lens Studio - **Function**: Captures camera frames from AR glasses and sends to server - **To add new UI**: Edit `CoordinateFetcher.ts` in the `displayLabel()` method (lines 124-151) to modify visual markers and positioning ### AI Agent **Location**: Multiple files in root directory - **`ai_client.py`**: Gemini Live API and OpenRouter integration - `stream_to_gemini_live()`: Real-time streaming with Gemini - `send_to_openrouter()`: Multi-model support via OpenRouter - **`ollama_client.py`**: Local Ollama integration - `get_ollama_response()`: Vision-capable models (llava, llama3.2-vision) - **`grounding.py`**: YOLO-based object detection - `Grounding.detect()`: Run detection on frames - `Grounding.update_prompt()`: Modify detection classes dynamically **To add new AI tools**: Create functions in `ai_client.py` or `ollama_client.py` following existing patterns ### Backend Servers - **`serve.py`**: WebSocket server for AR glasses (port 5001) - **`app.py`**: Webcam webapp (port 5001) - Camera feed display - Frame analysis with AI models ## Installation ```bash # Install dependencies uv sync # Or with pip pip install -r requirements.txt ``` ## Usage ### Start AR Glasses Server ```bash python main.py ``` ### Start Webcam Webapp ```bash python app.py # Visit http://localhost:5001 ``` ## Requirements - Python 3.13+ - Snap Lens Studio (for AR glasses development) - Ollama (optional, for local AI models) - OpenRouter API key (optional, for cloud AI models) - Google API key (optional, for Gemini Live) ## Environment Variables - `GOOGLE_API_KEY`: For Gemini Live API - `OPENROUTER_API_KEY`: For OpenRouter multi-model access ## Detected evidence (automated analysis) Indexed codebase: 30 recognized source files, 195 KB. - HTML (language) — detected in the code - Python (language) — detected in the code - TypeScript (language) — detected in the code - Google Gemini (technology) — claimed on Devpost, not found in the code - AI coding agent: Claude Code — evidence: config files committed to the repository ## Codebase structure (from repository index) ### Files (66 of 66) ``` .claude/settings.local.json .DS_Store .gitignore .pre-commit-config.yaml .python-version agent_context.py ai_client.py app.py architecture.md grounding.py img_conversion.py load_env.py main copy.py main.py ollama_client.py pyproject.toml README.md serve.py snap/snap/.gitattributes snap/snap/.gitignore snap/snap/Assets/BoundBox.prefab snap/snap/Assets/BoundBox.prefab.meta snap/snap/Assets/bounding-box.png.meta snap/snap/Assets/box.png.meta snap/snap/Assets/BoxController.ts snap/snap/Assets/BoxController.ts.meta snap/snap/Assets/CoordinateFetcher.ts snap/snap/Assets/CoordinateFetcher.ts.meta snap/snap/Assets/Device Camera Texture.deviceCameraTexture snap/snap/Assets/Device Camera Texture.deviceCameraTexture.meta snap/snap/Assets/Echopark.hdr snap/snap/Assets/Echopark.hdr.meta snap/snap/Assets/flat.ss_graph snap/snap/Assets/flat.ss_graph.meta snap/snap/Assets/image_unlit.ss_graph snap/snap/Assets/image_unlit.ss_graph.meta snap/snap/Assets/Image.mat snap/snap/Assets/Image.mat.meta snap/snap/Assets/ImageMaterial.mat snap/snap/Assets/ImageMaterial.mat.meta snap/snap/Assets/Inconsolata-VariableFont_wdth,wght.ttf.meta snap/snap/Assets/Microphone Audio.micaudio snap/snap/Assets/Microphone Audio.micaudio.meta snap/snap/Assets/Render Target.renderTarget snap/snap/Assets/Render Target.renderTarget.meta snap/snap/Assets/Scene.scene snap/snap/Assets/Scene.scene.meta snap/snap/Packages/SpectaclesInteractionKit.lspkg snap/snap/Packages/SpectaclesInteractionKit.lspkg.meta snap/snap/snap.esproj templates/index.html test_context.py tethyr/__init__.py tethyr/email_supervisor.py tethyr/gemini_client.py tethyr/grounding.py tethyr/handlers.py tethyr/prompts.py tethyr/scratchpad_tools.py tethyr/scratchpad.py tethyr/server.py tethyr/slack_tool.py tethyr/storage.py tethyr/types.py tethyr/utils.py uv.lock ``` ### Dependencies - pyproject.toml: clip, google-genai[aiohttp]@>=1.46.0, livekit-agents[google]@~=1.2, loguru@>=0.7.3, numpy@>=2.2.6, opencv-contrib-python@>=4.12.0.88, opencv-python@>=4.12.0.88, pillow@>=12.0.0, scipy@>=1.16.2, slack-sdk@>=3.33.4, ultralytics@>=8.3.221, websockets@>=15.0.1 ### Recent commits (newest first) - prompts file for testing - Merge pull request #1 from raghavrajsah/plan-scratchpad - scratchpad integrated - fix: :bug: Fix the bug with slack bot reference that's probably shouldn't be here - improvements on gemini_client, prompt, detection - merge: branch 'main' of https://github.com/raghavrajsah/tethyr - feat: :sparkles: Refactor and improve gemini client - prompt changes - Merge remote-tracking branch 'origin/feat/email' - scratchpad for real because I forgot to add the files - prompt changes + added scratchpad with tools for agent - feat: :sparkles: Add slack tool to the gemini client - feat: :sparkles: Text buffer + gemini client prompt fix - fix: Remove wirte a light bulb - Merge branch 'main' of https://github.com/raghavrajsah/tethyr - fix: Improve stability in weird TCP environment - feat: :sparkles: Fix the BBOX - Merge remote-tracking branch 'origin/main' - minor changes - feat: :sparkles: Email ## Key source files (fetched from GitHub, selected and truncated for size) ### architecture.md ```markdown ┌─────────────────┐ │ Snap Spectacles │ (Camera captures frames) └────────┬────────┘ │ WiFi ▼ ┌─────────────────┐ │ Flask Server │ (Receives frames) │ • OpenCV │ │ • YOLO │ ← Grounding/object detection │ • Ollama/Gemini│ ← Vision AI └────────┬────────┘ │ ▼ ┌─────────────────┐ │ Agent Logic │ (The "brain") │ • State tracker│ ← What step are we on? │ • Vision model │ ← What do I see? │ • Planner │ ← What should I show next? └────────┬────────┘ │ ▼ ┌─────────────────┐ │ Overlay Data │ (JSON response) │ { │ │ step: 2, │ │ text: "...", │ │ arrow: {x,y}, │ │ highlight: {} │ │ } │ └────────┬────────┘ │ WiFi ▼ ┌─────────────────┐ │ Lens Studio │ (Renders overlays) │ (Your JS code) │ └─────────────────┘ ``` ### pyproject.toml ``` [project] name = "tethyr" version = "0.1.0" description = "AR Processing Server with Gemini Live integration for smart glasses" readme = "README.md" requires-python = ">=3.13" dependencies = [ "websockets>=15.0.1", "numpy>=2.2.6", "pillow>=12.0.0", "loguru>=0.7.3", "scipy>=1.16.2", "ultralytics>=8.3.221", "opencv-python>=4.12.0.88", "opencv-contrib-python>=4.12.0.88", "clip", "google-genai[aiohttp]>=1.46.0", "livekit-agents[google]~=1.2", "slack-sdk>=3.33.4", ] [dependency-groups] dev = ["ruff>=0.8.4", "pre-commit>=4.0.1"] [tool.setuptools] py-modules = [] [tool.ruff] line-length = 150 target-version = "py313" [tool.ruff.lint] select = [ "E", # pycodestyle errors "W", # pycodestyle warnings "F", # pyflakes "I", # isort "B", # flake8-bugbear "C4", # flake8-comprehensions "UP", # pyupgrade ] ignore = [] [tool.ruff.format] quote-style = "double" indent-style = "space" skip-magic-trailing-comma = false line-ending = "auto" [tool.uv.sources] clip = { git = "https://github.com/ultralytics/CLIP.git" } ``` ### app.py ```python from flask import Flask, render_template, Response, request, jsonify import cv2 import threading import time from grounding import Grounding app = Flask(__name__) class Camera: def __init__(self): self.camera = None self.frame = None self.is_running = False self.grounding = None def start(self): """Start the camera capture""" self.camera = cv2.VideoCapture(0) # 0 is the default camera on macOS if not self.camera.isOpened(): raise Exception("Could not open camera") # Set camera properties for better performance self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, 640) self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) self.camera.set(cv2.CAP_PROP_FPS, 30) # Initialize Grounding model self.grounding = Grounding("yoloe-11s-seg.pt", initial_prompt="person") self.is_running = True # Start a thread to continuously capture frames self.capture_thread = threading.Thread(target=self._capture_frames) self.capture_thread.daemon = True self.capture_thread.start() def _capture_frames(self): """Continuously capture frames from the camera""" while self.is_running: ret, frame = self.camera.read() if ret: # Flip the frame horizontally for mirror effect frame = cv2.flip(frame, 1) # Run YOLOE detection if enabled if self.grounding: frame = self._process_frame_with_yoloe(frame) self.frame = frame time.sleep(0.03) # ~30 FPS def _process_frame_with_yoloe(self, frame): """Process frame with YOLOE and draw bounding boxes""" try: # Get detection results from Grounding results = self.grounding.detect(frame) # Draw bounding boxes on the frame if len(results) > 0: result = results[0] # Draw boxes and labels if result.boxes is not None and len(result.boxes) > 0: boxes = result.boxes.xyxy.cpu().numpy() # Get box coordinates confidences = result.boxes.conf.cpu().numpy() # Get confidence scores class_ids = result.boxes.cls.cpu().numpy() if result.boxes.cls is not None else None for i, box in enumerate(boxes): x1, y1, x2, y2 = map(int, box) confidence = confidences[i] # Get class name if class_ids is not None and result.names: class_id = int(class_ids[i]) # Handle both list and dict formats if isinstance(result.names, dict): label = result.names.get(class_id, self.grounding.prompt_text) elif isinstance(result.names, list): label = result.names[class_id] if class_id < len(result.names) else self.grounding.prompt_text else: label = self.grounding.prompt_text else: label = self.grounding.prompt_text # Draw bounding box color = (0, 255, 0) # Green color thickness = 2 cv2.rectangle(frame, (x1, y1), (x2, y2), color, thickness) # Draw label with confidence label_text = f"{label}: {confidence:.2f}" font = cv2.FONT_HERSHEY_SIMPLEX font_scale = 0.6 font_thickness = 2 # Get text size for background (text_width, text_height), baseline = cv2.getTextSize( label_text, font, font_scale, font_thickness ) # Draw background rectangle for text cv2.rectangle( frame, (x1, y1 - text_height - 10), (x1 + text_width, y1), color, -1 # Filled rectangle ) # Draw text cv2.putText( frame, label_text, (x1, y1 - 5), font, font_scale, (0, 0, 0), # Black text font_thickness ) except Exception as e: # Log errors but don't crash the frame processing print(f"Error in YOLOE processing: {e}") return frame def get_frame(self): """Get the latest frame""" return self.frame def stop(self): """Stop the camera capture""" self.is_running = False if self.camera: self.camera.release() # Global camera instance camera = Camera() def generate_frames(): """Generate video frames for streaming""" while True: frame = camera.get_frame() if frame is not None: # Encode frame as JPEG ret, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85]) if ret: frame_bytes = buffer.tobytes() yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n') @app.route('/') d [truncated — 1369 more characters] ``` ### main.py ```python import asyncio import base64 import io import json from dataclasses import asdict, dataclass from datetime import datetime from pathlib import Path from typing import Literal import numpy as np import scipy.io.wavfile as wavfile from loguru import logger from PIL import Image from websockets import ConnectionClosed from websockets.asyncio.server import ServerConnection, serve import load_env # noqa: E402, F401 from agent_context import context_manager from ai_client import stream_to_gemini_live_sync logger.add( "logs/server.log", rotation="100 MB", retention="10 days", level="TRACE", encoding="utf-8", ) # ============================================================================ # REPAIR PLANS AND SAFETY WARNINGS # ============================================================================ # Predefined repair plans for different objects REPAIR_PLANS = { "light fixture": [ "Turn off power at the circuit breaker", "Remove the old light fixture cover", "Disconnect the wiring from the old fixture", "Connect wiring to the new fixture (match wire colors)", "Secure the new fixture to the ceiling box", "Attach the fixture cover and restore power", ], "faucet": [ "Turn off water supply valves under the sink", "Remove faucet handle by unscrewing the set screw", "Unscrew and remove the old cartridge or valve", "Install the new cartridge (ensure proper alignment)", "Reattach the faucet handle", "Turn on water supply and test for leaks", ], "door hinge": [ "Open the door and support it with a wedge", "Remove the hinge pin using a hammer and nail punch", "Unscrew the old hinge from the door", "Align and screw the new hinge to the door", "Reattach the door and insert the hinge pin", "Test door swing and adjust if needed", ], "outlet": [ "Turn off power at the circuit breaker", "Remove the outlet cover plate", "Unscrew the outlet from the electrical box", "Disconnect wires from the old outlet (note positions)", "Connect wires to the new outlet (match positions)", "Screw outlet into box and replace cover plate", ], } # Safety warnings for different repair types SAFETY_WARNINGS = { "light fixture": "⚠️ SAFETY: Ensure power is OFF at circuit breaker before starting!", "faucet": "⚠️ SAFETY: Turn off water supply before starting to avoid flooding!", "door hinge": "⚠️ SAFETY: Support the door to prevent it from falling!", "outlet": "⚠️ SAFETY: Ensure power is OFF at circuit breaker before starting!", } # ============================================================================ # DATA CLASSES # ============================================================================ @dataclass class Resolution: width: int height: int @dataclass class Color: r: float g: float b: float a: float @dataclass class Position: x: float y: float @dataclass class BoundingBox: x: int y: int width: int height: int label: str confidence: float | None = None @dataclass class HandshakeMessage: type: Literal["handshake"] timestamp: int device: str capabilities: dict[str, bool] | None = None @dataclass class VideoFrameMessage: type: Literal["video_frame"] data: str # base64 encoded image timestamp: int frame_number: int resolution: Resolution @dataclass class AudioChunkMessage: type: Literal["audio_chunk"] data: str # base64 encoded audio timestamp: int frame_number: int sample_rate: int samples: int channels: int ClientMessage = HandshakeMessage | VideoFrameMessage | AudioChunkMessage @dataclass class HandshakeAckMessage: type: Literal["handshake_ack"] server: str timestamp: str @dataclass class OverlayMessage: type: Literal["overlay"] text: str timestamp: str color: Color | None = None position: Position | None = None @dataclass class BBoxMessage: type: Literal["bbox"] bbox: BoundingBox timestamp: str color: Color | None = None @dataclass class ClearMessage: type: Literal["clear"] timestamp: str ServerMessage = HandshakeAckMessage | OverlayMessage | BBoxMessage | ClearMessage @dataclass class ClientState: websocket: ServerConnection client_id: str frame_count: int = 0 audio_buffer: list[np.ndarray] = None last_detection: dict | None = None output_dir: Path | None = None audio_sample_rate: int = 16000 audio_channels: int = 1 def __post_init__(self): if self.audio_buffer is None: self.audio_buffer = [] clients: dict[str, ClientState] = {} def setup_output_directory(client_id: str) -> Path: """ Create output directory for this client session Args: client_id: Unique client identifier Returns: Path to the output directory """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_dir = Path("output") / f"session_{client_id}_{timestamp}" output_dir.mkdir(parents=True, exist_ok=True) # Create subdirectories for frames (output_dir / "frames").mkdir(exist_ok=True) logger.info(f"Created output directory: {output_dir}") return output_dir def save_audio_buffer(client_state: ClientState) -> None: """ Save accumulated audio buffer as a .wav file Args: client_state: Client state containing audio buffer and metadata """ if not client_state.audio_buffer or not client_state.output_dir: logger.warning(f"No audio to save for client {client_state.client_id}") return try: # Concatenate all audio chunks audio_data = np.concatenate(client_state.audio_buffer) # Reshape for multi-channel audio if needed if client_state.audio_channels > 1: # Interleaved format: [L, [truncated — 18744 more characters] ``` ### tethyr/server.py ```python """ Tethyr WebSocket server for AR smart glasses Handles video/audio streaming and connects to Gemini Live API """ import asyncio import json import os from loguru import logger from websockets import ConnectionClosed from websockets.asyncio.server import ServerConnection, serve from .gemini_client import GeminiSessionManager from .handlers import handle_audio_chunk, handle_handshake, handle_video_frame from .storage import StorageMiddleware from .types import AudioChunkMessage, ClientState, HandshakeMessage, VideoFrameMessage from .utils import clean_json_message, parse_client_message # Configure logging logger.add( "logs/server.log", rotation="100 MB", retention="10 days", level="TRACE", encoding="utf-8", ) # Global state clients: dict[str, ClientState] = {} async def handle_client( websocket: ServerConnection, gemini_manager: GeminiSessionManager, storage: StorageMiddleware | None = None, ): """ Main handler for WebSocket client connection Args: websocket: WebSocket connection gemini_manager: Gemini session manager storage: Optional storage middleware for debug/recording """ client_id = str(id(websocket)) client_state = ClientState(websocket=websocket, client_id=client_id) clients[client_id] = client_state logger.info(f"Client {client_id} connected. Total clients: {len(clients)}") try: async for raw_message in websocket: logger.trace(f"Raw message from client {client_id}: {raw_message}") try: cleaned_message = clean_json_message(raw_message) if cleaned_message != raw_message: logger.trace( f"Cleaned message from client {client_id}: " f"removed {len(raw_message) - len(cleaned_message)} trailing characters" ) while cleaned_message: try: message_dict = json.loads(cleaned_message) break except json.JSONDecodeError: cleaned_message = clean_json_message(cleaned_message[:-1]) else: raise json.JSONDecodeError() message = parse_client_message(message_dict) if message is None: continue # Route to appropriate handler match message: case HandshakeMessage(): logger.debug(f"Handshake from client {client_id}") await handle_handshake( message, client_state, gemini_manager, storage, ) case VideoFrameMessage(): # logger.debug(f"Video frame from client {client_id}") await handle_video_frame( message, client_state, gemini_manager, storage, ) case AudioChunkMessage(): # logger.debug(f"Audio chunk from client {client_id}") await handle_audio_chunk( message, client_state, gemini_manager, storage, ) except json.JSONDecodeError as e: logger.opt(exception=e).error(f"Invalid JSON received from client {client_id}") except Exception as e: logger.opt(exception=e).error(f"Error handling message from client {client_id}") except ConnectionClosed: logger.info(f"Client {client_id} disconnected") finally: if client_id in clients: if storage and storage.enabled: storage.save_session(client_state) await gemini_manager.close_session(client_id) logger.info( f"Session complete for client {client_id}: " f"{client_state.frame_count} video frames, " f"{client_state.audio_chunk_count} audio chunks" ) del clients[client_id] logger.info(f"Client {client_id} removed. Total clients: {len(clients)}") async def forever(): await asyncio.Future() async def main(enable_storage: bool = False): """ Start the Tethyr WebSocket server Args: enable_storage: Whether to enable frame/audio storage for debugging """ # Initialize Gemini session manager gemini_manager = GeminiSessionManager() # Initialize storage middleware storage = StorageMiddleware(enabled=enable_storage) async with serve( lambda ws: handle_client(ws, gemini_manager, storage), "0.0.0.0", 5001, max_size=10_000_000, ping_interval=20, ping_timeout=10, ): logger.info( "============================================================\n" "Tethyr AR Server with Gemini Live\n" "============================================================\n" "WebSocket Server: ws://0.0.0.0:5001\n" "Waiting for Spectacles to connect...\n" f"Storage middleware: {'ENABLED' if enable_storage else 'DISABLED'}\n" "============================================================" ) await forever() def run_server(): """ Entry point for the tethyr server command Reads TETHYR_ENABLE_STORAGE environment variable to enable debug storage """ enable_storage = os.getenv("TETHYR_ENABLE_STORAGE", "false").lower() in ( "true", "1", "yes", ) asyncio.run(main(enable_storage=enable_storage)) if __name__ == "__main__": run_server() ``` ### serve.py ```python import load_env # noqa: E402, F401 from tethyr.server import run_server if __name__ == "__main__": run_server() ``` ### .pre-commit-config.yaml ```yaml repos: - repo: https://github.com/astral-sh/ruff-pre-commit rev: v0.8.4 hooks: # Run the linter - id: ruff args: [--fix] # Run the formatter - id: ruff-format ``` ### img_conversion.py ```python import base64 import cv2 def frame_to_base64(frame: any) -> str: """ Convert OpenCV frame to base64 string for Ollama Args: frame: OpenCV frame (numpy array) Returns: Base64 encoded string """ ret, buffer = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85]) if not ret: raise ValueError("Failed to encode frame") return base64.b64encode(buffer).decode("utf-8") ``` ### load_env.py ```python """Load environment variables from .env file""" import os def load_env(): """Load .env file into environment variables""" env_path = os.path.join(os.path.dirname(__file__), ".env") if not os.path.exists(env_path): return with open(env_path) as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: key, value = line.split("=", 1) os.environ[key.strip()] = value.strip() # Auto-load on import load_env() ``` ### grounding.py ```python import threading from ultralytics import YOLO class Grounding: def __init__(self, model_path, initial_prompt="person"): """Initialize the YOLOE model with a default prompt""" self.model = None self.model_lock = threading.Lock() print("Loading YOLOE model...") self.model = YOLO(model_path) self.update_prompt(initial_prompt) print(f"YOLOE model loaded with prompt: '{self.prompt_text}'") def detect(self, frame): """Run detection on a frame and return results""" with self.model_lock: results = self.model.predict(frame, verbose=False, conf=0.1, iou=0.5) return results def update_prompt(self, prompt): """Update the text prompt for detection""" self.prompt_text = prompt with self.model_lock: classes = [cls.strip() for cls in prompt.split(",")] print(f"Updating YOLOE to detect: {classes}") text_embeddings = self.model.get_text_pe(classes) self.model.set_classes(classes, text_embeddings) print(f"Successfully updated detection prompt to: {classes}") ``` [20 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]