Spaces:
Running on Zero
Running on Zero
File size: 1,902 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 | 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()
|