Upload narrator.py with huggingface_hub

narrator.py

@@ -2,39 +2,66 @@
 """
 narrator.py
 -----------
-Turnkey documentary voiceover generator for Kokoro Grounded Documentary.
-Can be imported directly into any Python project or run via CLI.
+Core VoiceoverNarrator class for the kokoro-voiceover package.
+Provides broadcast-grade documentary and long-form speech synthesis using Kokoro-ONNX
+and the tuned Grounded Documentary voice profile (55% bm_george + 45% bm_daniel).
 """
 
 import os
+import re
+import json
+import time
 import subprocess
 from pathlib import Path
-from typing import Optional, Union
+from typing import Optional, Union, List, Tuple, Dict, Any
 import numpy as np
 import soundfile as sf
 from kokoro_onnx import Kokoro
 
-try:
-    from huggingface_hub import hf_hub_download
-except ImportError:
-    hf_hub_download = None
-
-DEFAULT_BROADCAST_FILTER = (
-    "highpass=f=80,"
-    "equalizer=f=160:t=q:w=1.0:g=2.8,"
-    "equalizer=f=260:t=q:w=1.2:g=2.0,"
-    "equalizer=f=450:t=q:w=2.0:g=-2.0,"
-    "equalizer=f=3200:t=q:w=2.0:g=-1.6,"
-    "equalizer=f=5800:t=q:w=2.5:g=-3.5,"
+from .pronunciation import apply_pronunciation_lexicon
+
+# Gemini-Calibrated Broadcast Mastering Filter Chain (Warm sub-chest resonance + surgical de-essing)
+BROADCAST_AUDIO_FILTER = (
+    "highpass=f=75:p=2,"
+    "equalizer=f=180:t=q:w=1.2:g=3.2,"
+    "equalizer=f=480:t=q:w=2.0:g=-1.8,"
+    "equalizer=f=3200:t=q:w=1.5:g=1.2,"
     "equalizer=f=7200:t=q:w=2.5:g=-3.0,"
     "lowpass=f=11000,"
-    "deesser=i=0.45:m=0.5:f=0.55:s=o,"
-    "compand=attacks=0.03:decays=0.3:points=-80/-80|-35/-25|-16/-13|0/-3,"
-    "loudnorm=I=-16:TP=-1.5:LRA=8.5"
+    "deesser=i=0.7:m=0.5:f=0.65:s=o,"
+    "compand=attacks=0.02:decays=0.25:points=-70/-70|-45/-32|-20/-14|0/-6:soft-knee=6,"
+    "loudnorm=I=-16:TP=-1.5:LRA=9"
 )
 
 
-class GroundedNarrator:
+def clean_text_base(raw_text: str, apply_lexicon: bool = True) -> str:
+    """Sanitizes raw markdown script text for natural TTS delivery and applies phonetic lexicon."""
+    t = raw_text
+    if apply_lexicon:
+        t = apply_pronunciation_lexicon(t)
+    t = re.sub(r"\*\*([^*]+)\*\*", r"\1", t)
+    t = re.sub(r"\*([^*]+)\*", r"\1", t)
+    t = re.sub(r"`([^`]+)`", r"\1", t)
+    t = re.sub(r"#{1,6}\s*", "", t)
+    t = re.sub(r"\[(?:PAUSE|BREATH|BEAT|SILENCE|MUSIC|SFX|STAGE DIRECTION)[^\]]*\]", "", t, flags=re.IGNORECASE)
+    t = re.sub(r"\[[^\]]*\]", "", t)
+    t = re.sub(r"\((?:pause|breath|beat|silence)[^)]*\)", "", t, flags=re.IGNORECASE)
+    t = t.replace("\ufffd", " - ")
+    t = t.replace("’", "'").replace("‘", "'").replace("“", '"').replace("”", '"')
+    t = t.replace("—", " - ").replace("–", " - ")
+    t = re.sub(r"\s*\(([^)]+)\)\s*", r" — \1 — ", t)
+    t = t.replace(";", " — ")
+    t = re.sub(r":\s+", " — ", t)
+    t = re.sub(r"\.{3,}", "...", t)
+    t = re.sub(r"\.\.\.(?=[A-Za-z])", "... ", t)
+    t = re.sub(r"\s*—\s*", " — ", t)
+    t = re.sub(r"\s*-\s*", " - ", t)
+    t = re.sub(r"( —\s*)+", " — ", t)
+    t = re.sub(r"[ \t]+", " ", t)
+    return t.strip()
+
+
+class VoiceoverNarrator:
     """
     Broadcast-grade documentary narrator using Kokoro-ONNX and the Grounded Documentary profile.
     """
@@ -44,89 +71,159 @@ class GroundedNarrator:
         model_path: Optional[Union[str, Path]] = None,
         voices_path: Optional[Union[str, Path]] = None,
         profile_path: Optional[Union[str, Path]] = None,
-        repo_id: Optional[str] = None
     ):
-        base_dir = Path(__file__).resolve().parent
-
-        if model_path is None:
-            local_model = base_dir / "kokoro-v1.0.onnx"
-            if local_model.exists():
-                model_path = local_model
-            elif repo_id and hf_hub_download:
-                model_path = hf_hub_download(repo_id=repo_id, filename="kokoro-v1.0.onnx")
-            else:
-                raise FileNotFoundError("kokoro-v1.0.onnx not found. Provide model_path or repo_id.")
-
-        if voices_path is None:
-            local_voices = base_dir / "voices-v1.0.bin"
-            if local_voices.exists():
-                voices_path = local_voices
-            elif repo_id and hf_hub_download:
-                voices_path = hf_hub_download(repo_id=repo_id, filename="voices-v1.0.bin")
-            else:
-                raise FileNotFoundError("voices-v1.0.bin not found. Provide voices_path or repo_id.")
-
-        if profile_path is None:
-            local_prof = base_dir / "grounded_documentary.npy"
-            if local_prof.exists():
-                profile_path = local_prof
-            elif repo_id and hf_hub_download:
-                profile_path = hf_hub_download(repo_id=repo_id, filename="grounded_documentary.npy")
-            else:
-                raise FileNotFoundError("grounded_documentary.npy not found.")
-
-        self.kokoro = Kokoro(str(model_path), str(voices_path))
-        self.voice_style = np.load(str(profile_path))
-
-    def synthesize(
+        root_dir = Path(__file__).resolve().parents[4]  # repo root
+        default_model = root_dir / "models" / "kokoro" / "kokoro-v1.0.onnx"
+        default_voices = root_dir / "models" / "kokoro" / "voices-v1.0.bin"
+        default_profile = root_dir / "models" / "kokoro" / "grounded_documentary.npy"
+
+        self.model_path = Path(model_path) if model_path else default_model
+        self.voices_path = Path(voices_path) if voices_path else default_voices
+        self.profile_path = Path(profile_path) if profile_path else default_profile
+
+        if not self.model_path.exists():
+            raise FileNotFoundError(f"Kokoro model not found at: {self.model_path}")
+        if not self.voices_path.exists():
+            raise FileNotFoundError(f"Voices library not found at: {self.voices_path}")
+
+        self.kokoro = Kokoro(str(self.model_path), str(self.voices_path))
+
+        if self.profile_path.exists():
+            self.voice_style = np.load(str(self.profile_path))
+            self.voice_label = "grounded_documentary (loaded profile)"
+        else:
+            v_daniel = self.kokoro.get_voice_style("bm_daniel")
+            v_george = self.kokoro.get_voice_style("bm_george")
+            self.voice_style = 0.55 * v_george + 0.45 * v_daniel
+            self.voice_label = "grounded_documentary (55% bm_george + 45% bm_daniel)"
+
+    def synthesize_chunks(
+        self,
+        chunks: List[Tuple[str, float]],
+        out_wav: Path,
+        out_mp3: Path,
+        out_json: Optional[Path] = None,
+        section_name: str = "Voiceover",
+        speed: float = 0.91,
+        sentence_pause: float = 0.65,
+        clause_pause: float = 0.30,
+        lang: str = "en-gb",
+        apply_mastering: bool = True,
+    ) -> float:
+        """
+        Synthesizes structured chunks into a broadcast-mastered MP3 audio track.
+        Each chunk is (text, pause_after_seconds).
+        """
+        out_wav.parent.mkdir(parents=True, exist_ok=True)
+        out_mp3.parent.mkdir(parents=True, exist_ok=True)
+
+        audio_segments = []
+        sample_rate = 24000
+        meta_segments = []
+        current_time = 0.0
+        t_start = time.time()
+        total_words = sum(len(text.split()) for text, _ in chunks)
+
+        print(f"\n--- Synthesizing: {section_name} ({len(chunks)} chunks, {total_words} words) ---")
+
+        for idx, (chunk_text, pause_after) in enumerate(chunks, 1):
+            cleaned = clean_text_base(chunk_text)
+            if not cleaned:
+                continue
+            words = len(cleaned.split())
+            print(f"  [{idx}/{len(chunks)}] ({words}w, pause {pause_after:.2f}s): {cleaned[:60]}...")
+
+            seg_audio, sr = self.kokoro.create(
+                cleaned,
+                voice=self.voice_style,
+                speed=speed,
+                lang=lang,
+                sentence_pause=sentence_pause,
+                clause_pause=clause_pause,
+                continuous=False
+            )
+            sample_rate = sr
+            seg_dur = len(seg_audio) / sr
+
+            meta_segments.append({
+                "unit_index": idx,
+                "text": cleaned,
+                "word_count": words,
+                "start_time": round(current_time, 3),
+                "end_time": round(current_time + seg_dur, 3),
+                "duration": round(seg_dur, 3),
+                "pause_after": round(pause_after, 3)
+            })
+            audio_segments.append(seg_audio)
+            current_time += seg_dur
+
+            if pause_after > 0:
+                silence_samples = int(sample_rate * pause_after)
+                audio_segments.append(np.zeros(silence_samples, dtype=np.float32))
+                current_time += pause_after
+
+        full_audio = np.concatenate(audio_segments)
+        total_dur = len(full_audio) / sample_rate
+        elapsed = time.time() - t_start
+
+        # 1. Save WAV
+        sf.write(str(out_wav), full_audio, sample_rate)
+
+        # 2. Master to MP3
+        cmd_mp3 = ["ffmpeg", "-y", "-i", str(out_wav)]
+        if apply_mastering:
+            cmd_mp3.extend(["-af", BROADCAST_AUDIO_FILTER])
+        cmd_mp3.extend([
+            "-c:a", "libmp3lame",
+            "-b:a", "192k",
+            str(out_mp3)
+        ])
+        subprocess.run(cmd_mp3, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+        # 3. Save JSON metadata
+        if out_json:
+            meta = {
+                "section": section_name,
+                "voice_profile": self.voice_label,
+                "lang": lang,
+                "speed": speed,
+                "sentence_pause": sentence_pause,
+                "clause_pause": clause_pause,
+                "sample_rate": sample_rate,
+                "duration_seconds": round(total_dur, 3),
+                "word_count": total_words,
+                "generation_time_seconds": round(elapsed, 2),
+                "mastered": apply_mastering,
+                "units": meta_segments
+            }
+            with open(out_json, "w", encoding="utf-8") as jf:
+                json.dump(meta, jf, indent=2)
+
+        print(f"-> Finished {section_name}: {total_dur:.2f}s (~{total_dur/60:.2f} min) in {elapsed:.1f}s")
+        return total_dur
+
+    def synthesize_text(
         self,
         text: str,
         output_path: Union[str, Path],
-        speed: float = 0.92,
-        sentence_pause: float = 0.42,
-        clause_pause: float = 0.20,
+        speed: float = 0.91,
+        sentence_pause: float = 0.65,
+        clause_pause: float = 0.30,
         apply_mastering: bool = True
     ) -> float:
-        """
-        Synthesizes text to WAV/MP3 with deliberate pacing and broadcast mastering.
-        """
+        """Single-string simple synthesis helper."""
         out_p = Path(output_path)
-        out_p.parent.mkdir(parents=True, exist_ok=True)
         wav_path = out_p.with_suffix(".wav")
-
-        audio, sr = self.kokoro.create(
-            text=text,
-            voice=self.voice_style,
+        mp3_path = out_p.with_suffix(".mp3")
+        chunks = [(p.strip(), 0.85) for p in text.split("\n\n") if p.strip()]
+        return self.synthesize_chunks(
+            chunks=chunks,
+            out_wav=wav_path,
+            out_mp3=mp3_path,
+            out_json=out_p.with_suffix(".json"),
+            section_name=out_p.stem,
             speed=speed,
-            lang="en-gb",
             sentence_pause=sentence_pause,
             clause_pause=clause_pause,
-            continuous=False
+            apply_mastering=apply_mastering
         )
-
-        sf.write(str(wav_path), audio, sr)
-        dur = len(audio) / sr
-
-        if out_p.suffix.lower() == ".mp3" or apply_mastering:
-            mp3_path = out_p.with_suffix(".mp3")
-            cmd = ["ffmpeg", "-y", "-i", str(wav_path)]
-            if apply_mastering:
-                cmd.extend(["-af", DEFAULT_BROADCAST_FILTER])
-            cmd.extend(["-c:a", "libmp3lame", "-b:a", "192k", str(mp3_path)])
-            subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-            if out_p.suffix.lower() == ".mp3" and wav_path != out_p:
-                wav_path.unlink(missing_ok=True)
-
-        return dur
-
-
-if __name__ == "__main__":
-    import argparse
-    parser = argparse.ArgumentParser(description="Generate broadcast documentary narration")
-    parser.add_argument("text", type=str, help="Text to speak")
-    parser.add_argument("-o", "--output", type=str, default="narration.mp3", help="Output path (.mp3 or .wav)")
-    args = parser.parse_args()
-
-    narrator = GroundedNarrator()
-    dur = narrator.synthesize(args.text, args.output)
-    print(f"Generated {dur:.2f}s of mastered voiceover -> {args.output}")