(status);
useEffect(() => {
statusRef.current = status;
}, [status]);
// API key check
useEffect(() => {
const key = process.env.GEMINI_API_KEY; // Keep as you had it
if (key && key.trim() !== "") {
setApiKeyPresent(true);
setStatus(DetectionStatus.IDLE);
setStatusMessage("API Key found. Connecting to Pi...");
} else {
setApiKeyPresent(false);
setStatus(DetectionStatus.API_KEY_MISSING);
setStatusMessage("Error: GEMINI_API_KEY environment variable is not set.");
}
}, []);
// ADD: WebSocket connection management
const connectWebSocket = useCallback(() => {
if (wsRef.current?.readyState === WebSocket.OPEN) {
return; // Already connected
}
setStatus(DetectionStatus.WEBSOCKET_CONNECTING);
setStatusMessage("Connecting to Pi sensors...");
try {
wsRef.current = new WebSocket(WEBSOCKET_URL);
wsRef.current.onopen = () => {
console.log('π WebSocket connected to Pi');
setWsConnected(true);
setStatus(DetectionStatus.IDLE);
setStatusMessage("Connected to Pi sensors. Ready to start detection.");
// Clear any reconnection timeout
if (reconnectTimeoutRef.current) {
clearTimeout(reconnectTimeoutRef.current);
reconnectTimeoutRef.current = null;
}
};
wsRef.current.onmessage = (event) => {
try {
const sensorPackage: SensorPackage = JSON.parse(event.data);
setSensorData(sensorPackage);
// FIXED: Store frame data if available
if (sensorPackage.camera?.image_data) {
setLastFrameData(sensorPackage.camera.image_data);
}
// FIXED: Process frame if detection is active and we have frame data
if (isDetecting && sensorPackage.camera?.image_data && !processingFrameRef.current) {
processFrameWithSensorContext(sensorPackage);
}
} catch (error) {
console.error('Error parsing sensor data:', error);
}
};
wsRef.current.onclose = () => {
console.log('π WebSocket disconnected from Pi');
setWsConnected(false);
setStatus(DetectionStatus.WEBSOCKET_DISCONNECTED);
setStatusMessage("Disconnected from Pi. Attempting to reconnect...");
// Attempt to reconnect after 3 seconds
reconnectTimeoutRef.current = setTimeout(() => {
connectWebSocket();
}, 3000);
};
wsRef.current.onerror = (error) => {
console.error('WebSocket error:', error);
setStatus(DetectionStatus.WEBSOCKET_DISCONNECTED);
setStatusMessage("Connection error. Check if Pi sensor service is running.");
};
} catch (error) {
console.error('Failed to create WebSocket connection:', error);
setStatus(DetectionStatus.WEBSOCKET_DISCONNECTED);
setStatusMessage("Failed to connect to Pi. Is the sensor service running?");
}
}, [isDetecting]);
// ADD: Connect to WebSocket when component mounts and API key is available
useEffect(() => {
if (apiKeyPresent) {
connectWebSocket();
}
return () => {
if (wsRef.current) {
wsRef.current.close();
}
if (reconnectTimeoutRef.current) {
clearTimeout(reconnectTimeoutRef.current);
}
};
}, [apiKeyPresent, connectWebSocket]);
// UPDATE: Camera status handler for fallback mode
const handleCameraStatusChange = useCallback((cameraStatus: DetectionStatus, message?: string) => {
// Only use webcam fallback if not connected to Pi
if (!wsConnected && statusRef.current !== DetectionStatus.API_KEY_MISSING) {
setStatus(cameraStatus);
if (message) setStatusMessage(message);
if (cameraStatus === DetectionStatus.CAMERA_ERROR) {
setIsDetecting(false);
} else if (cameraStatus === DetectionStatus.IDLE && apiKeyPresent) {
setStatusMessage(message || "Camera ready. Click 'Start Detection'.");
}
}
}, [wsConnected, apiKeyPresent]);
// ENHANCE: Speech function with sensor context
const speak = (text: string, hazardTy
[truncated β 14324 more characters]
```
### test.html
```html
Pi WebSocket Test
π€ SentientSight Pi Connection Test
π Connecting to Pi...
Network Quality Test Complete! 0% packet loss detected.
Waiting for sensor data...
```
### pi_camera_service.py
```python
#!/usr/bin/env python3
"""
Simple Pi Camera Service for Indoor Navigation
Provides camera images via HTTP API for the navigation agent
"""
import asyncio
import base64
import io
import json
from flask import Flask, jsonify, request
from flask_cors import CORS
import cv2
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = Flask(__name__)
CORS(app)
# Global camera object
camera = None
def initialize_camera():
"""Initialize camera (Pi Camera or USB camera)"""
global camera
try:
# Try Pi Camera first
try:
import picamera
camera = picamera.PiCamera()
camera.resolution = (640, 480)
logger.info("Pi Camera initialized")
return True
except ImportError:
logger.info("Pi Camera not available, trying USB camera")
# Fallback to USB camera via OpenCV
camera = cv2.VideoCapture(0)
if camera.isOpened():
camera.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
logger.info("USB Camera initialized")
return True
else:
logger.error("No camera available")
return False
except Exception as e:
logger.error(f"Error initializing camera: {e}")
return False
def capture_image_pi_camera():
"""Capture image using Pi Camera"""
try:
stream = io.BytesIO()
camera.capture(stream, format='jpeg')
stream.seek(0)
image_data = stream.read()
return base64.b64encode(image_data).decode('utf-8')
except Exception as e:
logger.error(f"Error capturing Pi Camera image: {e}")
return None
def capture_image_usb_camera():
"""Capture image using USB camera"""
try:
ret, frame = camera.read()
if ret:
# Encode frame as JPEG
_, buffer = cv2.imencode('.jpg', frame)
image_data = buffer.tobytes()
return base64.b64encode(image_data).decode('utf-8')
else:
logger.error("Failed to capture frame from USB camera")
return None
except Exception as e:
logger.error(f"Error capturing USB camera image: {e}")
return None
@app.route('/capture', methods=['GET'])
def capture_image():
"""Capture and return current camera image as base64"""
try:
if camera is None:
return jsonify({'error': 'Camera not initialized'}), 500
# Determine camera type and capture accordingly
if hasattr(camera, 'capture'): # Pi Camera
image_base64 = capture_image_pi_camera()
else: # USB Camera
image_base64 = capture_image_usb_camera()
if image_base64:
return jsonify({
'status': 'success',
'image_data': image_base64,
'timestamp': asyncio.get_event_loop().time()
})
else:
return jsonify({'error': 'Failed to capture image'}), 500
except Exception as e:
logger.error(f"Error in capture endpoint: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/status', methods=['GET'])
def camera_status():
"""Get camera status"""
return jsonify({
'camera_available': camera is not None,
'camera_type': 'pi_camera' if hasattr(camera, 'capture') else 'usb_camera'
})
@app.route('/health', methods=['GET'])
def health_check():
"""Health check endpoint"""
return jsonify({'status': 'healthy', 'service': 'pi_camera_service'})
def cleanup_camera():
"""Cleanup camera resources"""
global camera
if camera:
try:
if hasattr(camera, 'close'): # Pi Camera
camera.close()
else: # USB Camera
camera.release()
logger.info("Camera cleaned up")
except Exception as e:
logger.error(f"Error cleaning up camera: {e}")
if __name__ == '__main__':
try:
logger.info("Starting Pi Camera Service for Indoor Navigation")
if initialize_camera():
logger.info("Camera service ready")
# Run Flask app
app.run(host='0.0.0.0', port=8000, debug=False)
else:
logger.error("Failed to initialize camera. Exiting.")
except KeyboardInterrupt:
logger.info("Service interrupted by user")
except Exception as e:
logger.error(f"Service error: {e}")
finally:
cleanup_camera()
logger.info("Pi Camera Service shutdown complete")
```
### photo_analyzer.py
```python
#!/usr/bin/env python3
"""
Photo Analysis Script for TAN OAK Room Navigation
Uses CLIP embeddings to find similar reference images for position detection
"""
import os
import json
import base64
import numpy as np
from pathlib import Path
from typing import Dict, List, Tuple
import logging
from sentence_transformers import SentenceTransformer
from PIL import Image
import io
import requests
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class PhotoAnalyzer:
def __init__(self, reference_images_dir: str):
self.reference_dir = Path(reference_images_dir)
# Use CLIP model for image embeddings
self.model = SentenceTransformer('clip-ViT-B-32')
self.reference_embeddings = {}
self.reference_metadata = {}
def process_reference_images(self):
"""Process all reference images and create embeddings"""
logger.info("Processing reference images...")
# Support both .jpg and .jpeg extensions
image_files = list(self.reference_dir.glob("*.jpg")) + list(self.reference_dir.glob("*.jpeg"))
for image_file in image_files:
try:
# Load image
with open(image_file, "rb") as f:
image_data = f.read()
# Create embedding
image = Image.open(io.BytesIO(image_data))
embedding = self.model.encode(image)
# Store embedding and metadata
self.reference_embeddings[image_file.stem] = embedding
self.reference_metadata[image_file.stem] = {
"filename": image_file.name,
"position": self._extract_position_from_filename(image_file.stem),
"landmarks": [] # To be filled manually or via AI
}
logger.info(f"Processed: {image_file.name}")
except Exception as e:
logger.error(f"Error processing {image_file.name}: {e}")
def _extract_position_from_filename(self, filename: str) -> str:
"""Extract position info from filename if available"""
# Example: "chair_1_facing_screen.jpg" β "sitting at chair 1 facing screen"
# Example: "door_entrance.jpg" β "near door entrance"
filename_lower = filename.lower()
if "chair" in filename_lower:
return f"sitting at {filename_lower.replace('_', ' ')}"
elif "door" in filename_lower:
return f"near {filename_lower.replace('_', ' ')}"
elif "center" in filename_lower:
return f"at {filename_lower.replace('_', ' ')}"
else:
return filename_lower.replace('_', ' ')
def find_similar_position(self, current_image_base64: str, top_k: int = 3) -> List[Tuple[str, float, str]]:
"""Find most similar reference images to current view"""
try:
# Decode current image
image_data = base64.b64decode(current_image_base64)
current_image = Image.open(io.BytesIO(image_data))
# Get embedding for current image
current_embedding = self.model.encode(current_image)
# Calculate similarities
similarities = []
for ref_name, ref_embedding in self.reference_embeddings.items():
similarity = np.dot(current_embedding, ref_embedding) / (
np.linalg.norm(current_embedding) * np.linalg.norm(ref_embedding)
)
position = self.reference_metadata[ref_name]["position"]
similarities.append((ref_name, similarity, position))
# Sort by similarity (highest first)
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_k]
except Exception as e:
logger.error(f"Error finding similar position: {e}")
return []
def analyze_position(self, current_image_base64: str) -> Dict:
"""Analyze current position using similarity search"""
similar_positions = self.find_similar_position(current_image_base64)
if not similar_positions:
return {"position": "unknown", "confidence": "low", "matches": []}
# Get best match
best_match = similar_positions[0]
best_name, best_similarity, best_position = best_match
# Determine confidence
if best_similarity > 0.8:
confidence = "high"
elif best_similarity > 0.6:
confidence = "medium"
else:
confidence = "low"
return {
"position": best_position,
"confidence": confidence,
"similarity_score": float(best_similarity),
"best_match": best_name,
"top_matches": [
{"name": name, "similarity": float(sim), "position": pos}
for name, sim, pos in similar_positions
]
}
def save_embeddings(self, filepath: str):
"""Save embeddings and metadata to file"""
data = {
"embeddings": {k: v.tolist() for k, v in self.reference_embeddings.items()},
"metadata": self.reference_metadata
}
with open(filepath, 'w') as f:
json.dump(data, f, indent=2)
logger.info(f"Saved embeddings to {filepath}")
def load_embeddings(self, filepath: str):
"""Load embeddings and metadata from file"""
if not os.path.exists(filepath):
logger.warning(f"Embeddings file not found: {filepath}")
return
with open(filepath, 'r') as f:
data = json.load(f)
# Convert lists back to numpy arrays
self.reference_embeddings = {
k: np.array(v) for k,
[truncated β 1334 more characters]
```
### indoor_navigation_agent.py
```python
import asyncio
import logging
from dotenv import load_dotenv
import os
import json
import base64
import requests
from typing import Optional, Dict, List
import datetime
from pathlib import Path
from photo_analyzer import PhotoAnalyzer
from livekit import agents
from livekit.agents import Agent, AgentSession
import livekit.plugins.openai as openai_plugin
from livekit.plugins import silero
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Reference Images Directory
REFERENCE_IMAGES_DIR = Path(__file__).parent / "reference_images" / "tan_oak"
# Room Database - TAN OAK Room (4th Floor MLK)
ROOM_DATABASE = {
"tan_oak_room": {
"name": "TAN OAK Conference Room",
"floor": "4th Floor",
"building": "MLK Student Union",
"landmarks": {
"conference_table": {
"description": "Large wooden conference table with chairs",
"visual_cues": ["rectangular table", "multiple chairs", "dark wooden surface"],
"position": "center"
},
"projector_screen": {
"description": "Wall-mounted projector screen",
"visual_cues": ["white screen", "mounted on wall", "rectangular"],
"position": "front_wall"
},
"windows": {
"description": "Large windows with view outside",
"visual_cues": ["natural light", "glass panels", "window frames"],
"position": "exterior_wall"
},
"whiteboards": {
"description": "Wall-mounted whiteboards",
"visual_cues": ["white surface", "marker trays", "rectangular"],
"position": "side_wall"
},
"door_entrance": {
"description": "Main room entrance door",
"visual_cues": ["door handle", "door frame", "exit sign"],
"position": "entrance_wall"
}
},
"exits": {
"main_door": {
"description": "Main entrance/exit door",
"location": "entrance_wall",
"leads_to": "hallway_to_elevator",
"visual_cues": ["door handle", "exit sign", "hallway visible"]
}
},
"navigation_instructions": {
"from_center_to_exit": {
"if_facing_screen": "Turn around 180 degrees, walk straight to the door",
"if_facing_windows": "Turn left 90 degrees, walk straight to the door",
"if_facing_door": "Walk straight ahead to the door",
"if_back_to_door": "Turn around 180 degrees, walk straight to the door"
},
"from_chair_to_exit": {
"default": "Stand up, face the door, and walk straight to the exit"
}
},
"reference_positions": {
# We'll map specific viewpoints from your photos
# Format: "photo_name": {"position": "description", "landmarks_visible": [...]}
}
}
}
class ConvexClient:
def __init__(self, deployment_url: str):
"""Initialize Convex client for getting navigation images"""
self.base_url = deployment_url.rstrip('/')
def get_latest_navigation_image(self) -> Optional[str]:
"""Get the latest navigation image as base64 string"""
url = f"{self.base_url}/get_latest_navigation_image"
try:
response = requests.post(
url,
json={},
headers={'Content-Type': 'application/json'},
timeout=10
)
response.raise_for_status()
result = response.json()
if result.get('success') and result.get('image_data'):
logger.info(f"πΈ Got latest image: {result.get('image_id')}")
return result['image_data'] # Already base64 encoded
else:
logger.warning(f"No image available: {result.get('message', 'Unknown error')}")
return None
except requests.exceptions.RequestException as e:
logger.error(f"Error getting latest navigation image: {e}")
return None
def get_navigation_service_status(self) -> Dict:
"""Get status of the navigation camera service"""
url = f"{self.base_url}/get_navigation_service_status"
try:
response = requests.post(
url,
json={},
headers={'Content-Type': 'application/json'},
timeout=5
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Error getting service status: {e}")
return {"service_active": False, "error": str(e)}
def load_reference_images() -> Dict:
"""Load reference images from the photos directory"""
reference_images = {}
if REFERENCE_IMAGES_DIR.exists():
for image_file in REFERENCE_IMAGES_DIR.glob("*.jpg"):
try:
with open(image_file, "rb") as f:
image_data = base64.b64encode(f.read()).decode('utf-8')
reference_images[image_file.stem] = image_data
logger.info(f"Loaded reference image: {image_file.name}")
except Exception as e:
logger.error(f"Error loading {image_file.name}: {e}")
return reference_images
class IndoorNavigationAgent(Agent):
"""Indoor Navigation Assistant for TAN OAK Room"""
def __init__(self) -> None:
super().__init__(
instructions=(
"You are an indoor navigation assistant helping blind users navigate the TAN OAK conference room. "
"Yo
[truncated β 11431 more characters]
```
### .idea/vcs.xml
```xml
```
[19 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]