diff --git a/README.md b/README.md index 1be57a2..fc42a19 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,16 @@ from pythaitts import TTS tts = TTS(pretrained="fastthaig2p") file = tts.tts("สวัสดีครับ", speaker_idx="thai_som", filename="output.wav") +# Use MakHamTTS Paradee (standalone, single-voice 8MB quantized INT8 ONNX model, offline without download) +# MakHamTTS Paradee Sample Rate is 24000 Hz +tts = TTS(pretrained="makham_paradee") +file = tts.tts("สวัสดีค่ะ", filename="output.wav") + +# Use MakHamTTS (Kokoro-82M Thai base model, voices: newvoice_female, newvoice_male) +# MakHamTTS Sample Rate is 24000 Hz +tts = TTS(pretrained="makham") +file = tts.tts("สวัสดีค่ะ", speaker_idx="newvoice_female", filename="output.wav") + # Use Lunarlist ONNX # Sample Rate is 22050 Hz tts = TTS(pretrained="lunarlist_onnx") diff --git a/demo_realtime.py b/demo_realtime.py index 165142d..98bf8c9 100644 --- a/demo_realtime.py +++ b/demo_realtime.py @@ -23,7 +23,7 @@ def main(): # 1. Initialize TTS model print("Initializing FastThaiG2P TTS model...") t_start = time.time() - tts = TTS(pretrained="fastthaig2p") + tts = TTS(pretrained="makham_paradee") print(f"✓ Model loaded in {time.time() - t_start:.2f}s") print() diff --git a/docs/index.rst b/docs/index.rst index 405f8eb..d90b80f 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -40,6 +40,16 @@ PyThaiTTS provides English-to-Thai-IPA Grapheme-to-Phoneme (G2P) conversion for .. autofunction:: en_ipa_to_thai_ipa +MakHamTTS Models +---------------- +MakHamTTS models powered by Kokoro-82M Thai and Paradee 1.1 distillation. + +.. autoclass:: MakHamTTS + :members: + +.. autoclass:: MakHamTTSParadee + :members: + FastThaiG2P Components ---------------------- diff --git a/example_realtime.py b/example_realtime.py index e9b7729..9f3b9ed 100644 --- a/example_realtime.py +++ b/example_realtime.py @@ -14,7 +14,7 @@ def main(): print("-" * 50) # Initialize FastThaiG2P TTS - tts = TTS(pretrained="fastthaig2p") + tts = TTS(pretrained="makham_paradee") # 1. Stream speech from text string chunk-by-chunk text = "สวัสดีครับ ยินดีต้อนรับสู่ระบบสังเคราะห์เสียงภาษาไทยแบบเรียลไทม์" diff --git a/output.wav b/output.wav new file mode 100644 index 0000000..809de7c Binary files /dev/null and b/output.wav differ diff --git a/pythaitts/__init__.py b/pythaitts/__init__.py index 4b838fd..fe371e8 100644 --- a/pythaitts/__init__.py +++ b/pythaitts/__init__.py @@ -13,6 +13,10 @@ expand_maiyamok, load_custom_dict, ) +from pythaitts.pretrained.makham import ( + MakHamTTS, + MakHamTTSParadee, +) from pythaitts.pretrained.fastthaig2p.en_g2p import ( EnglishG2P, en_g2p, @@ -40,10 +44,10 @@ def __init__( **kwargs, ) -> None: """ - :param str pretrained: TTS pretrained (fastthaig2p, lunarlist_onnx, khanomtan, lunarlist, vachana) - :param str mode: pretrained mode (fastthaig2p, lunarlist_onnx, and vachana don't support) + :param str pretrained: TTS pretrained (fastthaig2p, makham, makham_paradee, lunarlist_onnx, khanomtan, lunarlist, vachana) + :param str mode: pretrained mode (fastthaig2p, makham, makham_paradee, lunarlist_onnx, and vachana don't support) :param str version: model version (default is 1.0 or 1.1) - :param str device: device for running model. (fastthaig2p, lunarlist_onnx, and vachana support CPU only.) + :param str device: device for running model. (fastthaig2p, makham, makham_paradee, lunarlist_onnx, and vachana support CPU only.) :param custom_dict: Optional custom word dictionary for custom Thai pronunciation in words **Options for mode** @@ -65,6 +69,9 @@ def __init__( For fastthaig2p tts model, \ You can see more about fastthaig2p at `https://github.com/awslabs/FastThaiG2P `_ + + For makham and makham_paradee tts models, \ + You can see more about MakHamTTS at `https://huggingface.co/pythainlp/MakHamTTS `_ """ self.pretrained = pretrained self.mode = mode @@ -93,6 +100,12 @@ def load_pretrained(self, version="1.0", **kwargs): elif self.pretrained in ("fastthaig2p", "FastThaiG2P"): from pythaitts.pretrained.fastthaig2p import FastThaiG2P self.model = FastThaiG2P(device=self.device, **kwargs) + elif self.pretrained in ("makham", "MakHamTTS"): + from pythaitts.pretrained.makham import MakHamTTS + self.model = MakHamTTS(**kwargs) + elif self.pretrained in ("makham_paradee", "MakHamTTS_paradee"): + from pythaitts.pretrained.makham import MakHamTTSParadee + self.model = MakHamTTSParadee(**kwargs) else: raise NotImplementedError( "PyThaiTTS doesn't support %s pretrained." % self.pretrained @@ -169,7 +182,7 @@ def tts( if self.pretrained == "lunarlist" or self.pretrained == "lunarlist_onnx": return self.model(text=text, return_type=return_type, filename=filename) elif self.pretrained == "vachana": - if speaker_idx in ("thai_som", "Linda", None): + if speaker_idx in ("thai_som", "Linda", "newvoice_female", None): speaker_idx = "th_f_1" return self.model( text=text, @@ -179,7 +192,7 @@ def tts( **kwargs, ) elif self.pretrained in ("fastthaig2p", "FastThaiG2P"): - if speaker_idx in ("Linda", None): + if speaker_idx in ("Linda", "newvoice_female", None): speaker_idx = "thai_som" call_kwargs = dict(kwargs) if effective_custom_dict is not None: @@ -191,7 +204,31 @@ def tts( filename=filename, **call_kwargs, ) - if speaker_idx in ("thai_som", None): + elif self.pretrained in ("makham", "MakHamTTS"): + if speaker_idx in ("Linda", "thai_som", None): + speaker_idx = "newvoice_female" + call_kwargs = dict(kwargs) + if effective_custom_dict is not None: + call_kwargs["custom_dict"] = effective_custom_dict + return self.model( + text=text, + speaker_idx=speaker_idx, + return_type=return_type, + filename=filename, + **call_kwargs, + ) + elif self.pretrained in ("makham_paradee", "MakHamTTS_paradee"): + call_kwargs = dict(kwargs) + if effective_custom_dict is not None: + call_kwargs["custom_dict"] = effective_custom_dict + return self.model( + text=text, + speaker_idx=speaker_idx, + return_type=return_type, + filename=filename, + **call_kwargs, + ) + if speaker_idx in ("thai_som", "newvoice_female", None): speaker_idx = "Linda" return self.model( text=text, @@ -230,7 +267,7 @@ def stream( ) if self.pretrained in ("fastthaig2p", "FastThaiG2P"): - if speaker_idx in ("Linda", None): + if speaker_idx in ("Linda", "newvoice_female", None): speaker_idx = "thai_som" return self.model.stream( text=text, @@ -242,6 +279,30 @@ def stream( custom_dict=effective_custom_dict, **kwargs, ) + elif self.pretrained in ("makham", "MakHamTTS"): + if speaker_idx in ("Linda", "thai_som", None): + speaker_idx = "newvoice_female" + return self.model.stream( + text=text, + speaker_idx=speaker_idx, + return_type=return_type, + play=play, + preprocess=preprocess, + max_phonemes=max_phonemes, + custom_dict=effective_custom_dict, + **kwargs, + ) + elif self.pretrained in ("makham_paradee", "MakHamTTS_paradee"): + return self.model.stream( + text=text, + speaker_idx=speaker_idx, + return_type=return_type, + play=play, + preprocess=preprocess, + max_phonemes=max_phonemes, + custom_dict=effective_custom_dict, + **kwargs, + ) else: from pythaitts.realtime import stream_text_to_chunks @@ -276,6 +337,8 @@ def _gen(): "en_g2p", "en_to_ipa", "en_ipa_to_thai_ipa", + "MakHamTTS", + "MakHamTTSParadee", "preprocess_text", "load_custom_dict", "num_to_thai", diff --git a/pythaitts/pretrained/makham/__init__.py b/pythaitts/pretrained/makham/__init__.py new file mode 100644 index 0000000..983bc1b --- /dev/null +++ b/pythaitts/pretrained/makham/__init__.py @@ -0,0 +1,8 @@ +# -*- coding: utf-8 -*- +""" +MakHamTTS Models +""" +from pythaitts.pretrained.makham.makham_tts import MakHamTTS +from pythaitts.pretrained.makham.paradee_tts import MakHamTTSParadee + +__all__ = ["MakHamTTS", "MakHamTTSParadee"] diff --git a/pythaitts/pretrained/makham/data/config.json b/pythaitts/pretrained/makham/data/config.json new file mode 100644 index 0000000..719423b --- /dev/null +++ b/pythaitts/pretrained/makham/data/config.json @@ -0,0 +1,177 @@ +{ + "istftnet": { + "upsample_kernel_sizes": [ + 20, + 12 + ], + "upsample_rates": [ + 10, + 6 + ], + "gen_istft_hop_size": 5, + "gen_istft_n_fft": 20, + "resblock_dilation_sizes": [ + [ + 1, + 3, + 5 + ], + [ + 1, + 3, + 5 + ], + [ + 1, + 3, + 5 + ] + ], + "resblock_kernel_sizes": [ + 3, + 7, + 11 + ], + "upsample_initial_channel": 512 + }, + "dim_in": 64, + "dropout": 0.2, + "hidden_dim": 512, + "max_conv_dim": 512, + "max_dur": 50, + "multispeaker": true, + "n_layer": 3, + "n_mels": 80, + "n_token": 178, + "style_dim": 128, + "text_encoder_kernel_size": 5, + "plbert": { + "hidden_size": 768, + "num_attention_heads": 12, + "intermediate_size": 2048, + "max_position_embeddings": 512, + "num_hidden_layers": 12, + "dropout": 0.1 + }, + "vocab": { + ";": 1, + ":": 2, + ",": 3, + ".": 4, + "!": 5, + "?": 6, + "—": 9, + "…": 10, + "\"": 11, + "(": 12, + ")": 13, + "“": 14, + "”": 15, + " ": 16, + "̃": 17, + "ʣ": 18, + "ʥ": 19, + "ʦ": 20, + "ʨ": 21, + "ᵝ": 22, + "ꭧ": 23, + "A": 24, + "I": 25, + "O": 31, + "Q": 33, + "S": 35, + "T": 36, + "W": 39, + "Y": 41, + "ᵊ": 42, + "a": 43, + "b": 44, + "c": 45, + "d": 46, + "e": 47, + "f": 48, + "h": 50, + "i": 51, + "j": 52, + "k": 53, + "l": 54, + "m": 55, + "n": 56, + "o": 57, + "p": 58, + "q": 59, + "r": 60, + "s": 61, + "t": 62, + "u": 63, + "v": 64, + "w": 65, + "x": 66, + "y": 67, + "z": 68, + "ɑ": 69, + "ɐ": 70, + "ɒ": 71, + "æ": 72, + "β": 75, + "ɔ": 76, + "ɕ": 77, + "ç": 78, + "ɖ": 80, + "ð": 81, + "ʤ": 82, + "ə": 83, + "ɚ": 85, + "ɛ": 86, + "ɜ": 87, + "ɟ": 90, + "ɡ": 92, + "ɥ": 99, + "ɨ": 101, + "ɪ": 102, + "ʝ": 103, + "ɯ": 110, + "ɰ": 111, + "ŋ": 112, + "ɳ": 113, + "ɲ": 114, + "ɴ": 115, + "ø": 116, + "ɸ": 118, + "θ": 119, + "œ": 120, + "ɹ": 123, + "ɾ": 125, + "ɻ": 126, + "ʁ": 128, + "ɽ": 129, + "ʂ": 130, + "ʃ": 131, + "ʈ": 132, + "ʧ": 133, + "ʊ": 135, + "ʋ": 136, + "ʌ": 138, + "ɣ": 139, + "ɤ": 140, + "χ": 142, + "ʎ": 143, + "ʒ": 147, + "ʔ": 148, + "ˈ": 156, + "ˌ": 157, + "ː": 158, + "ʰ": 162, + "ʲ": 164, + "↓": 169, + "→": 171, + "↗": 172, + "↘": 173, + "ᵻ": 177, + "↑": 170 + }, + "language": "th", + "model_type": "paradee-1.1", + "num_parameters": 7877918, + "architecture": "Paradee 1.1 (TextStudent s+mlp + VDecoder vhead v2)" +} \ No newline at end of file diff --git a/pythaitts/pretrained/makham/data/paradee_thai_int8.onnx b/pythaitts/pretrained/makham/data/paradee_thai_int8.onnx new file mode 100644 index 0000000..34e69f1 Binary files /dev/null and b/pythaitts/pretrained/makham/data/paradee_thai_int8.onnx differ diff --git a/pythaitts/pretrained/makham/makham_tts.py b/pythaitts/pretrained/makham/makham_tts.py new file mode 100644 index 0000000..539f0d4 --- /dev/null +++ b/pythaitts/pretrained/makham/makham_tts.py @@ -0,0 +1,269 @@ +# -*- coding: utf-8 -*- +""" +MakHamTTS +""" +from typing import Optional, Union, Iterable +from pathlib import Path +import os +import json +import wave +import numpy as np +import onnxruntime as ort +from huggingface_hub import hf_hub_download + +from pythaitts.pretrained.fastthaig2p.g2p import G2P +from pythaitts.pretrained.fastthaig2p.kokoro import ipa_to_kokoro +from pythaitts.pretrained.fastthaig2p.tts import _trim_boundary_tokens, _load_voicepack + +_HF_REPO = "pythainlp/MakHamTTS" + + +class MakHamTTS: + """MakHamTTS: Thai TTS powered by Kokoro-82M Thai and MakHamTTS voices. + + Downloads model, voicepack, and config from pythainlp/MakHamTTS on first use. + """ + + MAX_PHONEMES = 510 + SUPPORTED_VOICES = ["newvoice_female", "newvoice_male"] + + def __init__( + self, + model_path: Optional[Union[str, Path]] = None, + voicepack_path: Optional[Union[str, Path]] = None, + config_path: Optional[Union[str, Path]] = None, + speed: float = 1.0, + intra_op_threads: int = 0, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + speaker_idx: str = "newvoice_female", + **kwargs, + ): + if model_path is None: + model_path = hf_hub_download( + repo_id=_HF_REPO, + filename="fastthaig2p_model/kokoro_thai.onnx", + ) + if config_path is None: + config_path = hf_hub_download( + repo_id=_HF_REPO, + filename="fastthaig2p_model/config.json", + ) + + model_path = Path(model_path) + if not model_path.exists(): + raise FileNotFoundError(f"Not found: {model_path}") + + options = ort.SessionOptions() + if intra_op_threads: + options.intra_op_num_threads = intra_op_threads + self._session = ort.InferenceSession( + str(model_path), options, providers=["CPUExecutionProvider"] + ) + + with open(config_path, encoding="utf-8") as f: + self._vocab = json.load(f)["vocab"] + + self._speaker_idx = speaker_idx + self._voicepacks = {} + if voicepack_path is not None: + vp = Path(voicepack_path) + if not vp.exists(): + raise FileNotFoundError(f"Not found: {vp}") + self._voicepacks[speaker_idx] = _load_voicepack(vp) + else: + self._get_voicepack(speaker_idx) + + self._g2p = G2P(custom_dict=custom_dict) + self.sample_rate = 24000 + self.speed = speed + + def _get_voicepack(self, speaker_idx: str) -> np.ndarray: + if speaker_idx in self._voicepacks: + return self._voicepacks[speaker_idx] + if speaker_idx in self.SUPPORTED_VOICES: + vp_path = hf_hub_download( + repo_id=_HF_REPO, + filename=f"fastthaig2p_model/{speaker_idx}.npy", + ) + vp = _load_voicepack(Path(vp_path)) + self._voicepacks[speaker_idx] = vp + return vp + if os.path.exists(speaker_idx): + vp = _load_voicepack(Path(speaker_idx)) + self._voicepacks[speaker_idx] = vp + return vp + raise ValueError( + f"Unsupported voice '{speaker_idx}'. Supported voices are: {', '.join(self.SUPPORTED_VOICES)}" + ) + + def add_custom_dict( + self, custom_dict: Union[dict, str, Path, Iterable] + ) -> None: + """Add custom word pronunciations to the underlying G2P engine.""" + self._g2p.add_custom_dict(custom_dict) + + def set_custom_dict( + self, custom_dict: Union[dict, str, Path, Iterable] + ) -> None: + """Replace all custom word pronunciations in the underlying G2P engine.""" + self._g2p.set_custom_dict(custom_dict) + + def _text_to_phonemes( + self, + text: str, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + ) -> str: + return ipa_to_kokoro(self._g2p.convert(text, custom_dict=custom_dict)) + + def generate( + self, + text: str, + speaker_idx: Optional[str] = None, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + ) -> np.ndarray: + """Convert Thai text to float32 audio array (24 kHz mono).""" + speaker = speaker_idx or self._speaker_idx + voice = self._get_voicepack(speaker) + + phonemes = self._text_to_phonemes(text, custom_dict=custom_dict) + if not phonemes: + return np.array([], dtype=np.float32) + if len(phonemes) > self.MAX_PHONEMES: + raise ValueError( + f"Text produces {len(phonemes)} phonemes; max {self.MAX_PHONEMES}. " + "Split the input into shorter chunks." + ) + ref_s = voice[len(phonemes) - 1] + + ids = [self._vocab[c] for c in phonemes if c in self._vocab] + input_ids = np.array([[0, *ids, 0]], dtype=np.int64) + ref = np.asarray(ref_s, dtype=np.float32) + audio, durations = self._session.run( + None, + { + "input_ids": input_ids, + "ref_s": ref, + "speed": np.full(1, self.speed, dtype=np.float32), + }, + ) + return _trim_boundary_tokens(audio, durations) + + def synthesize( + self, + text: str, + output_path: Union[str, Path], + speaker_idx: Optional[str] = None, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + ) -> Path: + """Convert Thai text to WAV file.""" + output_path = Path(output_path) + audio = self.generate(text, speaker_idx=speaker_idx, custom_dict=custom_dict) + audio_int16 = (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16) + with wave.open(str(output_path), "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(self.sample_rate) + wf.writeframes(audio_int16.tobytes()) + return output_path + + def __call__( + self, + text: str, + speaker_idx: Optional[str] = "newvoice_female", + return_type: str = "file", + filename: Optional[str] = None, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + **kwargs, + ): + """Generate speech from text using MakHamTTS. + + :param str text: Input text to synthesize + :param str speaker_idx: Voice to use ('newvoice_female' or 'newvoice_male', or voicepack path) + :param str return_type: Return type ('file' or 'waveform') + :param str filename: Output filename for the generated audio + :param custom_dict: Optional custom word dictionary + :param kwargs: Additional parameters (e.g. speed) + :return: File path if return_type is 'file', otherwise audio waveform (np.ndarray) + """ + if speaker_idx in ("Linda", "thai_som", None): + speaker_idx = "newvoice_female" + + if "speed" in kwargs: + self.speed = kwargs["speed"] + + if return_type == "waveform": + return self.generate(text, speaker_idx=speaker_idx, custom_dict=custom_dict) + else: + import tempfile + + if filename is None: + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as fp: + filename = fp.name + + self.synthesize(text, filename, speaker_idx=speaker_idx, custom_dict=custom_dict) + return str(filename) + + def stream( + self, + text: Union[str, Iterable[str]], + speaker_idx: Optional[str] = "newvoice_female", + return_type: str = "waveform", + play: bool = False, + preprocess: bool = True, + max_phonemes: int = 400, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + **kwargs, + ): + """Stream Thai speech generation chunk by chunk in real-time.""" + if speaker_idx in ("Linda", "thai_som", None): + speaker_idx = "newvoice_female" + + 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=lambda t: self._text_to_phonemes(t, custom_dict=custom_dict), + ) + for chunk in chunks: + audio = self.generate(chunk, speaker_idx=speaker_idx, custom_dict=custom_dict) + 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: + self.speed = orig_speed + if player is not None: + player.close() + + tts_stream = stream diff --git a/pythaitts/pretrained/makham/paradee_tts.py b/pythaitts/pretrained/makham/paradee_tts.py new file mode 100644 index 0000000..a119a33 --- /dev/null +++ b/pythaitts/pretrained/makham/paradee_tts.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +""" +MakHamTTS Paradee model +""" +from typing import Optional, Union, Iterable +from pathlib import Path +import os +import json +import wave +import numpy as np +import onnxruntime as ort + +from pythaitts.pretrained.fastthaig2p.g2p import G2P +from pythaitts.pretrained.fastthaig2p.kokoro import ipa_to_kokoro + +_PARADEE_DIR = Path(__file__).parent / "data" +_DEFAULT_MODEL_PATH = _PARADEE_DIR / "paradee_thai_int8.onnx" +_DEFAULT_CONFIG_PATH = _PARADEE_DIR / "config.json" + + +class MakHamTTSParadee: + """MakHamTTS Paradee: Standalone, single-voice Thai ONNX TTS model (female voice). + + This model runs locally using the included quantized INT8 ONNX file without download. + """ + + MAX_PHONEMES = 510 + + def __init__( + self, + model_path: Optional[Union[str, Path]] = None, + config_path: Optional[Union[str, Path]] = None, + speed: float = 1.0, + intra_op_threads: int = 0, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + **kwargs, + ): + model_path = Path(model_path) if model_path else _DEFAULT_MODEL_PATH + config_path = Path(config_path) if config_path else _DEFAULT_CONFIG_PATH + + if not model_path.exists(): + raise FileNotFoundError(f"Paradee ONNX model not found: {model_path}") + if not config_path.exists(): + raise FileNotFoundError(f"Paradee config not found: {config_path}") + + options = ort.SessionOptions() + if intra_op_threads: + options.intra_op_num_threads = intra_op_threads + self._session = ort.InferenceSession( + str(model_path), options, providers=["CPUExecutionProvider"] + ) + + with open(config_path, encoding="utf-8") as f: + self._vocab = json.load(f)["vocab"] + + self._g2p = G2P(custom_dict=custom_dict) + self.sample_rate = 24000 + self.speed = speed + + def add_custom_dict( + self, custom_dict: Union[dict, str, Path, Iterable] + ) -> None: + """Add custom word pronunciations to the underlying G2P engine.""" + self._g2p.add_custom_dict(custom_dict) + + def set_custom_dict( + self, custom_dict: Union[dict, str, Path, Iterable] + ) -> None: + """Replace all custom word pronunciations in the underlying G2P engine.""" + self._g2p.set_custom_dict(custom_dict) + + def _text_to_phonemes( + self, + text: str, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + ) -> str: + return ipa_to_kokoro(self._g2p.convert(text, custom_dict=custom_dict)) + + def generate( + self, + text: str, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + ) -> np.ndarray: + """Convert Thai text to float32 audio array (24 kHz mono).""" + phonemes = self._text_to_phonemes(text, custom_dict=custom_dict) + if not phonemes: + return np.array([], dtype=np.float32) + if len(phonemes) > self.MAX_PHONEMES: + raise ValueError( + f"Text produces {len(phonemes)} phonemes; max {self.MAX_PHONEMES}. " + "Split the input into shorter chunks." + ) + + ids = [0] + [self._vocab[c] for c in phonemes if c in self._vocab] + [0] + audio = self._session.run( + None, + { + "input_ids": np.array([ids], dtype=np.int64), + "speed": np.array([self.speed], dtype=np.float32), + }, + )[0][0] + return audio + + def synthesize( + self, + text: str, + output_path: Union[str, Path], + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + ) -> Path: + """Convert Thai text to WAV file.""" + output_path = Path(output_path) + audio = self.generate(text, custom_dict=custom_dict) + audio_int16 = (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16) + with wave.open(str(output_path), "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(self.sample_rate) + wf.writeframes(audio_int16.tobytes()) + return output_path + + def __call__( + self, + text: str, + speaker_idx: Optional[str] = None, + return_type: str = "file", + filename: Optional[str] = None, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + **kwargs, + ): + """Generate speech from text using MakHamTTS Paradee. + + :param str text: Input text to synthesize + :param str speaker_idx: Unused (single voice) + :param str return_type: Return type ('file' or 'waveform') + :param str filename: Output filename for the generated audio + :param custom_dict: Optional custom word dictionary + :param kwargs: Additional parameters (e.g. speed) + :return: File path if return_type is 'file', otherwise audio waveform (np.ndarray) + """ + if "speed" in kwargs: + self.speed = kwargs["speed"] + + if return_type == "waveform": + return self.generate(text, custom_dict=custom_dict) + else: + import tempfile + + if filename is None: + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as fp: + filename = fp.name + + self.synthesize(text, filename, custom_dict=custom_dict) + return str(filename) + + 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, + custom_dict: Optional[Union[dict, str, Path, Iterable]] = None, + **kwargs, + ): + """Stream Thai speech generation chunk by chunk in real-time.""" + 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=lambda t: self._text_to_phonemes(t, custom_dict=custom_dict), + ) + for chunk in chunks: + audio = self.generate(chunk, custom_dict=custom_dict) + 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: + self.speed = orig_speed + if player is not None: + player.close() + + tts_stream = stream diff --git a/setup.py b/setup.py index 8cd6564..db31033 100644 --- a/setup.py +++ b/setup.py @@ -19,6 +19,7 @@ packages=find_packages(), package_data={ "pythaitts.pretrained.fastthaig2p": ["data/*", "data/fallback/*"], + "pythaitts.pretrained.makham": ["data/*"], }, include_package_data=True, python_requires=">=3.6", diff --git a/tests/test_makham.py b/tests/test_makham.py new file mode 100644 index 0000000..f12a3bc --- /dev/null +++ b/tests/test_makham.py @@ -0,0 +1,88 @@ +# -*- coding: utf-8 -*- +""" +Unit tests for MakHamTTS and MakHamTTS Paradee models +""" +import os +import tempfile +import unittest +import numpy as np + +from pythaitts import TTS, MakHamTTS, MakHamTTSParadee + + +class TestMakHamTTS(unittest.TestCase): + """Test MakHamTTS and MakHamTTSParadee models""" + + def test_paradee_direct_init(self): + tts = MakHamTTSParadee() + self.assertEqual(tts.sample_rate, 24000) + self.assertEqual(tts.speed, 1.0) + audio = tts.generate("สวัสดีครับ") + self.assertIsInstance(audio, np.ndarray) + self.assertGreater(len(audio), 0) + + def test_paradee_empty_text(self): + tts = MakHamTTSParadee() + audio = tts.generate("") + self.assertEqual(len(audio), 0) + + def test_paradee_file_output(self): + tts = MakHamTTSParadee() + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as fp: + temp_path = fp.name + try: + out_file = tts(text="สวัสดีครับ", return_type="file", filename=temp_path) + self.assertEqual(out_file, temp_path) + self.assertTrue(os.path.exists(temp_path)) + self.assertGreater(os.path.getsize(temp_path), 44) + finally: + if os.path.exists(temp_path): + os.remove(temp_path) + + def test_paradee_tts_wrapper(self): + tts = TTS(pretrained="makham_paradee") + wav = tts.tts("สวัสดีครับ", return_type="waveform") + self.assertIsInstance(wav, np.ndarray) + self.assertGreater(len(wav), 0) + + tts_alias = TTS(pretrained="MakHamTTS_paradee") + wav_alias = tts_alias.tts("สวัสดีครับ", return_type="waveform") + self.assertIsInstance(wav_alias, np.ndarray) + self.assertGreater(len(wav_alias), 0) + + def test_paradee_custom_dict(self): + tts = TTS(pretrained="makham_paradee", custom_dict={"ทดสอบ": "ทบ-สอบ"}) + wav = tts.tts("ทดสอบ", return_type="waveform") + self.assertIsInstance(wav, np.ndarray) + self.assertGreater(len(wav), 0) + + def test_paradee_stream(self): + tts = TTS(pretrained="makham_paradee") + chunks = list(tts.stream("สวัสดีครับ ยินดีต้อนรับ")) + self.assertGreater(len(chunks), 0) + for chunk in chunks: + self.assertIsInstance(chunk, np.ndarray) + self.assertGreater(len(chunk), 0) + + def test_makham_base_tts_wrapper(self): + tts = TTS(pretrained="makham") + wav = tts.tts("สวัสดีครับ", speaker_idx="newvoice_female", return_type="waveform") + self.assertIsInstance(wav, np.ndarray) + self.assertGreater(len(wav), 0) + + tts_alias = TTS(pretrained="MakHamTTS") + wav_alias = tts_alias.tts("สวัสดีครับ", return_type="waveform") + self.assertIsInstance(wav_alias, np.ndarray) + self.assertGreater(len(wav_alias), 0) + + def test_makham_base_stream(self): + tts = TTS(pretrained="makham") + chunks = list(tts.stream("สวัสดีครับ ยินดีต้อนรับ", speaker_idx="newvoice_female")) + self.assertGreater(len(chunks), 0) + for chunk in chunks: + self.assertIsInstance(chunk, np.ndarray) + self.assertGreater(len(chunk), 0) + + +if __name__ == "__main__": + unittest.main()