(args: argparse.Namespace, case: dict[str, Any], case_dir: Path)
| 306 | def write_sequence_file(case: dict[str, Any], case_dir: Path) -> Path: |
| 307 | sequence_path = case_dir / "requests.json" |
| 308 | requests = [materialize_request_paths(request) for request in case["requests"]] |
| 309 | with sequence_path.open("w", encoding="utf-8") as handle: |
| 310 | json.dump({"requests": requests}, handle, ensure_ascii=False, indent=2) |
| 311 | handle.write("\n") |
| 312 | return sequence_path |
| 313 | |
| 314 | |
| 315 | def extract_text_outputs(case: dict[str, Any], stdout: str) -> list[dict[str, str]]: |
| 316 | request_ids = [str(request.get("id", f"request_{index}")) for index, request in enumerate(case.get("requests", []))] |
| 317 | outputs: list[dict[str, str]] = [] |
| 318 | request_id = "" |
| 319 | for line in stdout.splitlines(): |
| 320 | if line.startswith("request_id="): |
| 321 | request_id = line[len("request_id=") :] |
| 322 | continue |
| 323 | if not line.startswith("text_output="): |
| 324 | continue |
| 325 | fallback_id = request_ids[len(outputs)] if len(outputs) < len(request_ids) else f"text_output_{len(outputs)}" |
| 326 | outputs.append({ |
| 327 | "id": request_id if request_id else fallback_id, |
| 328 | "text": line[len("text_output=") :], |
| 329 | }) |
| 330 | request_id = "" |
| 331 | return outputs |
| 332 | |
| 333 | |
| 334 | def write_text_outputs(case: dict[str, Any], case_dir: Path, stdout: str) -> None: |
| 335 | outputs = extract_text_outputs(case, stdout) |
| 336 | with (case_dir / "text_outputs.json").open("w", encoding="utf-8") as handle: |
| 337 | json.dump({"outputs": outputs}, handle, ensure_ascii=False, indent=2) |
| 338 | handle.write("\n") |
| 339 | |
| 340 | |
| 341 | def resolve_model_path(models_root: Path, value: str) -> Path: |
| 342 | path = Path(value) |
| 343 | if path.is_absolute(): |
| 344 | return path |
| 345 | if path.parts and path.parts[0] == "models": |
| 346 | return models_root.joinpath(*path.parts[1:]) |
| 347 | if path.parts and path.parts[0] == "assets": |
| 348 | return REPO_ROOT / path |
| 349 | return models_root / path |
| 350 | |
| 351 | |
| 352 | def resolve_model_override(value: Path | None) -> Path | None: |
| 353 | if value is None: |
| 354 | return None |
| 355 | if value.is_absolute(): |
| 356 | return value |
| 357 | return REPO_ROOT / value |
| 358 | |
| 359 | |
| 360 | def build_command(args: argparse.Namespace, case: dict[str, Any], case_dir: Path) -> list[str]: |
| 361 | model_path = resolve_model_override(args.model_path) or resolve_model_path(args.models_root, case["model"]) |
| 362 | command = [ |
no test coverage detected