Spaces:
Running on Zero
Running on Zero
File size: 3,011 Bytes
88fe4e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | from src.vision.florence import FlorenceVisionEngine
from src.conversation.context import CONTEXT
from src.speech.tts import text_to_speech
from src.speech.audio_manager import AudioQueue
from src.vision.scene_change import compute_hash, hash_distance
from src.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}"
|