#!/usr/bin/env python3 """Unified local CosyVoice production renderer for Claude and Codex plugins.""" from __future__ import annotations import argparse import hashlib import json import math import os import platform import re import shutil import subprocess import sys import time from difflib import SequenceMatcher from pathlib import Path from typing import Any PLUGIN_ROOT = Path(__file__).resolve().parent.parent PROFILE_DIR = PLUGIN_ROOT / "profiles" PROMPT_PREFIX = "You are a helpful assistant.<|endofprompt|>" MODEL_RELATIVE = Path("models") / "CosyVoice3-0.5B-Candle" ID_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*$") class VoiceError(RuntimeError): """A production failure with an actionable message.""" def platform_default_runtime() -> Path: configured = os.environ.get("LOCAL_VOICE_RUNTIME") if configured: return Path(configured).expanduser() if platform.system() == "Windows": base = Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local")) return base / "MPM Local Voice" / "runtime" if platform.system() == "Darwin": return ( Path.home() / "Library" / "Application Support" / "MPM Local Voice" / "runtime" ) return Path.home() / ".local" / "share" / "mpm-local-voice" / "runtime" def json_read(path: Path) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise VoiceError(f"Missing file: {path}") from exc except json.JSONDecodeError as exc: raise VoiceError(f"Invalid JSON in {path}: {exc}") from exc def json_write(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8") def normalized_tokens(text: str) -> list[str]: return re.findall(r"[a-z0-9]+", text.casefold()) def word_count(text: str) -> int: return len(normalized_tokens(text)) def clean_markdown(text: str) -> str: text = re.sub(r"^\s{0,3}#{1,6}\s+.*$", "", text, flags=re.MULTILINE) text = re.sub(r"^\s*[-*+]\s+", "", text, flags=re.MULTILINE) text = re.sub(r"^\s*\d+[.)]\s+", "", text, flags=re.MULTILINE) text = re.sub(r"^\s*>\s?", "", text, flags=re.MULTILINE) text = re.sub(r"!\[[^\]]*\]\([^)]+\)", "", text) text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text) text = re.sub(r"[*_~`]", "", text) return text def paragraphs_from_markdown(path: Path) -> list[str]: cleaned = clean_markdown(path.read_text(encoding="utf-8")) paragraphs = [ re.sub(r"\s+", " ", block).strip() for block in re.split(r"\n\s*\n", cleaned) ] return [block for block in paragraphs if block] def last_sentence(text: str, maximum_words: int = 10) -> str: parts = re.split(r"(?<=[.!?])\s+", text.strip()) candidate = parts[-1] if parts else text.strip() words = candidate.split() return " ".join(words[-maximum_words:]) def first_sentence(text: str, maximum_words: int = 10) -> str: parts = re.split(r"(?<=[.!?])\s+", text.strip()) selected: list[str] = [] for part in parts or [text.strip()]: selected.extend(part.split()) if len(selected) >= 6: break return " ".join(selected[:maximum_words]) def slug(value: str) -> str: value = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").casefold() return value or "segment" def profile_path(voice: str) -> Path: return PROFILE_DIR / f"{voice}.json" def load_profile(voice: str) -> dict[str, Any]: profile = json_read(profile_path(voice)) if profile.get("id") != voice: raise VoiceError(f"Profile ID mismatch in {profile_path(voice)}") return profile def available_profiles() -> list[dict[str, Any]]: profiles = [] for path in sorted(PROFILE_DIR.glob("*.json")): profile = json_read(path) profiles.append( { "id": profile["id"], "display_name": profile["display_name"], "adapter": profile["adapter"], "roles": sorted(profile["roles"]), } ) return profiles def validate_plan_data(plan: dict[str, Any]) -> dict[str, Any]: if plan.get("schema_version") != "1.0": raise VoiceError("Render plan schema_version must be '1.0'") job_id = str(plan.get("job_id", "")) if not ID_PATTERN.fullmatch(job_id): raise VoiceError("job_id must be filesystem-safe") voice = str(plan.get("voice", "")) profile = load_profile(voice) output = Path(str(plan.get("output", ""))).expanduser() if not output.is_absolute(): raise VoiceError("Render plan output must be an absolute path") if output.suffix.casefold() not in {".wav", ".mp3"}: raise VoiceError("Render plan output must end in .wav or .mp3") segments = plan.get("segments") if not isinstance(segments, list) or not segments: raise VoiceError("Render plan must contain at least one segment") seen: set[str] = set() roles = set(profile["roles"]) deliveries = set(profile.get("deliveries", {})) for index, segment in enumerate(segments, start=1): if not isinstance(segment, dict): raise VoiceError(f"Segment {index} must be an object") segment_id = str(segment.get("id", "")) if not ID_PATTERN.fullmatch(segment_id): raise VoiceError(f"Segment {index} has invalid id {segment_id!r}") if segment_id in seen: raise VoiceError(f"Duplicate segment id: {segment_id}") seen.add(segment_id) text = str(segment.get("text", "")).strip() if not text: raise VoiceError(f"{segment_id}: canonical text is required") role = segment.get("role") if role is not None and role not in roles: raise VoiceError(f"{segment_id}: unknown role {role!r}") if profile["adapter"] == "dialogue" and role is None: raise VoiceError(f"{segment_id}: dialogue segments require a role") delivery = segment.get("delivery") if delivery is not None and delivery not in deliveries: raise VoiceError(f"{segment_id}: unknown delivery {delivery!r}") fixed = segment.get("fixed_asset") pool = segment.get("asset_pool") if fixed and pool: raise VoiceError(f"{segment_id}: choose fixed_asset or asset_pool") if pool is not None and ( not isinstance(pool, list) or not all(isinstance(x, str) for x in pool) ): raise VoiceError(f"{segment_id}: asset_pool must be a list of paths") if segment.get("target_wpm") and segment.get("tempo_multiplier"): raise VoiceError( f"{segment_id}: target_wpm and tempo_multiplier are mutually exclusive" ) return profile def build_plan(args: argparse.Namespace) -> dict[str, Any]: profile = load_profile(args.voice) max_words = int(profile["generation"]["maximum_words"]) short_max = int(profile["generation"]["short_native_max_words"]) raw_blocks = paragraphs_from_markdown(args.script) if not raw_blocks: raise VoiceError("Script contains no speakable text") segments: list[dict[str, Any]] = [] if profile["adapter"] == "dialogue": role_pattern = re.compile( r"^(?:\*\*)?(PRODUCER(?:\s+GUY)?|EXEC(?:\s+GUY)?|" r"WRITER(?:\s+GUY)?)(?:\*\*)?\s*:\s*(.+)$", re.IGNORECASE, ) for block in raw_blocks: match = role_pattern.match(block) if not match: raise VoiceError( "Ryan scripts require Producer/Writer labels on every paragraph" ) label, spoken = match.groups() role = ( "writer" if label.casefold().startswith("writer") else "producer" ) segments.append( { "id": f"{len(segments) + 1:02d}-{role}", "role": role, "text": spoken.strip(), } ) else: packed: list[str] = [] for index, block in enumerate(raw_blocks): is_short_frame = ( word_count(block) <= short_max and index in {0, len(raw_blocks) - 1} ) if is_short_frame: if packed: segments.append( { "id": f"{len(segments) + 1:02d}-passage", "text": " ".join(packed), "delivery": "narrative" if "narrative" in profile.get("deliveries", {}) else None, } ) packed = [] segments.append( { "id": f"{len(segments) + 1:02d}-frame", "text": block, "native_tempo": True, } ) continue proposed = " ".join([*packed, block]) if packed and word_count(proposed) > max_words: segments.append( { "id": f"{len(segments) + 1:02d}-passage", "text": " ".join(packed), "delivery": "narrative" if "narrative" in profile.get("deliveries", {}) else None, } ) packed = [block] else: packed.append(block) if packed: segments.append( { "id": f"{len(segments) + 1:02d}-passage", "text": " ".join(packed), "delivery": "narrative" if "narrative" in profile.get("deliveries", {}) else None, } ) for segment in segments: if segment.get("delivery") is None: segment.pop("delivery", None) for index, segment in enumerate(segments): if index: segment["context_before"] = last_sentence( segments[index - 1]["text"] ) if index + 1 < len(segments): segment["context_after"] = first_sentence( segments[index + 1]["text"] ) segment.setdefault( "pause_after_ms", int(profile["assembly"]["default_pause_ms"]), ) plan = { "schema_version": "1.0", "job_id": args.job_id or slug(args.script.stem), "voice": args.voice, "output": str(args.output_audio.resolve()), "final_transcript_qa": True, "segments": segments, } validate_plan_data(plan) json_write(args.output_plan, plan) return plan def resolve_runtime(path: str | None) -> Path: return ( Path(path).expanduser().resolve() if path else platform_default_runtime().resolve() ) def executable(name: str, preferred: str | None = None) -> str: if preferred: candidate = Path(preferred).expanduser() if candidate.exists(): return str(candidate) found = shutil.which(name) if not found: raise VoiceError(f"Required executable is unavailable: {name}") return found def resolve_asset(relative: str, runtime: Path) -> Path: candidate = Path(relative).expanduser() if candidate.is_absolute() and candidate.exists(): return candidate for base in (runtime, PLUGIN_ROOT): resolved = base / candidate if resolved.exists(): return resolved raise VoiceError(f"Missing authorized asset: {relative}") def select_asset(segment: dict[str, Any], plan: dict[str, Any]) -> str | None: if segment.get("fixed_asset"): return str(segment["fixed_asset"]) pool = segment.get("asset_pool") if not pool: return None digest = hashlib.sha256( f"{plan['job_id']}:{segment['id']}".encode() ).digest() return str(pool[int.from_bytes(digest[:4], "big") % len(pool)]) def role_for_segment( segment: dict[str, Any], profile: dict[str, Any], ) -> tuple[str, dict[str, Any], dict[str, Any]]: delivery = profile.get("deliveries", {}).get(segment.get("delivery"), {}) role_name = segment.get("role") or delivery.get("role") or "default" role = profile["roles"][role_name] return role_name, role, delivery def job_paths(plan: dict[str, Any], runtime: Path) -> dict[str, Path]: root = runtime / "jobs" / plan["job_id"] return { "root": root, "raw": root / "raw", "alignment": root / "alignment", "processed": root / "processed", "qa": root / "qa", } def import_audio_stack() -> tuple[Any, Any]: try: import numpy as np import soundfile as sf except ImportError as exc: raise VoiceError( "numpy and soundfile must be installed in the Local Voice environment" ) from exc return np, sf def import_model() -> tuple[Any, Any]: try: from cosyvoice3 import CosyVoice3, PyDevice except ImportError as exc: raise VoiceError( "cosyvoice3 is not installed in this Python environment" ) from exc return CosyVoice3, PyDevice def validate_generated(samples: Any, segment_id: str, np: Any) -> None: samples = np.asarray(samples).squeeze() if samples.ndim != 1 or samples.size == 0: raise VoiceError(f"{segment_id}: unexpected audio shape {samples.shape}") if not np.isfinite(samples).all(): raise VoiceError(f"{segment_id}: generated non-finite audio") def generate_segments( plan: dict[str, Any], profile: dict[str, Any], runtime: Path, paths: dict[str, Path], *, device_name: str, resume: bool, ) -> None: np, sf = import_audio_stack() CosyVoice3, PyDevice = import_model() model_dir = runtime / MODEL_RELATIVE if not model_dir.exists(): raise VoiceError(f"Missing Candle model directory: {model_dir}") generated = [ segment for segment in plan["segments"] if not select_asset(segment, plan) ] pending = [ segment for segment in generated if not (resume and (paths["raw"] / f"{segment['id']}.wav").exists()) ] if not pending: print("All raw speech segments already exist.") return print(f"Loading CosyVoice3 on {device_name}...") model = CosyVoice3( str(model_dir), device=PyDevice(device_name), use_f16=False, ) for index, segment in enumerate(pending, start=1): _, role, _ = role_for_segment(segment, profile) reference_audio = resolve_asset(role["reference_audio"], runtime) reference_text = resolve_asset(role["reference_transcript"], runtime) context = segment.get("context_before") if context is None: context = role.get( "default_context", profile["generation"].get("default_context", ""), ) target = str(segment.get("tts_text", segment["text"])).strip() spoken = target if context: spoken = f"{str(context).strip()} ... ... ... {spoken}" if segment.get("context_after"): spoken += f" ... ... ... {str(segment['context_after']).strip()}" prompt_text = PROMPT_PREFIX + reference_text.read_text( encoding="utf-8" ).strip() started = time.perf_counter() audio = model.inference_zero_shot( text=spoken, prompt_text=prompt_text, prompt_wav=str(reference_audio), ) elapsed = time.perf_counter() - started samples = np.asarray(audio, dtype=np.float32).squeeze() validate_generated(samples, segment["id"], np) destination = paths["raw"] / f"{segment['id']}.wav" sf.write(destination, samples, model.sample_rate, subtype="PCM_16") duration = samples.size / model.sample_rate print( f"[{index}/{len(pending)}] {segment['id']}: " f"{duration:.2f}s in {elapsed:.2f}s" ) def run_whisper( source: Path, output_dir: Path, *, whisper_path: str | None, ) -> Path: whisper = executable("whisper", whisper_path) output_dir.mkdir(parents=True, exist_ok=True) subprocess.run( [ whisper, str(source), "--model", "base", "--device", "cpu", "--fp16", "False", "--language", "en", "--word_timestamps", "True", "--output_format", "json", "--output_dir", str(output_dir), "--verbose", "False", ], check=True, ) result = output_dir / f"{source.stem}.json" if not result.exists(): raise VoiceError(f"Whisper did not create {result}") return result def align_segments( plan: dict[str, Any], runtime: Path, paths: dict[str, Path], *, whisper_path: str | None, ) -> None: for segment in plan["segments"]: if select_asset(segment, plan): continue raw = paths["raw"] / f"{segment['id']}.wav" if not raw.exists(): raise VoiceError(f"{segment['id']}: missing raw WAV") alignment = paths["alignment"] / f"{segment['id']}.json" if not alignment.exists() or raw.stat().st_mtime > alignment.stat().st_mtime: run_whisper( raw, paths["alignment"], whisper_path=whisper_path, ) def aligned_words(path: Path) -> list[dict[str, Any]]: data = json_read(path) return [ word for segment in data.get("segments", []) for word in segment.get("words", []) if normalized_tokens(str(word.get("word", ""))) ] def locate_phrase( phrase: str, words: list[dict[str, Any]], *, start_at: int = 0, minimum_score: float = 0.55, ) -> tuple[int, int, float]: expected = normalized_tokens(phrase)[:10] if len(expected) < 2: raise VoiceError(f"Phrase too short for alignment: {phrase!r}") heard = [normalized_tokens(str(word["word"]))[0] for word in words] best = (-1.0, -1, -1) for start in range(start_at, len(heard)): for width in range(max(2, len(expected) - 3), len(expected) + 4): candidate = heard[start:start + width] if len(candidate) < 2: continue score = SequenceMatcher( None, "".join(expected), "".join(candidate), ).ratio() if score > best[0]: best = (score, start, width) if best[0] < minimum_score: raise VoiceError( f"Could not locate {phrase!r}; confidence={best[0]:.3f}" ) return best[1], best[2], best[0] def dbfs(value: float) -> float: return 20 * math.log10(max(value, 1e-10)) def quiet_cut( audio: Any, sample_rate: int, *, lower: float, upper: float, target: float, ceiling: float, np: Any, ) -> tuple[int, float]: lo = max(0, round(lower * sample_rate)) hi = min(len(audio), round(upper * sample_rate)) window = max(1, round(0.010 * sample_rate)) if hi - lo < window: fallback_lo = max(0, round((upper - 0.14) * sample_rate)) fallback_hi = min( len(audio), max(fallback_lo + 1, round((upper - 0.035) * sample_rate)), ) section = np.abs(audio[fallback_lo:fallback_hi]) chosen = fallback_lo + int(np.argmin(section)) local = audio[ max(0, chosen - window // 2):min(len(audio), chosen + window // 2) ] return chosen, dbfs(float(np.sqrt(np.mean(local ** 2)))) squared = audio[lo:hi] ** 2 rms = np.sqrt( np.convolve(squared, np.ones(window) / window, mode="valid") ) centers = np.arange(len(rms)) + window // 2 candidates = np.flatnonzero(rms <= 10 ** (ceiling / 20)) if not candidates.size: chosen = int(np.argmin(rms)) return lo + int(centers[chosen]), dbfs(float(rms[chosen])) desired = round(target * sample_rate) - lo chosen = candidates[int(np.argmin(np.abs(centers[candidates] - desired)))] return lo + int(centers[chosen]), dbfs(float(rms[chosen])) def transcode_asset( source: Path, destination: Path, *, ffmpeg_path: str | None, ) -> None: ffmpeg = executable("ffmpeg", ffmpeg_path) destination.parent.mkdir(parents=True, exist_ok=True) subprocess.run( [ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-i", str(source), "-filter:a", ( "loudnorm=I=-18:LRA=7:TP=-1.5," "afade=t=in:st=0:d=0.008," "areverse,afade=t=in:st=0:d=0.008,areverse" ), "-ar", "24000", "-ac", "1", "-codec:a", "pcm_s16le", str(destination), ], check=True, ) def tempo_for_segment( segment: dict[str, Any], profile: dict[str, Any], role: dict[str, Any], delivery: dict[str, Any], *, body_duration: float, target_words: int, ) -> tuple[float, float | None, float]: raw_wpm = target_words / max(body_duration, 0.001) * 60 short_limit = int(profile["generation"]["short_native_max_words"]) cadence = profile["cadence"] minimum_words = int(cadence["minimum_words_for_normalization"]) if ( segment.get("native_tempo") or target_words <= short_limit or target_words < minimum_words ): return 1.0, None, raw_wpm multiplier = segment.get("tempo_multiplier") target_wpm = segment.get("target_wpm") if multiplier is None and target_wpm is None: multiplier = delivery.get("tempo_multiplier") target_wpm = delivery.get("target_wpm") if multiplier is None and target_wpm is None: multiplier = role.get("tempo_multiplier") if multiplier is None and target_wpm is None: mode = cadence.get("default_mode", "native") if mode == "multiplier": multiplier = cadence.get("default_tempo_multiplier", 1.0) elif mode == "wpm": target_wpm = cadence.get("default_wpm") if target_wpm is not None: tempo = float(target_wpm) / raw_wpm else: tempo = float(multiplier if multiplier is not None else 1.0) minimum = float(cadence["minimum_tempo"]) maximum = float(cadence["maximum_tempo"]) if not minimum <= tempo <= maximum: raise VoiceError( f"{segment['id']}: cadence correction {tempo:.3f} " f"outside validated range {minimum:.2f}-{maximum:.2f}" ) return tempo, float(target_wpm) if target_wpm is not None else None, raw_wpm def apply_processing( source: Path, destination: Path, *, tempo: float, ffmpeg_path: str | None, ) -> None: ffmpeg = executable("ffmpeg", ffmpeg_path) subprocess.run( [ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-i", str(source), "-filter:a", ( f"atempo={tempo:.6f}," "loudnorm=I=-18:LRA=7:TP=-1.5," "afade=t=in:st=0:d=0.008," "areverse,afade=t=in:st=0:d=0.008,areverse" ), "-ar", "24000", "-ac", "1", "-codec:a", "pcm_s16le", str(destination), ], check=True, ) def append_suffix( body: Any, suffix: Any, *, gap_ms: int, sample_rate: int, np: Any, ) -> Any: gap = np.zeros(round(gap_ms / 1000 * sample_rate), dtype=np.float32) return np.concatenate([body, gap, suffix]) def process_generated_segment( segment: dict[str, Any], profile: dict[str, Any], runtime: Path, paths: dict[str, Path], *, ffmpeg_path: str | None, ) -> tuple[Any, dict[str, Any]]: np, sf = import_audio_stack() raw = paths["raw"] / f"{segment['id']}.wav" alignment = paths["alignment"] / f"{segment['id']}.json" words = aligned_words(alignment) target_text = str(segment.get("tts_text", segment["text"])) context = segment.get("context_before") role_name, role, delivery = role_for_segment(segment, profile) if context is None: context = role.get( "default_context", profile["generation"].get("default_context", ""), ) start_index, _, match_confidence = locate_phrase( target_text, words, start_at=2 if context else 0, ) expected_tokens = normalized_tokens(target_text) ending_marker = " ".join(expected_tokens[-min(7, len(expected_tokens)):]) ending_start, ending_width, marker_confidence = locate_phrase( ending_marker, words, start_at=max(start_index, start_index + len(expected_tokens) - 12), minimum_score=0.45, ) after_index = min(len(words), ending_start + ending_width) after_confidence: float | None = None if segment.get("context_after"): try: context_start, _, after_confidence = locate_phrase( str(segment["context_after"]), words, start_at=after_index, minimum_score=0.45, ) after_index = context_start except VoiceError: after_confidence = None target_words = words[start_index:after_index] if not target_words: raise VoiceError(f"{segment['id']}: alignment selected no target words") body_start = float(target_words[0]["start"]) body_end = float(target_words[-1]["end"]) audio, sample_rate = sf.read(raw, dtype="float32") audio = np.asarray(audio).reshape(-1) if start_index == 0: start_cut = 0 start_level = dbfs( float(np.sqrt(np.mean(audio[:max(1, sample_rate // 100)] ** 2))) ) else: prior_end = float(words[start_index - 1]["end"]) start_cut, start_level = quiet_cut( audio, sample_rate, lower=prior_end, upper=body_start, target=body_start - 0.10, ceiling=-30.0, np=np, ) start_cut = max(start_cut, round(prior_end * sample_rate)) if after_index == len(words): end_cut = len(audio) end_level = dbfs( float(np.sqrt(np.mean(audio[-max(1, sample_rate // 100):] ** 2))) ) else: after_start = float(words[after_index]["start"]) end_cut, end_level = quiet_cut( audio, sample_rate, lower=body_end + 0.030, upper=after_start, target=body_end + int(profile["assembly"]["minimum_natural_tail_ms"]) / 1000, ceiling=-38.0, np=np, ) minimum_end = min( len(audio), round( ( body_end + min( 0.08, int(profile["assembly"]["minimum_natural_tail_ms"]) / 1000, ) ) * sample_rate ), ) end_cut = max(end_cut, minimum_end) selected = audio[start_cut:end_cut] expected = normalized_tokens(target_text) heard = [ normalized_tokens(str(word["word"]))[0] for word in target_words if normalized_tokens(str(word["word"])) ] matcher = SequenceMatcher(None, expected, heard, autojunk=False) matched = sum(block.size for block in matcher.get_matching_blocks()) coverage = matched / max(1, len(expected)) ending_expected = expected[-min(7, len(expected)):] ending_heard = heard[-min(9, len(heard)):] ending = SequenceMatcher( None, ending_expected, ending_heard, autojunk=False, ).ratio() if coverage < 0.65 or ending < 0.50: raise VoiceError( f"{segment['id']}: incomplete target; " f"coverage={coverage:.3f}, ending={ending:.3f}" ) tempo, target_wpm, raw_wpm = tempo_for_segment( segment, profile, role, delivery, body_duration=body_end - body_start, target_words=len(expected), ) trimmed = paths["processed"] / f"{segment['id']}-trimmed.wav" ready = paths["processed"] / f"{segment['id']}.wav" sf.write(trimmed, selected, sample_rate, subtype="PCM_16") apply_processing( trimmed, ready, tempo=tempo, ffmpeg_path=ffmpeg_path, ) ready_audio, ready_rate = sf.read(ready, dtype="float32") if ready_rate != 24000: raise VoiceError(f"{segment['id']}: processed sample rate is not 24 kHz") result = np.asarray(ready_audio).reshape(-1) if segment.get("suffix_asset"): suffix_source = resolve_asset(str(segment["suffix_asset"]), runtime) suffix_ready = paths["processed"] / f"{segment['id']}-suffix.wav" transcode_asset( suffix_source, suffix_ready, ffmpeg_path=ffmpeg_path, ) suffix, suffix_rate = sf.read(suffix_ready, dtype="float32") if suffix_rate != 24000: raise VoiceError(f"{segment['id']}: suffix sample rate mismatch") result = append_suffix( result, np.asarray(suffix).reshape(-1), gap_ms=int(segment.get("suffix_gap_ms", 90)), sample_rate=24000, np=np, ) sf.write(ready, result, 24000, subtype="PCM_16") return result, { "id": segment["id"], "source": "generated", "role": role_name, "delivery": segment.get("delivery"), "target_match_confidence": round(match_confidence, 3), "after_match_confidence": ( round(after_confidence, 3) if after_confidence is not None else None ), "transcript_coverage": round(coverage, 3), "ending_confidence": round(ending, 3), "ending_marker_confidence": round(marker_confidence, 3), "retained_pre_word_ms": round( (body_start - start_cut / sample_rate) * 1000, 1, ), "retained_post_word_ms": round( (end_cut / sample_rate - body_end) * 1000, 1, ), "start_cut_dbfs": round(start_level, 1), "end_cut_dbfs": round(end_level, 1), "raw_wpm": round(raw_wpm, 1), "target_wpm": round(target_wpm, 1) if target_wpm else None, "applied_tempo": round(tempo, 4), "native_tempo": tempo == 1.0, "suffix_asset": segment.get("suffix_asset"), } def process_asset_segment( segment: dict[str, Any], plan: dict[str, Any], runtime: Path, paths: dict[str, Path], *, ffmpeg_path: str | None, ) -> tuple[Any, dict[str, Any]]: np, sf = import_audio_stack() selected = select_asset(segment, plan) if not selected: raise VoiceError(f"{segment['id']}: asset selection failed") source = resolve_asset(selected, runtime) ready = paths["processed"] / f"{segment['id']}.wav" transcode_asset(source, ready, ffmpeg_path=ffmpeg_path) audio, sample_rate = sf.read(ready, dtype="float32") if sample_rate != 24000: raise VoiceError(f"{segment['id']}: fixed asset is not 24 kHz") return np.asarray(audio).reshape(-1), { "id": segment["id"], "source": "fixed_asset", "asset": str(source), "native_tempo": True, "duration_seconds": round(len(audio) / sample_rate, 3), } def assemble( plan: dict[str, Any], profile: dict[str, Any], runtime: Path, paths: dict[str, Path], *, ffmpeg_path: str | None, whisper_path: str | None, final_qa: bool, ) -> dict[str, Any]: np, sf = import_audio_stack() processed: list[Any] = [] reports: list[dict[str, Any]] = [] for segment in plan["segments"]: if select_asset(segment, plan): audio, report = process_asset_segment( segment, plan, runtime, paths, ffmpeg_path=ffmpeg_path, ) else: audio, report = process_generated_segment( segment, profile, runtime, paths, ffmpeg_path=ffmpeg_path, ) processed.append(audio) reports.append(report) crossfade = round( int(profile["assembly"]["crossfade_ms"]) / 1000 * 24000 ) joined = processed[0] seam_times: list[float] = [] for index, right in enumerate(processed[1:], start=1): pause_ms = int( plan["segments"][index - 1].get( "pause_after_ms", profile["assembly"]["default_pause_ms"], ) ) if pause_ms > 0: joined = np.concatenate( [ joined, np.zeros(round(pause_ms / 1000 * 24000), dtype=np.float32), right, ] ) seam_times.append((len(joined) - len(right)) / 24000) else: overlap = min(crossfade, len(joined), len(right)) if overlap: theta = np.linspace( 0, np.pi / 2, overlap, dtype=np.float32, ) seam_times.append((len(joined) - overlap) / 24000) joined = np.concatenate( [ joined[:-overlap], joined[-overlap:] * np.cos(theta) + right[:overlap] * np.sin(theta), right[overlap:], ] ) else: seam_times.append(len(joined) / 24000) joined = np.concatenate([joined, right]) output = Path(plan["output"]) output.parent.mkdir(parents=True, exist_ok=True) wav_output = output if output.suffix.casefold() == ".wav" else output.with_suffix( ".wav" ) sf.write(wav_output, joined, 24000, subtype="PCM_16") if output.suffix.casefold() == ".mp3": ffmpeg = executable("ffmpeg", ffmpeg_path) subprocess.run( [ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-i", str(wav_output), "-codec:a", "libmp3lame", "-q:a", "2", str(output), ], check=True, ) derivative = np.abs(np.diff(joined)) seam_report = [] for seam in seam_times: center = round(seam * 24000) lo = max(0, center - round(0.08 * 24000)) hi = min(len(derivative), center + round(0.14 * 24000)) boundary_index = min(max(0, center - 1), len(derivative) - 1) boundary_step = ( float(derivative[boundary_index]) if len(derivative) else 0.0 ) seam_report.append( { "time_seconds": round(seam, 3), "boundary_step": round(boundary_step, 6), "click_risk": boundary_step > 0.08, "maximum_derivative": round( float(derivative[lo:hi].max()) if hi > lo else 0.0, 6, ), } ) report: dict[str, Any] = { "schema_version": "1.0", "job_id": plan["job_id"], "voice": plan["voice"], "output": str(output), "wav_output": str(wav_output), "duration_seconds": round(len(joined) / 24000, 3), "segments": reports, "seams": seam_report, "global_maximum_derivative": round(float(derivative.max()), 6), } if final_qa: final_alignment = run_whisper( output, paths["qa"], whisper_path=whisper_path, ) data = json_read(final_alignment) heard_text = " ".join( str(segment.get("text", "")).strip() for segment in data.get("segments", []) ).strip() expected = normalized_tokens( " ".join(segment["text"] for segment in plan["segments"]) ) heard = normalized_tokens(heard_text) matcher = SequenceMatcher(None, expected, heard, autojunk=False) matched = sum(block.size for block in matcher.get_matching_blocks()) coverage = matched / max(1, len(expected)) precision = matched / max(1, len(heard)) passed = coverage >= 0.60 and precision >= 0.75 report["final_transcript_qa"] = { "coverage": round(coverage, 3), "precision": round(precision, 3), "heard_text": heard_text, "passed": passed, } if not passed: json_write(paths["root"] / "qa-report.json", report) raise VoiceError( "Final transcript QA failed with " f"coverage={coverage:.3f}, precision={precision:.3f}" ) report_path = paths["root"] / "qa-report.json" json_write(report_path, report) report["report_path"] = str(report_path) return report def doctor(runtime: Path, device_name: str) -> int: checks: list[tuple[str, bool, str]] = [] checks.append( ( "Python", sys.version_info >= (3, 10), platform.python_version(), ) ) for name in ("ffmpeg", "whisper"): path = shutil.which(name) checks.append((name, path is not None, path or "not found")) for module in ("numpy", "soundfile", "cosyvoice3"): try: __import__(module) checks.append((module, True, "importable")) except ImportError: checks.append((module, False, "not importable")) model = runtime / MODEL_RELATIVE required_model_files = [ "llm.safetensors", "flow.safetensors", "hift.safetensors", "campplus.onnx", "speech_tokenizer_v3.onnx", "config.json", ] missing_model = [name for name in required_model_files if not (model / name).exists()] checks.append( ( "Candle model", not missing_model, str(model) if not missing_model else "missing " + ", ".join(missing_model), ) ) for summary in available_profiles(): try: profile = load_profile(summary["id"]) missing = [] for role in profile["roles"].values(): for field in ("reference_audio", "reference_transcript"): try: resolve_asset(role[field], runtime) except VoiceError: missing.append(role[field]) checks.append( ( summary["display_name"], not missing, "ready" if not missing else "missing assets: " + ", ".join(missing), ) ) except VoiceError as exc: checks.append((summary["display_name"], False, str(exc))) print(f"Runtime: {runtime}") print(f"Requested device: {device_name}") for name, passed, detail in checks: print(f"{'PASS' if passed else 'FAIL'} {name}: {detail}") return 0 if all(item[1] for item in checks) else 1 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(prog="local_voice") subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("list", help="List installed voice profiles") validate_parser = subparsers.add_parser( "validate", help="Validate a render plan without generating audio", ) validate_parser.add_argument("plan", type=Path) plan_parser = subparsers.add_parser( "plan", help="Create a basic render plan from Markdown", ) plan_parser.add_argument("--voice", required=True) plan_parser.add_argument("--script", required=True, type=Path) plan_parser.add_argument("--output-plan", required=True, type=Path) plan_parser.add_argument("--output-audio", required=True, type=Path) plan_parser.add_argument("--job-id") doctor_parser = subparsers.add_parser( "doctor", help="Check the installed runtime and authorized voice assets", ) doctor_parser.add_argument("--runtime-root") doctor_parser.add_argument("--device", default="metal") render_parser = subparsers.add_parser( "render", help="Generate, align, process, and assemble a render plan", ) render_parser.add_argument("plan", type=Path) render_parser.add_argument("--runtime-root") render_parser.add_argument( "--device", choices=("metal", "cpu", "cuda"), default="metal" if platform.system() == "Darwin" else "cpu", ) render_parser.add_argument("--resume", action="store_true") render_parser.add_argument("--assemble-only", action="store_true") render_parser.add_argument("--skip-final-qa", action="store_true") render_parser.add_argument("--ffmpeg-path") render_parser.add_argument("--whisper-path") return parser.parse_args() def main() -> int: args = parse_args() if args.command == "list": print(json.dumps(available_profiles(), indent=2)) return 0 if args.command == "validate": plan = json_read(args.plan) profile = validate_plan_data(plan) print( json.dumps( { "valid": True, "voice": profile["id"], "segments": len(plan["segments"]), "output": plan["output"], }, indent=2, ) ) return 0 if args.command == "plan": plan = build_plan(args) print(json.dumps(plan, indent=2)) return 0 if args.command == "doctor": return doctor(resolve_runtime(args.runtime_root), args.device) if args.command == "render": plan = json_read(args.plan) profile = validate_plan_data(plan) runtime = resolve_runtime(args.runtime_root) paths = job_paths(plan, runtime) for path in paths.values(): path.mkdir(parents=True, exist_ok=True) if not args.assemble_only: generate_segments( plan, profile, runtime, paths, device_name=args.device, resume=args.resume, ) align_segments( plan, runtime, paths, whisper_path=args.whisper_path, ) report = assemble( plan, profile, runtime, paths, ffmpeg_path=args.ffmpeg_path, whisper_path=args.whisper_path, final_qa=bool( plan.get("final_transcript_qa", True) and not args.skip_final_qa ), ) print(json.dumps(report, indent=2)) return 0 raise VoiceError(f"Unsupported command: {args.command}") if __name__ == "__main__": try: raise SystemExit(main()) except VoiceError as exc: print(f"ERROR: {exc}", file=sys.stderr) raise SystemExit(2)