Spaces:
Runtime error
Runtime error
File size: 3,079 Bytes
aacc29a | 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 | """CLI: build synthetic GEC pairs by running Gipformer over TTS audio (paper §4.1 Step 3).
python scribe/training/scripts/make_synth_pairs.py \
--input artifacts/synthetic/synthetic_audio_manifest_smoke.jsonl \
--output artifacts/gec_pairs/darag_synthetic_pairs_smoke.jsonl \
--datastore artifacts/retrieval/term_datastore_smoke.json --limit 10 --resume
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "scribe" / "training"))
sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "scribe"))
from gec.cliutil import configure_stdout # noqa: E402
configure_stdout()
from carepath.config import Settings # noqa: E402
from gec.data import build_synthetic_pairs, read_jsonl, validate_gec_pair # noqa: E402
from gec.retrieval import build_ne_retriever # noqa: E402
from carepath.services.asr import build_asr_service # noqa: E402
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input", required=True, help="JSONL from voice_clone_tts.py")
parser.add_argument("--output", default="artifacts/gec_pairs/darag_synthetic_pairs.jsonl")
parser.add_argument("--datastore", default="data/medical_lexicon.json")
parser.add_argument("--retrieval-backend", default="lexical", choices=["lexical", "semantic", "hybrid"])
parser.add_argument("--limit", type=int, default=None)
parser.add_argument(
"--n-best",
type=int,
default=1,
help="compatibility option; >1 adds acoustic variants, not decoder beam N-best",
)
parser.add_argument("--resume", action="store_true")
args = parser.parse_args()
rows = read_jsonl(Path(args.input))
if args.limit:
rows = rows[: args.limit]
settings = Settings.from_env()
asr = build_asr_service(settings)
retriever = build_ne_retriever(
Path(args.datastore), backend=args.retrieval_backend, top_k=settings.retrieval_top_k
)
output_path = Path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
completed = _completed(output_path) if args.resume else set()
mode = "a" if args.resume and output_path.exists() else "w"
written = 0
with output_path.open(mode, encoding="utf-8") as handle:
for pair in build_synthetic_pairs(
rows, retriever=retriever, asr=asr, completed_ids=completed, n_best=args.n_best
):
result = validate_gec_pair(pair)
if not result.ok:
raise ValueError(f"Invalid synthetic pair {pair.get('audio_id')}: {result.errors}")
handle.write(json.dumps(pair, ensure_ascii=False) + "\n")
handle.flush()
written += 1
print(f"Wrote {written} synthetic GEC pairs to {output_path}")
def _completed(path: Path) -> set[str]:
if not path.exists():
return set()
return {row["audio_id"] for row in read_jsonl(path) if row.get("audio_id")}
if __name__ == "__main__":
main()
|