diff --git a/Cortex root folder/chatterbox_tts_module.py b/Cortex root folder/chatterbox_tts_module.py new file mode 100644 index 0000000..d20fc82 --- /dev/null +++ b/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) + + diff --git a/Cortex root folder/edge_tts_module.py b/Cortex root folder/edge_tts_module.py new file mode 100644 index 0000000..5591866 --- /dev/null +++ b/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) + + diff --git a/Cortex root folder/google_tts_module.py b/Cortex root folder/google_tts_module.py new file mode 100644 index 0000000..7dd2333 --- /dev/null +++ b/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("") + 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) + + diff --git a/Cortex root folder/tests/test_tts.py b/Cortex root folder/tests/test_tts.py new file mode 100644 index 0000000..57b662c --- /dev/null +++ b/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() + +