# Project export: Iris
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: TreeHacks 2025
- Tagline: Real-time vision for the blind.
- Devpost: https://devpost.com/software/iris-vlnzp0
- GitHub: https://github.com/RoyceAroc/iris
- Video: https://www.youtube.com/embed/3Y1PsWWNvb4?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Result: winner (Edge AI Prize: Most Creative On-Device AI Deployment ($10k Cash))
- Team: 3 GitHub contributor(s) — Ayush (5 commits), RoyceAroc (5 commits), Saniya Savla (4 commits)
## Devpost submission (written by the team)
### Inspiration
🤔 Have you ever wondered how blind people cross the road? How imminent hazards like a "wet floor sign" or a pothole to their right might inflict harm? Here are some scary statistics: A survey of 300 blind people found that about 40% experienced head height collisions at least once a year. source 68% of people with visual impairment had been directly exposed to at least one serious life event. source There is a 46% increased risk of road traffic crash among those with visual impairment. source This hackathon we decided to build Iris.
### What it does
+
### How we built it
🛠️ Iris is a sleek headset built for the blind and visually impaired, designed to provide real-time guidance through the world. By attaching an iPhone as a camera to an affordable 3D-printed headset, users can navigate their surroundings with confidence. Iris doesn’t just describe the environment – it actively guides the wearer through A.I. agent specialists. Using real-time models such as captioning, classification, image segmentation agents, and depth estimation agents, Iris detects hazards in under 100 milliseconds. When a danger is identified, it instantly provides a haptic buzz, signaling the user of danger. From there, Iris delivers step-by-step spoken instructions via TTS, directing the wearer on how to safely navigate around obstacles, cross streets, or reorient themselves in complex environments. Taking a VLM model, we were able to tune it for real-time workloads via aggressive PyTorch compiler optimization, cudNN utilization, specific matmul precision, and usage of optimal attention kernels like flash / paged attention. To achieve high network throughput, we developed a custom userspace protocol to stream tokens to clients, and to reduce agent latency, we quantized models to architecture-specific dtypes. Built using Expo, the mobile app is compatible on both Android and iOS devices. Iris is also able to take STT commands and decode it into agentic workflows such as hazard detection, scene similarity scoring, and object detection (via Groq LPU APIs) which are carried out in real-time. Want to know when the pedestrian sign turns white for you to walk across? Want to know where your water bottle is? Iris has your back.
### Challenges we ran into
😿 Even the slightest delay in detecting a hazard could mean the difference between safe navigation and serious injury for a blind user. Latency was a critical factor, so we spent a significant amount of time tuning our models to minimize Time to Detection (what we call TTD). We implemented attention optimizations, leveraged cuDNN for accelerated inference, and fine-tuned every aspect of our pipeline. But no amount of optimization could make up for the fact that we were hitting a hard wall – our compute resources simply weren’t enough. That’s when we reached out to Brev.dev, which gave us access to the compute power we needed to push our models further, drastically improving both training and inference speed. Compute wasn’t our only bottleneck. Network bandwidth became a serious constraint. Our initial protocol relied on JSON, which, while easy to work with, was bloated – filled with unnecessary delimiters and whitespace that slowed down transmission. Every millisecond mattered. To eliminate this overhead, we designed a custom datagram protocol, stripping out redundant characters and ensuring that every byte carried meaningful information. The result was a lean, high-speed data stream that helped us shave off critical delays. Finally, there was the challenge of hardware. One of our core goals was affordability – if our solution wasn’t accessible to blind users, it wasn’t a solution at all. Off-the-shelf headsets were too expensive, so we designed our own. We developed a custom STL model for 3D printing a headset that could securely hold an iPhone, ensuring a low-cost but effective hardware setup. This way, users could leverage the sensors and processing power of their existing devices without needing expensive proprietary hardware.
### Accomplishments we're proud of
💯 Iris started as a simple idea: an intelligent headset that could actively guide blind users through the world. But for us, it was also personal. One of our team members has a blind relative, and hearing firsthand about the daily challenges of navigating city streets without reliable assistance pushed us to make Iris as fast and reliable as possible. What we ended up with was a fully functional, multimodal A.I. system capable of real-time hazard detection and step-by-step navigation. Here are some of Iris' capabilties: Iris can chain together vision-language models, depth estimation, and object detection to dynamically respond to user queries in real time. We designed and 3D-printed an affordable, durable headset that transforms an iPhone into an AI-powered navigation assistant Users can issue STT commands like “Where’s the door?” or “Can I cross the street?” and receive responses near-instantly.
### What we learned
💭 This project pushed us deep into the weeds of HPC optimization, hardware, and networking in exciting ways. One of the biggest lessons came from wrestling with the PyTorch compiler. We thought we understood it – until we started hitting obscure bottlenecks that weren’t obvious at first glance. Some layers weren’t fusing correctly, some optimizations were actually slowing things down, and cuDNN didn’t always behave the way we assumed it would. We spent hours profiling CUDA graphs, attempting optimal Tensor core and register usage, tweaking VRAM memory layouts, and rewriting parts of our model just to squeeze out those last few milliseconds. On the hardware side, we learned that designing an STL file isn’t just about making a model that looks right – it has to print right. Our first 3D print warped mid-print. Through trial and error, we figured out how to design for material constraints, adjust print settings for strength, and optimize for minimal waste. Eventually, we had a sleek, durable headset that could securely hold an iPhone – something that felt incredibly rewarding after our early failures.
### What's next
🚀 Iris has proven its ability to provide real-time hazard detection and navigation, but we’re just getting started. Here’s what’s next: Right now, Iris warns of hazards through a single vibration alert. We plan to implement a more advanced haptic feedback system that provides directional guidance such as buzzing stronger on the left or right to indicate safer paths or signaling when the user should stop. Beyond hazard detection, Iris will integrate more specialized agents like sign language interpretation and personalized object recognition (i.e., users can “teach” Iris what their personal items look like, so it can help locate them.) We want to put Iris in the hands of real users. Our next step is testing with more visually impaired individuals to refine the experience, gather feedback, and iterate on both software and hardware. We truly see this as something that changes the lives of blind individuals for the better.
## README (from the GitHub repository)
## Running our Application
```
cd frontend && npx expo run
```
```
cd backend && uv run src/main.py
```
## Inspiration 🤔
Have you ever wondered how blind people cross the road? How imminent hazards like a "wet floor sign" or a pothole to their right might inflict harm? Here are some scary statistics:
- A survey of 300 blind people found that about 40% experienced head height collisions at least once a year. [source](https://www.researchgate.net/publication/228828914_Mobility-related_accidents_experienced_by_people_with_visual_impairment#:~:text=A%20survey%20of%20300%20blind,once%20a%20month%20%5B2%5D%20.)
- 68% of people with visual impairment had been directly exposed to at least one serious life event. [source](https://pmc.ncbi.nlm.nih.gov/articles/PMC8583190/)
- There is a 46% increased risk of road traffic crash among those with visual impairment. [source](The summary estimate revealed a 46% increased risk of road traffic crash among those with visual impairment)
This hackathon we decided to build Iris.
## What it does + How we built it 🛠️
Iris is a sleek headset built for the blind and visually impaired, designed to provide real-time guidance through the world. By attaching an iPhone as a camera to an affordable 3D-printed headset, users can navigate their surroundings with confidence. Iris doesn’t just describe the environment – it actively guides the wearer through A.I. agent specialists. Using real-time models such as captioning, classification, image segmentation agents, and depth estimation agents, Iris detects hazards in under 100 milliseconds. When a danger is identified, it instantly provides a haptic buzz, signaling the user of danger. From there, Iris delivers step-by-step spoken instructions via TTS, directing the wearer on how to safely navigate around obstacles, cross streets, or reorient themselves in complex environments.
Taking a VLM model, we were able to tune it for real-time workloads via aggressive PyTorch compiler optimization, cudNN utilization, specific matmul precision, and usage of optimal attention kernels like flash / paged attention. To achieve high network throughput, we developed a custom userspace protocol to stream tokens to clients, and to reduce agent latency, we quantized models to architecture-specific dtypes.
Built using Expo, the mobile app is compatible on both Android and iOS devices. Iris is also able to take STT commands and decode it into agentic workflows such as hazard detection, scene similarity scoring, and object detection (via Groq LPU APIs) which are carried out in real-time. Want to know when the pedestrian sign turns white for you to walk across? Want to know where your water bottle is? Iris has your back.
## Challenges we ran into 😿
Even the slightest delay in detecting a hazard could mean the difference between safe navigation and serious injury for a blind user. Latency was a critical factor, so we spent a significant amount of time tuning our models to minimize Time to Detection (what we call TTD). We implemented attention optimizations, leveraged cuDNN for accelerated inference, and fine-tuned every aspect of our pipeline. But no amount of optimization could make up for the fact that we were hitting a hard wall – our compute resources simply weren’t enough. That’s when we reached out to Brev.dev, which gave us access to the compute power we needed to push our models further, drastically improving both training and inference speed.
Compute wasn’t our only bottleneck. Network bandwidth became a serious constraint. Our initial protocol relied on JSON, which, while easy to work with, was bloated – filled with unnecessary delimiters and whitespace that slowed down transmission. Every millisecond mattered. To eliminate this overhead, we designed a custom datagram protocol, stripping out redundant characters and ensuring that every byte carried meaningful information. The result was a lean, high-speed data stream that helped us shave off critical delays.
Finally, there was the challenge of hardware. One of our core goals was affordability – if our solution wasn’t accessible to blind users, it wasn’t a solution at all. Off-the-shelf headsets were too expensive, so we designed our own. We developed a custom STL model for 3D printing a headset that could securely hold an iPhone, ensuring a low-cost but effective hardware setup. This way, users could leverage the sensors and processing power of their existing devices without needing expensive proprietary hardware.
## Accomplishments that we're proud of 💯
Iris started as a simple idea: an intelligent headset that could actively guide blind users through the world. But for us, it was also personal. One of our team members has a blind relative, and hearing firsthand about the daily challenges of navigating city streets without reliable assistance pushed us to make Iris as fast and reliable as possible. What we ended up with was a fully functional, multimodal A.I. system capable of real-time hazard detection and step-by-step navigation. Here are some of Iris' capabilties:
- Iris can chain together vision-language models, depth estimation, and object detection to dynamically respond to user queries in real time.
- We designed and 3D-printed an affordable, durable headset that transforms an iPhone into an AI-powered navigation assistant
- Users can issue STT commands like “Where’s the door?” or “Can I cross the street?” and receive responses near-instantly.
## What we learned 💭
This project pushed us deep into the weeds of HPC optimization, hardware, and networking in exciting ways. One of the biggest lessons came from wrestling with the PyTorch compiler. We thought we understood it – until we started hitting obscure bottlenecks that weren’t obvious at first glance. Some layers weren’t fusing correctly, some optimizations were actually slowing things down, and cuDNN didn’t always behave the way we assumed it would. We spent hours profiling CUDA graphs, attempting optimal Tensor core and register usage, tweaking VRAM memory layouts, and rewriting parts of our model just to squeeze out those last few milliseconds.
On the hardware side, we learned that designing an STL file isn’t just about making a model that looks right – it has to print right. Our first 3D print warped mid-print. Through trial and error, we figured out how to design for material constraints, adjust print settings for strength, and optimize for minimal waste. Eventually, we had a sleek, durable headset that could securely hold an iPhone – something that felt incredibly rewarding after our early failures.
## What's next for Iris 🚀
Iris has proven its ability to provide real-time hazard detection and navigation, but we’re just getting started. Here’s what’s next:
- Right now, Iris warns of hazards through a single vibration alert. We plan to implement a more advanced haptic feedback system that provides directional guidance such as buzzing stronger on the left or right to indicate safer paths or signaling when the user should stop.
- Beyond hazard detection, Iris will integrate more specialized agents like sign language interpretation and personalized object recognition (i.e., users can “teach” Iris what their personal items look like, so it can help locate them.)
- We want to put Iris in the hands of real users. Our next step is testing with more visually impaired individuals to refine the experience, gather feedback, and iterate on both software and hardware. We truly see this as something that changes the lives of blind individuals for the better.
## Detected evidence (automated analysis)
Indexed codebase: 13 recognized source files, 37 KB.
- Hugging Face (technology) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — detected in the code
- React (technology) — detected in the code
- TypeScript (language) — detected in the code
## Codebase structure (from repository index)
### Files (23 of 23)
```
backend/.gitignore
backend/.python-version
backend/pyproject.toml
backend/README.md
backend/src/main.py
backend/src/models/__init__.py
backend/src/models/base.py
backend/src/models/blip.py
backend/src/models/groq.py
backend/src/models/llava.py
backend/src/prompts/caption.txt
backend/src/prompts/classify.txt
backend/test/image.py
backend/test/video.py
frontend/.gitignore
frontend/android/.gradle/8.10.2/gc.properties
frontend/app.json
frontend/app/_layout.tsx
frontend/app/index.tsx
frontend/package.json
frontend/README.md
frontend/tsconfig.json
README.md
```
### Dependencies
- backend/pyproject.toml: accelerate@>=1.3.0, aiohttp@>=3.11.12, imagehash@>=4.3.2, opencv-python@>=4.11.0.86, picows@>=1.7.2, pillow@>=11.1.0, pytest@>=8.3.4, pytest-asyncio@>=0.25.3, torch@>=2.5.1, transformers@>=4.48.1, uvloop@>=0.21.0
- frontend/package.json: @babel/core@^7.25.2, @expo/vector-icons@^14.0.2, @react-navigation/bottom-tabs@^7.2.0, @react-navigation/native@^7.0.14, @types/jest@^29.5.12, @types/react@~18.3.12, @types/react-test-renderer@^18.3.0, @types/uuid@^10.0.0, expo@~52.0.35, expo-av@~15.0.2, expo-blur@~14.0.3, expo-camera@~16.0.16, expo-constants@~17.0.6, expo-font@~13.0.3, expo-haptics@~14.0.1, expo-image-manipulator@~13.0.6, expo-linking@~7.0.5, expo-router@~4.0.17, expo-speech@~13.0.1, expo-splash-screen@~0.29.22, expo-status-bar@~2.0.1, expo-symbols@~0.2.2, expo-system-ui@~4.0.8, expo-web-browser@~14.0.2, jest@^29.2.1, jest-expo@~52.0.4, react@18.3.1, react-dom@18.3.1, react-native@0.76.7, react-native-gesture-handler@~2.20.2, react-native-get-random-values@^1.11.0, react-native-reanimated@~3.16.1, react-native-safe-area-context@4.12.0, react-native-screens@~4.4.0, react-native-tts@^4.1.1, react-native-vision-camera@^4.6.3, react-native-web@~0.19.13, react-native-webview@13.12.5, react-test-renderer@18.3.1, typescript@^5.3.3
### Recent commits (newest first)
- Update groq.py
- Add devpost instructions
- Add gitignore to backend
- Add VLM implementation
- Update README.md
- Update README.md
- Update README.md
- Update README.md
- Merge branch 'main' of https://github.com/RoyceAroc/iris
- added resizing
- fuck u ayush
- Improve object detection accuracy and add perceptual hashing
- Add backend
- token locks
- yo
- yuh
- f
- haptics
- yo
- added haptics
## Key source files (fetched from GitHub, selected and truncated for size)
### backend/pyproject.toml
```
[project]
name = "iris"
version = "0.1.0"
description = "Firmware for the Iris headset"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"picows>=1.7.2",
"pillow>=11.1.0",
"pytest-asyncio>=0.25.3",
"pytest>=8.3.4",
"torch>=2.5.1",
"transformers>=4.48.1",
"accelerate>=1.3.0",
"uvloop>=0.21.0",
"opencv-python>=4.11.0.86",
"aiohttp>=3.11.12",
"imagehash>=4.3.2",
]
[tool.pyright]
typeCheckingMode = "off"
```
### frontend/package.json
```
{
"name": "treehacks",
"main": "expo-router/entry",
"version": "1.0.0",
"scripts": {
"start": "expo start",
"reset-project": "node ./scripts/reset-project.js",
"android": "expo start --android",
"ios": "expo start --ios",
"web": "expo start --web",
"test": "jest --watchAll",
"lint": "expo lint"
},
"jest": {
"preset": "jest-expo"
},
"dependencies": {
"@expo/vector-icons": "^14.0.2",
"@react-navigation/bottom-tabs": "^7.2.0",
"@react-navigation/native": "^7.0.14",
"expo": "~52.0.35",
"expo-blur": "~14.0.3",
"expo-camera": "~16.0.16",
"expo-constants": "~17.0.6",
"expo-font": "~13.0.3",
"expo-haptics": "~14.0.1",
"expo-linking": "~7.0.5",
"expo-router": "~4.0.17",
"expo-speech": "~13.0.1",
"expo-splash-screen": "~0.29.22",
"expo-status-bar": "~2.0.1",
"expo-symbols": "~0.2.2",
"expo-system-ui": "~4.0.8",
"expo-web-browser": "~14.0.2",
"react": "18.3.1",
"react-dom": "18.3.1",
"react-native": "0.76.7",
"react-native-gesture-handler": "~2.20.2",
"react-native-get-random-values": "^1.11.0",
"react-native-reanimated": "~3.16.1",
"react-native-safe-area-context": "4.12.0",
"react-native-screens": "~4.4.0",
"react-native-tts": "^4.1.1",
"react-native-vision-camera": "^4.6.3",
"react-native-web": "~0.19.13",
"react-native-webview": "13.12.5",
"expo-av": "~15.0.2",
"expo-image-manipulator": "~13.0.6"
},
"devDependencies": {
"@babel/core": "^7.25.2",
"@types/jest": "^29.5.12",
"@types/react": "~18.3.12",
"@types/react-test-renderer": "^18.3.0",
"@types/uuid": "^10.0.0",
"jest": "^29.2.1",
"jest-expo": "~52.0.4",
"react-test-renderer": "18.3.1",
"typescript": "^5.3.3"
},
"private": true
}
```
### frontend/app/_layout.tsx
```typescript
import { Stack } from "expo-router";
export default function RootLayout() {
return ;
}
```
### backend/src/main.py
```python
import transformers
import imagehash
import asyncio
import picows
import uvloop
from models.base import Frame, DeviceModel, VendorModel, FrameStatus
from models.groq import LlamaVisionModel
from models.llava import LlavaModel
from collections import deque
transformers.logging.set_verbosity_error()
class Server(picows.WSListener):
def __init__(
self,
caption_model: DeviceModel | VendorModel,
classify_model: DeviceModel | VendorModel,
) -> None:
self.caption_model = caption_model
self.classify_model = classify_model
self.hashes = deque(maxlen=3)
self.similarity_threshold = 20
self.different_threshold = 50
super().__init__()
def handle_frame(
self, transport: picows.WSTransport, scene_frame: Frame, caption_id: str
) -> None:
# Compute perceptual hash of the new frame
new_hash = imagehash.phash(scene_frame.as_image())
# Check if the new frame is too similar to any of the last 3 frames
for cached_hash in self.hashes:
if (cached_hash - new_hash) <= self.similarity_threshold:
print(f"Skipping similar frame for {caption_id}")
return
# Check if the new frame is "very different" (>= different_threshold) from at least one of the last 3
significant_difference_found = any(
abs(cached_hash - new_hash) >= self.different_threshold
for cached_hash in self.hashes
)
self.hashes.append(new_hash)
classification = self.classify_model.classify(scene_frame)
print(f"Classification result for {caption_id}: {classification}")
# Decide whether to stream inference
if classification == FrameStatus.Safe and not significant_difference_found:
print(
"Skipping frame since classification is safe or significant difference found"
)
return
# Stream inference tokens
for token in self.caption_model.caption(scene_frame):
payload = f"{caption_id}|{token}".encode("utf-8")
transport.send(picows.WSMsgType.TEXT, payload)
# Send end token
end_payload = f"{caption_id}|".encode("utf-8")
transport.send(picows.WSMsgType.TEXT, end_payload)
def on_ws_connected(self, transport: picows.WSTransport) -> None:
print("New client connected")
def on_ws_frame(self, transport: picows.WSTransport, frame: picows.WSFrame) -> None:
if frame.msg_type == picows.WSMsgType.BINARY:
data = frame.get_payload_as_bytes()
parts = data.split(b"|", 1)
caption_id = parts[0].decode("utf-8")
image_bytes = parts[1]
scene_frame = Frame(image_bytes)
# Stream captions on main thread while classification runs in background
self.handle_frame(transport, scene_frame, caption_id)
elif frame.msg_type == picows.WSMsgType.CLOSE:
transport.send_close(frame.get_close_code(), frame.get_close_message())
transport.disconnect()
elif frame.msg_type == picows.WSMsgType.PING:
transport.send_pong(frame.get_payload_as_bytes())
async def main():
print("Beginning model warmup")
caption_model = LlavaModel()
caption_model.warmup()
classify_model = LlamaVisionModel()
server = await picows.ws_create_server(
lambda _: Server(caption_model, classify_model), "0.0.0.0", 2222
)
for s in server.sockets:
print(f"Server started on {s.getsockname()}")
await server.serve_forever()
if __name__ == "__main__":
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main())
```
### frontend/app/index.tsx
```typescript
import { CameraView, CameraType } from "expo-camera";
import { useState, useEffect, useRef } from "react";
import { View } from "react-native";
import { Audio } from "expo-av";
import * as Haptics from "expo-haptics";
import "react-native-get-random-values";
import { v4 as uuidv4 } from "uuid";
import * as Speech from "expo-speech";
import * as ImageManipulator from 'expo-image-manipulator';
function base64ToArrayBuffer(base64: string) {
const binaryString = atob(base64);
const len = binaryString.length;
const bytes = new Uint8Array(len);
for (let i = 0; i < len; i++) {
bytes[i] = binaryString.charCodeAt(i);
}
return bytes.buffer;
}
export default function Index() {
const [facing, setFacing] = useState("back");
const cameraRef = useRef(null);
const [isRecording, setIsRecording] = useState(false);
const ws = useRef(null);
const [captions, setCaptions] = useState>({});
const visitedUIDs = useRef>(new Set());
const speechQueue = useRef([]);
const isSpeaking = useRef(false);
useEffect(() => {
ws.current = new WebSocket("ws://209.20.159.34:2222");
ws.current.onopen = () => {
console.log("Connected to WebSocket server");
};
ws.current.onmessage = (event) => {
const message = event.data;
const [uid, token] = message.split("|");
if (!visitedUIDs.current.has(uid)) {
visitedUIDs.current.add(uid);
Haptics.notificationAsync(Haptics.NotificationFeedbackType.Error);
}
if (token.trim() == "") {
startSpeech(captions[uid]);
}
if (captions[uid]) {
captions[uid] = captions[uid] + " " + token;
} else {
captions[uid] = token;
}
};
ws.current.onerror = (error) => {
console.error("WebSocket error:", error);
};
ws.current.onclose = () => {
console.log("WebSocket connection closed");
};
return () => {
if (ws.current) {
ws.current.close();
}
};
}, []);
function startSpeech(text: string) {
if (!text || isSpeaking.current) return;
isSpeaking.current = true;
Audio.setAudioModeAsync({
playsInSilentModeIOS: true,
allowsRecordingIOS: false,
shouldDuckAndroid: true,
});
Speech.speak(text, {
rate: 1.0,
language: "en",
onDone: () => {
isSpeaking.current = false;
},
onError: () => {
isSpeaking.current = false;
},
});
}
useEffect(() => {
let interval: string | number | NodeJS.Timeout | undefined;
if (isRecording) {
interval = setInterval(async () => {
if (cameraRef.current) {
try {
const uniqueId = uuidv4();
const photo = await cameraRef.current.takePictureAsync({
quality: 0.5,
base64: true,
skipProcessing: true,
shutterSound: false,
});
if (photo && photo.base64) {
const resizedPhoto = await ImageManipulator.manipulateAsync(
photo.uri,
[{ resize: { width: 128, height: 128 } }],
{ base64: true }
);
if (resizedPhoto && resizedPhoto.base64) {
const imageBuffer = base64ToArrayBuffer(resizedPhoto.base64);
const idBuffer = new TextEncoder().encode(uniqueId + "|");
const combinedBuffer = new Uint8Array(
idBuffer.length + imageBuffer.byteLength
);
combinedBuffer.set(idBuffer, 0);
combinedBuffer.set(new Uint8Array(imageBuffer), idBuffer.length);
if (ws.current && ws.current.readyState === WebSocket.OPEN) {
console.log("Sending binary data to WebSocket");
ws.current.send(combinedBuffer);
}
}
}
} catch (error) {
console.error("Error taking pic:", error);
}
}
}, 2000);
}
return () => clearInterval(interval);
}, [isRecording]);
return (
setIsRecording((prev) => !prev)}
/>
);
}
```
### backend/test/image.py
```python
import asyncio
import time
import picows
import uvloop
import argparse
from pathlib import Path
class Client(picows.WSListener):
def __init__(self, image_path: str):
super().__init__()
self.image_path = image_path
self.start_time = None
self.first_token_time = None
self.transport = None
def on_ws_connected(self, transport: picows.WSTransport):
self.transport = transport
try:
if not Path(self.image_path).is_file():
print(f"Error: Image file not found: {self.image_path}")
transport.disconnect()
return
with open(self.image_path, "rb") as f:
image_bytes = f.read()
print("Connected to server, sending image...")
self.start_time = time.perf_counter()
transport.send(picows.WSMsgType.BINARY, image_bytes)
except Exception as e:
print(f"Error reading image: {e}")
transport.disconnect()
def on_ws_frame(self, transport: picows.WSTransport, frame: picows.WSFrame):
if frame.msg_type == picows.WSMsgType.TEXT:
token = frame.get_payload_as_ascii_text()
if token == "":
total_time = time.perf_counter() - self.start_time
print(f"\nTime for full caption: {total_time:.3f}s")
transport.send_close(picows.WSCloseCode.OK)
transport.disconnect()
return
if self.first_token_time is None:
self.first_token_time = time.perf_counter() - self.start_time
print(f"Time to first token: {self.first_token_time:.3f}s")
print(token, end="", flush=True)
elif frame.msg_type == picows.WSMsgType.CLOSE:
transport.send_close(frame.get_close_code(), frame.get_close_message())
transport.disconnect()
elif frame.msg_type == picows.WSMsgType.PING:
transport.send_pong(frame.get_payload_as_bytes())
async def main(image_path: str):
transport, _ = await picows.ws_connect(
lambda: Client(image_path), "ws://127.0.0.1:9001"
)
await transport.wait_disconnected()
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("image_path", help="Path to the image file to caption")
args = parser.parse_args()
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main(args.image_path))
```
### backend/test/video.py
```python
import picows
import asyncio
import uvloop
import uuid
import cv2
class VideoClient(picows.WSListener):
def __init__(self) -> None:
super().__init__()
self.transport = None
# Caption metadata
self.captions = {}
self.curr_caption_id = None
async def send_frames(self, transport: picows.WSTransport) -> None:
# Open the video file
capture = cv2.VideoCapture("data/video/broll.mp4")
if not capture.isOpened():
raise ValueError("Error opening video file")
# Retrieve FPS of video
fps = capture.get(cv2.CAP_PROP_FPS) if capture.get(cv2.CAP_PROP_FPS) > 0 else 30
interval = int(round(fps * 2))
print(f"Video FPS: {fps}, frames to skip: {interval}")
current_frame = 0
# Begin sending frames to server at 2 FPS
while True:
# Read the next frame from the video
ret, frame = capture.read()
if not ret:
break
# Send the current frame once per second
if current_frame % interval == 0:
ret, buffer = cv2.imencode(".jpg", frame)
if ret:
self.curr_caption_id = str(uuid.uuid4())
transport.send(
picows.WSMsgType.BINARY,
f"{self.curr_caption_id}|".encode() + buffer.tobytes(),
)
current_frame += 1
capture.release()
def on_ws_connected(self, transport: picows.WSTransport) -> None:
print("Established connection to server")
self.transport = transport
asyncio.create_task(self.send_frames(transport))
def on_ws_frame(self, transport: picows.WSTransport, frame: picows.WSFrame) -> None:
if frame.msg_type == picows.WSMsgType.TEXT:
payload = frame.get_payload_as_ascii_text()
caption_id, token = payload.split("|")
token = token.strip()
# If token is the token, the caption has been completed
if token == "":
if caption_id == self.curr_caption_id:
print("Received final caption, initiating disconnection")
transport.disconnect()
return
if caption_id not in self.captions:
self.captions[caption_id] = token
else:
self.captions[caption_id] += " " + token
print(f"---\n{self.captions}")
async def main() -> None:
transport, _ = await picows.ws_connect(VideoClient, "ws://0.0.0.0:2222")
await transport.wait_disconnected()
if __name__ == "__main__":
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main())
```
### backend/src/models/__init__.py
```python
from .base import DeviceModel, Frame, FrameStatus, VendorModel
from .blip import BlipModel
from .llava import LlavaModel
from .groq import LlamaVisionModel
__all__ = [
"Frame",
"FrameStatus",
"DeviceModel",
"VendorModel",
"BlipModel",
"LlavaModel",
"LlamaVisionModel",
]
```
### backend/src/models/base.py
```python
import base64
import io
from abc import ABC, abstractmethod
from collections.abc import Generator, AsyncGenerator
from enum import Enum
from PIL import Image
class Frame:
def __init__(self, data: bytes) -> None:
self.data = data
self.image = None
self.encoded = None
def as_image(self) -> Image.Image:
if not self.image:
self.image = (
Image.open(io.BytesIO(self.data)).convert("RGB").resize((128, 128))
)
return self.image
def as_encoded(self) -> str:
self.encoded = self.encoded or base64.b64encode(self.data).decode("utf-8")
return self.encoded
class FrameStatus(Enum):
Hazard = "hazard"
Safe = "safe"
class DeviceModel(ABC):
@abstractmethod
def warmup(self) -> None:
pass
@abstractmethod
def classify(self, frame: Frame) -> FrameStatus:
pass
@abstractmethod
def caption(self, frame: Frame) -> Generator[str, None, None]:
pass
class VendorModel(ABC):
@abstractmethod
def classify(self, frame: Frame) -> FrameStatus:
pass
@abstractmethod
def caption(self, frame: Frame) -> AsyncGenerator[str, None]:
pass
```
### backend/src/models/blip.py
```python
import io
import torch
import threading
from PIL import Image
from typing import AsyncGenerator
from transformers import (
BlipForConditionalGeneration,
BlipProcessor,
TextIteratorStreamer,
)
from .base import DeviceModel, Frame, FrameStatus
class BlipModel(DeviceModel):
def __init__(self) -> None:
self.model_id = "Salesforce/blip-image-captioning-base"
self.device = torch.device("cuda:0")
# Enable Tensor Cores and cuDNN optimizations
torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.allow_tf32 = True
torch.set_float32_matmul_precision("medium")
torch.backends.cudnn.benchmark = True
# Initialize BLIP model and processor
self.model = BlipForConditionalGeneration.from_pretrained(
self.model_id,
torch_dtype=torch.float16,
).to(self.device)
self.processor = BlipProcessor.from_pretrained(self.model_id)
# Put model in eval/compiled mode
self.model.eval()
self.model = torch.compile(
self.model,
fullgraph=True,
dynamic=True,
mode="max-autotune",
)
# Load model prompts from files
with open("src/prompts/classify.txt", "r") as f:
self.classify_prompt = f.read().strip()
with open("src/prompts/caption.txt", "r") as f:
self.caption_prompt = f.read().strip()
def _process_input(self, frame: Frame, text: str = None):
# Process the frame, using the resized pillow image
inputs = self.processor(
images=frame.as_image(),
text=text,
return_tensors="pt",
)
return inputs.to(self.device, dtype=torch.float16, non_blocking=True)
@torch.inference_mode()
def warmup(self) -> None:
# Create a dummy 128x128 image
dummy = Image.new("RGB", (128, 128), color="white")
buffer = io.BytesIO()
dummy.save(buffer, format="PNG")
# Pass the frame through processor
frame = Frame(buffer.getvalue())
inputs = self._process_input(frame)
for _ in range(3):
_ = self.model.generate(
**inputs,
max_new_tokens=10,
num_beams=1,
do_sample=False,
use_cache=True,
)
torch.cuda.synchronize()
@torch.inference_mode()
async def classify(self, _: Frame) -> FrameStatus:
raise NotImplementedError()
@torch.inference_mode()
async def caption(self, frame: Frame) -> AsyncGenerator[str, None]:
streamer = TextIteratorStreamer(
tokenizer=self.processor.tokenizer,
skip_special_tokens=True,
)
# Prepare generation config (limit to 25 tokens)
generation_config = dict(
max_new_tokens=25,
num_beams=1,
do_sample=False,
min_length=5,
length_penalty=1.0,
use_cache=True,
streamer=streamer,
)
inputs = self._process_input(frame, text=self.caption_prompt)
# Generate the model response in a separate thread
def generate():
self.model.generate(**inputs, **generation_config)
generator = threading.Thread(target=generate)
generator.start()
# Stream tokens in the current thread
for token_text in streamer:
yield token_text
generator.join()
```
[2 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]