File size: 7,772 Bytes
cc9dbda | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | """Generate the Coherence Dial training corpus.
For each (sentence, voice, level): corrupt the sentence's phonemes at p(level), synthesize the
result with the BASE F5-TTS in that voice, and write a wav + a row in manifest.jsonl. The
manifest pairs each synthesized clip with its ORIGINAL sentence + level + voice — that's the
(input, level) -> audio mapping the LoRA fine-tune later learns.
Honest scaling note (read before running on Colab):
F5-TTS ~5-7s per generated clip on A100. So:
500 sentences x 1 voice x 5 levels = 2,500 clips ~ 3-4 h (single-voice spike)
500 sentences x 3 voices x 5 levels = 7,500 clips ~ 10-12 h
1500 sentences x 3 voices x 5 levels = 22,500 clips ~ 30-40 h (full scale; needs multi-session)
Default below is the single-voice spike that fits a 10h wall-clock budget alongside everything
else. Override --max-sentences and pass multiple --voice for larger runs.
Layout under --out:
data/coherence/
clip_00000_v1_lv0.wav
clip_00000_v1_lv0.json # per-clip metadata
manifest.jsonl # one line per clip
SUMMARY.json
"""
import argparse
import json
import sys
from pathlib import Path
import numpy as np
def load_sentences(path: Path, n: int):
lines = [ln.strip() for ln in path.read_text(encoding="utf-8", errors="ignore").splitlines()
if ln.strip() and not ln.startswith("#")]
if n and n < len(lines):
lines = lines[:n]
return lines
def main():
p = argparse.ArgumentParser()
p.add_argument("--sentences", required=True, help="text file, one sentence per line")
p.add_argument("--voice", action="append", required=True,
help="voice id + wav path + optional ref text, e.g. v1:data/voices/v1.wav:data/voices/v1.txt; pass multiple times")
p.add_argument("--lm", default="data/phoneme_lm.npz")
p.add_argument("--out", default="data/coherence")
p.add_argument("--max-sentences", type=int, default=500,
help="cap sentence count (default 500 for single-voice spike)")
p.add_argument("--levels", type=int, default=5)
p.add_argument("--input-mode", choices=["pseudo", "ipa", "text", "mondegreen"], default="pseudo",
help="pseudo/ipa = phoneme corruption (Tongues mode targets); mondegreen = "
"real-English-words phonetic ghost (Ghost mode targets); text = no corruption.")
p.add_argument("--seed-base", type=int, default=42)
p.add_argument("--model", default="F5TTS_v1_Base", help="F5-TTS variant identifier")
p.add_argument("--remove-silence", action="store_true")
p.add_argument("--resume", action="store_true", help="skip clips whose wav already exists")
args = p.parse_args()
# ---- inputs ----
sentences = load_sentences(Path(args.sentences), args.max_sentences)
voices = []
for v in args.voice:
parts = v.split(":")
vid = parts[0]
wav = parts[1]
ref_text_path = parts[2] if len(parts) > 2 else None
ref_text = ""
if ref_text_path and Path(ref_text_path).exists():
ref_text = Path(ref_text_path).read_text(encoding="utf-8").strip()
voices.append({"id": vid, "wav": wav, "ref_text": ref_text})
out = Path(args.out); out.mkdir(parents=True, exist_ok=True)
sys.path.insert(0, str(Path(__file__).resolve().parent))
from corrupt_phonemes import load_lm, corrupt_sentence, LEVEL_P
lm = load_lm(Path(args.lm))
# Mondegreen index lazily loaded (only when --input-mode=mondegreen). Heavy: ~5s.
mondegreen_idx = None
if args.input_mode == "mondegreen":
from mondegreen import MondegreenIndex
cmu_path = Path(args.lm).parent / "cmudict.dict"
mondegreen_idx = MondegreenIndex(cmu_path)
print(f"Mondegreen index: {mondegreen_idx.size} words", file=sys.stderr)
total = len(sentences) * len(voices) * args.levels
print(f"Generating {total} clips: {len(sentences)} sentences x {len(voices)} voices x {args.levels} levels",
file=sys.stderr)
# ---- F5-TTS init ----
try:
from f5_tts.api import F5TTS
except ImportError:
print("ERROR: f5-tts is not installed. `pip install f5-tts` (typically on Colab GPU).", file=sys.stderr)
sys.exit(1)
tts = F5TTS(model=args.model)
manifest_path = out / "manifest.jsonl"
manifest_f = manifest_path.open("a" if args.resume else "w")
idx = 0
written = 0
skipped = 0
for si, sentence in enumerate(sentences):
# cache corrupted texts per (sentence, level) — same across voices
corrupted_by_level = {}
for lv in range(args.levels):
seed = args.seed_base + si * 31 + lv
arpa, ipa, pseudo, display = corrupt_sentence(sentence, lv, lm, seed=seed)
mond = mondegreen_idx.substitute(sentence, lv, seed=seed) if mondegreen_idx else ""
corrupted_by_level[lv] = {"arpabet": " ".join(t for t in arpa if t.strip()),
"ipa": ipa, "pseudo": pseudo, "display": display,
"mondegreen": mond}
for voice in voices:
for lv in range(args.levels):
name = f"clip_{idx:05d}_{voice['id']}_lv{lv}"
wav_path = out / f"{name}.wav"
meta_path = out / f"{name}.json"
idx += 1
if args.resume and wav_path.exists():
skipped += 1
continue
gen = corrupted_by_level[lv]
gen_text = {"pseudo": gen["pseudo"], "ipa": gen["ipa"],
"text": sentence, "mondegreen": gen["mondegreen"]}[args.input_mode]
if not gen_text.strip():
continue
try:
tts.infer(
ref_file=voice["wav"], ref_text=voice["ref_text"],
gen_text=gen_text, file_wave=str(wav_path),
seed=args.seed_base + si * 7 + lv, remove_silence=args.remove_silence,
)
except Exception as e:
print(f" TTS fail on {name}: {e}", file=sys.stderr); continue
meta = {
"name": name, "sentence_id": si, "sentence": sentence,
"voice": voice["id"], "voice_wav": voice["wav"],
"level": lv, "level_p": LEVEL_P[lv],
"input_mode": args.input_mode, "gen_text": gen_text,
"arpabet": gen["arpabet"], "ipa": gen["ipa"],
"pseudo": gen["pseudo"], "display": gen["display"],
}
meta_path.write_text(json.dumps(meta, indent=2))
manifest_f.write(json.dumps({
"audio_path": str(wav_path.relative_to(out.parent)),
"sentence": sentence, "level": lv,
"voice": voice["id"], "voice_wav": voice["wav"],
"voice_ref_text": voice["ref_text"], "input_text": gen_text,
}) + "\n")
manifest_f.flush()
written += 1
if written % 25 == 0:
print(f" written {written}/{total - skipped} (skipped {skipped})", file=sys.stderr)
manifest_f.close()
summary = {
"sentences": len(sentences), "voices": [v["id"] for v in voices],
"levels": args.levels, "input_mode": args.input_mode,
"written": written, "skipped": skipped, "total_planned": total,
"manifest": str(manifest_path),
}
(out / "SUMMARY.json").write_text(json.dumps(summary, indent=2))
print(f"\nDONE: wrote {written}, skipped {skipped} -> {out}")
if __name__ == "__main__":
main()
|