118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
|
|
def _install_fake_livekit() -> None:
|
|
if "livekit.agents" in sys.modules:
|
|
return
|
|
|
|
try:
|
|
importlib.import_module("livekit.agents")
|
|
return
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
livekit_pkg = importlib.import_module("livekit")
|
|
except ImportError:
|
|
livekit_pkg = types.ModuleType("livekit")
|
|
livekit_pkg.__path__ = []
|
|
sys.modules["livekit"] = livekit_pkg
|
|
|
|
agents_module = types.ModuleType("livekit.agents")
|
|
types_module = types.ModuleType("livekit.agents.types")
|
|
|
|
class TTSCapabilities:
|
|
def __init__(self, *, streaming: bool, aligned_transcript: bool) -> None:
|
|
self.streaming = streaming
|
|
self.aligned_transcript = aligned_transcript
|
|
|
|
class AudioEmitter:
|
|
def __init__(self) -> None:
|
|
self._data = bytearray()
|
|
self.sample_rate = 0
|
|
self.num_channels = 0
|
|
|
|
def initialize(self, *, request_id: str, sample_rate: int, num_channels: int, mime_type: str) -> None:
|
|
self.request_id = request_id
|
|
self.sample_rate = sample_rate
|
|
self.num_channels = num_channels
|
|
self.mime_type = mime_type
|
|
|
|
def push(self, data: bytes) -> None:
|
|
if data:
|
|
self._data.extend(data)
|
|
|
|
def flush(self) -> None:
|
|
return None
|
|
|
|
def snapshot(self):
|
|
return SimpleNamespace(
|
|
sample_rate=self.sample_rate,
|
|
num_channels=self.num_channels,
|
|
data=bytes(self._data),
|
|
)
|
|
|
|
class BaseTTS:
|
|
def __init__(self, *, capabilities: TTSCapabilities, sample_rate: int, num_channels: int) -> None:
|
|
self.capabilities = capabilities
|
|
self.sample_rate = sample_rate
|
|
self.num_channels = num_channels
|
|
|
|
class BaseChunkedStream:
|
|
def __init__(self, *, tts: BaseTTS, input_text: str, conn_options) -> None:
|
|
self._tts = tts
|
|
self._input_text = input_text
|
|
self._conn_options = conn_options
|
|
|
|
async def collect(self):
|
|
emitter = AudioEmitter()
|
|
await self._run(emitter)
|
|
return emitter.snapshot()
|
|
|
|
class APIConnectOptions:
|
|
def __init__(self, **kwargs) -> None:
|
|
self.kwargs = kwargs
|
|
|
|
tts_module = SimpleNamespace(
|
|
TTS=BaseTTS,
|
|
TTSCapabilities=TTSCapabilities,
|
|
ChunkedStream=BaseChunkedStream,
|
|
AudioEmitter=AudioEmitter,
|
|
)
|
|
|
|
agents_module.tts = tts_module
|
|
agents_module.utils = SimpleNamespace(shortuuid=lambda: "req-test")
|
|
|
|
types_module.APIConnectOptions = APIConnectOptions
|
|
types_module.DEFAULT_API_CONNECT_OPTIONS = APIConnectOptions()
|
|
|
|
setattr(livekit_pkg, "agents", agents_module)
|
|
sys.modules["livekit.agents"] = agents_module
|
|
sys.modules["livekit.agents.types"] = types_module
|
|
|
|
|
|
_install_fake_livekit()
|
|
|
|
from app.livekit.adapters.fake_tts import FakeTTS
|
|
|
|
|
|
class FakeLiveKitTTSTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_synthesize_collects_pcm_audio(self) -> None:
|
|
tts = FakeTTS()
|
|
stream = tts.synthesize("teste fake")
|
|
frame = await stream.collect()
|
|
|
|
self.assertEqual(frame.sample_rate, 16000)
|
|
self.assertEqual(frame.num_channels, 1)
|
|
self.assertGreater(len(frame.data), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|