voice_feedback.py

import time
import threading
from dataclasses import dataclass, field
from typing import List
from collections import deque

import config


@dataclass
class VoiceMessage:
    text: str
    priority: int
    timestamp: float = field(default_factory=time.time)


class VoiceFeedbackManager:
    def __init__(
        self,
        cooldown_seconds: float = config.VOICE_COOLDOWN_SECONDS,
        max_queue_size: int = config.VOICE_MAX_QUEUE_SIZE,
    ):
        self.cooldown_seconds = cooldown_seconds
        self.max_queue_size = max_queue_size
        self.last_spoken: dict = {}
        self.message_queue: deque = deque(maxlen=max_queue_size)
        self.lock = threading.Lock()

    def enqueue(self, text: str, severity: str = "warning"):
        """Add a voice message to the queue (called from callback thread).
        Applies throttling and priority filtering."""
        if self._is_throttled(text):
            return

        priority = config.VOICE_PRIORITY.get(severity, 1)
        msg = VoiceMessage(text=text, priority=priority)

        with self.lock:
            if len(self.message_queue) >= self.max_queue_size:
                # Find lowest priority (highest number) in queue
                worst = max(self.message_queue, key=lambda m: m.priority)
                if priority < worst.priority:
                    self.message_queue.remove(worst)
                    self.message_queue.append(msg)
            else:
                self.message_queue.append(msg)

    def get_pending_messages(self) -> List[VoiceMessage]:
        """Get and clear all pending messages (called from main thread).
        Returns messages sorted by priority (highest first)."""
        with self.lock:
            messages = sorted(self.message_queue, key=lambda m: m.priority)
            self.message_queue.clear()
        return messages

    def generate_tts_html(self, messages: List[VoiceMessage]) -> str:
        """Generate HTML/JS snippet using Web Speech API to speak messages."""
        if not messages:
            return ""

        # Build JS array of message texts
        js_messages = ", ".join(f'"{m.text}"' for m in messages)

        return f"""
        <script>
        (function() {{
            if (!('speechSynthesis' in window)) {{
                console.warn('Text-to-speech not supported in this browser');
                return;
            }}
            const synth = window.speechSynthesis;
            synth.cancel();
            const messages = [{js_messages}];
            messages.forEach((msg, i) => {{
                setTimeout(() => {{
                    const utterance = new SpeechSynthesisUtterance(msg);
                    utterance.rate = 1.0;
                    utterance.pitch = 1.0;
                    utterance.volume = 1.0;
                    synth.speak(utterance);
                }}, i * 2500);
            }});
        }})();
        </script>
        """

    def _is_throttled(self, text: str) -> bool:
        """Check if this message was spoken recently."""
        now = time.time()
        with self.lock:
            if text in self.last_spoken:
                if now - self.last_spoken[text] < self.cooldown_seconds:
                    return True
            self.last_spoken[text] = now
        return False