320 lines
13 KiB
Python
320 lines
13 KiB
Python
"""Speech orchestration.
|
|
|
|
The one deliberate design decision worth stating: POST /api/robot/speak returns as
|
|
soon as the *robot has acknowledged* the utterance - not when speaking finishes.
|
|
The remaining lifecycle (speaking -> completed) streams over the WebSocket.
|
|
|
|
That matters for a live demo. Blocking the HTTP response until the robot stops
|
|
talking would make a 12-second sentence look like a 12-second-slow button, and it
|
|
would tie a browser request to the speaking duration for no benefit.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
from ..config.settings import Settings
|
|
from ..core.events import EventBus, EventType
|
|
from ..core.text import normalise
|
|
from ..robot.base import (
|
|
RobotBusy,
|
|
RobotError,
|
|
RobotUnreachable,
|
|
SpeechProgress,
|
|
SpeechRequest,
|
|
SpeechResult,
|
|
SpeechStage,
|
|
)
|
|
from ..robot.manager import RobotManager
|
|
from .history import SpeechHistory
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ValidationError(RobotError):
|
|
code = "invalid_text"
|
|
user_message = "Please enter some text for the robot to say."
|
|
|
|
|
|
def _swallow_exception(task: "asyncio.Task") -> None:
|
|
"""Retrieve a finished task's exception so asyncio stops complaining."""
|
|
if not task.cancelled():
|
|
task.exception()
|
|
|
|
|
|
class SpeechService:
|
|
def __init__(self, settings: Settings, manager: RobotManager, bus: EventBus) -> None:
|
|
self._settings = settings
|
|
self._manager = manager
|
|
self._bus = bus
|
|
self.history = SpeechHistory(settings.speech.history_limit)
|
|
self._lock = asyncio.Lock()
|
|
self._active_task: Optional[asyncio.Task] = None
|
|
self._active_id: Optional[str] = None
|
|
|
|
def update_settings(self, settings: Settings) -> None:
|
|
"""Adopt reloaded configuration (limits, interrupt policy) in place."""
|
|
self._settings = settings
|
|
|
|
# -- state --------------------------------------------------------------- #
|
|
@property
|
|
def is_busy(self) -> bool:
|
|
return self._active_task is not None and not self._active_task.done()
|
|
|
|
@property
|
|
def active_request_id(self) -> Optional[str]:
|
|
return self._active_id if self.is_busy else None
|
|
|
|
# -- validation ---------------------------------------------------------- #
|
|
def validate(self, text: Optional[str]) -> str:
|
|
cleaned = normalise(text or "")
|
|
limits = self._settings.speech
|
|
if len(cleaned) < max(1, limits.min_length):
|
|
raise ValidationError("empty text")
|
|
if len(cleaned) > limits.max_length:
|
|
raise ValidationError(
|
|
"text too long: {0} chars".format(len(cleaned)),
|
|
user_message=(
|
|
"Text is too long ({0} characters). The limit is {1}. "
|
|
"Split it into shorter sentences.".format(len(cleaned), limits.max_length)
|
|
),
|
|
)
|
|
return cleaned
|
|
|
|
# -- speak --------------------------------------------------------------- #
|
|
async def speak(self, text: Optional[str], voice: Optional[str] = None,
|
|
language: Optional[str] = None) -> Dict[str, Any]:
|
|
cleaned = self.validate(text)
|
|
|
|
if not self._manager.is_connected:
|
|
status = self._manager.status_dict()
|
|
raise RobotUnreachable(
|
|
"robot not connected (state={0})".format(status["state"]),
|
|
user_message=(
|
|
status.get("error")
|
|
or "Robot is offline. Check the robot IP address and network connection."
|
|
),
|
|
)
|
|
|
|
async with self._lock:
|
|
if self.is_busy:
|
|
if not self._settings.speech.allow_interrupt:
|
|
raise RobotBusy()
|
|
logger.info("interrupting utterance %s", self._active_id)
|
|
await self._cancel_active()
|
|
|
|
request = SpeechRequest(text=cleaned, voice=voice, language=language)
|
|
entry = self.history.add(request.id, cleaned)
|
|
self._bus.publish(EventType.HISTORY_UPDATED, {"entry": entry, "action": "add"})
|
|
|
|
ack = asyncio.Event()
|
|
state: Dict[str, Any] = {"ackLatencyMs": None, "stage": SpeechStage.SENDING}
|
|
|
|
async def on_progress(progress: SpeechProgress) -> None:
|
|
state["stage"] = progress.stage
|
|
if progress.stage is not SpeechStage.SENDING:
|
|
if state["ackLatencyMs"] is None:
|
|
state["ackLatencyMs"] = progress.elapsed_ms
|
|
ack.set()
|
|
self._bus.publish(EventType.SPEECH_PROGRESS, progress.to_dict())
|
|
self.history.update(request.id, stage=progress.stage.value)
|
|
|
|
started = time.perf_counter()
|
|
self._active_id = request.id
|
|
self._active_task = asyncio.create_task(
|
|
self._run(request, on_progress), name="speak-" + request.id
|
|
)
|
|
# The HTTP call usually returns before the task finishes, so nobody is
|
|
# left to await it. Consume any exception here to keep asyncio from
|
|
# logging "Task exception was never retrieved" - it is already
|
|
# reported to the browser as a speech.progress / speech.result event.
|
|
self._active_task.add_done_callback(_swallow_exception)
|
|
|
|
# Wait only for the acknowledgement, then hand control back to the UI.
|
|
ack_wait = asyncio.ensure_future(ack.wait())
|
|
done, _ = await asyncio.wait(
|
|
{ack_wait, self._active_task},
|
|
timeout=self._settings.robot.request_timeout + 2,
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
if not ack_wait.done():
|
|
ack_wait.cancel()
|
|
|
|
# The utterance failed before it was ever acknowledged - surface it now
|
|
# as an HTTP error rather than only as a WebSocket event.
|
|
if self._active_task in done:
|
|
exc = self._active_task.exception()
|
|
if exc is not None:
|
|
raise exc
|
|
result = self._active_task.result()
|
|
return self._response(request, result.stage, result.ack_latency_ms, started)
|
|
|
|
if not ack.is_set():
|
|
await self._cancel_active()
|
|
raise RobotError(
|
|
"no acknowledgement within timeout",
|
|
user_message="The robot did not respond in time. The request was cancelled.",
|
|
)
|
|
|
|
return self._response(request, state["stage"], state["ackLatencyMs"], started)
|
|
|
|
def _response(self, request: SpeechRequest, stage: Any, ack_ms: Optional[int],
|
|
started: float) -> Dict[str, Any]:
|
|
stage_value = stage.value if hasattr(stage, "value") else str(stage)
|
|
return {
|
|
"success": stage_value not in (SpeechStage.FAILED.value, SpeechStage.CANCELLED.value),
|
|
"status": stage_value,
|
|
"requestId": request.id,
|
|
"text": request.text,
|
|
"ackLatencyMs": ack_ms if ack_ms is not None else int((time.perf_counter() - started) * 1000),
|
|
}
|
|
|
|
async def _run(self, request: SpeechRequest, on_progress) -> SpeechResult:
|
|
"""Drive the utterance to completion and record the outcome."""
|
|
try:
|
|
result = await self._manager.adapter.speak(request, on_progress)
|
|
except asyncio.CancelledError:
|
|
entry = self.history.update(
|
|
request.id, stage=SpeechStage.CANCELLED.value, success=False, error="Stopped."
|
|
)
|
|
self._publish_history(entry)
|
|
raise
|
|
except RobotError as exc:
|
|
logger.warning("speech %s failed: %s", request.id, exc)
|
|
await on_progress(
|
|
SpeechProgress(
|
|
request.id,
|
|
SpeechStage.FAILED,
|
|
exc.user_message,
|
|
error_code=exc.code,
|
|
)
|
|
)
|
|
entry = self.history.update(
|
|
request.id,
|
|
stage=SpeechStage.FAILED.value,
|
|
success=False,
|
|
error=exc.user_message,
|
|
)
|
|
self._publish_history(entry)
|
|
self._bus.publish(
|
|
EventType.SPEECH_RESULT,
|
|
{
|
|
"requestId": request.id,
|
|
"success": False,
|
|
"stage": SpeechStage.FAILED.value,
|
|
"errorCode": exc.code,
|
|
"error": exc.user_message,
|
|
},
|
|
)
|
|
raise
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
logger.exception("unexpected speech failure")
|
|
await on_progress(
|
|
SpeechProgress(request.id, SpeechStage.FAILED, "Speech request failed.",
|
|
error_code="internal", detail=str(exc))
|
|
)
|
|
entry = self.history.update(
|
|
request.id, stage=SpeechStage.FAILED.value, success=False,
|
|
error="Speech request failed.",
|
|
)
|
|
self._publish_history(entry)
|
|
raise
|
|
|
|
entry = self.history.update(
|
|
request.id,
|
|
stage=result.stage.value,
|
|
success=result.success,
|
|
ackLatencyMs=result.ack_latency_ms,
|
|
totalMs=result.total_ms,
|
|
error=result.error,
|
|
)
|
|
self._publish_history(entry)
|
|
self._bus.publish(EventType.SPEECH_RESULT, result.to_dict())
|
|
return result
|
|
|
|
def _publish_history(self, entry: Optional[Dict[str, Any]]) -> None:
|
|
if entry is not None:
|
|
self._bus.publish(EventType.HISTORY_UPDATED, {"entry": entry, "action": "update"})
|
|
|
|
# -- stop ---------------------------------------------------------------- #
|
|
async def stop(self) -> Dict[str, Any]:
|
|
stopped_robot = False
|
|
try:
|
|
stopped_robot = await self._manager.adapter.stop_speaking()
|
|
except Exception as exc:
|
|
logger.warning("robot stop failed: %s", exc)
|
|
|
|
had_active = self.is_busy
|
|
if had_active:
|
|
# Give the adapter a moment to unwind cleanly before force-cancelling.
|
|
try:
|
|
await asyncio.wait_for(asyncio.shield(self._active_task), timeout=1.5)
|
|
except (asyncio.TimeoutError, Exception):
|
|
await self._cancel_active()
|
|
|
|
return {
|
|
"success": True,
|
|
"status": "stopped",
|
|
"hadActiveSpeech": had_active,
|
|
"robotAcknowledged": stopped_robot,
|
|
}
|
|
|
|
async def _cancel_active(self) -> None:
|
|
task = self._active_task
|
|
if task is None or task.done():
|
|
self._active_task = None
|
|
return
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
finally:
|
|
self._active_task = None
|
|
|
|
# -- history ------------------------------------------------------------- #
|
|
def annotate_saved(self, items):
|
|
"""Mark history entries whose audio is already on disk.
|
|
|
|
Lets the dashboard show a replay button only where playback really is
|
|
instant, instead of promising something it cannot deliver.
|
|
"""
|
|
try:
|
|
from .audio_library import get_audio_library
|
|
|
|
library = get_audio_library()
|
|
mock = self._settings.mock
|
|
if mock.voice_engine != "gemini":
|
|
return items
|
|
from ..core.pronunciation import build as build_pronouncer
|
|
from ..config.settings import PROJECT_ROOT
|
|
from ..robot.gemini_voice import split_sentences
|
|
|
|
pronouncer = build_pronouncer(PROJECT_ROOT, mock.pronunciation)
|
|
for item in items:
|
|
text = item.get("text") or ""
|
|
if pronouncer is not None:
|
|
text = pronouncer.apply(text)
|
|
chunks = split_sentences(text)
|
|
entries = [
|
|
library.find(c, mock.gemini_voice, mock.gemini_model, mock.gemini_style)
|
|
for c in chunks
|
|
]
|
|
found = [e for e in entries if e]
|
|
item["audioSaved"] = bool(chunks) and len(found) == len(chunks)
|
|
item["audioIds"] = [e["id"] for e in found]
|
|
except Exception: # pragma: no cover - never break history over this
|
|
logger.debug("could not annotate saved audio", exc_info=True)
|
|
return items
|
|
|
|
def clear_history(self) -> Dict[str, Any]:
|
|
removed = self.history.clear()
|
|
self._bus.publish(EventType.HISTORY_UPDATED, {"action": "clear"})
|
|
return {"success": True, "removed": removed}
|
|
|
|
async def shutdown(self) -> None:
|
|
await self._cancel_active()
|