Function main

examples/migration/benchmark_funasr.py:220–274

()

def main() -> None:
    args = parse_args()
    extensions = normalize_extensions(args.extensions)
    files = iter_audio_files(args.input, extensions, args.recursive)
    if not files:
        print(f"No audio files found under {args.input} for extensions {extensions}", file=sys.stderr)
        sys.exit(2)

    vad_model = None if args.vad_model.lower() == "none" else args.vad_model
    model_kwargs: Dict[str, Any] = {"model": args.model, "vad_model": vad_model, "device": args.device}
    if args.spk_model:
        model_kwargs["spk_model"] = args.spk_model

    print(f"Loading FunASR model: {args.model} on {args.device}")
    load_start = time.perf_counter()
    model = AutoModel(**model_kwargs)
    model_load_seconds = time.perf_counter() - load_start
    print(f"Model loaded in {model_load_seconds:.3f}s")

    rows: List[Dict[str, Any]] = []
    for index, audio_path in enumerate(files, start=1):
        display = str(audio_path if args.input.is_file() else audio_path.relative_to(args.input))
        print(f"[{index}/{len(files)}] {display}")
        duration = audio_duration_seconds(audio_path)
        start = time.perf_counter()
        row: Dict[str, Any] = {
            "input": display,
            "path": str(audio_path),
            "duration_seconds": duration,
            "model": args.model,
            "device": args.device,
            "language": args.language,
        }
        try:
            result = model.generate(input=str(audio_path), language=args.language, batch_size=args.batch_size)
            elapsed = time.perf_counter() - start
            row["elapsed_seconds"] = elapsed
            row["realtime_factor"] = (duration / elapsed) if duration and elapsed else None
            row["text"] = extract_text(result)
            print(f"  ok: {elapsed:.3f}s" + (f", {row['realtime_factor']:.3f}x" if row["realtime_factor"] else ""))
        except Exception as exc:  # keep benchmarking other files
            elapsed = time.perf_counter() - start
            row["elapsed_seconds"] = elapsed
            row["realtime_factor"] = None
            row["error"] = repr(exc)
            print(f"  error: {exc}", file=sys.stderr)
        rows.append(row)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    results_path = args.output_dir / "results.jsonl"
    summary_path = args.output_dir / "summary.md"
    write_jsonl(results_path, rows)
    summary_path.write_text(markdown_summary(rows, args, model_load_seconds), encoding="utf-8")
    print(f"\nWrote {results_path}")
    print(f"Wrote {summary_path}")


if __name__ == "__main__":