From fd71d5faeb2dd5d03a0e499dff49458c9aebfba5 Mon Sep 17 00:00:00 2001 From: Wannaphong Date: Sat, 3 Oct 2026 16:49:45 +0700 Subject: [PATCH 1/2] Add realtime TTS --- README.md | 61 ++ demo_realtime.py | 139 +++++ example_realtime.py | 30 + pythaitts/__init__.py | 77 +++ pythaitts/pretrained/fastthaig2p/__init__.py | 22 +- pythaitts/pretrained/fastthaig2p/tts.py | 80 ++- pythaitts/realtime.py | 562 +++++++++++++++++++ setup.py | 1 + tests/test_realtime.py | 294 ++++++++++ 9 files changed, 1264 insertions(+), 2 deletions(-) create mode 100644 demo_realtime.py create mode 100644 example_realtime.py create mode 100644 pythaitts/realtime.py create mode 100644 tests/test_realtime.py diff --git a/README.md b/README.md index da1fa3e..c8401bd 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,67 @@ tts = TTS(pretrained="khanomtan") file = tts.tts("ภาษาไทย", speaker_idx="Linda", filename="output.wav") ``` +### Real-time / Streaming TTS (FastThaiG2P) + +PyThaiTTS supports low-latency, real-time streaming speech synthesis with FastThaiG2P, making it ideal for conversational voice agents and LLM streaming: + +#### 1. Streaming Audio from Text + +Synthesize chunk-by-chunk in real time: + +```python +from pythaitts import TTS + +tts = TTS(pretrained="fastthaig2p") + +# Stream audio chunks as 24kHz float32 NumPy arrays +for audio_chunk in tts.stream("สวัสดีครับ ยินดีต้อนรับสู่ระบบเรียลไทม์ทีทีเอส"): + print(f"Audio chunk shape: {audio_chunk.shape}") + +# Stream raw 16-bit PCM bytes (for WebSockets or PyAudio) +for pcm_bytes in tts.stream("สวัสดีครับ", return_type="bytes"): + # send over websocket or write to audio stream + pass +``` + +#### 2. Streaming from an LLM Token Stream + +Feed tokens directly from an LLM or generator into `tts.stream()`: + +```python +from pythaitts import TTS + +tts = TTS() + +def token_stream(): + tokens = ["สวัสดี", "ครับ", " ", "นี่", "คือ", "การ", "สตรีม", "มิ่ง"] + for tok in tokens: + yield tok + +for audio_chunk in tts.stream(token_stream()): + # Process or play chunk with low latency + pass +``` + +#### 3. Integration with the RealtimeTTS Library + +You can use FastThaiG2P as an engine with [KoljaB/RealtimeTTS](https://github.com/KoljaB/RealtimeTTS): + +```sh +pip install pythaitts[realtime] +``` + +```python +from RealtimeTTS import TextToAudioStream +from pythaitts.realtime import FastThaiG2PEngine + +engine = FastThaiG2PEngine() +stream = TextToAudioStream(engine) +stream.feed("สวัสดีครับ วันนี้อากาศดีมาก") +stream.play() +``` + + ### Text Preprocessing PyThaiTTS includes automatic text preprocessing to improve TTS quality: diff --git a/demo_realtime.py b/demo_realtime.py new file mode 100644 index 0000000..165142d --- /dev/null +++ b/demo_realtime.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +Real-time Text-to-Speech Demo for FastThaiG2P in PyThaiTTS. + +Demonstrates: +1. Low-latency streaming TTS from text string (chunk by chunk). +2. Streaming TTS from an LLM-like token generator. +3. Streaming 16-bit PCM bytes for audio pipelines/WebSockets. +4. FastThaiG2PEngine integration with KoljaB/RealtimeTTS. +""" + +import time +from pythaitts import TTS, RealtimeTTS, FastThaiG2PEngine + + +def main(): + print("=" * 65) + print("PyThaiTTS - Real-time TTS Demo (FastThaiG2P)") + print("=" * 65) + print() + + # 1. Initialize TTS model + print("Initializing FastThaiG2P TTS model...") + t_start = time.time() + tts = TTS(pretrained="fastthaig2p") + print(f"✓ Model loaded in {time.time() - t_start:.2f}s") + print() + + # 2. Streaming from a full text string + sample_text = ( + "สวัสดีครับ ยินดีต้อนรับสู่ระบบสังเคราะห์เสียงภาษาไทยแบบเรียลไทม์ " + "ระบบนี้ช่วยให้สร้างเสียงพูดได้อย่างรวดเร็วและต่อเนื่อง " + "โดยเริ่มส่งสัญญาณเสียงได้ทันทีตั้งแต่ข้อความส่วนแรกประมวลผลเสร็จ" + ) + print("-----------------------------------------------------------------") + print("Demo 1: Real-time Streaming from Full Text") + print("-----------------------------------------------------------------") + print(f"Input text:\n{sample_text}\n") + + t0 = time.time() + total_audio_samples = 0 + chunk_count = 0 + ttfa = None # Time to First Audio + + for audio_chunk in tts.stream(sample_text, return_type="waveform"): + chunk_count += 1 + elapsed = time.time() - t0 + if ttfa is None: + ttfa = elapsed + print(f"⚡ Time to First Audio (TTFA): {ttfa * 1000:.1f} ms!") + + duration = len(audio_chunk) / 24000.0 + total_audio_samples += len(audio_chunk) + print( + f" [Chunk {chunk_count}] Received {len(audio_chunk)} samples " + f"({duration:.2f}s of audio) at +{elapsed:.2f}s" + ) + + total_time = time.time() - t0 + total_audio_sec = total_audio_samples / 24000.0 + rtf = total_time / total_audio_sec if total_audio_sec > 0 else 0 + print(f"\n✓ Generated {total_audio_sec:.2f}s of audio across {chunk_count} chunks in {total_time:.2f}s") + print(f" Real-Time Factor (RTF): {rtf:.3f} (< 1.0 means faster than real-time)") + print() + + # 3. Streaming from simulated LLM token stream + print("-----------------------------------------------------------------") + print("Demo 2: Real-time Streaming from LLM Token Stream") + print("-----------------------------------------------------------------") + + def simulate_llm_stream(): + tokens = [ + "สวัสดี", "ครับ", " ", "นี่", "คือ", "การ", "ทดสอบ", " ", + "การ", "สตรีม", "มิ่ง", " ", "ข้อความ", "จาก", " ", "โมเดล", + "ภาษา", "ขนาด", "ใหญ่", " ", "แบบ", "เรียล", "ไทม์", "ครับ" + ] + print("Streaming tokens from LLM: ", end="", flush=True) + for tok in tokens: + print(tok, end="", flush=True) + time.sleep(0.04) # simulate LLM generation delay + yield tok + print("\n") + + t0 = time.time() + stream_chunks = 0 + for audio_chunk in tts.stream(simulate_llm_stream(), return_type="waveform"): + stream_chunks += 1 + elapsed = time.time() - t0 + dur = len(audio_chunk) / 24000.0 + print(f" [Audio Chunk {stream_chunks}] Duration: {dur:.2f}s at +{elapsed:.2f}s") + + print(f"✓ LLM streaming synthesis complete in {time.time() - t0:.2f}s\n") + + # 4. Streaming 16-bit PCM bytes + print("-----------------------------------------------------------------") + print("Demo 3: Streaming 16-bit PCM Bytes (for WebSockets / PyAudio)") + print("-----------------------------------------------------------------") + byte_chunks = 0 + total_bytes = 0 + for pcm in tts.stream("ระบบเสียงภาษาไทย คุณภาพสูง", return_type="bytes"): + byte_chunks += 1 + total_bytes += len(pcm) + print(f" [PCM Chunk {byte_chunks}] Received {len(pcm)} bytes of raw 16-bit PCM") + + print(f"✓ Total raw PCM data: {total_bytes} bytes (24 kHz, 16-bit, mono)\n") + + # 5. RealtimeTTS Library compatibility + print("-----------------------------------------------------------------") + print("Demo 4: RealtimeTTS Engine (KoljaB/RealtimeTTS compatibility)") + print("-----------------------------------------------------------------") + try: + from RealtimeTTS import TextToAudioStream + print("RealtimeTTS library detected. Initializing FastThaiG2PEngine...") + engine = FastThaiG2PEngine() + stream = TextToAudioStream(engine) + print("✓ FastThaiG2PEngine ready with RealtimeTTS TextToAudioStream!") + except ImportError: + print("KoljaB/RealtimeTTS library is not installed.") + print("To use with RealtimeTTS:") + print(" pip install pythaitts[realtime]") + print(" or: pip install RealtimeTTS") + print("\nFastThaiG2PEngine can then be used directly:") + print(" from RealtimeTTS import TextToAudioStream") + print(" from pythaitts.realtime import FastThaiG2PEngine") + print(" engine = FastThaiG2PEngine()") + print(" stream = TextToAudioStream(engine)") + print(" stream.feed('สวัสดีครับ')") + print(" stream.play()") + + print() + print("=" * 65) + print("Real-time TTS Demo completed successfully!") + print("=" * 65) + return 0 + + +if __name__ == "__main__": + exit(main()) diff --git a/example_realtime.py b/example_realtime.py new file mode 100644 index 0000000..e9b7729 --- /dev/null +++ b/example_realtime.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +Example for Real-time TTS using FastThaiG2P. + +Usage: + python example_realtime.py +""" + +from pythaitts import TTS + +def main(): + print("PyThaiTTS Realtime TTS Example (fastthaig2p)") + print("-" * 50) + + # Initialize FastThaiG2P TTS + tts = TTS(pretrained="fastthaig2p") + + # 1. Stream speech from text string chunk-by-chunk + text = "สวัสดีครับ ยินดีต้อนรับสู่ระบบสังเคราะห์เสียงภาษาไทยแบบเรียลไทม์" + print(f"Streaming text: {text}\n") + + for i, audio_chunk in enumerate(tts.stream(text, return_type="waveform"), start=1): + duration = len(audio_chunk) / 24000.0 + print(f" Chunk {i}: {len(audio_chunk)} samples ({duration:.2f}s of audio)") + + print("\nDone!") + +if __name__ == "__main__": + main() diff --git a/pythaitts/__init__.py b/pythaitts/__init__.py index 6042e6d..4aa8295 100644 --- a/pythaitts/__init__.py +++ b/pythaitts/__init__.py @@ -4,7 +4,17 @@ """ __version__ = "0.6.0" +from typing import Union, Iterable, Iterator + from pythaitts.preprocess import preprocess_text, num_to_thai, expand_maiyamok +from pythaitts.realtime import ( + RealtimeTTS, + FastThaiG2PEngine, + PyThaiTTSEngine, + RealtimeTTSEngine, + chunk_text, + stream_text_to_chunks, +) class TTS: @@ -100,3 +110,70 @@ def tts(self, text: str, speaker_idx: str = "thai_som", language_idx: str = "th- return_type=return_type, filename=filename ) + + def stream( + self, + text: Union[str, Iterable[str]], + speaker_idx: str = "thai_som", + return_type: str = "waveform", + play: bool = False, + preprocess: bool = True, + max_phonemes: int = 400, + **kwargs, + ): + """ + Stream speech synthesis in real-time. + + :param Union[str, Iterable[str]] text: Input text or stream of text tokens (e.g. from LLM) + :param str speaker_idx: Voice to use (default: "thai_som" for fastthaig2p) + :param str return_type: Return format ("waveform", "bytes", "raw", "file") + :param bool play: Whether to play audio chunks in real-time to speakers + :param bool preprocess: Whether to preprocess text (numbers to words, ๆ) + :param int max_phonemes: Maximum phonemes per synthesized chunk + :param kwargs: Additional parameters passed to the model + :return: Generator yielding audio chunks + """ + if self.pretrained in ("fastthaig2p", "FastThaiG2P"): + if speaker_idx in ("Linda", None): + speaker_idx = "thai_som" + return self.model.stream( + text=text, + speaker_idx=speaker_idx, + return_type=return_type, + play=play, + preprocess=preprocess, + max_phonemes=max_phonemes, + **kwargs, + ) + else: + from pythaitts.realtime import stream_text_to_chunks + + def _gen(): + for chunk in stream_text_to_chunks( + text, max_phonemes=max_phonemes, preprocess=preprocess + ): + yield self.tts( + text=chunk, + speaker_idx=speaker_idx, + return_type=return_type, + preprocess=False, + **kwargs, + ) + + return _gen() + + tts_stream = stream + + +__all__ = [ + "TTS", + "RealtimeTTS", + "FastThaiG2PEngine", + "PyThaiTTSEngine", + "RealtimeTTSEngine", + "preprocess_text", + "num_to_thai", + "expand_maiyamok", + "chunk_text", + "stream_text_to_chunks", +] diff --git a/pythaitts/pretrained/fastthaig2p/__init__.py b/pythaitts/pretrained/fastthaig2p/__init__.py index dffa10c..242c1af 100644 --- a/pythaitts/pretrained/fastthaig2p/__init__.py +++ b/pythaitts/pretrained/fastthaig2p/__init__.py @@ -23,7 +23,18 @@ from .normalizer import normalize from .tokenizer import Tokenizer -__all__ = ["G2P", "Tokenizer", "normalize", "ipa_to_kokoro", "TTS", "FastThaiG2P"] +__all__ = [ + "G2P", + "Tokenizer", + "normalize", + "ipa_to_kokoro", + "TTS", + "FastThaiG2P", + "FastThaiG2PEngine", + "FastThaiG2PVoice", + "chunk_text", + "stream_text_to_chunks", +] def __getattr__(name): @@ -31,5 +42,14 @@ def __getattr__(name): from .tts import TTS, FastThaiG2P return FastThaiG2P if name == "FastThaiG2P" else TTS + if name in ("FastThaiG2PEngine", "FastThaiG2PVoice", "chunk_text", "stream_text_to_chunks"): + from pythaitts.realtime import ( + FastThaiG2PEngine, + FastThaiG2PVoice, + chunk_text, + stream_text_to_chunks, + ) + + return locals()[name] raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/pythaitts/pretrained/fastthaig2p/tts.py b/pythaitts/pretrained/fastthaig2p/tts.py index 002e47d..6c2b8e2 100644 --- a/pythaitts/pretrained/fastthaig2p/tts.py +++ b/pythaitts/pretrained/fastthaig2p/tts.py @@ -32,7 +32,7 @@ import os import wave from pathlib import Path -from typing import Optional +from typing import Optional, Union, Iterable, Iterator import numpy as np @@ -277,6 +277,84 @@ def __call__( self.synthesize(text, filename) return str(filename) + def stream( + self, + text: Union[str, Iterable[str]], + speaker_idx: str = "thai_som", + return_type: str = "waveform", + play: bool = False, + preprocess: bool = True, + max_phonemes: int = 400, + **kwargs, + ): + """Stream Thai speech generation chunk by chunk in real-time. + + :param Union[str, Iterable[str]] text: Input text or stream of text tokens (e.g. from LLM) + :param str speaker_idx: Voice to use (default: "thai_som" or path to custom voicepack) + :param str return_type: Return format ("waveform", "bytes", "raw", "file") + :param bool play: Whether to play audio chunks in real-time to speakers + :param bool preprocess: Whether to preprocess text (numbers to words, ๆ) + :param int max_phonemes: Maximum phonemes per synthesized chunk (default: 400) + :param kwargs: Additional parameters (e.g., speed) + :yield: Audio chunk (np.ndarray float32 waveform, int16 bytes, or wav file path) + """ + if speaker_idx in ("Linda", None): + speaker_idx = "thai_som" + + if speaker_idx not in self.SUPPORTED_VOICES and not os.path.exists(str(speaker_idx)): + raise ValueError( + f"Unsupported voice '{speaker_idx}'. Supported voices are: {', '.join(self.SUPPORTED_VOICES)}" + ) + + orig_speed = self.speed + if "speed" in kwargs: + self.speed = kwargs["speed"] + + from pythaitts.realtime import _AudioPlayer, stream_text_to_chunks + + player = _AudioPlayer(sample_rate=self.sample_rate) if play else None + + try: + chunks = stream_text_to_chunks( + text, + max_phonemes=max_phonemes, + preprocess=preprocess, + g2p_converter=self._text_to_phonemes, + ) + for chunk in chunks: + audio = self.generate(chunk) + if len(audio) == 0: + continue + + if player is not None: + player.play(audio) + + if return_type == "waveform": + yield audio + elif return_type in ("bytes", "raw"): + yield (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16).tobytes() + elif return_type == "file": + import tempfile + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as fp: + temp_name = fp.name + audio_int16 = (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16) + with wave.open(temp_name, "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(self.sample_rate) + wf.writeframes(audio_int16.tobytes()) + yield temp_name + else: + raise ValueError( + f"Unsupported return_type '{return_type}'. Choose 'waveform', 'bytes', or 'file'." + ) + finally: + if player is not None: + player.close() + self.speed = orig_speed + + tts_stream = stream + FastThaiG2P = TTS diff --git a/pythaitts/realtime.py b/pythaitts/realtime.py new file mode 100644 index 0000000..95552f5 --- /dev/null +++ b/pythaitts/realtime.py @@ -0,0 +1,562 @@ +# -*- coding: utf-8 -*- +""" +Real-time Text-to-Speech (TTS) support for PyThaiTTS. + +Provides: +- RealtimeTTS: High-level streaming TTS class. +- FastThaiG2PEngine: Engine for integration with the RealtimeTTS library (KoljaB/RealtimeTTS). +- chunk_text / stream_text_to_chunks: Text chunking utilities for streaming Thai synthesis. +- _AudioPlayer: Live speaker playback for audio chunks. +""" + +from __future__ import annotations + +import os +import queue +import tempfile +import wave +from pathlib import Path +from typing import Any, Callable, Iterable, Iterator, List, Optional, Union + +import numpy as np + +# Try importing BaseEngine from RealtimeTTS library +try: + from RealtimeTTS.engines.base_engine import BaseEngine, TimingInfo + _HAS_REALTIMETTS = True +except ImportError: + _HAS_REALTIMETTS = False + + class TimingInfo: + def __init__(self, start_time: float, end_time: float, word: str): + self.start_time = start_time + self.end_time = end_time + self.word = word + + def __str__(self): + return f"Word: {self.word}, Start Time: {self.start_time}, End Time: {self.end_time}" + + class BaseEngine: + """Fallback BaseEngine compatible with RealtimeTTS's BaseEngine.""" + + def __init__(self): + import multiprocessing as mp + + self.engine_name = "unknown" + self.can_consume_generators = False + self.preload_sentence_tokenizer = False + self.queue = queue.Queue() + self.timings = queue.Queue() + self.provides_word_timings = False + self.on_audio_chunk = None + self.on_playback_start = None + self.stop_synthesis_event = mp.Event() + self._trim_silence_start_pending = None + self.audio_duration = 0 + + def post_init(self): + pass + + def reset_audio_duration(self): + self.audio_duration = 0 + + def verify_sample_rate(self, sample_rate: int) -> int: + return sample_rate if sample_rate > 0 else 24000 + + def get_stream_info(self): + raise NotImplementedError + + def synthesize(self, text: str, sentence_count: int = 0) -> bool: + self.stop_synthesis_event.clear() + self._trim_silence_start_pending = True + return True + + def get_voices(self): + raise NotImplementedError + + def set_voice(self, voice: Union[str, object]): + raise NotImplementedError + + def set_voice_parameters(self, **voice_parameters): + pass + + def stop(self): + self.stop_synthesis_event.set() + + def shutdown(self): + self.stop() + + +class FastThaiG2PVoice: + """Voice descriptor for FastThaiG2P in RealtimeTTS.""" + + def __init__(self, name: str = "thai_som"): + self.name = name + + def __repr__(self): + return f"FastThaiG2PVoice(name={self.name})" + + def __str__(self): + return self.name + + +def chunk_text( + text: str, + max_phonemes: int = 400, + preprocess: bool = True, + g2p_converter: Optional[Callable[[str], str]] = None, +) -> List[str]: + """Split Thai text into chunks suitable for FastThaiG2P synthesis. + + Ensures each chunk does not exceed max_phonemes to prevent Kokoro context + overflow (limit 510). + + :param str text: Thai text to split + :param int max_phonemes: Maximum phonemes per chunk (default: 400) + :param bool preprocess: Whether to preprocess text (numbers to words, ๆ) + :param Callable g2p_converter: Optional custom function converting text to phonemes + :return: List of text chunks + """ + if not text or not str(text).strip(): + return [] + + if preprocess: + from pythaitts.preprocess import preprocess_text + + text = preprocess_text(str(text)) + + try: + from pythainlp.tokenize import sent_tokenize + + sentences = sent_tokenize(text, engine="crfcut") + except Exception: + sentences = [s.strip() for s in text.splitlines() if s.strip()] + if not sentences: + sentences = [text] + + if g2p_converter is None: + from pythaitts.pretrained.fastthaig2p import G2P, ipa_to_kokoro + + _g2p = G2P() + + def _to_phonemes(t: str) -> str: + return ipa_to_kokoro(_g2p.convert(t)) + + g2p_converter = _to_phonemes + + chunks: List[str] = [] + for sent in sentences: + s = sent.strip() + if not s: + continue + try: + phonemes = g2p_converter(s) + p_len = len(phonemes) + except Exception: + p_len = len(s) * 2 # conservative estimate + + if p_len <= max_phonemes: + chunks.append(s) + else: + # Sub-divide by words using PyThaiNLP + try: + from pythainlp.tokenize import word_tokenize + + words = word_tokenize(s, engine="newmm") + except Exception: + words = s.split(" ") + + cur_words: List[str] = [] + for w in words: + cur_words.append(w) + candidate = "".join(cur_words) + try: + cand_phonemes = g2p_converter(candidate) + cand_len = len(cand_phonemes) + except Exception: + cand_len = len(candidate) * 2 + + if cand_len > max_phonemes: + if len(cur_words) > 1: + chunks.append("".join(cur_words[:-1])) + cur_words = [w] + else: + chunks.append(w) + cur_words = [] + if cur_words: + chunks.append("".join(cur_words)) + + return chunks + + +def stream_text_to_chunks( + text_or_stream: Union[str, Iterable[str]], + max_phonemes: int = 400, + max_buffer_chars: int = 120, + preprocess: bool = True, + g2p_converter: Optional[Callable[[str], str]] = None, +) -> Iterator[str]: + """Yield text chunks from either a full string or a streaming iterable of text tokens. + + Buffers incoming tokens from LLM streams and emits complete sentences/clauses + as soon as they are ready. + + :param Union[str, Iterable[str]] text_or_stream: String or generator of text tokens + :param int max_phonemes: Maximum phonemes per chunk + :param int max_buffer_chars: Maximum characters to buffer before breaking on word boundary + :param bool preprocess: Whether to preprocess text + :param Callable g2p_converter: Optional phoneme conversion function + :yield: Text chunks ready for synthesis + """ + if isinstance(text_or_stream, str): + for c in chunk_text( + text_or_stream, + max_phonemes=max_phonemes, + preprocess=preprocess, + g2p_converter=g2p_converter, + ): + yield c + return + + buffer = "" + try: + from pythainlp.tokenize import sent_tokenize, word_tokenize + except ImportError: + sent_tokenize = None + word_tokenize = None + + for token in text_or_stream: + if not token: + continue + buffer += str(token) + + # Handle explicit newlines as immediate sentence breaks + if "\n" in buffer: + parts = buffer.split("\n") + for part in parts[:-1]: + if part.strip(): + for c in chunk_text( + part, + max_phonemes=max_phonemes, + preprocess=preprocess, + g2p_converter=g2p_converter, + ): + yield c + buffer = parts[-1] + + # Check for sentence boundaries + sents = [] + if sent_tokenize is not None and len(buffer) > 20: + try: + sents = sent_tokenize(buffer, engine="crfcut") + except Exception: + sents = [buffer] + + if len(sents) > 1: + for s in sents[:-1]: + if s.strip(): + for c in chunk_text( + s, + max_phonemes=max_phonemes, + preprocess=preprocess, + g2p_converter=g2p_converter, + ): + yield c + buffer = sents[-1] + elif len(buffer) >= max_buffer_chars: + # Buffer is long, break at word boundary + if word_tokenize is not None: + try: + words = word_tokenize(buffer, engine="newmm") + except Exception: + words = buffer.split(" ") + else: + words = buffer.split(" ") + + if len(words) > 1: + cutoff = max(1, int(len(words) * 0.75)) + to_emit = "".join(words[:cutoff]) + if to_emit.strip(): + for c in chunk_text( + to_emit, + max_phonemes=max_phonemes, + preprocess=preprocess, + g2p_converter=g2p_converter, + ): + yield c + buffer = "".join(words[cutoff:]) + + # Flush remainder of buffer + if buffer.strip(): + for c in chunk_text( + buffer, + max_phonemes=max_phonemes, + preprocess=preprocess, + g2p_converter=g2p_converter, + ): + yield c + + +class _AudioPlayer: + """Helper for streaming audio playback through speaker devices.""" + + def __init__(self, sample_rate: int = 24000): + self.sample_rate = sample_rate + self.pa_instance = None + self.pa_stream = None + self.sd = None + + # Try pyaudio first (allows continuous stream writes) + try: + import pyaudio + + self.pa_instance = pyaudio.PyAudio() + self.pa_stream = self.pa_instance.open( + format=pyaudio.paInt16, + channels=1, + rate=self.sample_rate, + output=True, + ) + return + except Exception: + pass + + # Fallback to sounddevice + try: + import sounddevice as sd + + self.sd = sd + return + except ImportError: + pass + + raise ImportError( + "Live audio playback requires 'pyaudio' or 'sounddevice'. " + "Please install one via: pip install pyaudio or pip install sounddevice" + ) + + def play(self, audio: np.ndarray): + if self.pa_stream is not None: + pcm = (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16).tobytes() + self.pa_stream.write(pcm) + elif self.sd is not None: + self.sd.play(audio, samplerate=self.sample_rate) + self.sd.wait() + + def close(self): + if self.pa_stream is not None: + try: + self.pa_stream.stop_stream() + self.pa_stream.close() + except Exception: + pass + self.pa_stream = None + if self.pa_instance is not None: + try: + self.pa_instance.terminate() + except Exception: + pass + self.pa_instance = None + + +class FastThaiG2PEngine(BaseEngine): + """RealtimeTTS engine implementation using FastThaiG2P + Kokoro-82M. + + Seamlessly integrates with KoljaB/RealtimeTTS: + ```python + from RealtimeTTS import TextToAudioStream + from pythaitts.realtime import FastThaiG2PEngine + + engine = FastThaiG2PEngine() + stream = TextToAudioStream(engine) + stream.feed("สวัสดีครับ วันนี้อากาศดีมาก") + stream.play() + ``` + """ + + SUPPORTED_VOICES = ["thai_som"] + + def __init__( + self, + voice: Union[str, FastThaiG2PVoice] = "thai_som", + speed: float = 1.0, + model_path: Optional[str | Path] = None, + voicepack_path: Optional[str | Path] = None, + config_path: Optional[str | Path] = None, + device: Optional[str] = None, + preprocess: bool = True, + max_phonemes: int = 400, + debug: bool = False, + ): + super().__init__() + self.engine_name = "fastthaig2p" + self.debug = debug + self.preprocess = preprocess + self.max_phonemes = max_phonemes + self.speed = speed + + from pythaitts.pretrained.fastthaig2p.tts import FastThaiG2P + + self.model = FastThaiG2P( + model_path=model_path, + voicepack_path=voicepack_path, + config_path=config_path, + device=device, + speed=speed, + ) + self.sample_rate = self.model.sample_rate + self.set_voice(voice) + + def post_init(self): + self.engine_name = "fastthaig2p" + + def get_stream_info(self): + """Returns audio stream format, channels, and sample rate.""" + try: + import pyaudio + + pa_int16 = pyaudio.paInt16 + except ImportError: + pa_int16 = 1 # Standard paInt16 enum value + + return pa_int16, 1, self.sample_rate + + def synthesize(self, text: str, sentence_count: int = 0) -> bool: + """Synthesizes text and pushes 16-bit PCM chunks into self.queue.""" + if hasattr(super(), "synthesize"): + super().synthesize(text, sentence_count) + self.stop_synthesis_event.clear() + + if not text or not str(text).strip(): + return True + + chunks = chunk_text( + str(text), + max_phonemes=self.max_phonemes, + preprocess=self.preprocess, + g2p_converter=self.model._text_to_phonemes, + ) + + for chunk in chunks: + if self.stop_synthesis_event.is_set(): + if self.debug: + print("[FastThaiG2PEngine] Synthesis stopped by event.") + return False + + try: + audio = self.model.generate(chunk) + if len(audio) == 0: + continue + + pcm_bytes = ( + (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16).tobytes() + ) + self.queue.put(pcm_bytes) + self.audio_duration += len(audio) / float(self.sample_rate) + + if self.on_audio_chunk is not None: + try: + self.on_audio_chunk(pcm_bytes) + except Exception: + pass + except Exception as e: + if self.debug: + print(f"[FastThaiG2PEngine] Error synthesizing chunk: {e}") + return False + + return True + + def get_voices(self) -> List[FastThaiG2PVoice]: + """Returns list of supported voices.""" + return [FastThaiG2PVoice(v) for v in self.SUPPORTED_VOICES] + + def set_voice(self, voice: Union[str, FastThaiG2PVoice]): + """Sets the active voice.""" + voice_str = voice.name if isinstance(voice, FastThaiG2PVoice) else str(voice) + if voice_str in ("Linda", None, ""): + voice_str = "thai_som" + + if voice_str not in self.SUPPORTED_VOICES and not os.path.exists(voice_str): + raise ValueError( + f"Unsupported voice '{voice_str}'. Supported voices are: {', '.join(self.SUPPORTED_VOICES)}" + ) + self.current_voice = voice_str + + def set_voice_parameters(self, **voice_parameters): + """Sets optional voice parameters (e.g. speed).""" + if "speed" in voice_parameters: + self.speed = float(voice_parameters["speed"]) + self.model.speed = self.speed + + def stop(self): + """Stops ongoing synthesis.""" + self.stop_synthesis_event.set() + + def shutdown(self): + """Shuts down the engine.""" + self.stop() + + +# Aliases for convenience +PyThaiTTSEngine = FastThaiG2PEngine +RealtimeTTSEngine = FastThaiG2PEngine + + +class RealtimeTTS: + """High-level Real-time Text-to-Speech manager for PyThaiTTS. + + Supports streaming generation from text or token streams with low latency. + """ + + def __init__( + self, + pretrained: str = "fastthaig2p", + speaker_idx: str = "thai_som", + speed: float = 1.0, + device: str = "cpu", + **kwargs, + ): + from pythaitts import TTS + + self.tts = TTS(pretrained=pretrained, device=device, **kwargs) + self.speaker_idx = speaker_idx + self.speed = speed + + def stream( + self, + text: Union[str, Iterable[str]], + speaker_idx: Optional[str] = None, + return_type: str = "waveform", + play: bool = False, + preprocess: bool = True, + max_phonemes: int = 400, + **kwargs, + ) -> Iterator[np.ndarray | bytes | str]: + """Stream speech synthesis in real-time. + + :param Union[str, Iterable[str]] text: Input text or token generator + :param str speaker_idx: Voice to use (default: initialized voice) + :param str return_type: Return format ("waveform", "bytes", "raw", "file") + :param bool play: Whether to play audio chunks in real-time + :param bool preprocess: Whether to preprocess text + :param int max_phonemes: Maximum phonemes per synthesized chunk + :param kwargs: Additional parameters passed to the model + :yield: Audio chunk (numpy array, PCM bytes, or wav file path) + """ + voice = speaker_idx or self.speaker_idx + speed = kwargs.get("speed", self.speed) + return self.tts.stream( + text=text, + speaker_idx=voice, + return_type=return_type, + play=play, + preprocess=preprocess, + max_phonemes=max_phonemes, + speed=speed, + **kwargs, + ) + + def synthesize(self, text: str, **kwargs): + """Standard batch synthesis.""" + voice = kwargs.pop("speaker_idx", self.speaker_idx) + return self.tts.tts(text, speaker_idx=voice, **kwargs) diff --git a/setup.py b/setup.py index 30ada88..f1c3daa 100644 --- a/setup.py +++ b/setup.py @@ -26,6 +26,7 @@ extras_require={ "vachanatts": ["vachanatts"], "vachana": ["vachanatts"], + "realtime": ["RealtimeTTS"], }, license="Apache Software License 2.0", zip_safe=False, diff --git a/tests/test_realtime.py b/tests/test_realtime.py new file mode 100644 index 0000000..b4e4f07 --- /dev/null +++ b/tests/test_realtime.py @@ -0,0 +1,294 @@ +# -*- coding: utf-8 -*- +""" +Unit tests for realtime TTS integration with FastThaiG2P +""" +import os +import unittest +from unittest.mock import Mock, patch +import numpy as np + +from pythaitts import TTS, RealtimeTTS, FastThaiG2PEngine +from pythaitts.realtime import ( + chunk_text, + stream_text_to_chunks, + FastThaiG2PVoice, + _AudioPlayer, +) + + +class TestRealtimeChunking(unittest.TestCase): + """Test text chunking and streaming utilities for realtime synthesis.""" + + def test_chunk_text_empty(self): + self.assertEqual(chunk_text(""), []) + self.assertEqual(chunk_text(" "), []) + self.assertEqual(chunk_text(None), []) + + def test_chunk_text_basic(self): + chunks = chunk_text("สวัสดีครับ ยินดีต้อนรับ") + self.assertTrue(len(chunks) >= 1) + self.assertIn("สวัสดีครับ", chunks[0]) + + def test_chunk_text_multiple_sentences(self): + text = "สวัสดีครับ วันนี้อากาศดีมาก เราไปเที่ยวกันเถอะ" + chunks = chunk_text(text) + self.assertTrue(len(chunks) >= 2) + + def test_chunk_text_preprocessing(self): + text = "มี 5 คนๆ" + chunks_preprocessed = chunk_text(text, preprocess=True) + self.assertTrue(len(chunks_preprocessed) > 0) + combined = " ".join(chunks_preprocessed) + self.assertIn("ห้า", combined) + self.assertIn("คนคน", combined) + self.assertNotIn("5", combined) + self.assertNotIn("ๆ", combined) + + chunks_no_preprocess = chunk_text(text, preprocess=False) + self.assertTrue(len(chunks_no_preprocess) > 0) + combined_raw = " ".join(chunks_no_preprocess) + self.assertIn("5", combined_raw) + self.assertIn("ๆ", combined_raw) + + def test_chunk_text_long_sentence(self): + # Long Thai sentence without spaces + long_sentence = "นี่คือตัวอย่างของข้อความภาษาไทยที่เขียนติดต่อกันยาวมากโดยไม่มีการเว้นวรรคเลยแม้แต่น้อยเพื่อทดสอบการแบ่งส่วน" * 4 + chunks = chunk_text(long_sentence, max_phonemes=300) + self.assertTrue(len(chunks) > 1) + from pythaitts.pretrained.fastthaig2p import G2P, ipa_to_kokoro + g2p = G2P() + for c in chunks: + phonemes = ipa_to_kokoro(g2p.convert(c)) + self.assertLessEqual(len(phonemes), 300) + + def test_stream_text_to_chunks_from_string(self): + text = "สวัสดีครับ วันนี้อากาศดี" + chunks = list(stream_text_to_chunks(text)) + self.assertTrue(len(chunks) >= 1) + + def test_stream_text_to_chunks_from_tokens(self): + tokens = ["สวัสดี", "ครับ", " ", "ยินดี", "ต้อนรับ", "สู่", "พายไทย", "ทีทีเอส"] + chunks = list(stream_text_to_chunks(tokens)) + self.assertTrue(len(chunks) >= 1) + combined = "".join(chunks) + self.assertIn("สวัสดี", combined) + self.assertIn("พายไทย", combined) + + def test_stream_text_to_chunks_with_newlines(self): + tokens = ["ข้อความที่หนึ่ง\n", "ข้อความที่สอง\n"] + chunks = list(stream_text_to_chunks(tokens)) + self.assertTrue(len(chunks) >= 2) + + +class TestFastThaiG2PStream(unittest.TestCase): + """Test FastThaiG2P streaming synthesis methods.""" + + @patch('onnxruntime.InferenceSession') + @patch('pythaitts.pretrained.fastthaig2p.tts._default_asset') + def setUp(self, mock_asset, mock_session): + import tempfile, json + # Dummy assets + self.temp_files = [] + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as fp: + json.dump({"vocab": {"a": 1, "b": 2}}, fp) + self.dummy_config = fp.name + self.temp_files.append(fp.name) + with tempfile.NamedTemporaryFile(suffix=".npy", delete=False) as fp: + np.save(fp.name, np.zeros((510, 1, 256), dtype=np.float32)) + self.dummy_voice = fp.name + self.temp_files.append(fp.name) + with tempfile.NamedTemporaryFile(suffix=".onnx", delete=False) as fp: + self.dummy_onnx = fp.name + self.temp_files.append(fp.name) + + from pythaitts.pretrained.fastthaig2p import FastThaiG2P + self.model = FastThaiG2P( + model_path=self.dummy_onnx, + voicepack_path=self.dummy_voice, + config_path=self.dummy_config, + ) + + def tearDown(self): + for f in self.temp_files: + if os.path.exists(f): + os.unlink(f) + + def test_stream_waveform_return(self): + dummy_audio = np.array([0.1, -0.1, 0.2], dtype=np.float32) + self.model.generate = Mock(return_value=dummy_audio) + + chunks = list(self.model.stream("สวัสดีครับ วันนี้อากาศดี", return_type="waveform")) + self.assertTrue(len(chunks) >= 1) + for chunk in chunks: + self.assertIsInstance(chunk, np.ndarray) + self.assertTrue(np.array_equal(chunk, dummy_audio)) + + def test_stream_bytes_return(self): + dummy_audio = np.array([0.5, -0.5], dtype=np.float32) + self.model.generate = Mock(return_value=dummy_audio) + + chunks = list(self.model.stream("สวัสดีครับ", return_type="bytes")) + self.assertTrue(len(chunks) >= 1) + for chunk in chunks: + self.assertIsInstance(chunk, bytes) + self.assertEqual(len(chunk), 4) # 2 samples * 2 bytes + + def test_stream_file_return(self): + dummy_audio = np.zeros(2400, dtype=np.float32) + self.model.generate = Mock(return_value=dummy_audio) + + chunks = list(self.model.stream("สวัสดีครับ", return_type="file")) + self.assertTrue(len(chunks) >= 1) + for fpath in chunks: + self.assertTrue(os.path.exists(fpath)) + self.assertTrue(fpath.endswith(".wav")) + os.unlink(fpath) + + def test_stream_from_generator(self): + dummy_audio = np.array([0.1], dtype=np.float32) + self.model.generate = Mock(return_value=dummy_audio) + + def token_gen(): + yield "สวัสดี" + yield "ครับ" + + chunks = list(self.model.stream(token_gen(), return_type="waveform")) + self.assertTrue(len(chunks) >= 1) + + def test_stream_invalid_speaker(self): + with self.assertRaises(ValueError): + list(self.model.stream("สวัสดี", speaker_idx="unknown_voice")) + + def test_stream_speaker_mapping(self): + self.model.generate = Mock(return_value=np.zeros(10, dtype=np.float32)) + list(self.model.stream("สวัสดี", speaker_idx="Linda")) + list(self.model.stream("สวัสดี", speaker_idx=None)) + + def test_stream_invalid_return_type(self): + self.model.generate = Mock(return_value=np.zeros(10, dtype=np.float32)) + with self.assertRaises(ValueError): + list(self.model.stream("สวัสดี", return_type="invalid_type")) + + +class TestTTSIntegration(unittest.TestCase): + """Test TTS.stream integration in pythaitts/__init__.py""" + + @patch('pythaitts.pretrained.fastthaig2p.FastThaiG2P') + def test_tts_stream_delegates_to_fastthaig2p(self, mock_fastthaig2p_cls): + mock_instance = Mock() + mock_instance.stream.return_value = iter([np.zeros(10, dtype=np.float32)]) + mock_fastthaig2p_cls.return_value = mock_instance + + tts = TTS(pretrained="fastthaig2p") + chunks = list(tts.stream("สวัสดีครับ")) + mock_instance.stream.assert_called_once() + self.assertEqual(len(chunks), 1) + + @patch('pythaitts.pretrained.fastthaig2p.FastThaiG2P') + def test_tts_stream_alias(self, mock_fastthaig2p_cls): + mock_instance = Mock() + mock_instance.stream.return_value = iter([np.zeros(10, dtype=np.float32)]) + mock_fastthaig2p_cls.return_value = mock_instance + + tts = TTS(pretrained="fastthaig2p") + self.assertEqual(tts.stream, tts.tts_stream) + chunks = list(tts.tts_stream("สวัสดีครับ")) + mock_instance.stream.assert_called_once() + + +class TestRealtimeTTSClass(unittest.TestCase): + """Test RealtimeTTS class.""" + + @patch('pythaitts.TTS') + def test_realtimetts_init_and_stream(self, mock_tts_cls): + mock_tts_instance = Mock() + mock_tts_instance.stream.return_value = iter([np.zeros(20, dtype=np.float32)]) + mock_tts_cls.return_value = mock_tts_instance + + rt = RealtimeTTS(pretrained="fastthaig2p", speed=1.2) + chunks = list(rt.stream("สวัสดีครับ")) + self.assertEqual(len(chunks), 1) + mock_tts_instance.stream.assert_called_once() + call_kwargs = mock_tts_instance.stream.call_args.kwargs + self.assertEqual(call_kwargs['speed'], 1.2) + self.assertEqual(call_kwargs['speaker_idx'], "thai_som") + + @patch('pythaitts.TTS') + def test_realtimetts_synthesize(self, mock_tts_cls): + mock_tts_instance = Mock() + mock_tts_instance.tts.return_value = "output.wav" + mock_tts_cls.return_value = mock_tts_instance + + rt = RealtimeTTS() + out = rt.synthesize("สวัสดีครับ", filename="test.wav") + self.assertEqual(out, "output.wav") + mock_tts_instance.tts.assert_called_once_with("สวัสดีครับ", speaker_idx="thai_som", filename="test.wav") + + +class TestFastThaiG2PEngine(unittest.TestCase): + """Test FastThaiG2PEngine for RealtimeTTS library compatibility.""" + + @patch('pythaitts.pretrained.fastthaig2p.tts.FastThaiG2P') + def test_engine_init_and_stream_info(self, mock_fastthaig2p_cls): + mock_instance = Mock() + mock_instance.sample_rate = 24000 + mock_instance.SUPPORTED_VOICES = ["thai_som"] + mock_fastthaig2p_cls.return_value = mock_instance + + engine = FastThaiG2PEngine() + self.assertEqual(engine.engine_name, "fastthaig2p") + fmt, channels, rate = engine.get_stream_info() + self.assertEqual(channels, 1) + self.assertEqual(rate, 24000) + + @patch('pythaitts.pretrained.fastthaig2p.tts.FastThaiG2P') + def test_engine_voices(self, mock_fastthaig2p_cls): + mock_instance = Mock() + mock_instance.sample_rate = 24000 + mock_instance.SUPPORTED_VOICES = ["thai_som"] + mock_fastthaig2p_cls.return_value = mock_instance + + engine = FastThaiG2PEngine() + voices = engine.get_voices() + self.assertEqual(len(voices), 1) + self.assertEqual(str(voices[0]), "thai_som") + + engine.set_voice("thai_som") + self.assertEqual(engine.current_voice, "thai_som") + + with self.assertRaises(ValueError): + engine.set_voice("unsupported_voice") + + @patch('pythaitts.pretrained.fastthaig2p.tts.FastThaiG2P') + def test_engine_synthesize_puts_to_queue(self, mock_fastthaig2p_cls): + mock_instance = Mock() + mock_instance.sample_rate = 24000 + mock_instance.SUPPORTED_VOICES = ["thai_som"] + mock_instance._text_to_phonemes.return_value = "sa.wat.di" + # 100 samples float32 + dummy_audio = np.ones(100, dtype=np.float32) * 0.5 + mock_instance.generate.return_value = dummy_audio + mock_fastthaig2p_cls.return_value = mock_instance + + engine = FastThaiG2PEngine() + success = engine.synthesize("สวัสดีครับ") + self.assertTrue(success) + self.assertFalse(engine.queue.empty()) + pcm_chunk = engine.queue.get() + self.assertEqual(len(pcm_chunk), 200) # 100 samples * 2 bytes + + @patch('pythaitts.pretrained.fastthaig2p.tts.FastThaiG2P') + def test_engine_stop(self, mock_fastthaig2p_cls): + mock_instance = Mock() + mock_instance.sample_rate = 24000 + mock_instance.SUPPORTED_VOICES = ["thai_som"] + mock_fastthaig2p_cls.return_value = mock_instance + + engine = FastThaiG2PEngine() + self.assertFalse(engine.stop_synthesis_event.is_set()) + engine.stop() + self.assertTrue(engine.stop_synthesis_event.is_set()) + + +if __name__ == '__main__': + unittest.main() From 6915becee4f859b30c481e7633b887a4939d0371 Mon Sep 17 00:00:00 2001 From: Wannaphong Date: Sat, 3 Oct 2026 17:15:44 +0700 Subject: [PATCH 2/2] Fix tests --- pythaitts/realtime.py | 37 +++++++++++++++++++++++++++++++++---- tests/test_realtime.py | 8 ++++++++ 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/pythaitts/realtime.py b/pythaitts/realtime.py index 95552f5..be02af8 100644 --- a/pythaitts/realtime.py +++ b/pythaitts/realtime.py @@ -125,14 +125,32 @@ def chunk_text( text = preprocess_text(str(text)) + sentences = [] try: from pythainlp.tokenize import sent_tokenize - sentences = sent_tokenize(text, engine="crfcut") + try: + sentences = sent_tokenize(text, engine="crfcut") + except Exception: + sentences = sent_tokenize(text, engine="whitespace+newline") except Exception: + pass + + if not sentences or len(sentences) <= 1: + if " " in text or "\n" in text: + try: + from pythainlp.tokenize import sent_tokenize + + ws_sents = sent_tokenize(text, engine="whitespace+newline") + if len(ws_sents) > 1: + sentences = ws_sents + except Exception: + pass + + if not sentences: sentences = [s.strip() for s in text.splitlines() if s.strip()] - if not sentences: - sentences = [text] + if not sentences: + sentences = [text] if g2p_converter is None: from pythaitts.pretrained.fastthaig2p import G2P, ipa_to_kokoro @@ -250,7 +268,18 @@ def stream_text_to_chunks( try: sents = sent_tokenize(buffer, engine="crfcut") except Exception: - sents = [buffer] + try: + sents = sent_tokenize(buffer, engine="whitespace+newline") + except Exception: + sents = [buffer] + + if len(sents) <= 1 and (" " in buffer or "\n" in buffer): + try: + ws_sents = sent_tokenize(buffer, engine="whitespace+newline") + if len(ws_sents) > 1: + sents = ws_sents + except Exception: + pass if len(sents) > 1: for s in sents[:-1]: diff --git a/tests/test_realtime.py b/tests/test_realtime.py index b4e4f07..746189e 100644 --- a/tests/test_realtime.py +++ b/tests/test_realtime.py @@ -34,6 +34,14 @@ def test_chunk_text_multiple_sentences(self): chunks = chunk_text(text) self.assertTrue(len(chunks) >= 2) + def test_chunk_text_without_pycrfsuite(self): + import sys + # Simulate environment where python-crfsuite is not installed + with patch.dict(sys.modules, {"pycrfsuite": None}): + text = "สวัสดีครับ วันนี้อากาศดีมาก เราไปเที่ยวกันเถอะ" + chunks = chunk_text(text) + self.assertTrue(len(chunks) >= 2) + def test_chunk_text_preprocessing(self): text = "มี 5 คนๆ" chunks_preprocessed = chunk_text(text, preprocess=True)