feat: TTS end-to-end intelligibility harness

ovoscope/__init__.py

@@ -9,7 +9,7 @@
 from ovos_bus_client.session import SessionManager, Session
 from ovos_config.config import Configuration
 from ovos_config.models import LocalConf
 from ovos_core.intent_services import IntentService
 from ovos_core.skill_manager import SkillManager
 from ovos_plugin_manager.skills import find_skill_plugins
 from ovos_utils.fakebus import FakeBus
@@ -719,7 +719,7 @@
     ###########################
     track_bus_coverage: bool = False  # enable BusCoverageTracker for this test
     print_bus_coverage: bool = False  # print inline summary after execute()
     bus_coverage_report: Optional["BusCoverageReport"] = dataclasses.field(default=None, init=False, repr=False)
 
     ###########################
     # test runner internals
@@ -916,14 +916,14 @@
             if self.verbose:
                 print(f"💡 final session: {last_sess.serialize()}")
                 print(f"> expected: {expected_sess.serialize()}")
             assert {s[0] for s in last_sess.active_skills} == {s[0] for s in expected_sess.active_skills}, f"❌ final session active_skills doesn't match"
             assert sess.lang == expected_sess.lang, f"❌ final session lang doesn't match"
             assert sess.pipeline == expected_sess.pipeline, f"❌ final session pipeline doesn't match"
             assert sess.system_unit == expected_sess.system_unit, f"❌ final session system_unit doesn't match"
             assert sess.date_format == expected_sess.date_format, f"❌ final session date_format doesn't match"
             assert sess.time_format == expected_sess.time_format, f"❌ final session time_format doesn't match"
             assert sess.site_id == expected_sess.site_id, f"❌ final session site_id doesn't match"
             assert sess.session_id == expected_sess.session_id, f"❌ final session session_id doesn't match"
             assert set(sess.blacklisted_skills) == set(expected_sess.blacklisted_skills), f"❌ final session blacklisted_skills doesn't match"
             assert set(sess.blacklisted_intents) == set(expected_sess.blacklisted_intents), f"❌ final session blacklisted_intents doesn't match"
             if self.verbose:
@@ -1095,6 +1095,28 @@
     else:
         raise
 
+try:
+    from ovoscope.tts_intelligibility import (  # noqa: F401
+        TTSIntelligibilityHarness,
+        IntelligibilityReport,
+        UtteranceScore,
+        score_tts_intelligibility,
+    )
+except ImportError as e:
+    # Optional [tts] extra. Silence only when the missing module is one of the
+    # optional TTS-scoring deps; a logic error in a present lib must re-raise.
+    _TTS_OPTIONAL_MODULES = (
+        "jiwer",
+        "ovos_audio", "ovos_audio.audio",
+        "ovos_utterance_normalizer",
+        "ovos_stt_plugin_fasterwhisper",
+        "faster_whisper",
+    )
+    if isinstance(e, ModuleNotFoundError) and e.name in _TTS_OPTIONAL_MODULES:
+        pass
+    else:
+        raise
+
 try:
     from ovoscope.listener import (  # noqa: F401
         MiniListener,

ovoscope/audio.py

@@ -518,21 +518,32 @@ class PlaybackServiceHarness:
     Args:
         validate_source: Enable session-source validation in the service.
         disable_ocp: Disable legacy OCP in the encapsulated AudioService.
+        tts: TTS instance to drive the PlaybackService with. Defaults to a
+            fresh ``MockTTS()`` (backward compatible). Pass a real TTS plugin
+            to synthesise actual audio — the rendered WAV path of each
+            utterance is captured in :attr:`captured_wavs`.
     """
 
     def __init__(self, validate_source: bool = False,
-                 disable_ocp: bool = True) -> None:
+                 disable_ocp: bool = True,
+                 tts: Optional[TTS] = None) -> None:
         """Initialise harness parameters.
 
         Args:
             validate_source: Enable session-source validation.
             disable_ocp: Disable OCP audio plugin.
+            tts: TTS instance to inject. Defaults to ``MockTTS()`` when None.
         """
         self.validate_source: bool = validate_source
         self.disable_ocp: bool = disable_ocp
         self.bus: Optional[FakeBus] = None
         self.svc = None  # PlaybackService instance
-        self.mock_tts: Optional[MockTTS] = None
+        # ``mock_tts`` keeps its historic name for backward compatibility but
+        # holds whatever TTS was injected (real plugin or MockTTS).
+        self.tts: Optional[TTS] = tts
+        self.mock_tts: Optional[TTS] = None
+        # Paths captured from the ``play_audio`` side_effect, in playback order.
+        self.captured_wavs: List[str] = []
         self._play_audio_patcher = None
         self._audio_enabled_patcher = None
         self._audio_output_start = threading.Event()
@@ -558,15 +569,25 @@ def __enter__(self) -> "PlaybackServiceHarness":
         TTS.queue = Queue()
 
         self.bus = FakeBus()
-        self.mock_tts = MockTTS()
+        # Inject the provided TTS (real plugin) or fall back to MockTTS.
+        self.mock_tts = self.tts if self.tts is not None else MockTTS()
 
-        # Patch play_audio so no real audio device is accessed
+        # Patch play_audio so no real audio device is accessed. The side_effect
+        # records the first positional arg — the rendered WAV path
+        # (ovos_audio/playback.py: ``self.p = play_audio(data)``) — so callers
+        # can round-trip the synthesised audio through a reference STT.
         mock_proc = MagicMock()
         mock_proc.communicate.return_value = (b"", b"")
         mock_proc.wait.return_value = 0
 
+        self.captured_wavs = []
+
+        def _capture_play_audio(data, *args, **kwargs):
+            self.captured_wavs.append(data)
+            return mock_proc
+
         self._play_audio_patcher = patch(
-            "ovos_audio.playback.play_audio", return_value=mock_proc
+            "ovos_audio.playback.play_audio", side_effect=_capture_play_audio
         )
         self._play_audio_patcher.start()