| """Explicit unknown-phone pause policy; no ONNX/model loading.""" |
| import json |
| from pathlib import Path |
| import tempfile |
| import threading |
| import unittest |
|
|
| import lameenc |
| import numpy as np |
|
|
| from tts.engine import SpeechEngine |
| from tts.text import SpeechError, cache_key, playable_phonemes, playback_segments, requested_segments, validate_text |
| from tts.pronunciation import create_frontend |
| from tests.test_pronunciation import FakeContext |
|
|
|
|
| class FallbackTests(unittest.TestCase): |
| def test_optional_phrase_plan_keeps_original_unicode_numeric_atoms_and_cache_identity(self): |
| text = "校长读长文。𠮷后文继续。" |
| result = requested_segments(text, [6, 13], allow_pause=True) |
| self.assertEqual("".join(source for _, _, source in result), text) |
| with self.assertRaises(SpeechError): requested_segments("日期2026-10-18。", [6, 13], allow_pause=True) |
| self.assertNotEqual(cache_key(text, "zf_001", 1, "assets", segment_ends=[6, 13]), cache_key(text, "zf_001", 1, "assets", segment_ends=[13])) |
|
|
| def test_only_unknown_phonemes_become_commas_with_explicit_opt_in(self): |
| vocab = set("ab, .123") |
| self.assertEqual(playable_phonemes("a❓★b\n", vocab, unsupported_policy="pause"), ("a,,b ", 2)) |
| with self.assertRaises(SpeechError): |
| playable_phonemes("a❓b", vocab) |
| self.assertNotEqual(cache_key("正文", "zf_001", 1, "assets"), cache_key("正文", "zf_001", 1, "assets", unsupported_policy="pause")) |
|
|
| def test_symbol_control_and_blank_source_spans_are_preserved_only_in_pause_mode(self): |
| for text in ("😀※★", "甲\x0c乙", " \n\t", "𰻞。"): |
| validate_text(text, unsupported_policy="pause") |
| parts = playback_segments(text, allow_pause=True) |
| self.assertEqual("".join(value for _, _, value in parts), text) |
| for text in ("😀※★", "甲\x0c乙", " \n\t"): |
| with self.assertRaises(SpeechError): validate_text(text) |
| with self.assertRaises(SpeechError): validate_text("\ud800", unsupported_policy="pause") |
|
|
| def test_real_misaki_preserves_supported_readings_around_unknown_symbols(self): |
| frontend = create_frontend(model=FakeContext()) |
| phones = frontend("甲𠮷★α😀乙。")[0] |
| supported = set(frontend("甲，乙。")[0]) |
| replaced, count = playable_phonemes(phones, supported, unsupported_policy="pause") |
| self.assertEqual(count, 4) |
| self.assertEqual(replaced, "ㄐ压3,,,,ㄧ3.") |
|
|
| def test_engine_publishes_real_encoded_pause_without_inference_for_unknown_only(self): |
| engine = SpeechEngine.__new__(SpeechEngine) |
| engine.np, engine.lameenc, engine.sample_rate = np, lameenc, 24000 |
| engine.vocab = {char: index for index, char in enumerate(" ,.")} |
| engine.g2p = create_frontend(model=FakeContext()) |
| event = threading.Event() |
| with tempfile.TemporaryDirectory() as directory: |
| file = Path(directory) / "pause.mp3" |
| published = [] |
| result = engine.synthesize("😀★𰻞。", "zf_001", 1, file, event, progress=published.append, unsupported_policy="pause") |
| self.assertGreater(file.stat().st_size, 0) |
| self.assertEqual(result["unsupported_phoneme_count"], 3) |
| self.assertEqual(result["unsupported_policy"], "pause") |
| self.assertEqual(result["segments"][-1]["end"], 6) |
| self.assertEqual(len(published), 1) |
| |
| self.assertAlmostEqual(result["audio_seconds"], .40, places=3) |
| self.assertEqual(result["batches"], 0) |
| with self.assertRaises(SpeechError): engine.synthesize("甲😀乙。", "zf_001", 1, file, event) |
|
|
