268 lines
11 KiB
Python
268 lines
11 KiB
Python
import importlib.util
|
||
import pathlib
|
||
import unittest
|
||
from array import array
|
||
|
||
|
||
MODULE_PATH = pathlib.Path(__file__).with_name("tts_gateway.py")
|
||
SPEC = importlib.util.spec_from_file_location("tts_gateway", MODULE_PATH)
|
||
gateway = importlib.util.module_from_spec(SPEC)
|
||
SPEC.loader.exec_module(gateway)
|
||
|
||
|
||
class LanguageSegmentationTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.original_code_switch = gateway.CODE_SWITCH_ENABLED
|
||
|
||
def tearDown(self):
|
||
gateway.CODE_SWITCH_ENABLED = self.original_code_switch
|
||
|
||
def test_german_only(self):
|
||
self.assertEqual(
|
||
gateway.segment_languages("Guten Abend, wie warm ist es heute?"),
|
||
[("de", "Guten Abend, wie warm ist es heute?")],
|
||
)
|
||
|
||
def test_english_only(self):
|
||
text = "This is a short test and it is running on the local server."
|
||
self.assertEqual(gateway.segment_languages(text), [("en", text)])
|
||
|
||
def test_mixed_compounds(self):
|
||
gateway.CODE_SWITCH_ENABLED = True
|
||
text = "Ich öffne das Unraid-Dashboard und prüfe die Docker-Container."
|
||
self.assertEqual(
|
||
gateway.segment_languages(text),
|
||
[
|
||
("de", "Ich öffne das "),
|
||
("en", "Unraid-Dashboard"),
|
||
("de", " und prüfe die "),
|
||
("en", "Docker-Container"),
|
||
("de", "."),
|
||
],
|
||
)
|
||
|
||
def test_operational_vocabulary_is_not_read_as_german(self):
|
||
gateway.CODE_SWITCH_ENABLED = True
|
||
text = (
|
||
"31 von 60 Containern laufen, alle sind up und healthy. "
|
||
"Die Health-Checks zeigen keine Crashes oder Restart-Loops. "
|
||
"Immich meldet unsupported image format und premature end of JPEG."
|
||
)
|
||
segments = gateway.segment_languages(text)
|
||
english = [part.lower() for language, part in segments if language == "en"]
|
||
self.assertIn("containern", english)
|
||
self.assertIn("up", english)
|
||
self.assertIn("healthy", english)
|
||
self.assertIn("health-checks", english)
|
||
self.assertIn("crashes", english)
|
||
self.assertIn("restart-loops", english)
|
||
self.assertIn("unsupported image format", english)
|
||
self.assertIn("premature end of jpeg", english)
|
||
prepared = gateway.prepare_segments(text)
|
||
self.assertTrue(all(any(char.isalnum() for char in part) for _, part in prepared))
|
||
|
||
def test_long_text_is_cleaned_and_chunked(self):
|
||
text = "✅ **Status**\n" + ("Ein langer deutscher Diagnosesatz. " * 30)
|
||
segments = gateway.prepare_segments(text)
|
||
self.assertTrue(segments)
|
||
self.assertTrue(all(len(part) <= gateway.CHUNK_CHARS for _, part in segments))
|
||
self.assertNotIn("✅", "".join(part for _, part in segments))
|
||
|
||
def test_inflected_container_has_english_pronunciation(self):
|
||
gateway.CODE_SWITCH_ENABLED = True
|
||
prepared = gateway.prepare_segments("Ich prüfe die Containern.")
|
||
self.assertIn(("en", "containers."), prepared)
|
||
|
||
def test_code_switch_is_disabled_by_default(self):
|
||
gateway.CODE_SWITCH_ENABLED = False
|
||
text = "Docker-Container laufen, die Health-Checks melden healthy."
|
||
self.assertEqual(gateway.segment_languages(text), [("de", text)])
|
||
|
||
def test_weather_notation_is_spoken_naturally(self):
|
||
text = (
|
||
"Rastatt (76437), Sonntag 23.08.: 22° / 10°, "
|
||
"0 % Regen. Quellen: wetter.com und wetteronline.de"
|
||
)
|
||
spoken = gateway.clean_for_speech(text)
|
||
self.assertIn("Postleitzahl sieben sechs vier drei sieben", spoken)
|
||
self.assertIn("23. August", spoken)
|
||
self.assertIn("Höchstwert 22 Grad, Tiefstwert 10 Grad", spoken)
|
||
self.assertIn("0 Prozent Regen", spoken)
|
||
self.assertIn("wetter Punkt C O M", spoken)
|
||
self.assertIn("wetteronline Punkt de", spoken)
|
||
|
||
def test_full_date_and_raw_url_are_normalized(self):
|
||
spoken = gateway.clean_for_speech(
|
||
"Am 23.08.2026 steht es auf https://www.example.org/path?q=1."
|
||
)
|
||
self.assertIn("23. August 2026", spoken)
|
||
self.assertIn("example Punkt O R G", spoken)
|
||
self.assertNotIn("https", spoken)
|
||
|
||
def test_wind_speed_is_spoken_as_a_unit(self):
|
||
for text in (
|
||
"Böen bis etwa 29 km/h.",
|
||
"Böen bis etwa 29 KM / H.",
|
||
"Wind mit 29 kmh.",
|
||
):
|
||
with self.subTest(text=text):
|
||
spoken = gateway.clean_for_speech(text)
|
||
self.assertIn("29 Kilometer pro Stunde", spoken)
|
||
self.assertNotIn("km", spoken.lower())
|
||
|
||
def test_qwen_normalizes_ipv4_time_date_and_count(self):
|
||
spoken = gateway.prepare_for_qwen_speech(
|
||
"full_kiosk (192.168.1.5): 2× Timeout um 02:14 seit 01.09."
|
||
)
|
||
self.assertIn("full kiosk", spoken)
|
||
self.assertIn("192 Punkt 168 Punkt 1 Punkt 5", spoken)
|
||
self.assertIn("2 mal Timeout", spoken)
|
||
self.assertIn("2 Uhr 14", spoken)
|
||
self.assertIn("1. September", spoken)
|
||
|
||
def test_qwen_speaks_temperature_and_time_ranges_with_bis(self):
|
||
spoken = gateway.prepare_for_qwen_speech(
|
||
"15°–23 °C, 93 %, ca. 12 mm, v.a. um 05–06 Uhr, "
|
||
"Wind max. 16 km/h."
|
||
)
|
||
self.assertIn("15 bis 23 Grad", spoken)
|
||
self.assertIn("93 Prozent", spoken)
|
||
self.assertIn("circa 12 mm", spoken)
|
||
self.assertIn("vor allem", spoken)
|
||
self.assertIn("5 bis 6 Uhr", spoken)
|
||
self.assertIn("Wind maximal 16 Kilometer pro Stunde", spoken)
|
||
|
||
def test_qwen_keeps_prosody_punctuation(self):
|
||
spoken = gateway.prepare_for_qwen_speech(
|
||
"Ist das gut? Ja! SarahTV: erreichbar."
|
||
)
|
||
self.assertEqual(spoken, "Ist das gut? Ja! SarahTV: erreichbar.")
|
||
|
||
def test_visual_punctuation_becomes_natural_pauses(self):
|
||
spoken = gateway.clean_for_speech(
|
||
"Status: stabil – keine Fehler; Docker-Container laufen."
|
||
)
|
||
self.assertEqual(
|
||
spoken,
|
||
"Status, stabil, keine Fehler, Docker Container laufen.",
|
||
)
|
||
self.assertNotRegex(spoken, r"[:;\-‐‑‒–—−]")
|
||
|
||
def test_question_and_exclamation_marks_are_not_sent_to_xtts(self):
|
||
spoken = gateway.clean_for_speech("Wie geht es dir? Wirklich gut!")
|
||
self.assertEqual(spoken, "Wie geht es dir. Wirklich gut.")
|
||
self.assertNotRegex(spoken, r"[!?]")
|
||
|
||
def test_punctuation_only_segments_are_never_synthesized(self):
|
||
segments = gateway.prepare_segments("Status: – alles läuft.")
|
||
self.assertTrue(segments)
|
||
self.assertTrue(all(any(char.isalnum() for char in part) for _, part in segments))
|
||
|
||
def test_weather_summary_keeps_complete_sentences_together(self):
|
||
text = (
|
||
"Heute in Rastatt: teils sonnig, trocken, 10 bis 22 Grad. "
|
||
"Abends wolkiger, 16 bis 21 Grad. Böen bis 29 km/h. "
|
||
"Kein Regen erwartet. Quelle: wetter.com, wetteronline.de"
|
||
)
|
||
segments = gateway.prepare_segments(text)
|
||
spoken = " ".join(part for _, part in segments)
|
||
self.assertEqual(len(segments), 1)
|
||
self.assertTrue(all(len(part) <= gateway.CHUNK_CHARS
|
||
for _, part in segments))
|
||
self.assertIn("29 Kilometer pro Stunde", spoken)
|
||
self.assertIn("Kein Regen erwartet", spoken)
|
||
self.assertIn("wetteronline Punkt de", spoken)
|
||
|
||
def test_short_followup_sentence_shares_the_same_xtts_request(self):
|
||
text = (
|
||
"Ehrlich gesagt habe ich keine echten Gefühle wie Menschen, aber "
|
||
"ich bin wach, aufmerksam und motiviert, dir zu helfen. "
|
||
"Klingt das gut?"
|
||
)
|
||
segments = gateway.prepare_segments(text)
|
||
self.assertEqual(len(segments), 1)
|
||
self.assertIn("dir zu helfen. Klingt das gut.", segments[0][1])
|
||
|
||
def test_xtts_receives_a_stable_terminal_stop_cue(self):
|
||
self.assertEqual(
|
||
gateway._stabilize_xtts_ending("Heute bleibt es trocken."),
|
||
"Heute bleibt es trocken;",
|
||
)
|
||
self.assertEqual(
|
||
gateway._stabilize_xtts_ending("Klingt das gut."),
|
||
"Klingt das gut;",
|
||
)
|
||
|
||
def test_normal_sentence_is_not_broken_into_word_sized_requests(self):
|
||
text = (
|
||
"Die automatische Komprimierung ist in OpenClaw standardmäßig "
|
||
"aktiviert und lässt sich über die Konfigurationsdatei steuern."
|
||
)
|
||
self.assertEqual(gateway.prepare_segments(text), [("de", text)])
|
||
|
||
def test_overlong_sentence_prefers_clause_boundaries(self):
|
||
clause = "dieser natürlich gesprochene Teilsatz bleibt zusammen,"
|
||
text = " ".join([clause] * 12) + " und endet hier."
|
||
segments = gateway.prepare_segments(text)
|
||
self.assertGreater(len(segments), 1)
|
||
self.assertTrue(all(len(part) <= gateway.CHUNK_CHARS
|
||
for _, part in segments))
|
||
self.assertTrue(all(len(part.split()) > 4 for _, part in segments))
|
||
|
||
|
||
class FallbackTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.original_xtts = gateway.synthesize_xtts
|
||
self.original_piper = gateway.synthesize_piper
|
||
|
||
def tearDown(self):
|
||
gateway.synthesize_xtts = self.original_xtts
|
||
gateway.synthesize_piper = self.original_piper
|
||
|
||
def test_piper_is_used_when_xtts_fails(self):
|
||
def fail(*_args):
|
||
raise RuntimeError("synthetic XTTS failure")
|
||
|
||
gateway.synthesize_xtts = fail
|
||
gateway.synthesize_piper = lambda *_args: (b"piper", "audio/wav")
|
||
self.assertEqual(
|
||
gateway.synthesize("synthetic test", "wav", 1.0),
|
||
(b"piper", "audio/wav"),
|
||
)
|
||
|
||
|
||
class AudioJoinTests(unittest.TestCase):
|
||
@staticmethod
|
||
def pcm(values):
|
||
samples = array("h", values)
|
||
return samples.tobytes()
|
||
|
||
def test_trim_removes_long_silent_edges(self):
|
||
pcm = self.pcm(([0] * 1000) + ([500] * 100) + ([0] * 1000))
|
||
trimmed = gateway._trim_pcm(pcm)
|
||
self.assertLess(len(trimmed), len(pcm))
|
||
self.assertGreater(len(trimmed), 100 * 2)
|
||
|
||
def test_join_inserts_pause_only_after_sentence(self):
|
||
spoken = self.pcm([500] * 1000)
|
||
inline = gateway._join_pcm([("Docker", spoken), ("Container", spoken)])
|
||
sentence = gateway._join_pcm([("Fertig.", spoken), ("Weiter", spoken)])
|
||
self.assertGreater(len(sentence), len(inline))
|
||
|
||
def test_long_internal_silence_is_shortened(self):
|
||
spoken = self.pcm([900] * 1000)
|
||
long_silence = self.pcm([0] * int(24000 * 1.5))
|
||
result = gateway._compress_internal_silence(spoken + long_silence + spoken)
|
||
expected_keep = int(24000 * gateway.INTERNAL_SILENCE_KEEP_MS / 1000) * 2
|
||
self.assertEqual(len(result), len(spoken) * 2 + expected_keep)
|
||
|
||
def test_natural_short_pause_is_preserved(self):
|
||
spoken = self.pcm([900] * 1000)
|
||
short_silence = self.pcm([0] * int(24000 * 0.3))
|
||
original = spoken + short_silence + spoken
|
||
self.assertEqual(gateway._compress_internal_silence(original), original)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|