tts: lazy-load heavy deps in TTS modules (refs #2)

Cortex root folder/chatterbox_tts_module.py

@@ -0,0 +1,27 @@
+"""Chatterbox TTS with lazy torch/model import to keep import light."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+
+def speak(text: str, voice: Optional[str] = None, speaking_rate: Optional[float] = None) -> None:
+    if not text or not text.strip():
+        return
+
+    # Lazy imports of heavy deps
+    import torch  # type: ignore
+    import soundfile as sf  # type: ignore
+    from chatterbox.tts import ChatterboxTTS  # type: ignore
+    import tempfile
+    from playsound import playsound  # type: ignore
+
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    model = ChatterboxTTS.from_pretrained(device=device)
+    waveform, sample_rate = model.generate(text=text)
+
+    with tempfile.NamedTemporaryFile(delete=True, suffix=".wav") as tmp:
+        sf.write(tmp.name, waveform, sample_rate)
+        playsound(tmp.name)
+
+

Cortex root folder/edge_tts_module.py

@@ -0,0 +1,36 @@
+"""Edge TTS module with lazy imports to avoid import-time failures.
+
+The heavy dependency `edge_tts` is imported inside the call path so that
+module import stays lightweight and test collection remains stable.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Optional
+
+
+async def _generate_and_save(text: str, voice: str, rate: str, out_path: str) -> None:
+    # Lazy import heavy dependency
+    import edge_tts  # type: ignore
+
+    communicate = edge_tts.Communicate(text, voice=voice, rate=rate)
+    await communicate.save(out_path)
+
+
+def speak(text: str, voice: Optional[str] = None, speaking_rate: Optional[float] = None) -> None:
+    if not text or not text.strip():
+        return
+
+    # Defaults (keep simple to avoid config imports)
+    voice_id = voice or "en-US-AriaNeural"
+    rate_str = f"+{speaking_rate:.1f}%" if speaking_rate is not None else "+0%"
+
+    import tempfile
+    from playsound import playsound  # type: ignore
+
+    with tempfile.NamedTemporaryFile(delete=True, suffix=".mp3") as tmp:
+        asyncio.run(_generate_and_save(text, voice_id, rate_str, tmp.name))
+        playsound(tmp.name)
+
+

Cortex root folder/google_tts_module.py

@@ -0,0 +1,43 @@
+"""Google TTS module with lazy imports to avoid import-time failures."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+
+def speak(text: str, voice: Optional[str] = None, speaking_rate: Optional[float] = None) -> None:
+    if not text or not text.strip():
+        return
+
+    # Lazy imports
+    from google.cloud import texttospeech  # type: ignore
+    from playsound import playsound  # type: ignore
+    import tempfile
+
+    voice_id = voice or "en-US-Wavenet-D"
+    rate = float(speaking_rate) if speaking_rate is not None else 1.0
+    language_code = "-".join(voice_id.split("-")[:2])
+
+    client = texttospeech.TextToSpeechClient()
+    voice_params = texttospeech.VoiceSelectionParams(language_code=language_code, name=voice_id)
+    audio_config = texttospeech.AudioConfig(
+        audio_encoding=texttospeech.AudioEncoding.MP3,
+        speaking_rate=rate,
+    )
+
+    synthesis_input = (
+        texttospeech.SynthesisInput(ssml=text)
+        if text.strip().startswith("<speak>")
+        else texttospeech.SynthesisInput(text=text)
+    )
+
+    response = client.synthesize_speech(
+        input=synthesis_input, voice=voice_params, audio_config=audio_config
+    )
+
+    with tempfile.NamedTemporaryFile(delete=True, suffix=".mp3") as tmp:
+        tmp.write(response.audio_content)
+        tmp.flush()
+        playsound(tmp.name)
+
+

Cortex root folder/tests/test_tts.py

@@ -0,0 +1,97 @@
+"""Tests for TTS modules.
+
+Skips cleanly on environments where optional TTS dependencies are not usable.
+In particular, on Windows a broken Torch install can raise OSError during import.
+"""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+# Skip these tests if optional imports fail
+pytest.importorskip("edge_tts")
+pytest.importorskip("google.cloud.texttospeech")
+
+# Torch on Windows can raise OSError during binary/DLL load instead of ImportError.
+# Ensure we skip the entire module cleanly in that case.
+try:  # noqa: SIM105 - broad except to ensure test collection stability
+    import torch  # type: ignore  # noqa: F401
+except OSError as e:  # DLL load error or similar
+    pytest.skip(f"Skipping TTS tests: torch unavailable ({e})", allow_module_level=True)
+except Exception as e:
+    # Any other unexpected import-time exception should also skip these optional tests
+    pytest.skip(f"Skipping TTS tests: torch not usable ({e})", allow_module_level=True)
+
+
+def test_edge_tts_import():
+    """Test that edge_tts can be imported."""
+    from edge_tts_module import speak as edge_speak
+
+    assert callable(edge_speak)
+
+
+def test_google_tts_import():
+    """Test that google_tts can be imported."""
+    from google_tts_module import speak as google_speak
+
+    assert callable(google_speak)
+
+
+def test_chatterbox_tts_import():
+    """Test that chatterbox_tts can be imported."""
+    from chatterbox_tts_module import speak as chatterbox_speak
+
+    assert callable(chatterbox_speak)
+
+
+@patch("edge_tts.Communicate")
+def test_edge_tts_mock(mock_communicate):
+    """Test edge_tts with a mock."""
+    # Setup mock: edge-tts API uses .save; ensure the call path doesn't raise
+    mock_communicate.return_value.save = MagicMock()
+
+    from edge_tts_module import speak as edge_speak
+
+    # Test with mock
+    with patch("playsound.playsound") as mock_playsound:
+        edge_speak("Test text")
+        mock_playsound.assert_called_once()
+
+
+@patch("google.cloud.texttospeech.TextToSpeechClient")
+def test_google_tts_mock(mock_client):
+    """Test google_tts with a mock."""
+    # Setup mock
+    mock_instance = MagicMock()
+    mock_instance.synthesize_speech.return_value.audio_content = b"mock_audio_data"
+    mock_client.return_value = mock_instance
+
+    from google_tts_module import speak as google_speak
+
+    # Test with mock
+    with patch("playsound.playsound") as mock_playsound:
+        google_speak("Test text")
+        mock_playsound.assert_called_once()
+
+
+@patch("chatterbox_tts_module.ChatterboxTTS")
+def test_chatterbox_tts_mock(mock_chatterbox):
+    """Test chatterbox_tts with a mock."""
+    # Setup mock
+    mock_instance = MagicMock()
+    mock_instance.generate.return_value = (MagicMock(), 22050)  # waveform, sample_rate
+    mock_chatterbox.from_pretrained.return_value = mock_instance
+
+    # Mock soundfile write via the module alias used in implementation
+    with patch("chatterbox_tts_module.sf.write") as mock_sf_write, patch(
+        "playsound.playsound"
+    ) as mock_playsound:
+        from chatterbox_tts_module import speak as chatterbox_speak
+
+        # Test with mock
+        chatterbox_speak("Test text")
+
+        # Verify the mock was called
+        mock_sf_write.assert_called_once()
+        mock_playsound.assert_called_once()
+
+