tts_engine.py

import os
import struct
import tempfile
import wave

import numpy as np
import pyttsx3
from scipy.io import wavfile
from scipy.signal import resample

from voice_mapper import VoiceParameters


class TTSEngine:
    def __init__(self):
        self._engine = pyttsx3.init()

    def _apply_pitch_shift(self, audio_data: np.ndarray, sample_rate: int, semitones: int) -> np.ndarray:
        if semitones == 0:
            return audio_data

        factor = 2 ** (semitones / 12.0)
        original_length = len(audio_data)
        stretched = resample(audio_data, int(original_length / factor))
        resampled = resample(stretched, original_length)

        return resampled.astype(np.int16)

    def synthesize(self, text: str, params: VoiceParameters, output_path: str) -> str:
        self._engine.setProperty("rate", params.rate)
        self._engine.setProperty("volume", params.volume)

        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
            tmp_path = tmp.name

        try:
            self._engine.save_to_file(text, tmp_path)
            self._engine.runAndWait()

            if not os.path.exists(tmp_path) or os.path.getsize(tmp_path) == 0:
                raise RuntimeError("TTS engine failed to produce audio output.")

            if params.pitch_shift != 0:
                sample_rate, audio_data = wavfile.read(tmp_path)

                if audio_data.ndim > 1:
                    audio_data = audio_data[:, 0]

                shifted = self._apply_pitch_shift(audio_data, sample_rate, params.pitch_shift)
                wavfile.write(output_path, sample_rate, shifted)
            else:
                with open(tmp_path, "rb") as src, open(output_path, "wb") as dst:
                    dst.write(src.read())

        finally:
            if os.path.exists(tmp_path):
                os.remove(tmp_path)

        return output_path