Upload 41 files

.env.example

@@ -0,0 +1,11 @@
+# SIGHTLINE Configuration
+SIGHTLINE_MODEL_NAME=microsoft/Florence-2-base
+SIGHTLINE_CAPTURE_INTERVAL=3.0
+SIGHTLINE_SCENE_THRESHOLD=0.10
+SIGHTLINE_DEBOUNCE_MS=800
+SIGHTLINE_MAX_DIM=768
+SIGHTLINE_HASH_SIZE=16
+SIGHTLINE_TTS_TIMEOUT=12.0
+SIGHTLINE_TTS_RATE=+8%
+SIGHTLINE_AUDIO_FORMAT=mp3
+SIGHTLINE_MAX_QUEUE_SIZE=3

.gitattributes

@@ -1,35 +1,35 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,20 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    libgl1-mesa-glx \
+    libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENV PYTHONUNBUFFERED=1
+
+EXPOSE 7860
+
+CMD ["python", "-m", "app.main"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Lovnish Verma
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,13 +1,51 @@
----
-title: SightLine
-emoji: ⚡
-colorFrom: gray
-colorTo: purple
-sdk: gradio
-sdk_version: 6.25.0
-python_version: '3.12'
-app_file: app.py
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# SightLine: See the World Through AI
+
+SightLine is a voice-first multimodal AI assistant designed primarily for blind and low-vision users.
+
+## Features
+- **Real-Time Scene Understanding:** Continuously analyze the camera feed to detect people, vehicles, obstacles, etc.
+- **Voice-First Interaction:** Hands-free interaction via voice commands (e.g. "What do you see?", "Read the sign").
+- **Smart Scene Change Detection:** Triggers descriptions only when meaningful changes occur in the scene.
+- **Conversational Context:** Remembers recent objects and OCR text to provide natural context-aware responses.
+- **Accessibility First:** High contrast, ARIA labels, screen reader support, large controls.
+
+## Project Structure
+```text
+sightline/
+├── app/
+│   ├── main.py
+│   ├── config.py
+│   ├── vision/
+│   ├── speech/
+│   ├── conversation/
+│   ├── camera/
+│   ├── accessibility/
+│   └── ui/
+├── Dockerfile
+├── requirements.txt
+└── README.md
+```
+
+## Running Locally
+
+1. Create a virtual environment and install dependencies:
+   ```bash
+   python -m venv venv
+   source venv/bin/activate
+   pip install -r requirements.txt
+   pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121 --upgrade --force-reinstall
+   ```
+2. Start the application:
+   ```bash
+   python -m app.main
+   ```
+3. Open the UI at `http://localhost:7860`.
+
+## Docker Deployment
+```bash
+docker build -t sightline .
+docker run -p 7860:7860 sightline
+```
+
+## Privacy & Safety
+SightLine does not permanently store camera frames or microphone recordings. It processes imagery purely to provide instant auditory feedback to the user. **Note:** SightLine is not a replacement for a cane, guide dog, caregiver, or professional mobility aid.

app/accessibility/accessibility.py

@@ -0,0 +1,7 @@
+def create_aria_live_region() -> str:
+    """Returns HTML string for screen reader announcements."""
+    return '<div class="sr-only" aria-live="assertive" id="aria-live-region" role="status"></div>'
+
+def announce_status(message: str) -> str:
+    """Formats a message to be read by screen readers immediately."""
+    return f'<div id="sightline-status" role="status" aria-live="assertive">{message}</div>'

app/config.py

@@ -0,0 +1,42 @@
+import os
+from dataclasses import dataclass
+from typing import Dict
+
+@dataclass
+class Config:
+    """Central configuration for SightLine."""
+
+    # Timing
+    CAPTURE_INTERVAL: float = float(os.getenv("SIGHTLINE_CAPTURE_INTERVAL", "3.0"))
+    SCENE_THRESHOLD: float = float(os.getenv("SIGHTLINE_SCENE_THRESHOLD", "0.10"))
+    DEBOUNCE_MS: int = int(os.getenv("SIGHTLINE_DEBOUNCE_MS", "800"))
+    MAX_DIM: int = int(os.getenv("SIGHTLINE_MAX_DIM", "768"))
+    HASH_SIZE: int = int(os.getenv("SIGHTLINE_HASH_SIZE", "16"))
+
+    # Audio
+    TTS_TIMEOUT: float = float(os.getenv("SIGHTLINE_TTS_TIMEOUT", "12.0"))
+    TTS_RATE: str = os.getenv("SIGHTLINE_TTS_RATE", "+8%")
+    AUDIO_FORMAT: str = os.getenv("SIGHTLINE_AUDIO_FORMAT", "mp3")
+    MAX_QUEUE_SIZE: int = int(os.getenv("SIGHTLINE_MAX_QUEUE_SIZE", "3"))
+
+    # Model
+    MODEL_NAME: str = os.getenv("SIGHTLINE_MODEL_NAME", "microsoft/Florence-2-base")
+    
+    # Text generation limits for tasks
+    MAX_NEW_TOKENS: Dict[str, int] = None
+
+    # UI
+    APP_NAME: str = "SightLine"
+    APP_VERSION: str = "1.0"
+    
+    def __post_init__(self):
+        if self.MAX_NEW_TOKENS is None:
+            self.MAX_NEW_TOKENS = {
+                "<CAPTION>": 64,
+                "<DETAILED_CAPTION>": 120,
+                "<MORE_DETAILED_CAPTION>": 200,
+                "<OD>": 256,
+                "<OCR>": 300,
+            }
+
+CONFIG = Config()

app/conversation/assistant.py

@@ -0,0 +1,78 @@
+from app.vision.florence import FlorenceVisionEngine
+from app.conversation.context import CONTEXT
+from app.speech.tts import text_to_speech
+from app.speech.audio_manager import AudioQueue
+from app.vision.scene_change import compute_hash, hash_distance
+from app.config import CONFIG
+from PIL import Image
+
+AUDIO_QUEUE = AudioQueue(max_size=CONFIG.MAX_QUEUE_SIZE)
+
+class SightLineAssistant:
+    """Orchestrates vision, context, and speech."""
+    
+    def __init__(self):
+        self.vision = FlorenceVisionEngine()
+        self.audio_finish_time = 0.0
+        
+    def initialize(self):
+        self.vision.load()
+        
+    def process_image(self, image, task: str, voice_name: str, force: bool = False):
+        """Process an image and generate a response."""
+        if self.vision.model is None:
+            return "Model is initializing, please wait...", None, "⏳ Model Loading..."
+            
+        import time
+        if not force and time.time() < getattr(self, "audio_finish_time", 0.0):
+            return None, None, "🔊 Speaking..."
+            
+        import numpy as np
+        if isinstance(image, np.ndarray):
+            image = Image.fromarray(image)
+            
+        img_hash = compute_hash(image)
+        
+        # Map human-readable tasks to internal tokens
+        task_map = {
+            "Quick Glance": "<CAPTION>",
+            "Detailed Scene": "<DETAILED_CAPTION>",
+            "Immersive Description": "<MORE_DETAILED_CAPTION>",
+            "Read Text": "<OCR>"
+        }
+        internal_task = task_map.get(task, "<DETAILED_CAPTION>")
+        
+        # Debounce/Duplicate check if not forced
+        if not force and CONTEXT.is_duplicate(img_hash, internal_task):
+            text, audio = CONTEXT.get_last()
+            return text, audio, "Used cached result"
+            
+        # Inference based on task
+        if internal_task == "<OCR>":
+            response = self.vision.read_text(image)
+        elif internal_task == "<MORE_DETAILED_CAPTION>":
+            response = self.vision.describe_scene(image, detailed=True)
+        else:
+            response = self.vision.describe_scene(image, detailed=False)
+            
+        # TTS
+        audio_path = text_to_speech(response, voice_name)
+        
+        if audio_path:
+            import time
+            try:
+                from pydub import AudioSegment
+                duration = AudioSegment.from_file(audio_path).duration_seconds
+            except Exception as e:
+                print(f"Duration error: {e}")
+                duration = len(response) / 15.0  # Fallback rough estimate
+            self.audio_finish_time = time.time() + duration
+
+        # Update context
+        CONTEXT.update(img_hash, task, response, audio_path)
+        
+        # Enqueue audio
+        if audio_path:
+            AUDIO_QUEUE.enqueue(response, audio_path)
+            
+        return response, audio_path, f"Processed task: {task}"

app/conversation/context.py

@@ -0,0 +1,59 @@
+from dataclasses import dataclass, field
+import threading
+import time
+from typing import List, Dict, Any, Optional, Tuple
+
+@dataclass
+class SessionContext:
+    """Thread-safe application state and conversational context."""
+
+    # Scene hashing
+    last_hash: Optional[bytes] = None
+    last_task: str = ""
+    last_text: str = ""
+    last_audio: Optional[str] = None
+
+    # Realtime
+    realtime_active: bool = False
+    last_capture_time: float = 0.0
+
+    # History
+    history: List[Dict[str, Any]] = field(default_factory=list)
+    max_history: int = 50
+
+    # Lock
+    _lock: threading.Lock = field(default_factory=threading.Lock)
+
+    def update(self, hash_val: bytes, task: str, text: str, audio: Optional[str]):
+        """Update state with new capture results."""
+        with self._lock:
+            self.last_hash = hash_val
+            self.last_task = task
+            self.last_text = text
+            self.last_audio = audio
+            
+            # Add to history
+            self.history.insert(0, {
+                "time": time.strftime("%H:%M:%S"),
+                "task": task,
+                "text": text,
+            })
+            if len(self.history) > self.max_history:
+                self.history = self.history[: self.max_history]
+
+    def is_duplicate(self, hash_val: bytes, task: str) -> bool:
+        """Check if this hash+task combination was already processed."""
+        with self._lock:
+            return (
+                self.last_hash is not None
+                and self.last_hash == hash_val
+                and self.last_task == task
+                and self.last_text != ""
+            )
+
+    def get_last(self) -> Tuple[str, Optional[str]]:
+        """Get last description text and audio."""
+        with self._lock:
+            return self.last_text, self.last_audio
+
+CONTEXT = SessionContext()

app/conversation/intent.py

@@ -0,0 +1,15 @@
+import re
+
+def detect_intent(text: str) -> str:
+    """Map natural language to a vision task token."""
+    text = text.lower().strip()
+    
+    if re.search(r"read|sign|text|document|label|menu", text):
+        return "<OCR>"
+    elif re.search(r"where|object|nearby|around me|front of me", text):
+        return "<OD>"
+    elif re.search(r"detail|describe.*room", text):
+        return "<MORE_DETAILED_CAPTION>"
+    else:
+        # Default to standard caption
+        return "<DETAILED_CAPTION>"

app/main.py

@@ -0,0 +1,28 @@
+import gradio as gr
+from app.ui.components import build_ui
+from app.conversation.assistant import SightLineAssistant
+from app.config import CONFIG
+from app.ui.styles import CSS
+
+import threading
+
+def main():
+    print(f"Starting {CONFIG.APP_NAME} v{CONFIG.APP_VERSION}")
+    
+    assistant = SightLineAssistant()
+    # Load model in background thread for instant UI rendering
+    threading.Thread(target=assistant.initialize, daemon=True).start()
+    
+    demo = build_ui(assistant)
+    
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False,
+        debug=True,
+        css=CSS
+    )
+
+
+if __name__ == "__main__":
+    main()

app/speech/audio_manager.py

@@ -0,0 +1,63 @@
+import threading
+import os
+from collections import deque
+from typing import Tuple, Optional
+
+class AudioQueue:
+    """Thread-safe FIFO audio queue with interruption support."""
+
+    def __init__(self, max_size: int = 3):
+        self._queue: deque[Tuple[str, str]] = deque()  # (text, audio_path)
+        self._current: Optional[str] = None
+        self._lock = threading.Lock()
+        self._counter = 0
+        self._max_size = max_size
+
+    def enqueue(self, text: str, audio_path: str) -> Optional[str]:
+        """Add audio to queue. Returns the path to play (or None if queue full)."""
+        with self._lock:
+            if len(self._queue) >= self._max_size:
+                oldest = self._queue.popleft()
+                self._safe_delete(oldest[1])
+            self._queue.append((text, audio_path))
+            self._counter += 1
+            return audio_path
+
+    def dequeue(self) -> Optional[Tuple[str, str]]:
+        """Get next audio item."""
+        with self._lock:
+            if self._queue:
+                item = self._queue.popleft()
+                self._current = item[1]
+                return item
+            return None
+
+    def clear(self):
+        """Clear all queued audio and delete files."""
+        with self._lock:
+            for _, path in self._queue:
+                self._safe_delete(path)
+            self._queue.clear()
+            self._current = None
+
+    def interrupt(self):
+        """Interrupt current and clear queue."""
+        self.clear()
+
+    @property
+    def is_empty(self) -> bool:
+        with self._lock:
+            return len(self._queue) == 0
+
+    @property
+    def size(self) -> int:
+        with self._lock:
+            return len(self._queue)
+
+    @staticmethod
+    def _safe_delete(path: str):
+        try:
+            if path and os.path.exists(path):
+                os.unlink(path)
+        except OSError:
+            pass

app/speech/stt.py

@@ -0,0 +1,17 @@
+class STTEngine:
+    """Abstract interface for Speech-to-Text."""
+    
+    def load(self):
+        """Initialize the STT model/provider."""
+        pass
+        
+    def transcribe(self, audio_file: str) -> str:
+        """Transcribe an audio file into text."""
+        return ""
+
+class MockSTT(STTEngine):
+    """Fallback mock STT since we are focusing on UI/Vision first."""
+    def transcribe(self, audio_file: str) -> str:
+        return "describe the scene"
+
+# Future integration with faster-whisper goes here.

app/speech/tts.py

@@ -0,0 +1,53 @@
+import asyncio
+import threading
+import tempfile
+from typing import Optional
+import edge_tts
+from app.config import CONFIG
+
+# Voice mappings for supported locales
+VOICE_MAP = {
+    "English (US) - Aria": "en-US-AriaNeural",
+    "English (US) - Guy": "en-US-GuyNeural",
+    "English (UK) - Sonia": "en-GB-SoniaNeural",
+    "Hindi - Swara": "hi-IN-SwaraNeural",
+    "Hindi - Madhur": "hi-IN-MadhurNeural",
+}
+
+def init_tts_loop() -> asyncio.AbstractEventLoop:
+    """Create a dedicated event loop for TTS in a background thread."""
+    loop = asyncio.new_event_loop()
+
+    def _run():
+        asyncio.set_event_loop(loop)
+        loop.run_forever()
+
+    threading.Thread(target=_run, daemon=True).start()
+    return loop
+
+_TTS_LOOP = init_tts_loop()
+
+def text_to_speech(text: str, voice_name: str = "English (US) - Aria") -> Optional[str]:
+    """Convert text to speech, returning the audio file path."""
+    if not text or not text.strip():
+        return None
+
+    try:
+        voice_id = VOICE_MAP.get(voice_name, "en-US-AriaNeural")
+
+        async def _generate():
+            with tempfile.NamedTemporaryFile(delete=False, suffix=f".{CONFIG.AUDIO_FORMAT}") as f:
+                path = f.name
+            communicate = edge_tts.Communicate(
+                text.strip(),
+                voice=voice_id,
+                rate=CONFIG.TTS_RATE,
+            )
+            await communicate.save(path)
+            return path
+
+        future = asyncio.run_coroutine_threadsafe(_generate(), _TTS_LOOP)
+        return future.result(timeout=CONFIG.TTS_TIMEOUT)
+    except Exception as e:
+        print(f"TTS error: {e}")
+        return None

app/ui/components.py

@@ -0,0 +1,83 @@
+import gradio as gr
+from app.ui.styles import CSS
+from app.config import CONFIG
+
+def _status_html(msg: str) -> str:
+    """Wrap a status message in the styled status-bar div."""
+    return f'<div id="sightline-status" role="status" aria-live="polite">{msg}</div>'
+
+def build_ui(assistant) -> gr.Blocks:
+    auto_start_js = """
+    function() {
+        setInterval(function() {
+            let clickTarget = function(btn) {
+                let text = (btn.textContent || btn.innerText || '').toLowerCase().trim();
+                if (text === 'click to access webcam' || text === 'record') {
+                    btn.click();
+                }
+            };
+            
+            // Check Shadow DOMs
+            document.querySelectorAll('*').forEach(function(el) {
+                if (el.shadowRoot) {
+                    el.shadowRoot.querySelectorAll('button').forEach(clickTarget);
+                }
+            });
+            
+            // Check regular DOM
+            document.querySelectorAll('button').forEach(clickTarget);
+        }, 1000); // Check every second indefinitely
+    }
+    """
+    with gr.Blocks(title=f"{CONFIG.APP_NAME} v{CONFIG.APP_VERSION}", js=auto_start_js) as demo:
+        gr.HTML('<div class="sr-only" aria-live="assertive" id="aria-live-region" role="status"></div>')
+        
+        status_bar = gr.HTML(_status_html("✅ Ready — Press D to describe"))
+
+        gr.Markdown(f"# 👁️ {CONFIG.APP_NAME}")
+        
+        rt_state = gr.State(True)
+
+        with gr.Row():
+            with gr.Column(scale=1):
+                # Controls at the top
+                with gr.Row():
+                    describe_btn = gr.Button("🔍 Describe (D)", variant="primary", elem_id="btn-describe")
+                    realtime_btn = gr.Button("⚫ Stop Realtime (R)", variant="secondary", elem_id="btn-realtime")
+                
+                with gr.Row():
+                    task_radio = gr.Radio(
+                        choices=["Quick Glance", "Detailed Scene", "Immersive Description", "Read Text"],
+                        value="Detailed Scene",
+                        label="Mode"
+                    )
+                    voice_dropdown = gr.Dropdown(
+                        choices=["English (US) - Aria", "Hindi - Swara"],
+                        value="English (US) - Aria",
+                        label="Voice"
+                    )
+                
+                webcam = gr.Image(label="📷 Camera", type="numpy", sources=["webcam"], streaming=True)
+                with gr.Accordion("Upload Image", open=False):
+                    upload = gr.Image(label="📁 Upload", type="numpy", sources=["upload"])
+
+            with gr.Column(scale=1):
+                caption_box = gr.Textbox(label="📝 Description", lines=6, interactive=False)
+                audio_player = gr.Audio(label="🔊 Audio", type="filepath", autoplay=True)
+                
+                with gr.Row():
+                    repeat_btn = gr.Button("🔁 Repeat (P)", elem_id="btn-repeat")
+                    stop_btn = gr.Button("⏹ Stop (Esc)", elem_id="btn-stop")
+
+        components = {
+            "webcam": webcam, "upload": upload, "task_radio": task_radio,
+            "voice_dropdown": voice_dropdown, "describe_btn": describe_btn,
+            "realtime_btn": realtime_btn, "caption_box": caption_box,
+            "audio_player": audio_player, "status_bar": status_bar,
+            "rt_state": rt_state, "repeat_btn": repeat_btn, "stop_btn": stop_btn
+        }
+        
+        from app.ui.events import wire_events
+        wire_events(demo, components, assistant)
+        
+        return demo

app/ui/events.py

@@ -0,0 +1,66 @@
+import gradio as gr
+
+def wire_events(demo, components, assistant):
+    c = components
+
+    def handle_describe(webcam_img, upload_img, task, voice):
+        is_webcam = webcam_img is not None
+        img = webcam_img if is_webcam else upload_img
+        if img is None:
+            return gr.update(), gr.update(), f'<div id="sightline-status">Please provide an image.</div>'
+            
+        import numpy as np
+        import cv2
+        # If the image came from the webcam, it is likely horizontally mirrored by Gradio.
+        # We must un-mirror it so text and spatial relationships are correct!
+        if is_webcam and isinstance(img, np.ndarray):
+            img = cv2.flip(img, 1)
+            
+        text, audio, status = assistant.process_image(img, task, voice, force=True)
+        return text or gr.update(), audio or gr.update(), f'<div id="sightline-status">{status}</div>'
+
+    c["describe_btn"].click(
+        handle_describe,
+        inputs=[c["webcam"], c["upload"], c["task_radio"], c["voice_dropdown"]],
+        outputs=[c["caption_box"], c["audio_player"], c["status_bar"]]
+    )
+
+    def toggle_rt(is_active):
+        new_state = not is_active
+        btn_text = "⚫ Stop Realtime (R)" if new_state else "⚫ Start Realtime (R)"
+        status_msg = "Realtime Started" if new_state else "Realtime Paused"
+        return new_state, gr.update(value=btn_text), f'<div id="sightline-status">{status_msg}</div>'
+
+    c["realtime_btn"].click(
+        toggle_rt,
+        inputs=[c["rt_state"]],
+        outputs=[c["rt_state"], c["realtime_btn"], c["status_bar"]]
+    )
+
+    def handle_rt_stream(image, task, voice, is_active):
+        print(f"DEBUG stream: is_active={is_active}, task={task}, voice={voice}")
+        if is_active is False:
+            return gr.update(), gr.update(), f'<div id="sightline-status">Realtime Paused</div>'
+        if not is_active:
+            # Fallback if None
+            is_active = True
+            
+        if image is None:
+            return gr.update(), gr.update(), f'<div id="sightline-status">No Camera</div>'
+            
+        import numpy as np
+        import cv2
+        # Un-mirror the webcam feed for backend processing
+        if isinstance(image, np.ndarray):
+            image = cv2.flip(image, 1)
+            
+        text, audio, status = assistant.process_image(image, task, voice, force=False)
+        return text or gr.update(), audio or gr.update(), f'<div id="sightline-status">{status}</div>'
+
+    c["webcam"].stream(
+        handle_rt_stream,
+        inputs=[c["webcam"], c["task_radio"], c["voice_dropdown"], c["rt_state"]],
+        outputs=[c["caption_box"], c["audio_player"], c["status_bar"]],
+        stream_every=3.0
+    )
+

app/ui/styles.py

@@ -0,0 +1,62 @@
+CSS = """
+/* Base Variables */
+:root {
+    --accent: #2563eb;
+    --accent-hover: #1d4ed8;
+    --bg-primary: #ffffff;
+    --bg-secondary: #f8fafc;
+    --text-primary: #1e293b;
+    --border: #e2e8f0;
+    --radius: 12px;
+}
+
+/* High Contrast Support */
+body.hc {
+    filter: contrast(1.7) brightness(1.05);
+}
+
+.gr-button {
+    min-height: 80px !important; /* Larger touch targets */
+    font-size: 1.3em !important;
+    border-radius: var(--radius) !important;
+    font-weight: bold !important;
+    border: 4px solid transparent !important;
+    transition: all 0.2s ease-in-out;
+}
+
+.gr-button:focus, .gr-button:hover {
+    border: 4px solid #FFD700 !important; /* High contrast yellow focus ring */
+    outline: none !important;
+    transform: scale(1.02);
+}
+
+.gr-button-primary {
+    background: #000000 !important;
+    color: #FFFFFF !important;
+    border: 4px solid #FFFFFF !important;
+}
+
+#sightline-status {
+    background: #000000;
+    color: #00FF00; /* High contrast terminal green */
+    padding: 20px;
+    border-radius: var(--radius);
+    font-size: 1.4em;
+    font-weight: 800;
+    text-align: center;
+    border: 3px solid #00FF00;
+    box-shadow: 0 4px 6px rgba(0,255,0,0.2);
+}
+
+/* Screen reader only */
+.sr-only {
+    position: absolute;
+    width: 1px;
+    height: 1px;
+    padding: 0;
+    margin: -1px;
+    overflow: hidden;
+    clip: rect(0, 0, 0, 0);
+    border: 0;
+}
+"""

app/vision/captioning.py

@@ -0,0 +1,12 @@
+def format_caption(raw_caption: str) -> str:
+    """Format raw scene descriptions for speech clarity."""
+    caption = raw_caption.strip()
+    if not caption:
+        return "I couldn't understand what's in the image."
+    
+    # Capitalize first letter, ensure period at the end
+    if not caption.endswith('.'):
+        caption += '.'
+    caption = caption[0].upper() + caption[1:]
+    
+    return caption

app/vision/detection.py

@@ -0,0 +1,52 @@
+from typing import Dict, List, Tuple
+
+def format_object_detection(od_data: Dict) -> str:
+    """Format object detection results into natural spatial language."""
+    if not od_data or not od_data.get("labels"):
+        return "No objects detected."
+
+    labels = od_data.get("labels", [])
+    bboxes = od_data.get("bboxes", [])
+
+    if not labels:
+        return "No objects detected."
+
+    objects: List[Tuple[str, str]] = []
+    for label, bbox in zip(labels, bboxes):
+        x1, y1, x2, y2 = bbox
+        cx = (x1 + x2) / 2
+        cy = (y1 + y2) / 2
+        
+        # Florence uses 0-1000 coordinate space typically, or 0-999
+        if cx < 333:
+            pos = "on the left"
+        elif cx < 666:
+            pos = "in the center"
+        else:
+            pos = "on the right"
+            
+        objects.append((label.strip(), pos))
+
+    # Deduplicate (keep first occurrence of each label type)
+    seen: set = set()
+    unique: List[Tuple[str, str]] = []
+    for lbl, pos in objects:
+        key = lbl.lower()
+        if key and key not in seen:
+            seen.add(key)
+            unique.append((lbl, pos))
+
+    if not unique:
+        return "No objects detected."
+
+    if len(unique) == 1:
+        lbl, pos = unique[0]
+        return f"I see a {lbl} {pos}."
+
+    parts = [f"a {lbl} {pos}" for lbl, pos in unique]
+
+    if len(parts) <= 5:
+        return "I see " + ", ".join(parts[:-1]) + f", and {parts[-1]}."
+    else:
+        summary = ", ".join(parts[:5])
+        return f"I see {len(unique)} objects including: {summary}, and {len(unique) - 5} more."

app/vision/florence.py

@@ -0,0 +1,150 @@
+import torch
+from transformers import AutoModelForCausalLM, AutoProcessor
+from PIL import Image
+from typing import Optional, Dict, Any
+
+from app.vision.vision_engine import VisionEngine
+from app.config import CONFIG
+from app.vision.utils import preprocess_image, auto_enhance
+from app.vision.captioning import format_caption
+from app.vision.ocr import format_ocr
+from app.vision.detection import format_object_detection
+
+def get_device() -> str:
+    """Select best available device."""
+    if torch.cuda.is_available():
+        return "cuda"
+    elif torch.backends.mps.is_available():
+        return "mps"
+    return "cpu"
+
+DEVICE: str = get_device()
+DTYPE: torch.dtype = torch.float16 if DEVICE == "cuda" else torch.float32
+
+class FlorenceVisionEngine(VisionEngine):
+    def __init__(self):
+        self.model: Optional[AutoModelForCausalLM] = None
+        self.processor: Optional[AutoProcessor] = None
+        self.paddle_ocr = None
+
+    def load(self):
+        """Load the Florence-2 model."""
+        if self.model is not None:
+            return
+            
+        try:
+            print(f"Loading Florence-2 on {DEVICE.upper()}...")
+            self.model = AutoModelForCausalLM.from_pretrained(
+                CONFIG.MODEL_NAME,
+                trust_remote_code=True,
+                torch_dtype=DTYPE,
+            ).to(DEVICE).eval()
+
+            self.processor = AutoProcessor.from_pretrained(
+                CONFIG.MODEL_NAME,
+                trust_remote_code=True,
+            )
+            print("Florence-2 loaded successfully")
+            
+            try:
+                from paddleocr import PaddleOCR
+                print("Loading PaddleOCR...")
+                self.paddle_ocr = PaddleOCR(use_angle_cls=True, lang='en', show_log=False)
+                print("PaddleOCR loaded successfully")
+            except Exception as e:
+                print(f"PaddleOCR load failed: {e}")
+                
+            self._warmup()
+        except Exception as e:
+            print(f"Model loading failed: {e}")
+            raise
+
+    def _warmup(self):
+        """Run a dummy inference to warm up kernels."""
+        try:
+            dummy = Image.new("RGB", (224, 224), 128)
+            self._run_inference(dummy, "<CAPTION>")
+            print("Model warmed up")
+        except Exception as e:
+            print(f"Warmup warning: {e}")
+
+    def _run_inference(self, image: Image.Image, task_token: str) -> Dict[str, Any]:
+        """Core inference logic."""
+        if self.model is None or self.processor is None:
+            raise RuntimeError("Model not loaded. Call load() first.")
+
+        image = preprocess_image(image)
+        image = auto_enhance(image)
+        max_tokens = CONFIG.MAX_NEW_TOKENS.get(task_token, 64)
+
+        inputs = self.processor(
+            text=task_token,
+            images=image,
+            return_tensors="pt",
+        ).to(DEVICE)
+        
+        if "pixel_values" in inputs:
+            inputs["pixel_values"] = inputs["pixel_values"].to(DTYPE)
+
+        with torch.inference_mode():
+            output_ids = self.model.generate(
+                input_ids=inputs["input_ids"],
+                pixel_values=inputs["pixel_values"],
+                max_new_tokens=max_tokens,
+                do_sample=False,
+                num_beams=1,
+                use_cache=True,
+            )
+
+        raw_text = self.processor.batch_decode(output_ids, skip_special_tokens=False)[0]
+        result = self.processor.post_process_generation(
+            raw_text,
+            task=task_token,
+            image_size=(image.width, image.height),
+        )
+        return result
+
+    def describe_scene(self, image: Image.Image, detailed: bool = False) -> str:
+        task = "<MORE_DETAILED_CAPTION>" if detailed else "<DETAILED_CAPTION>"
+        try:
+            result = self._run_inference(image, task)
+            return format_caption(result.get(task, ""))
+        except Exception as e:
+            print(f"describe_scene error: {e}")
+            return "I couldn't analyze the scene right now."
+
+    def read_text(self, image: Image.Image) -> str:
+        try:
+            if hasattr(self, 'paddle_ocr') and self.paddle_ocr:
+                import numpy as np
+                # Convert PIL Image to RGB Numpy array for PaddleOCR
+                img_array = np.array(image.convert("RGB"))
+                result = self.paddle_ocr.ocr(img_array, cls=True)
+                
+                if not result or result[0] is None:
+                    return "I couldn't find any clear text in the image."
+                
+                lines = []
+                for line in result[0]:
+                    text = line[1][0]
+                    lines.append(text)
+                
+                final_text = " ".join(lines).strip()
+                if not final_text:
+                    return "I couldn't find any clear text."
+                return f"The text says: {final_text}"
+            else:
+                # Fallback to Florence-2 OCR
+                result = self._run_inference(image, "<OCR>")
+                return format_ocr(result.get("<OCR>", ""))
+        except Exception as e:
+            print(f"read_text error: {e}")
+            return "I couldn't read the text right now."
+
+    def analyze(self, image: Image.Image, task: str) -> str:
+        # Generic handler
+        try:
+            result = self._run_inference(image, task)
+            return str(result)
+        except Exception as e:
+            return f"Error: {e}"

app/vision/ocr.py

@@ -0,0 +1,8 @@
+def format_ocr(raw_text: str) -> str:
+    """Format OCR text for speech clarity."""
+    text_found = raw_text.strip()
+    if not text_found:
+        return "No text detected."
+    
+    # Add conversational framing
+    return f"The text says: {text_found}"

app/vision/scene_change.py

@@ -0,0 +1,20 @@
+from PIL import Image
+from typing import Optional
+
+def compute_hash(image: Image.Image, size: int = 16) -> bytes:
+    """Compute difference hash (dHash) for scene change detection."""
+    gray = image.resize((size + 1, size), Image.LANCZOS).convert("L")
+    pixels = list(gray.getdata())
+    return bytes(
+        1 if pixels[y * (size + 1) + x] > pixels[y * (size + 1) + x + 1] else 0
+        for y in range(size)
+        for x in range(size)
+    )
+
+def hash_distance(a: Optional[bytes], b: Optional[bytes]) -> float:
+    """Compute normalized Hamming distance between two hashes."""
+    if a is None or b is None:
+        return 1.0
+    if len(a) != len(b):
+        return 1.0
+    return sum(x != y for x, y in zip(a, b)) / len(a)

app/vision/utils.py

@@ -0,0 +1,16 @@
+from PIL import Image, ImageEnhance
+from app.config import CONFIG
+
+def preprocess_image(image: Image.Image) -> Image.Image:
+    """Resize image for inference while preserving aspect ratio."""
+    w, h = image.size
+    if max(w, h) <= CONFIG.MAX_DIM:
+        return image
+    scale = CONFIG.MAX_DIM / max(w, h)
+    new_size = (int(w * scale), int(h * scale))
+    return image.resize(new_size, Image.LANCZOS)
+
+def auto_enhance(image: Image.Image) -> Image.Image:
+    """Auto-enhance image for better vision model performance."""
+    enhancer = ImageEnhance.Contrast(image)
+    return enhancer.enhance(1.1)

app/vision/vision_engine.py

@@ -0,0 +1,26 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any
+from PIL import Image
+
+class VisionEngine(ABC):
+    """Abstract interface for vision models."""
+
+    @abstractmethod
+    def load(self):
+        """Load the vision model into memory."""
+        pass
+
+    @abstractmethod
+    def describe_scene(self, image: Image.Image, detailed: bool = False) -> str:
+        """Generate a description of the scene."""
+        pass
+
+    @abstractmethod
+    def read_text(self, image: Image.Image) -> str:
+        """Perform OCR and return formatted text."""
+        pass
+
+    @abstractmethod
+    def analyze(self, image: Image.Image, task: str) -> str:
+        """Run a raw generic analysis task on the image."""
+        pass

requirements.txt

@@ -0,0 +1,10 @@
+transformers==4.48.0
+timm
+torch>=2.1.0
+torchvision
+Pillow>=10.0.0
+einops
+edge-tts
+gradio>=5.0.0
+numpy
+accelerate