wsy182
4485cbf702
Split monolithic files into focused modules: - app/core: settings, logging, lifecycle - app/signaling: websocket server, ICE parser, message models - app/webrtc: peer session, video receiver, frame source - app/vision: pose landmarker wrapper, model config, pose types - app/exercises/dead_bug: detector, metrics, rules, state machine, types - app/rendering: skeleton renderer, status overlay, window display - app/audio: rep announcer - app/diagnostics: perf timer, crash handler - configs: environment-based settings - tests: unit tests for rules, state machine, ICE parser - run.py: entry point
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import queue
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from typing import Any
|
|
|
|
from loguru import logger
|
|
|
|
|
|
class RepAnnouncer:
|
|
def __init__(self, *, enabled: bool = True, rate: int = 185, volume: float = 1.0) -> None:
|
|
self.enabled = enabled
|
|
self.rate = rate
|
|
self.volume = volume
|
|
self._queue: queue.Queue[str | None] = queue.Queue()
|
|
self._thread: threading.Thread | None = None
|
|
self._engine: Any | None = None
|
|
self._use_macos_say = False
|
|
self._current_process: subprocess.Popen | None = None
|
|
|
|
if self.enabled:
|
|
self._start()
|
|
|
|
def announce_count(self, count: int) -> None:
|
|
if not self.enabled or count <= 0:
|
|
return
|
|
while True:
|
|
try:
|
|
self._queue.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
self._queue.put(str(count))
|
|
|
|
def close(self) -> None:
|
|
if not self.enabled:
|
|
return
|
|
self._queue.put(None)
|
|
if self._thread is not None:
|
|
self._thread.join(timeout=1.0)
|
|
if self._current_process is not None and self._current_process.poll() is None:
|
|
self._current_process.terminate()
|
|
|
|
def _start(self) -> None:
|
|
if sys.platform == "darwin":
|
|
self._use_macos_say = True
|
|
logger.info("Rep announcer initialized with macOS say")
|
|
else:
|
|
try:
|
|
import pyttsx3
|
|
|
|
self._engine = pyttsx3.init()
|
|
self._engine.setProperty("rate", self.rate)
|
|
self._engine.setProperty("volume", self.volume)
|
|
logger.info("Rep announcer initialized with pyttsx3")
|
|
except Exception as exc:
|
|
self.enabled = False
|
|
logger.warning("Rep announcer disabled, pyttsx3 unavailable: {}", exc)
|
|
return
|
|
|
|
self._thread = threading.Thread(target=self._run, name="RepAnnouncer", daemon=True)
|
|
self._thread.start()
|
|
|
|
def _run(self) -> None:
|
|
while True:
|
|
text = self._queue.get()
|
|
if text is None:
|
|
return
|
|
|
|
try:
|
|
if self._use_macos_say:
|
|
if self._current_process is not None and self._current_process.poll() is None:
|
|
self._current_process.terminate()
|
|
self._current_process = subprocess.Popen(
|
|
["say", "-r", str(self.rate), text],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
elif self._engine is not None:
|
|
self._engine.say(text)
|
|
self._engine.runAndWait()
|
|
except Exception as exc:
|
|
logger.warning("Failed to announce rep count {}: {}", text, exc)
|