(config: dict[str, Any], args: argparse.Namespace)
| 774 | raise RuntimeError( |
| 775 | f"need {args.requests_per_session} --text values, only received {len(args.texts)}" |
| 776 | ) |
| 777 | texts = args.texts[: args.requests_per_session] |
| 778 | request_manifest["texts"] = texts |
| 779 | return texts, request_manifest, scenario_config |
| 780 | |
| 781 | task_name = str(config["task"]) |
| 782 | catalog = load_omnivoice_case_catalog(REPO_ROOT / str(config["case_catalog"])) |
| 783 | task_cases = catalog.get(task_name) |
| 784 | if not isinstance(task_cases, dict): |
| 785 | raise RuntimeError(f"OmniVoice case catalog is missing task {task_name!r}") |
| 786 | |
| 787 | if len(args.case_names) > 1: |
| 788 | raise RuntimeError("OmniVoice warmbench accepts at most one --case-name because a case carries scenario metadata") |
| 789 | case_name = args.case_names[0] if args.case_names else str(config.get("default_case_name", "default")) |
| 790 | case = task_cases.get(case_name) |
| 791 | if not isinstance(case, dict): |
| 792 | available = ", ".join(sorted(task_cases)) |
| 793 | raise RuntimeError(f"unknown OmniVoice case name {case_name!r} for task {task_name}; available: {available}") |
| 794 | |
| 795 | texts = case.get("texts") |
| 796 | if not isinstance(texts, list) or not texts: |
| 797 | raise RuntimeError(f"OmniVoice case {task_name}.{case_name} has no texts") |
| 798 | if len(texts) < args.requests_per_session: |
| 799 | raise RuntimeError( |
| 800 | f"OmniVoice case {task_name}.{case_name} needs at least {args.requests_per_session} texts, only has {len(texts)}" |
| 801 | ) |
| 802 | texts = [str(text) for text in texts[: args.requests_per_session]] |
| 803 | for key in ("language", "voice_design_instruct", "reference_text", "whisper_model", "whisper_language", "asr_compact_lcs_min"): |
| 804 | if key in case: |
| 805 | scenario_config[key] = case[key] |
no test coverage detected