init

app/synthesis.py

@@ -0,0 +1,112 @@
+import numpy as np
+from dataclasses import dataclass, field
+from silero_tts.silero_tts import SileroTTS
+
+from app.segmentation import Segment
+from app.ssml import build_ssml_chunks
+
+SSML_MAX_CHARS = 500
+
+
+@dataclass
+class ModelEntry:
+    model_id: str
+    tts: SileroTTS
+    languages: list[str] = field(default_factory=list)
+
+
+@dataclass
+class SynthesisEngine:
+    models: list[ModelEntry] = field(default_factory=list)
+    sample_rate: int = 24000
+    use_ssml: bool = True
+    ssml_max_chars: int = SSML_MAX_CHARS
+
+    def load(self, models_config, sample_rate: int = 24000) -> None:
+        self.sample_rate = sample_rate
+        self.models = []
+        for m in models_config:
+            tts = SileroTTS(
+                model_id=m.file,
+                language=m.languages[0] if m.languages else "ru",
+                speaker="default",
+                sample_rate=sample_rate,
+                device="cpu",
+                num_threads=4,
+            )
+            self.models.append(ModelEntry(
+                model_id=m.file,
+                tts=tts,
+                languages=m.languages,
+            ))
+
+    def find_model(self, lang: str) -> ModelEntry | None:
+        for m in self.models:
+            if lang in m.languages:
+                return m
+        return None
+
+    def get_model(self, model_id: str) -> ModelEntry | None:
+        for m in self.models:
+            if m.model_id == model_id:
+                return m
+        return None
+
+    def available_speakers(self) -> dict[str, list[str]]:
+        return {m.model_id: list(m.tts.tts_model.speakers) for m in self.models}
+
+    def _synthesize_text(self, model: ModelEntry, speaker: str, text: str) -> np.ndarray:
+        audio = model.tts.tts_model.apply_tts(
+            text=text,
+            speaker=speaker,
+            sample_rate=model.tts.sample_rate,
+            put_accent=model.tts.put_accent,
+            put_yo=model.tts.put_yo,
+        )
+        return (audio.numpy() * 32767).astype(np.int16)
+
+    def _synthesize_ssml(self, model: ModelEntry, speaker: str, ssml_text: str) -> np.ndarray:
+        audio = model.tts.tts_model.apply_tts(
+            ssml_text=ssml_text,
+            speaker=speaker,
+            sample_rate=model.tts.sample_rate,
+            put_accent=True,
+            put_yo=model.tts.put_yo,
+        )
+        return (audio.numpy() * 32767).astype(np.int16)
+
+    def synthesize_segments(
+        self, segments: list[Segment], speaker: str, lang: str,
+        *, rate: str | None = None, pitch: str | None = None, use_ssml: bool | None = None,
+    ) -> bytes:
+        model = self.find_model(lang)
+        if model is None:
+            raise ValueError(f"no model supports language '{lang}'")
+        return self._synthesize_multi(segments, model, speaker, rate, pitch, use_ssml)
+
+    def synthesize_segments_model(
+        self, segments: list[Segment], model_id: str, speaker: str,
+        *, rate: str | None = None, pitch: str | None = None, use_ssml: bool | None = None,
+    ) -> bytes:
+        model = self.get_model(model_id)
+        if model is None:
+            raise ValueError(f"model '{model_id}' not loaded")
+        return self._synthesize_multi(segments, model, speaker, rate, pitch, use_ssml)
+
+    def _synthesize_multi(
+        self, segments: list[Segment], model: ModelEntry, speaker: str,
+        rate: str | None, pitch: str | None, use_ssml: bool | None,
+    ) -> bytes:
+        parts: list[np.ndarray] = []
+        ssml = self.use_ssml if use_ssml is None else use_ssml
+        if ssml:
+            for chunk in build_ssml_chunks(
+                segments, rate=rate, pitch=pitch, max_chars=self.ssml_max_chars
+            ):
+                parts.append(self._synthesize_ssml(model, speaker, chunk))
+        else:
+            for seg in segments:
+                parts.append(self._synthesize_text(model, speaker, seg.text))
+        if not parts:
+            return b""
+        return np.concatenate(parts).tobytes()