Update server.py

server.py

@@ -1,6 +1,7 @@
 """
 cantrell-kokoro-engine — StoryVoice TTS Backend
 Docker Space, Python 3.11, FastAPI only
+Supports voice blending, sentence-level silence padding, pronunciation map
 """
 
 import io
@@ -9,7 +10,7 @@ import time
 import numpy as np
 import soundfile as sf
 from fastapi import FastAPI, HTTPException
-from fastapi.responses import Response, JSONResponse
+from fastapi.responses import Response
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from kokoro import KPipeline
@@ -33,7 +34,9 @@ def get_pipeline(lang_code):
     return _pipelines[lang_code]
 
 def lang_for_voice(voice_id: str) -> str:
-    return "b" if voice_id.startswith("b") else "a"
+    # Use first voice in a blend to determine lang
+    first = voice_id.split('+')[0].strip().split(':')[0].strip()
+    return "b" if first.startswith("b") else "a"
 
 # ── Pronunciation map ─────────────────────────────────────────────────────────
 PRONUNCIATION = {
@@ -41,6 +44,12 @@ PRONUNCIATION = {
     "ybor":     "eebore",
     "breathed": "breethd",
     "Breathed": "Breethd",
+    "Sanae":    "Suh-nay",
+    "sanae":    "suh-nay",
+    "Nae":      "Nay",
+    "nae":      "nay",
+    "Keymoni":  "Keymoney",
+    "keymoni":  "keymoney",
 }
 
 def apply_pronunciation(text: str) -> str:
@@ -48,48 +57,44 @@ def apply_pronunciation(text: str) -> str:
         text = text.replace(word, replacement)
     return text
 
-# ── Sentence splitter ─────────────────────────────────────────────────────────
-def split_sentences(text: str) -> list:
-    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
-    return [s.strip() for s in sentences if s.strip()]
-
-# ── Audio helpers ─────────────────────────────────────────────────────────────
-SAMPLE_RATE = 24000
+# ── Voice blending ────────────────────────────────────────────────────────────
+# Blend format: "af_aoede:0.50 + af_sky:0.30 + af_nicole:0.20"
+# Single voice: "af_heart" or "af_heart:1.0"
 
-def make_silence(ms: int) -> np.ndarray:
-    return np.zeros(int(SAMPLE_RATE * ms / 1000), dtype=np.float32)
+def parse_blend(voice_str: str):
+    """Parse blend string into list of (voice_id, weight) tuples."""
+    parts = [p.strip() for p in voice_str.split('+')]
+    blend = []
+    for part in parts:
+        if ':' in part:
+            vid, w = part.rsplit(':', 1)
+            blend.append((vid.strip(), float(w.strip())))
+        else:
+            blend.append((part.strip(), 1.0))
+    # Normalize weights
+    total = sum(w for _, w in blend)
+    return [(v, w / total) for v, w in blend]
 
-def generate_sentence(pipeline, sentence: str, voice_id: str, speed: float) -> np.ndarray:
-    chunks = []
-    for _, _, audio in pipeline(sentence, voice=voice_id, speed=speed):
-        if audio is not None and len(audio) > 0:
-            chunks.append(audio)
-    if not chunks:
-        return np.array([], dtype=np.float32)
-    return np.concatenate(chunks) if len(chunks) > 1 else chunks[0]
+def blend_voices(blend_list):
+    """Create a blended voice tensor from a list of (voice_id, weight) tuples."""
+    import torch
+    blended = None
+    for voice_id, weight in blend_list:
+        # KPipeline loads voice pack internally; access via pipeline's voice method
+        lang = "b" if voice_id.startswith("b") else "a"
+        pipeline = get_pipeline(lang)
+        pack = pipeline.load_voice(voice_id)
+        if blended is None:
+            blended = pack * weight
+        else:
+            blended = blended + pack * weight
+    return blended
 
-def generate_audio(text: str, voice_id: str, speed: float = 1.0) -> bytes:
-    text = apply_pronunciation(text)
-    sentences = split_sentences(text)
-    lang = lang_for_voice(voice_id)
-    pipeline = get_pipeline(lang)
-    segments = []
-    for i, sentence in enumerate(sentences):
-        if not sentence:
-            continue
-        audio = generate_sentence(pipeline, sentence, voice_id, speed)
-        if len(audio) > 0:
-            segments.append(audio)
-            if i < len(sentences) - 1:
-                pause_ms = 250 if sentence.endswith(('!', '?')) else 150
-                segments.append(make_silence(pause_ms))
-    if not segments:
-        raise ValueError("No audio generated")
-    combined = np.concatenate(segments)
-    buf = io.BytesIO()
-    sf.write(buf, combined, SAMPLE_RATE, format="mp3")
-    buf.seek(0)
-    return buf.read()
+# ── Named presets ─────────────────────────────────────────────────────────��───
+PRESETS = {
+    "male_narrator":   "am_adam:0.60 + am_michael:0.30 + am_onyx:0.10",
+    "female_narrator": "af_aoede:0.50 + af_sky:0.30 + af_nicole:0.20",
+}
 
 # ── Voice registry ────────────────────────────────────────────────────────────
 VOICES = [
@@ -121,12 +126,20 @@ VOICES = [
     {"voice_id": "bm_fable",    "display_name": "Fable",    "gender": "male",   "accent": "british"},
     {"voice_id": "bm_george",   "display_name": "George",   "gender": "male",   "accent": "british"},
     {"voice_id": "bm_lewis",    "display_name": "Lewis",    "gender": "male",   "accent": "british"},
+    # Named presets shown as selectable voices
+    {"voice_id": "male_narrator",   "display_name": "Male Narrator (Blend)",   "gender": "male",   "accent": "american"},
+    {"voice_id": "female_narrator", "display_name": "Female Narrator (Blend)", "gender": "female", "accent": "american"},
 ]
 
 VOICE_MAP = {v["voice_id"]: v for v in VOICES}
 
+SAMPLE_RATE = 24000
+
 def resolve_voice(voice_id: str) -> str:
     voice_id = voice_id.replace(".mp3", "").strip()
+    # Check presets first
+    if voice_id in PRESETS:
+        return PRESETS[voice_id]
     if voice_id in VOICE_MAP:
         return voice_id
     matched = next(
@@ -135,8 +148,65 @@ def resolve_voice(voice_id: str) -> str:
     )
     return matched or "af_heart"
 
+def make_silence(ms: int) -> np.ndarray:
+    return np.zeros(int(SAMPLE_RATE * ms / 1000), dtype=np.float32)
+
+def split_sentences(text: str) -> list:
+    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
+    return [s.strip() for s in sentences if s.strip()]
+
+def generate_sentence(pipeline, sentence: str, voice, speed: float) -> np.ndarray:
+    chunks = []
+    for _, _, audio in pipeline(sentence, voice=voice, speed=speed):
+        if audio is not None and len(audio) > 0:
+            chunks.append(audio)
+    if not chunks:
+        return np.array([], dtype=np.float32)
+    return np.concatenate(chunks) if len(chunks) > 1 else chunks[0]
+
+def generate_audio(text: str, voice_id: str, speed: float = 1.0) -> bytes:
+    text = apply_pronunciation(text)
+    sentences = split_sentences(text)
+
+    # Resolve preset to blend string
+    voice_str = PRESETS.get(voice_id, voice_id)
+    lang = lang_for_voice(voice_str)
+    pipeline = get_pipeline(lang)
+
+    # Determine if blending needed
+    is_blend = '+' in voice_str or ':' in voice_str
+    if is_blend:
+        blend_list = parse_blend(voice_str)
+        try:
+            voice = blend_voices(blend_list)
+        except Exception:
+            # Fallback to first voice if blending fails
+            voice = blend_list[0][0]
+    else:
+        voice = voice_str.split(':')[0].strip()
+
+    segments = []
+    for i, sentence in enumerate(sentences):
+        if not sentence:
+            continue
+        audio = generate_sentence(pipeline, sentence, voice, speed)
+        if len(audio) > 0:
+            segments.append(audio)
+            if i < len(sentences) - 1:
+                pause_ms = 250 if sentence.endswith(('!', '?')) else 150
+                segments.append(make_silence(pause_ms))
+
+    if not segments:
+        raise ValueError("No audio generated")
+
+    combined = np.concatenate(segments)
+    buf = io.BytesIO()
+    sf.write(buf, combined, SAMPLE_RATE, format="mp3")
+    buf.seek(0)
+    return buf.read()
+
 # ── Routes ────────────────────────────────────────────────────────────────────
-@app.get("/", response_class=Response)
+@app.get("/")
 def index():
     html = """<!DOCTYPE html>
 <html>
@@ -183,9 +253,7 @@ def index():
   <div class="status" id="status"></div>
 </div>
 <script>
-var VOICES = [];
 fetch('/voices').then(r=>r.json()).then(function(data){
-  VOICES = data;
   var sel = document.getElementById('voice');
   data.forEach(function(v){
     var o = document.createElement('option');
@@ -232,15 +300,18 @@ function generate(){
 </html>"""
     return Response(content=html, media_type="text/html")
 
-
 @app.get("/health")
 def health():
-    return {"status": "ok", "engine": "kokoro-82m", "voices": len(VOICES), "timestamp": int(time.time())}
+    return {"status": "ok", "engine": "kokoro-82m", "voices": len(VOICES), "presets": list(PRESETS.keys()), "timestamp": int(time.time())}
 
 @app.get("/voices")
 def voices():
     return VOICES
 
+@app.get("/presets")
+def presets():
+    return PRESETS
+
 class GenerateRequest(BaseModel):
     text: str
     voice_id: str = "af_heart"
@@ -265,4 +336,4 @@ def tts_preview(req: PreviewRequest):
     return generate(GenerateRequest(text=req.text, voice_id=req.voice_id, speed=req.speed))
 
 if __name__ == "__main__":
-    uvicorn.run("server:app", host="0.0.0.0", port=7860)
+    uvicorn.run(app, host="0.0.0.0", port=7860)