From aa2d2e4c8fd8a4cf5f0a25b2cb37625fbfa0dea7 Mon Sep 17 00:00:00 2001 From: Nir Krakowski Date: Wed, 20 May 2026 17:32:56 +0300 Subject: [PATCH] feat: add deepdub tts extension --- .../extension/deepdub_tts_python/README.md | 28 ++ .../extension/deepdub_tts_python/__init__.py | 1 + .../extension/deepdub_tts_python/addon.py | 10 + .../extension/deepdub_tts_python/config.py | 72 ++++ .../deepdub_tts_python/deepdub_tts.py | 221 ++++++++++++ .../extension/deepdub_tts_python/extension.py | 326 ++++++++++++++++++ .../deepdub_tts_python/manifest.json | 72 ++++ .../deepdub_tts_python/property.json | 12 + .../deepdub_tts_python/requirements.txt | 2 + .../deepdub_tts_python/tests/__init__.py | 0 .../deepdub_tts_python/tests/bin/start | 9 + .../tests/configs/property_basic.json | 12 + .../tests/configs/property_dump.json | 14 + .../tests/configs/property_invalid.json | 9 + .../tests/configs/property_miss_required.json | 7 + .../deepdub_tts_python/tests/conftest.py | 75 ++++ .../deepdub_tts_python/tests/test_basic.py | 247 +++++++++++++ .../tests/test_error_msg.py | 142 ++++++++ .../deepdub_tts_python/tests/test_metrics.py | 103 ++++++ .../deepdub_tts_python/tests/test_params.py | 126 +++++++ .../tests/test_robustness.py | 123 +++++++ .../tests/test_state_machine.py | 126 +++++++ 22 files changed, 1737 insertions(+) create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/README.md create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/__init__.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/addon.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/config.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/deepdub_tts.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/extension.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/manifest.json create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/property.json create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/requirements.txt create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/__init__.py create mode 100755 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/bin/start create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_basic.json create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_dump.json create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_invalid.json create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_miss_required.json create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/conftest.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_basic.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_error_msg.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_metrics.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_params.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_robustness.py create mode 100644 ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_state_machine.py diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/README.md b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/README.md new file mode 100644 index 0000000000..0adb5f536d --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/README.md @@ -0,0 +1,28 @@ +# Deepdub TTS (Streaming) — Python + +TEN extension wrapping Deepdub's text-streaming WebSocket API. + +- One persistent WebSocket per extension instance, opened and configured during `on_init` (pre-warmed before the first sentence arrives). +- Each `tts_text_input` fragment is forwarded as a `stream-text` frame, so audio synthesis starts before the LLM has finished producing the full reply. +- The vendor's `isFinished` boundary is used to emit `tts_audio_end` and release the request lock, naturally serialising back-to-back TEN requests on a single connection. + +## Required configuration + +Provide via `property.json` `params` or environment variables: + +| Property | Env var | Notes | +| ----------------- | ------------------------------ | ------------------------------------------- | +| `api_key` | `DEEPDUB_API_KEY` | Account API key. | +| `url` | `DEEPDUB_WS_STREAMING_URL` | Text-streaming WS endpoint URL. | +| `voice_prompt_id` | `DEEPDUB_VOICE_PROMPT_ID` | Voice prompt to use for the session. | +| `model` | `DEEPDUB_MODEL` | Defaults to `dd-etts-3.2`. | +| `locale` | `DEEPDUB_LOCALE` | Defaults to `en-US`. | + +## Audio output + +Defaults to raw PCM (`s16le`) at 48 kHz, mono — fed directly to the framework's audio pipeline without header parsing. + +## Constraints + +- The streaming WS binds a single voice/locale/model to a connection (set via `stream-config`). Changing voice mid-session would require a reconnect. v1 assumes a single voice per agent session. +- `cancel_tts` (TEN flush) drops the in-flight stream and reconnects. diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/__init__.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/__init__.py new file mode 100644 index 0000000000..8e8b68bd8f --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/__init__.py @@ -0,0 +1 @@ +from . import addon # noqa: F401 diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/addon.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/addon.py new file mode 100644 index 0000000000..0eadf534ac --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/addon.py @@ -0,0 +1,10 @@ +from ten_runtime import Addon, register_addon_as_extension, TenEnv + + +@register_addon_as_extension("deepdub_tts_python") +class DeepdubTTSExtensionAddon(Addon): + def on_create_instance(self, ten_env: TenEnv, name: str, context) -> None: + from .extension import DeepdubTTSExtension + + ten_env.log_info("DeepdubTTSExtensionAddon on_create_instance") + ten_env.on_create_instance_done(DeepdubTTSExtension(name), context) diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/config.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/config.py new file mode 100644 index 0000000000..cfb58370ec --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/config.py @@ -0,0 +1,72 @@ +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + +from ten_ai_base import utils + + +class DeepdubTTSConfig(BaseModel): + api_key: str = "" + url: str = "" + model: str = "dd-etts-3.2" + locale: str = "en-US" + voice_prompt_id: str = "" + sample_rate: int = 48000 + channels: int = 1 + # PCM in network byte order, signed 16-bit LE — matches TEN's audio_frame expectations. + format: str = "s16le" + accept_emojis: bool = False + temperature: Optional[float] = None + variance: Optional[float] = None + tempo: Optional[float] = None + prompt_boost: Optional[bool] = None + keepalive_interval_seconds: float = 20.0 + dump: bool = False + dump_path: str = "" + params: Dict[str, Any] = Field(default_factory=dict) + black_list_params: List[str] = Field(default_factory=list) + + def update_params(self) -> None: + p = self.params + for k in ( + "api_key", + "url", + "model", + "locale", + "voice_prompt_id", + "sample_rate", + "channels", + "format", + "accept_emojis", + "temperature", + "variance", + "tempo", + "prompt_boost", + "keepalive_interval_seconds", + ): + if k in p: + setattr(self, k, p[k]) + del p[k] + + def validate_params(self) -> None: + missing = [ + n + for n in ("api_key", "url", "voice_prompt_id") + if not getattr(self, n) + ] + if missing: + raise ValueError( + f"required fields are missing or empty: params.{', params.'.join(missing)}" + ) + if self.sample_rate not in (8000, 16000, 22050, 24000, 44100, 48000): + raise ValueError(f"unsupported sample_rate: {self.sample_rate}") + if self.format not in ("s16le", "wav", "mp3", "opus", "mulaw"): + raise ValueError(f"unsupported format: {self.format}") + + def to_str(self, sensitive_handling: bool = False) -> str: + if not sensitive_handling: + return f"{self}" + c = self.copy(deep=True) + if c.api_key: + c.api_key = utils.encrypt(c.api_key) + return f"{c}" diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/deepdub_tts.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/deepdub_tts.py new file mode 100644 index 0000000000..8e571991e7 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/deepdub_tts.py @@ -0,0 +1,221 @@ +import asyncio +import base64 +import json +import time +from typing import Awaitable, Callable, Optional + +import websockets + +from ten_runtime import AsyncTenEnv + +from .config import DeepdubTTSConfig + + +class DeepdubTTSException(Exception): + def __init__(self, message: str, code: int = -1): + super().__init__(message) + self.message = message + self.code = code + + +AudioCb = Callable[[bytes], Awaitable[None]] +FinishCb = Callable[[], Awaitable[None]] +ErrorCb = Callable[[DeepdubTTSException], Awaitable[None]] + + +class DeepdubStreamingClient: + """One persistent text-streaming WebSocket per instance. + + Connect → recv status → send `stream-config` → push `stream-text` frames as + they arrive → receive interleaved audio and `isFinished` boundaries. + """ + + def __init__( + self, + config: DeepdubTTSConfig, + ten_env: Optional[AsyncTenEnv], + on_audio: AudioCb, + on_finish: FinishCb, + on_error: ErrorCb, + ): + self.config = config + self.ten_env = ten_env + self.on_audio = on_audio + self.on_finish = on_finish + self.on_error = on_error + + self._ws: Optional[websockets.ClientConnection] = None + self._ready = asyncio.Event() + self._stopping = False + self._discarding = False + self._reader_task: Optional[asyncio.Task] = None + self._supervisor_task: Optional[asyncio.Task] = None + self._keepalive_task: Optional[asyncio.Task] = None + self._stopped = asyncio.Event() + self._send_lock = asyncio.Lock() + + # ---------- lifecycle ---------- + + async def start(self) -> None: + self._supervisor_task = asyncio.create_task(self._supervise()) + + async def stop(self) -> None: + self._stopping = True + await self._close_ws() + if self._supervisor_task: + await self._stopped.wait() + + async def cancel(self) -> None: + """Drop the in-flight stream; reconnect from scratch.""" + self._discarding = True + await self._close_ws() + # supervisor will reconnect + + # ---------- public send API ---------- + + async def wait_ready(self, timeout: float = 10.0) -> None: + await asyncio.wait_for(self._ready.wait(), timeout=timeout) + + async def send_text(self, text: str) -> None: + if not text: + return + await self.wait_ready() + async with self._send_lock: + assert self._ws is not None + await self._ws.send( + json.dumps({"action": "stream-text", "data": {"text": text}}) + ) + + async def send_cancel(self) -> None: + async with self._send_lock: + if self._ws is not None: + try: + await self._ws.send(json.dumps({"action": "cancel"})) + except Exception: + pass + + # ---------- internals ---------- + + async def _close_ws(self) -> None: + ws = self._ws + self._ws = None + self._ready.clear() + if ws is not None: + try: + await ws.close() + except Exception: + pass + if self._keepalive_task: + self._keepalive_task.cancel() + self._keepalive_task = None + + async def _supervise(self) -> None: + backoff = 0.5 + while not self._stopping: + try: + await self._connect_and_run() + backoff = 0.5 + except Exception as e: + if self.ten_env: + self.ten_env.log_warn(f"deepdub stream loop error: {e}") + try: + await self.on_error( + e + if isinstance(e, DeepdubTTSException) + else DeepdubTTSException(str(e)) + ) + except Exception: + pass + if self._stopping: + break + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 5.0) + finally: + self._discarding = False + await self._close_ws() + self._stopped.set() + + async def _connect_and_run(self) -> None: + headers = {"x-api-key": self.config.api_key} + if self.ten_env: + self.ten_env.log_info("deepdub stream WS connecting") + t0 = time.time() + async with websockets.connect( + self.config.url, + additional_headers=headers, + max_size=1024 * 1024 * 16, + ) as ws: + self._ws = ws + # Initial status frame. + status_raw = await ws.recv() + status = json.loads(status_raw) + if status.get("action") == "error": + raise DeepdubTTSException( + status.get("message", "connection rejected") + ) + # Send stream-config (pre-warm). + cfg = { + "model": self.config.model, + "locale": self.config.locale, + "voicePromptId": self.config.voice_prompt_id, + "format": self.config.format, + "sampleRate": self.config.sample_rate, + "acceptEmojis": self.config.accept_emojis, + "temperature": self.config.temperature, + "variance": self.config.variance, + "tempo": self.config.tempo, + "promptBoost": self.config.prompt_boost, + } + await ws.send( + json.dumps({"action": "stream-config", "config": cfg}) + ) + self._ready.set() + if self.ten_env: + self.ten_env.log_info( + f"deepdub stream WS ready ({int((time.time()-t0)*1000)}ms)" + ) + self._keepalive_task = asyncio.create_task(self._keepalive()) + await self._read_loop(ws) + + async def _keepalive(self) -> None: + try: + while not self._stopping and self._ws is not None: + await asyncio.sleep(self.config.keepalive_interval_seconds) + if self._ws is None: + return + async with self._send_lock: + if self._ws is not None: + try: + await self._ws.send(json.dumps({"action": "ping"})) + except Exception: + return + except asyncio.CancelledError: + return + + async def _read_loop(self, ws: websockets.ClientConnection) -> None: + while not self._stopping: + raw = await ws.recv() + try: + msg = json.loads(raw) + except Exception: + continue + action = msg.get("action") + if action in ("pong", "status"): + continue + err = msg.get("error") + if err and not msg.get("generationId"): + await self.on_error(DeepdubTTSException(str(err))) + continue + if self._discarding: + # Drop frames belonging to the cancelled run. + continue + data = msg.get("data") + if data: + try: + pcm = base64.b64decode(data) + except Exception: + pcm = b"" + if pcm: + await self.on_audio(pcm) + if msg.get("isFinished"): + await self.on_finish() diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/extension.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/extension.py new file mode 100644 index 0000000000..d16ccb99cd --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/extension.py @@ -0,0 +1,326 @@ +import os +import traceback +from datetime import datetime +from typing import Optional + +from ten_ai_base.const import LOG_CATEGORY_KEY_POINT +from ten_ai_base.helper import PCMWriter +from ten_ai_base.message import ( + ModuleError, + ModuleErrorCode, + ModuleErrorVendorInfo, + ModuleType, + TTSAudioEndReason, +) +from ten_ai_base.struct import TTSTextInput +from ten_ai_base.tts2 import AsyncTTS2BaseExtension +from ten_runtime import AsyncTenEnv + +from .config import DeepdubTTSConfig +from .deepdub_tts import DeepdubStreamingClient, DeepdubTTSException + + +class DeepdubTTSExtension(AsyncTTS2BaseExtension): + def __init__(self, name: str) -> None: + super().__init__(name) + self.config: Optional[DeepdubTTSConfig] = None + self.client: Optional[DeepdubStreamingClient] = None + + self.current_request_id: Optional[str] = None + self.sent_ts: Optional[datetime] = None + self.first_chunk: bool = True + self.awaiting_finish: bool = False + self.current_request_finished: bool = False + self.total_audio_bytes: int = 0 + + self.recorder_map: dict[str, PCMWriter] = {} + + # ---------- lifecycle ---------- + + async def on_init(self, ten_env: AsyncTenEnv) -> None: + try: + await super().on_init(ten_env) + ten_env.log_debug("on_init") + + if self.config is None: + config_json, _ = await self.ten_env.get_property_to_json("") + if not config_json or config_json.strip() == "{}": + raise ValueError("Configuration is empty.") + self.config = DeepdubTTSConfig.model_validate_json(config_json) + self.config.update_params() + self.config.validate_params() + ten_env.log_info( + f"config: {self.config.to_str(sensitive_handling=True)}", + category=LOG_CATEGORY_KEY_POINT, + ) + + self.client = DeepdubStreamingClient( + config=self.config, + ten_env=ten_env, + on_audio=self._on_audio, + on_finish=self._on_finish, + on_error=self._on_error, + ) + await self.client.start() + ten_env.log_info("deepdub TTS client started (pre-warming)") + except Exception as e: + ten_env.log_error(f"on_init failed: {traceback.format_exc()}") + await self.send_tts_error( + self.current_request_id or "", + ModuleError( + message=f"init error: {e}", + module=ModuleType.TTS, + code=int(ModuleErrorCode.FATAL_ERROR), + vendor_info=None, + ), + ) + + async def on_stop(self, ten_env: AsyncTenEnv) -> None: + if self.client: + await self.client.stop() + self.client = None + for rid, rec in list(self.recorder_map.items()): + try: + await rec.flush() + except Exception as e: + ten_env.log_error(f"recorder flush error ({rid}): {e}") + self.recorder_map.clear() + await super().on_stop(ten_env) + + async def on_deinit(self, ten_env: AsyncTenEnv) -> None: + await super().on_deinit(ten_env) + + def vendor(self) -> str: + return "deepdub" + + def synthesize_audio_sample_rate(self) -> int: + return self.config.sample_rate if self.config else 48000 + + def synthesize_audio_channels(self) -> int: + return self.config.channels if self.config else 1 + + # ---------- request_tts ---------- + + async def request_tts(self, t: TTSTextInput) -> None: + try: + if self.client is None: + await self.send_tts_error( + t.request_id, + ModuleError( + message="client not initialized", + module=ModuleType.TTS, + code=int(ModuleErrorCode.FATAL_ERROR), + vendor_info=ModuleErrorVendorInfo(vendor=self.vendor()), + ), + ) + if t.text_input_end: + await self.finish_request( + t.request_id, reason=TTSAudioEndReason.ERROR + ) + return + + self.ten_env.log_info( + f"KEYPOINT request_tts: rid={t.request_id} end={t.text_input_end} text={t.text!r}" + ) + + if t.request_id != self.current_request_id: + self.current_request_id = t.request_id + self.sent_ts = datetime.now() + self.first_chunk = True + self.awaiting_finish = False + self.current_request_finished = False + self.total_audio_bytes = 0 + self._setup_recorder(t.request_id) + + if t.text: + self.metrics_add_output_characters(len(t.text)) + await self.client.send_text(t.text) + + if t.text_input_end: + self.current_request_finished = True + self.awaiting_finish = True + self.ten_env.log_info( + f"KEYPOINT awaiting isFinished for rid={t.request_id}" + ) + + except DeepdubTTSException as e: + await self._send_error(e, t) + except Exception as e: + self.ten_env.log_error( + f"request_tts error: {traceback.format_exc()}" + ) + await self.send_tts_error( + self.current_request_id or "", + ModuleError( + message=str(e), + module=ModuleType.TTS, + code=int(ModuleErrorCode.NON_FATAL_ERROR), + vendor_info=ModuleErrorVendorInfo(vendor=self.vendor()), + ), + ) + if t.text_input_end: + await self.finish_request( + self.current_request_id or "", + reason=TTSAudioEndReason.ERROR, + ) + + async def cancel_tts(self) -> None: + try: + if self.client: + await self.client.cancel() + if self.current_request_id and self.sent_ts: + interval = int( + (datetime.now() - self.sent_ts).total_seconds() * 1000 + ) + await self.send_tts_audio_end( + request_id=self.current_request_id, + request_event_interval_ms=interval, + request_total_audio_duration_ms=self._audio_ms(), + reason=TTSAudioEndReason.INTERRUPTED, + ) + await self.send_usage_metrics(self.current_request_id) + await self._reset_request_state() + except Exception as e: + self.ten_env.log_error(f"cancel_tts error: {e}") + + # ---------- callbacks from streaming client ---------- + + async def _on_audio(self, pcm: bytes) -> None: + if not pcm or not self.current_request_id: + return + self.total_audio_bytes += len(pcm) + self.metrics_add_recv_audio_chunks(pcm) + if self.first_chunk: + self.first_chunk = False + await self.send_tts_audio_start(request_id=self.current_request_id) + if self.sent_ts: + ttfb = int( + (datetime.now() - self.sent_ts).total_seconds() * 1000 + ) + await self.send_tts_ttfb_metrics( + request_id=self.current_request_id, + ttfb_ms=ttfb, + extra_metadata={ + "model": self.config.model if self.config else "", + "voice_prompt_id": ( + self.config.voice_prompt_id if self.config else "" + ), + }, + ) + rec = self.recorder_map.get(self.current_request_id) + if rec is not None: + try: + await rec.write(pcm) + except Exception as e: + self.ten_env.log_warn(f"recorder write error: {e}") + await self.send_tts_audio_data(pcm, 0) + + async def _on_finish(self) -> None: + # Vendor signalled end of audio for queued text. Only honour as a + # request boundary if TEN sent text_input_end and we're awaiting it. + if not self.awaiting_finish or not self.current_request_id: + return + rid = self.current_request_id + interval = ( + int((datetime.now() - self.sent_ts).total_seconds() * 1000) + if self.sent_ts + else 0 + ) + duration = self._audio_ms() + rec = self.recorder_map.get(rid) + if rec is not None: + try: + await rec.flush() + except Exception as e: + self.ten_env.log_warn(f"recorder flush error: {e}") + await self.send_tts_audio_end( + request_id=rid, + request_event_interval_ms=interval, + request_total_audio_duration_ms=duration, + ) + await self.send_usage_metrics(rid) + await self.finish_request(rid) + self.ten_env.log_info( + f"KEYPOINT tts_audio_end rid={rid} interval={interval}ms duration={duration}ms" + ) + await self._reset_request_state() + + async def _on_error(self, exc: DeepdubTTSException) -> None: + self.ten_env.log_error(f"deepdub stream error: {exc}") + if not self.current_request_id: + return + err = ModuleError( + message=str(exc), + module=ModuleType.TTS, + code=int(ModuleErrorCode.NON_FATAL_ERROR), + vendor_info=ModuleErrorVendorInfo( + vendor=self.vendor(), + code=str(exc.code), + message=exc.message, + ), + ) + await self.send_tts_error(self.current_request_id, err) + if self.current_request_finished: + await self.finish_request( + self.current_request_id, + reason=TTSAudioEndReason.ERROR, + error=err, + ) + await self._reset_request_state() + + # ---------- helpers ---------- + + def _audio_ms(self) -> int: + if not self.config: + return 0 + bps = 2 + ch = self.config.channels + sr = self.config.sample_rate + if sr <= 0: + return 0 + return int(self.total_audio_bytes / (sr * bps * ch) * 1000) + + def _setup_recorder(self, rid: str) -> None: + if not (self.config and self.config.dump and self.config.dump_path): + return + # Drop other recorders. + for old_rid in [k for k in self.recorder_map if k != rid]: + try: + # Best-effort sync close; the framework reuses ids per turn. + del self.recorder_map[old_rid] + except Exception: + pass + if rid in self.recorder_map: + return + path = os.path.join(self.config.dump_path, f"deepdub_dump_{rid}.pcm") + self.recorder_map[rid] = PCMWriter(path) + + async def _reset_request_state(self) -> None: + self.current_request_id = None + self.sent_ts = None + self.first_chunk = True + self.awaiting_finish = False + self.current_request_finished = False + self.total_audio_bytes = 0 + + async def _send_error( + self, exc: DeepdubTTSException, t: TTSTextInput + ) -> None: + err = ModuleError( + message=str(exc), + module=ModuleType.TTS, + code=int(ModuleErrorCode.NON_FATAL_ERROR), + vendor_info=ModuleErrorVendorInfo( + vendor=self.vendor(), + code=str(exc.code), + message=exc.message, + ), + ) + await self.send_tts_error(self.current_request_id or "", err) + if t.text_input_end: + await self.finish_request( + self.current_request_id or "", + reason=TTSAudioEndReason.ERROR, + error=err, + ) + await self._reset_request_state() diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/manifest.json b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/manifest.json new file mode 100644 index 0000000000..efbd3fcde3 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/manifest.json @@ -0,0 +1,72 @@ +{ + "type": "extension", + "name": "deepdub_tts_python", + "version": "0.1.0", + "display_name": { + "locales": { + "en-US": { "content": "Deepdub TTS (Streaming WebSocket) Python" } + } + }, + "description": { + "locales": { + "en-US": { + "content": "Deepdub TTS over the text-streaming WebSocket API. Pre-warmed connection, streams text fragments in, streams PCM out." + } + } + }, + "readme": { + "locales": { + "en-US": { "import_uri": "README.md" } + } + }, + "tags": ["python", "tts", "deepdub", "websocket", "streaming"], + "dependencies": [ + { "type": "system", "name": "ten_runtime_python", "version": "0.11" }, + { "type": "system", "name": "ten_ai_base", "version": "0.7" } + ], + "package": { + "include": [ + "manifest.json", + "property.json", + "BUILD.gn", + "**.tent", + "**.py", + "README.md", + "tests/**", + "requirements.txt" + ] + }, + "scripts": { + "test": "tests/bin/start" + }, + "api": { + "interface": [ + { "import_uri": "../../system/ten_ai_base/api/tts-interface.json" } + ], + "property": { + "properties": { + "params": { + "type": "object", + "properties": { + "api_key": { "type": "string" }, + "url": { "type": "string" }, + "model": { "type": "string" }, + "locale": { "type": "string" }, + "voice_prompt_id": { "type": "string" }, + "sample_rate": { "type": "int64" }, + "channels": { "type": "int64" }, + "format": { "type": "string" }, + "accept_emojis": { "type": "bool" }, + "temperature": { "type": "float64" }, + "variance": { "type": "float64" }, + "tempo": { "type": "float64" }, + "prompt_boost": { "type": "bool" }, + "keepalive_interval_seconds": { "type": "float64" } + } + }, + "dump": { "type": "bool" }, + "dump_path": { "type": "string" } + } + } + } +} diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/property.json b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/property.json new file mode 100644 index 0000000000..1a0e86e8d3 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/property.json @@ -0,0 +1,12 @@ +{ + "params": { + "api_key": "${env:DEEPDUB_API_KEY|}", + "url": "${env:DEEPDUB_WS_STREAMING_URL|}", + "model": "${env:DEEPDUB_MODEL|dd-etts-3.2}", + "locale": "${env:DEEPDUB_LOCALE|en-US}", + "voice_prompt_id": "${env:DEEPDUB_VOICE_PROMPT_ID|}", + "sample_rate": 48000, + "channels": 1, + "format": "s16le" + } +} diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/requirements.txt b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/requirements.txt new file mode 100644 index 0000000000..3805512cfb --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/requirements.txt @@ -0,0 +1,2 @@ +websockets>=12.0 +pydantic>=2.0 diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/__init__.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/bin/start b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/bin/start new file mode 100755 index 0000000000..8e78210572 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/bin/start @@ -0,0 +1,9 @@ +#!/bin/bash + +set -e + +cd "$(dirname "${BASH_SOURCE[0]}")/../.." + +export PYTHONPATH=.ten/app:.ten/app/ten_packages/system/ten_runtime_python/lib:.ten/app/ten_packages/system/ten_runtime_python/interface:.ten/app/ten_packages/system/ten_ai_base/interface:$PYTHONPATH + +pytest -s tests/ "$@" diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_basic.json b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_basic.json new file mode 100644 index 0000000000..6997da4099 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_basic.json @@ -0,0 +1,12 @@ +{ + "params": { + "api_key": "${env:DEEPDUB_API_KEY|test_api_key}", + "url": "${env:DEEPDUB_WS_STREAMING_URL|wss://example.invalid/ws}", + "voice_prompt_id": "${env:DEEPDUB_VOICE_PROMPT_ID|test_voice}", + "model": "dd-etts-3.2", + "locale": "en-US", + "sample_rate": 48000, + "channels": 1, + "format": "s16le" + } +} diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_dump.json b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_dump.json new file mode 100644 index 0000000000..1ff4b64954 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_dump.json @@ -0,0 +1,14 @@ +{ + "dump": true, + "dump_path": "./dump/", + "params": { + "api_key": "test_api_key", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "test_voice", + "model": "dd-etts-3.2", + "locale": "en-US", + "sample_rate": 48000, + "channels": 1, + "format": "s16le" + } +} diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_invalid.json b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_invalid.json new file mode 100644 index 0000000000..e37de434d5 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_invalid.json @@ -0,0 +1,9 @@ +{ + "params": { + "api_key": "test_api_key", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "test_voice", + "sample_rate": 12345, + "format": "s16le" + } +} diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_miss_required.json b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_miss_required.json new file mode 100644 index 0000000000..de36e22678 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/configs/property_miss_required.json @@ -0,0 +1,7 @@ +{ + "params": { + "api_key": "", + "url": "", + "voice_prompt_id": "" + } +} diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/conftest.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/conftest.py new file mode 100644 index 0000000000..b6feec6ec1 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/conftest.py @@ -0,0 +1,75 @@ +import json +import threading +from typing_extensions import override +import pytest +from ten_runtime import App, TenEnv + + +class FakeApp(App): + def __init__(self): + super().__init__() + self.event: threading.Event | None = None + + @override + def on_init(self, ten_env: TenEnv) -> None: + assert self.event + self.event.set() + ten_env.on_init_done() + + @override + def on_configure(self, ten_env: TenEnv) -> None: + ten_env.init_property_from_json( + json.dumps( + { + "ten": { + "log": { + "handlers": [ + { + "matchers": [{"level": "debug"}], + "formatter": { + "type": "plain", + "colored": True, + }, + "emitter": { + "type": "console", + "config": {"stream": "stdout"}, + }, + } + ] + } + } + } + ), + ) + ten_env.on_configure_done() + + +class FakeAppCtx: + def __init__(self, event: threading.Event): + self.fake_app: FakeApp | None = None + self.event = event + + +def run_fake_app(fake_app_ctx: FakeAppCtx): + app = FakeApp() + app.event = fake_app_ctx.event + fake_app_ctx.fake_app = app + app.run(False) + + +@pytest.fixture(scope="session", autouse=True) +def global_setup_and_teardown(): + event = threading.Event() + fake_app_ctx = FakeAppCtx(event) + + fake_app_thread = threading.Thread( + target=run_fake_app, args=(fake_app_ctx,) + ) + fake_app_thread.start() + event.wait() + assert fake_app_ctx.fake_app is not None + + yield + + fake_app_ctx.fake_app.close() + fake_app_thread.join() diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_basic.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_basic.py new file mode 100644 index 0000000000..bb8e0fbe92 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_basic.py @@ -0,0 +1,247 @@ +import sys +from pathlib import Path + +# Project root is 6 levels up from this file's parent. +project_root = str(Path(__file__).resolve().parents[6]) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +import asyncio +import filecmp +import json +import os +import shutil +import threading +from unittest.mock import patch, AsyncMock + +from ten_runtime import ExtensionTester, TenEnvTester, Data +from ten_ai_base.struct import TTSTextInput, TTSFlush + + +# Shared mock-wiring helper. +def _install_client_mock(mock_class): + mock_instance = mock_class.return_value + mock_instance.start = AsyncMock() + mock_instance.stop = AsyncMock() + mock_instance.cancel = AsyncMock() + mock_instance.send_text = AsyncMock() + mock_instance.wait_ready = AsyncMock() + mock_instance.send_cancel = AsyncMock() + + def ctor(config, ten_env, on_audio, on_finish, on_error): + mock_instance.on_audio = on_audio + mock_instance.on_finish = on_finish + mock_instance.on_error = on_error + return mock_instance + + mock_class.side_effect = ctor + return mock_instance + + +# ================ basic happy path + dump file comparison ================ +class ExtensionTesterDump(ExtensionTester): + def __init__(self): + super().__init__() + self.dump_dir = "./dump/" + self.test_dump_file_path = os.path.join( + self.dump_dir, "test_manual_dump.pcm" + ) + self.audio_end_received = False + self.received_audio_chunks = [] + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + ten_env_tester.log_info("Dump test started, sending TTS request.") + tts_input = TTSTextInput( + request_id="tts_request_1", + text="hello world, this is a deepdub test", + text_input_end=True, + ) + data = Data.create("tts_text_input") + data.set_property_from_json(None, tts_input.model_dump_json()) + ten_env_tester.send_data(data) + ten_env_tester.on_start_done() + + def on_data(self, ten_env: TenEnvTester, data) -> None: + if data.get_name() == "tts_audio_end": + self.audio_end_received = True + ten_env.stop_test() + + def on_audio_frame(self, ten_env: TenEnvTester, audio_frame): + buf = audio_frame.lock_buf() + try: + self.received_audio_chunks.append(bytes(buf)) + finally: + audio_frame.unlock_buf(buf) + + def write_test_dump_file(self): + with open(self.test_dump_file_path, "wb") as f: + for chunk in self.received_audio_chunks: + f.write(chunk) + + def find_tts_dump_file(self) -> str | None: + if not os.path.exists(self.dump_dir): + return None + for filename in os.listdir(self.dump_dir): + if filename.endswith(".pcm") and filename != os.path.basename( + self.test_dump_file_path + ): + return os.path.join(self.dump_dir, filename) + return None + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_dump_functionality(MockClient): + DUMP_PATH = "./dump/" + if os.path.exists(DUMP_PATH): + shutil.rmtree(DUMP_PATH) + os.makedirs(DUMP_PATH) + + mock_instance = _install_client_mock(MockClient) + + fake_chunk_1 = b"\x11\x22\x33\x44" * 20 + fake_chunk_2 = b"\xaa\xbb\xcc\xdd" * 20 + + async def stream_audio(text: str): + await mock_instance.on_audio(fake_chunk_1) + await asyncio.sleep(0.01) + await mock_instance.on_audio(fake_chunk_2) + await asyncio.sleep(0.01) + await mock_instance.on_finish() + + async def mock_send_text(text: str): + # text_input_end is set *after* send_text returns in request_tts; + # schedule the simulated stream so awaiting_finish is True when on_finish fires. + asyncio.create_task(stream_audio(text)) + + mock_instance.send_text.side_effect = mock_send_text + + tester = ExtensionTesterDump() + dump_config = { + "params": { + "api_key": "valid_key", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "valid_voice", + }, + "dump": True, + "dump_path": DUMP_PATH, + } + tester.set_test_mode_single("deepdub_tts_python", json.dumps(dump_config)) + + try: + tester.run() + assert tester.audio_end_received, "tts_audio_end was not received" + + tester.write_test_dump_file() + assert os.path.exists(tester.test_dump_file_path) + + tts_dump_file = tester.find_tts_dump_file() + assert ( + tts_dump_file is not None + ), f"Could not find TTS-generated dump file in {DUMP_PATH}" + + assert filecmp.cmp( + tts_dump_file, tester.test_dump_file_path, shallow=False + ), "Extension dump and test-collected audio differ." + finally: + if os.path.exists(DUMP_PATH): + shutil.rmtree(DUMP_PATH) + + +# ================ flush / cancel mid-stream ================ +class ExtensionTesterFlush(ExtensionTester): + def __init__(self): + super().__init__() + self.ten_env: TenEnvTester | None = None + self.audio_start_received = False + self.first_audio_frame_received = False + self.flush_start_received = False + self.audio_end_received = False + self.flush_end_received = False + self.audio_received_after_flush_end = False + self.received_audio_bytes = 0 + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + self.ten_env = ten_env_tester + tts_input = TTSTextInput( + request_id="flush_req", + text="A long enough sentence so we can interrupt mid-stream.", + text_input_end=True, + ) + data = Data.create("tts_text_input") + data.set_property_from_json(None, tts_input.model_dump_json()) + ten_env_tester.send_data(data) + ten_env_tester.on_start_done() + + def on_audio_frame(self, ten_env: TenEnvTester, audio_frame): + if self.flush_end_received: + self.audio_received_after_flush_end = True + if not self.first_audio_frame_received: + self.first_audio_frame_received = True + flush_data = Data.create("tts_flush") + flush_data.set_property_from_json( + None, TTSFlush(flush_id="flush_req").model_dump_json() + ) + ten_env.send_data(flush_data) + buf = audio_frame.lock_buf() + try: + self.received_audio_bytes += len(buf) + finally: + audio_frame.unlock_buf(buf) + + def on_data(self, ten_env: TenEnvTester, data) -> None: + name = data.get_name() + if name == "tts_audio_start": + self.audio_start_received = True + return + if name == "tts_flush_start": + self.flush_start_received = True + return + json_str, _ = data.get_property_to_json(None) + if not json_str: + return + if name == "tts_audio_end": + self.audio_end_received = True + elif name == "tts_flush_end": + self.flush_end_received = True + timer = threading.Timer(0.3, ten_env.stop_test) + timer.start() + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_flush_logic(MockClient): + mock_instance = _install_client_mock(MockClient) + + async def stream_long(text: str): + # Stream many chunks until cancel() is called by the extension. + for _ in range(40): + if mock_instance.cancel.called: + # Vendor finishes the burst with a finish boundary. + await mock_instance.on_finish() + return + await mock_instance.on_audio(b"\x11\x22\x33" * 100) + await asyncio.sleep(0.05) + await mock_instance.on_finish() + + async def mock_send_text(text: str): + asyncio.create_task(stream_long(text)) + + mock_instance.send_text.side_effect = mock_send_text + + config = { + "params": { + "api_key": "k", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v", + } + } + tester = ExtensionTesterFlush() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(config)) + tester.run() + + assert tester.audio_start_received + assert tester.first_audio_frame_received + assert tester.audio_end_received + assert tester.flush_end_received + assert not tester.audio_received_after_flush_end + # The extension must call client.cancel() in response to flush. + assert mock_instance.cancel.called, "client.cancel() was not invoked" diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_error_msg.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_error_msg.py new file mode 100644 index 0000000000..a3a7533b19 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_error_msg.py @@ -0,0 +1,142 @@ +import sys +from pathlib import Path + +project_root = str(Path(__file__).resolve().parents[6]) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +import json +from unittest.mock import patch, AsyncMock + +from ten_runtime import ExtensionTester, TenEnvTester, Data +from ten_ai_base.struct import TTSTextInput +from deepdub_tts_python.deepdub_tts import DeepdubTTSException + + +def _install_client_mock(mock_class): + mock_instance = mock_class.return_value + mock_instance.start = AsyncMock() + mock_instance.stop = AsyncMock() + mock_instance.cancel = AsyncMock() + mock_instance.send_text = AsyncMock() + mock_instance.wait_ready = AsyncMock() + mock_instance.send_cancel = AsyncMock() + + def ctor(config, ten_env, on_audio, on_finish, on_error): + mock_instance.on_audio = on_audio + mock_instance.on_finish = on_finish + mock_instance.on_error = on_error + return mock_instance + + mock_class.side_effect = ctor + return mock_instance + + +# ================ empty params → FATAL error from on_init ================ +class ExtensionTesterEmptyParams(ExtensionTester): + def __init__(self): + super().__init__() + self.error_received = False + self.error_code = None + self.error_message = None + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + ten_env_tester.on_start_done() + + def on_data(self, ten_env: TenEnvTester, data) -> None: + if data.get_name() != "error": + return + json_str, _ = data.get_property_to_json(None) + payload = json.loads(json_str) + self.error_received = True + self.error_code = payload.get("code") + self.error_message = payload.get("message", "") + ten_env.stop_test() + + +def test_empty_params_fatal_error(): + empty_params_config = { + "params": { + "api_key": "", + "url": "", + "voice_prompt_id": "", + } + } + tester = ExtensionTesterEmptyParams() + tester.set_test_mode_single( + "deepdub_tts_python", json.dumps(empty_params_config) + ) + tester.run() + + assert tester.error_received, "Expected error event" + assert ( + tester.error_code == -1000 + ), f"Expected FATAL_ERROR (-1000), got {tester.error_code}" + assert tester.error_message + + +# ================ vendor error during synthesis → NON_FATAL ================ +class ExtensionTesterVendorError(ExtensionTester): + def __init__(self): + super().__init__() + self.error_received = False + self.error_code = None + self.vendor_info = None + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + tts_input = TTSTextInput( + request_id="rid-err", + text="trigger vendor error", + text_input_end=True, + ) + d = Data.create("tts_text_input") + d.set_property_from_json(None, tts_input.model_dump_json()) + ten_env_tester.send_data(d) + ten_env_tester.on_start_done() + + def on_data(self, ten_env: TenEnvTester, data) -> None: + if data.get_name() != "error": + return + json_str, _ = data.get_property_to_json(None) + payload = json.loads(json_str) + self.error_received = True + self.error_code = payload.get("code") + self.vendor_info = payload.get("vendor_info", {}) + ten_env.stop_test() + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_vendor_error_propagated(MockClient): + import asyncio + + mock_instance = _install_client_mock(MockClient) + + async def fire_error(text: str): + await mock_instance.on_error( + DeepdubTTSException("vendor blew up", code=503) + ) + + async def mock_send_text(text: str): + asyncio.create_task(fire_error(text)) + + mock_instance.send_text.side_effect = mock_send_text + + config = { + "params": { + "api_key": "k", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v", + } + } + tester = ExtensionTesterVendorError() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(config)) + tester.run() + + assert tester.error_received + assert ( + tester.error_code == 1000 + ), f"Expected NON_FATAL_ERROR (1000), got {tester.error_code}" + assert tester.vendor_info is not None + assert tester.vendor_info.get("vendor") == "deepdub" + assert tester.vendor_info.get("code") == "503" + assert "vendor blew up" in tester.vendor_info.get("message", "") diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_metrics.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_metrics.py new file mode 100644 index 0000000000..8cb70341e3 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_metrics.py @@ -0,0 +1,103 @@ +import sys +from pathlib import Path + +project_root = str(Path(__file__).resolve().parents[6]) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +import asyncio +import json +from unittest.mock import patch, AsyncMock + +from ten_runtime import ExtensionTester, TenEnvTester, Data +from ten_ai_base.struct import TTSTextInput + + +def _install_client_mock(mock_class): + mock_instance = mock_class.return_value + mock_instance.start = AsyncMock() + mock_instance.stop = AsyncMock() + mock_instance.cancel = AsyncMock() + mock_instance.send_text = AsyncMock() + mock_instance.wait_ready = AsyncMock() + mock_instance.send_cancel = AsyncMock() + + def ctor(config, ten_env, on_audio, on_finish, on_error): + mock_instance.on_audio = on_audio + mock_instance.on_finish = on_finish + mock_instance.on_error = on_error + return mock_instance + + mock_class.side_effect = ctor + return mock_instance + + +class ExtensionTesterMetrics(ExtensionTester): + def __init__(self): + super().__init__() + self.ttfb_received = False + self.ttfb_value = -1 + self.audio_frame_received = False + self.audio_end_received = False + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + tts_input = TTSTextInput( + request_id="metrics_rid", + text="hello, this is a metrics test.", + text_input_end=True, + ) + d = Data.create("tts_text_input") + d.set_property_from_json(None, tts_input.model_dump_json()) + ten_env_tester.send_data(d) + ten_env_tester.on_start_done() + + def on_data(self, ten_env: TenEnvTester, data) -> None: + name = data.get_name() + if name == "metrics": + json_str, _ = data.get_property_to_json(None) + metrics_data = json.loads(json_str) + nested = metrics_data.get("metrics", {}) + if "ttfb" in nested: + self.ttfb_received = True + self.ttfb_value = nested.get("ttfb", -1) + elif name == "tts_audio_end": + self.audio_end_received = True + if self.ttfb_received: + ten_env.stop_test() + + def on_audio_frame(self, ten_env: TenEnvTester, audio_frame): + if not self.audio_frame_received: + self.audio_frame_received = True + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_ttfb_metric_is_sent(MockClient): + mock_instance = _install_client_mock(MockClient) + + async def stream_with_delay(text: str): + await asyncio.sleep(0.2) + await mock_instance.on_audio(b"\x11\x22\x33" * 32) + await mock_instance.on_finish() + + async def mock_send_text(text: str): + asyncio.create_task(stream_with_delay(text)) + + mock_instance.send_text.side_effect = mock_send_text + + config = { + "params": { + "api_key": "k", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v", + } + } + tester = ExtensionTesterMetrics() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(config)) + tester.run() + + assert tester.audio_frame_received + assert tester.audio_end_received + assert tester.ttfb_received + assert ( + 180 <= tester.ttfb_value <= 500 + ), f"TTFB out of expected band: {tester.ttfb_value}ms" diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_params.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_params.py new file mode 100644 index 0000000000..9a8b377ed1 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_params.py @@ -0,0 +1,126 @@ +import sys +from pathlib import Path + +project_root = str(Path(__file__).resolve().parents[6]) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +import json +from unittest.mock import patch, AsyncMock + +from ten_runtime import ( + ExtensionTester, + TenEnvTester, + Cmd, + CmdResult, + StatusCode, + TenError, +) + + +class ExtensionTesterForPassthrough(ExtensionTester): + """Trivial tester that just allows the extension to initialise so we can + assert how the underlying client was constructed.""" + + def check_hello(self, ten_env: TenEnvTester, result: CmdResult | None): + if result is None: + ten_env.stop_test(TenError(1, "CmdResult is None")) + return + if result.get_status_code() == StatusCode.OK: + ten_env.stop_test() + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + new_cmd = Cmd.create("hello_world") + ten_env_tester.send_cmd( + new_cmd, + lambda ten_env, result, _: self.check_hello(ten_env, result), + ) + ten_env_tester.on_start_done() + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_params_passthrough(MockClient): + """Config values land on the DeepdubTTSConfig passed to the client ctor.""" + mock_instance = MockClient.return_value + mock_instance.start = AsyncMock() + mock_instance.stop = AsyncMock() + + real_config = { + "params": { + "api_key": "the_key", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v-prompt-id", + "model": "dd-etts-3.2", + "locale": "es-ES", + "sample_rate": 24000, + "channels": 1, + "format": "s16le", + "temperature": 0.7, + "tempo": 1.1, + "prompt_boost": True, + } + } + tester = ExtensionTesterForPassthrough() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(real_config)) + tester.run() + + MockClient.assert_called_once() + _, kwargs = MockClient.call_args + cfg = kwargs.get("config") or MockClient.call_args.args[0] + + assert cfg.api_key == "the_key" + assert cfg.url == "wss://example.invalid/ws" + assert cfg.voice_prompt_id == "v-prompt-id" + assert cfg.model == "dd-etts-3.2" + assert cfg.locale == "es-ES" + assert cfg.sample_rate == 24000 + assert cfg.format == "s16le" + assert cfg.temperature == 0.7 + assert cfg.tempo == 1.1 + assert cfg.prompt_boost is True + + +def test_missing_required_params_rejected(): + """Empty required fields → on_init raises before client is constructed.""" + bad = {"params": {"api_key": "", "url": "", "voice_prompt_id": ""}} + received_error = {"flag": False} + + class T(ExtensionTester): + def on_start(self, ten_env_tester): + ten_env_tester.on_start_done() + + def on_data(self, ten_env, data): + if data.get_name() == "error": + received_error["flag"] = True + ten_env.stop_test() + + tester = T() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(bad)) + tester.run() + assert received_error["flag"], "Expected error event for missing params" + + +def test_invalid_sample_rate_rejected(): + bad = { + "params": { + "api_key": "k", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v", + "sample_rate": 12345, + } + } + received_error = {"flag": False} + + class T(ExtensionTester): + def on_start(self, ten_env_tester): + ten_env_tester.on_start_done() + + def on_data(self, ten_env, data): + if data.get_name() == "error": + received_error["flag"] = True + ten_env.stop_test() + + tester = T() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(bad)) + tester.run() + assert received_error["flag"], "Expected error for bad sample_rate" diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_robustness.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_robustness.py new file mode 100644 index 0000000000..c396aa45af --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_robustness.py @@ -0,0 +1,123 @@ +import sys +from pathlib import Path + +project_root = str(Path(__file__).resolve().parents[6]) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +import asyncio +import json +from typing import Any +from unittest.mock import patch, AsyncMock + +from ten_runtime import ExtensionTester, TenEnvTester, Data +from ten_ai_base.struct import TTSTextInput +from deepdub_tts_python.deepdub_tts import DeepdubTTSException + + +def _install_client_mock(mock_class): + mock_instance = mock_class.return_value + mock_instance.start = AsyncMock() + mock_instance.stop = AsyncMock() + mock_instance.cancel = AsyncMock() + mock_instance.send_text = AsyncMock() + mock_instance.wait_ready = AsyncMock() + mock_instance.send_cancel = AsyncMock() + + def ctor(config, ten_env, on_audio, on_finish, on_error): + mock_instance.on_audio = on_audio + mock_instance.on_finish = on_finish + mock_instance.on_error = on_error + return mock_instance + + mock_class.side_effect = ctor + return mock_instance + + +class ExtensionTesterRobustness(ExtensionTester): + def __init__(self): + super().__init__() + self.first_request_error: dict[str, Any] | None = None + self.second_request_successful = False + self.ten_env: TenEnvTester | None = None + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + self.ten_env = ten_env_tester + t1 = TTSTextInput( + request_id="rid_fail", + text="trigger a simulated drop", + text_input_end=True, + ) + d = Data.create("tts_text_input") + d.set_property_from_json(None, t1.model_dump_json()) + ten_env_tester.send_data(d) + ten_env_tester.on_start_done() + + def send_second_request(self): + if self.ten_env is None: + return + t2 = TTSTextInput( + request_id="rid_ok", + text="this one succeeds after reconnect", + text_input_end=True, + ) + d = Data.create("tts_text_input") + d.set_property_from_json(None, t2.model_dump_json()) + self.ten_env.send_data(d) + + def on_data(self, ten_env: TenEnvTester, data) -> None: + name = data.get_name() + json_str, _ = data.get_property_to_json(None) + if not json_str: + return + payload = json.loads(json_str) + + if name == "error" and payload.get("id") == "rid_fail": + self.first_request_error = payload + self.send_second_request() + + if payload.get("id") == "rid_ok" and name == "tts_audio_end": + self.second_request_successful = True + ten_env.stop_test() + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_recover_after_vendor_error(MockClient): + """First request fails via on_error; subsequent request still succeeds.""" + mock_instance = _install_client_mock(MockClient) + + call_count = {"n": 0} + + async def stream_first_then_succeed(text: str): + call_count["n"] += 1 + if call_count["n"] == 1: + await mock_instance.on_error( + DeepdubTTSException("Simulated vendor drop", code=503) + ) + else: + await mock_instance.on_audio(b"\x44\x55\x66" * 32) + await mock_instance.on_finish() + + async def mock_send_text(text: str): + asyncio.create_task(stream_first_then_succeed(text)) + + mock_instance.send_text.side_effect = mock_send_text + + config = { + "params": { + "api_key": "k", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v", + } + } + tester = ExtensionTesterRobustness() + tester.set_test_mode_single("deepdub_tts_python", json.dumps(config)) + tester.run() + + assert tester.first_request_error is not None + assert ( + tester.first_request_error.get("code") == 1000 + ), f"Expected NON_FATAL_ERROR, got {tester.first_request_error.get('code')}" + vendor_info = tester.first_request_error.get("vendor_info") + assert vendor_info and vendor_info.get("vendor") == "deepdub" + assert tester.second_request_successful diff --git a/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_state_machine.py b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_state_machine.py new file mode 100644 index 0000000000..6b734f3b66 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/deepdub_tts_python/tests/test_state_machine.py @@ -0,0 +1,126 @@ +import sys +from pathlib import Path + +project_root = str(Path(__file__).resolve().parents[6]) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +import asyncio +import json +from unittest.mock import patch, AsyncMock + +from ten_runtime import ExtensionTester, TenEnvTester, Data +from ten_ai_base.struct import TTSTextInput + + +class StateMachineExtensionTester(ExtensionTester): + def __init__(self): + super().__init__() + self.audio_start_events: list[str] = [] + self.audio_end_events: list[tuple[str, int]] = [] + self.request1_id = "state_req_1" + self.request2_id = "state_req_2" + self.test_completed = False + + def on_start(self, ten_env_tester: TenEnvTester) -> None: + for rid, text in ( + (self.request1_id, "First request text"), + (self.request2_id, "Second request text"), + ): + t = TTSTextInput(request_id=rid, text=text, text_input_end=True) + d = Data.create("tts_text_input") + d.set_property_from_json(None, t.model_dump_json()) + ten_env_tester.send_data(d) + ten_env_tester.on_start_done() + + def on_data(self, ten_env: TenEnvTester, data: Data) -> None: + name = data.get_name() + payload_str, _ = data.get_property_to_json("") + if not payload_str: + return + payload = json.loads(payload_str) + if name == "tts_audio_start": + self.audio_start_events.append(payload.get("request_id", "")) + elif name == "tts_audio_end": + self.audio_end_events.append( + (payload.get("request_id", ""), payload.get("reason", 0)) + ) + if len(self.audio_end_events) == 2: + self.test_completed = True + ten_env.stop_test() + + +@patch("deepdub_tts_python.extension.DeepdubStreamingClient") +def test_sequential_requests_state_machine(MockClient): + """Two requests sent back-to-back are processed in order; each one gets + its own audio_start/audio_end pair with the matching request_id.""" + mock_instance = MockClient.return_value + mock_instance.start = AsyncMock() + mock_instance.stop = AsyncMock() + mock_instance.cancel = AsyncMock() + mock_instance.send_text = AsyncMock() + mock_instance.wait_ready = AsyncMock() + mock_instance.send_cancel = AsyncMock() + + def ctor(config, ten_env, on_audio, on_finish, on_error): + mock_instance.on_audio = on_audio + mock_instance.on_finish = on_finish + mock_instance.on_error = on_error + return mock_instance + + MockClient.side_effect = ctor + + first_done = asyncio.Event() + + async def stream_first(): + await mock_instance.on_audio(b"\x01\x02" * 16) + await mock_instance.on_finish() + first_done.set() + + async def stream_second(): + await first_done.wait() + await mock_instance.on_audio(b"\x03\x04" * 16) + await mock_instance.on_finish() + + async def mock_send_text(text: str): + if "First" in text: + asyncio.create_task(stream_first()) + else: + asyncio.create_task(stream_second()) + + mock_instance.send_text.side_effect = mock_send_text + + tester = StateMachineExtensionTester() + config = { + "params": { + "api_key": "k", + "url": "wss://example.invalid/ws", + "voice_prompt_id": "v", + } + } + tester.set_test_mode_single("deepdub_tts_python", json.dumps(config)) + tester.run() + + assert tester.test_completed + assert len(tester.audio_start_events) == 2 + assert len(tester.audio_end_events) == 2 + # Both completed with success reason (== 1 in the TTSAudioEndReason enum). + assert tester.audio_end_events[0][1] == 1 + assert tester.audio_end_events[1][1] == 1 + + # Strict ordering: req1 starts and ends before req2 starts and ends. + i1 = tester.audio_start_events.index(tester.request1_id) + i2 = tester.audio_start_events.index(tester.request2_id) + assert i1 < i2 + + e1 = next( + i + for i, e in enumerate(tester.audio_end_events) + if e[0] == tester.request1_id + ) + e2 = next( + i + for i, e in enumerate(tester.audio_end_events) + if e[0] == tester.request2_id + ) + assert e1 < e2