| |
| """ |
| narrator.py |
| ----------- |
| 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, List, Tuple, Dict, Any |
| import numpy as np |
| import soundfile as sf |
| from kokoro_onnx import Kokoro |
|
|
| from .pronunciation import apply_pronunciation_lexicon |
|
|
| |
| 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.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" |
| ) |
|
|
|
|
| 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. |
| """ |
|
|
| def __init__( |
| self, |
| model_path: Optional[Union[str, Path]] = None, |
| voices_path: Optional[Union[str, Path]] = None, |
| profile_path: Optional[Union[str, Path]] = None, |
| ): |
| root_dir = Path(__file__).resolve().parents[4] |
| 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 |
|
|
| |
| sf.write(str(out_wav), full_audio, sample_rate) |
|
|
| |
| 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) |
|
|
| |
| 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.91, |
| sentence_pause: float = 0.65, |
| clause_pause: float = 0.30, |
| apply_mastering: bool = True |
| ) -> float: |
| """Single-string simple synthesis helper.""" |
| out_p = Path(output_path) |
| wav_path = out_p.with_suffix(".wav") |
| 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, |
| sentence_pause=sentence_pause, |
| clause_pause=clause_pause, |
| apply_mastering=apply_mastering |
| ) |
|
|
