64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
"""Speech history.
|
|
|
|
Deliberately in-memory and bounded: this is an operator convenience for a live
|
|
demo ("say that again"), not an audit log. It resets when the server restarts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
from typing import Any, Deque, Dict, List, Optional
|
|
|
|
from ..robot.base import SpeechStage
|
|
|
|
|
|
class SpeechHistory:
|
|
def __init__(self, limit: int = 100) -> None:
|
|
self._items: Deque[Dict[str, Any]] = deque(maxlen=max(1, limit))
|
|
self._index: Dict[str, Dict[str, Any]] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def add(self, request_id: str, text: str) -> Dict[str, Any]:
|
|
entry: Dict[str, Any] = {
|
|
"id": request_id,
|
|
"text": text,
|
|
"at": time.time(),
|
|
"stage": SpeechStage.SENDING.value,
|
|
"success": None,
|
|
"ackLatencyMs": None,
|
|
"totalMs": None,
|
|
"error": None,
|
|
}
|
|
with self._lock:
|
|
if len(self._items) == self._items.maxlen:
|
|
# Newest is at the left, so the deque drops the rightmost entry.
|
|
self._index.pop(self._items[-1]["id"], None)
|
|
self._items.appendleft(entry)
|
|
self._index[request_id] = entry
|
|
return entry
|
|
|
|
def update(self, request_id: str, **fields: Any) -> Optional[Dict[str, Any]]:
|
|
with self._lock:
|
|
entry = self._index.get(request_id)
|
|
if entry is None:
|
|
return None
|
|
entry.update(fields)
|
|
return dict(entry)
|
|
|
|
def list(self) -> List[Dict[str, Any]]:
|
|
with self._lock:
|
|
return [dict(item) for item in self._items]
|
|
|
|
def clear(self) -> int:
|
|
with self._lock:
|
|
count = len(self._items)
|
|
self._items.clear()
|
|
self._index.clear()
|
|
return count
|
|
|
|
def __len__(self) -> int:
|
|
with self._lock:
|
|
return len(self._items)
|