"""Generate the 20 Russian GLaDOS teacher clips with CosyVoice 3 VC. The English Portal WAV remains the speaker/timbre reference. Silero + RuAccent provides only Russian content, and Rubber Band first matches the reference duration. The optional pitch pass aligns the generated median F0 to the English reference while preserving formants. """ from __future__ import annotations import argparse import csv from dataclasses import asdict import json import os from pathlib import Path import sys import tempfile try: from glados_ru.align_pitch import align_median_f0 from glados_ru.pipeline_config import load_controls, resolve_line from glados_ru.prosody_profile import analyze, warp_duration except ModuleNotFoundError: # Support running the copied bundle directly. from align_pitch import align_median_f0 from pipeline_config import load_controls, resolve_line from prosody_profile import analyze, warp_duration ROOT = Path(__file__).parent DEFAULT_MODEL = os.environ.get( "COSYVOICE_MODEL_DIR", "pretrained_models/Fun-CosyVoice3-0.5B", ) DEFAULT_COSYVOICE_ROOT = Path(os.environ.get("COSYVOICE_ROOT", "CosyVoice")) DEFAULT_SOURCE = ROOT / "sources_silero" DEFAULT_REFERENCE = ROOT / "references" DEFAULT_PROFILE = ROOT / "profiles_batch" DEFAULT_WARPED = ROOT / "profiled_sources_batch" DEFAULT_OUTPUT = ROOT / "generated_cosyvoice_vc_batch" DEFAULT_PITCH = ROOT / "generated_cosyvoice_vc_batch_pitch" DEFAULT_CONTROLS = ROOT / "pipeline_controls.json" def ids_from_files(directory: Path) -> list[str]: return sorted(path.stem for path in directory.glob("*.wav")) def translations(root: Path) -> dict[str, str]: result: dict[str, str] = {} for path in sorted(root.glob("translations_*.tsv")): with path.open(encoding="utf-8", newline="") as handle: for row in csv.DictReader(handle, delimiter="\t"): result[row["id"].zfill(4)] = row["ru"] if not result: raise ValueError(f"no translations_*.tsv files found in {root}") return result def generate( *, model_dir: str, cosyvoice_root: Path, translations_dir: Path, source_dir: Path, reference_dir: Path, profile_dir: Path, warped_dir: Path, output_dir: Path, pitch_dir: Path, controls_path: Path, selected_ids: list[str], overwrite: bool, ) -> None: import torch import torchaudio sys.path.insert(0, str(cosyvoice_root)) sys.path.insert(0, str(cosyvoice_root / "third_party" / "Matcha-TTS")) from cosyvoice.cli.cosyvoice import AutoModel output_dir.mkdir(parents=True, exist_ok=True) pitch_dir.mkdir(parents=True, exist_ok=True) profile_dir.mkdir(parents=True, exist_ok=True) warped_dir.mkdir(parents=True, exist_ok=True) cosyvoice = AutoModel(model_dir=model_dir, fp16=False) translated = translations(translations_dir) controls = load_controls(controls_path) for item_id in selected_ids: canonical_text = translated.get(item_id, "") line = resolve_line(item_id, canonical_text, controls) source = source_dir / f"{item_id}.wav" reference = reference_dir / f"{item_id}.wav" speaker_reference = reference_dir / f"{line.speaker_reference}.wav" if not source.is_file() or not reference.is_file() or not speaker_reference.is_file(): print(f"skip {item_id}: missing source or reference", flush=True) continue profile_path = profile_dir / f"{item_id}.json" warped = warped_dir / f"{item_id}.wav" if overwrite or not profile_path.exists() or not warped.exists(): profile = analyze(reference, canonical_text) profile["source"] = analyze(source, line.tts_text) profile["pipeline_control"] = asdict(line) profile_path.write_text( json.dumps(profile, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) warp_duration(source, reference, warped) else: profile = json.loads(profile_path.read_text(encoding="utf-8")) expected_control = asdict(line) if profile.get("pipeline_control") != expected_control: profile["pipeline_control"] = expected_control profile_path.write_text( json.dumps(profile, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) target = output_dir / f"{item_id}.wav" if overwrite or not target.exists(): with torch.inference_mode(): results = cosyvoice.inference_vc( str(warped if line.duration_stage == "pre" else source), str(speaker_reference), stream=False, speed=1.0, ) first = next(iter(results)) if line.duration_stage == "pre": torchaudio.save( str(target), first["tts_speech"].cpu(), cosyvoice.sample_rate, ) else: with tempfile.TemporaryDirectory(prefix=f"glados-vc-{item_id}-") as temp_dir: raw_target = Path(temp_dir) / "raw-vc.wav" torchaudio.save( str(raw_target), first["tts_speech"].cpu(), cosyvoice.sample_rate, ) warp_duration(raw_target, reference, target) print(f"generated {target}", flush=True) else: print(f"skip existing {target}", flush=True) pitch_target = pitch_dir / f"{item_id}.wav" if overwrite or not pitch_target.exists(): result = align_median_f0(target, reference, pitch_target) print( f"pitch {item_id}: source={result.source_f0:.1f}Hz " f"reference={result.reference_f0:.1f}Hz output={result.output_f0:.1f}Hz " f"shift={result.semitones:.2f}st error={result.relative_error:.2%}", flush=True, ) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--model-dir", default=DEFAULT_MODEL) parser.add_argument("--cosyvoice-root", type=Path, default=DEFAULT_COSYVOICE_ROOT) parser.add_argument("--translations-dir", type=Path, default=ROOT) parser.add_argument("--source-dir", type=Path, default=DEFAULT_SOURCE) parser.add_argument("--reference-dir", type=Path, default=DEFAULT_REFERENCE) parser.add_argument("--profile-dir", type=Path, default=DEFAULT_PROFILE) parser.add_argument("--warped-dir", type=Path, default=DEFAULT_WARPED) parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT) parser.add_argument("--pitch-dir", type=Path, default=DEFAULT_PITCH) parser.add_argument("--controls", type=Path, default=DEFAULT_CONTROLS) parser.add_argument("--ids", nargs="*", default=None) parser.add_argument("--overwrite", action="store_true") args = parser.parse_args() selected = args.ids or ids_from_files(args.reference_dir) generate( model_dir=args.model_dir, cosyvoice_root=args.cosyvoice_root, translations_dir=args.translations_dir, source_dir=args.source_dir, reference_dir=args.reference_dir, profile_dir=args.profile_dir, warped_dir=args.warped_dir, output_dir=args.output_dir, pitch_dir=args.pitch_dir, controls_path=args.controls, selected_ids=selected, overwrite=args.overwrite, ) if __name__ == "__main__": main()