(
payload: dict[str, Any],
source_path: Path,
benchmark_index: dict[str, dict[str, Any]],
)
| 205 | |
| 206 | |
| 207 | def _normalize_legacy_result( |
| 208 | payload: dict[str, Any], |
| 209 | source_path: Path, |
| 210 | benchmark_index: dict[str, dict[str, Any]], |
| 211 | ) -> NormalizedResult: |
| 212 | legacy_metadata = _infer_legacy_metadata(source_path, benchmark_index) |
| 213 | benchmark_metadata = _merge_benchmark_metadata( |
| 214 | legacy_metadata["benchmark_name"], |
| 215 | {}, |
| 216 | benchmark_index, |
| 217 | ) |
| 218 | status = str(payload.get("flag", "")) |
| 219 | if status == "success": |
| 220 | normalized_status = "success" |
| 221 | elif status == "failed": |
| 222 | normalized_status = "failed" |
| 223 | else: |
| 224 | normalized_status = status |
| 225 | |
| 226 | return NormalizedResult( |
| 227 | status=normalized_status, |
| 228 | runtime=_float_or_none(payload.get("runtime", payload.get("time"))), |
| 229 | benchmark_name=legacy_metadata["benchmark_name"], |
| 230 | model_alias=legacy_metadata["model_alias"], |
| 231 | arch=legacy_metadata["arch"], |
| 232 | workflow_name="", |
| 233 | prompt_bundle_name="", |
| 234 | difficulty=str(benchmark_metadata.get("difficulty", "")), |
| 235 | category=str(benchmark_metadata.get("category", benchmark_metadata.get("type", ""))), |
| 236 | benchmark_target=str(benchmark_metadata.get("target", "")), |
| 237 | source_file=str(source_path), |
| 238 | source_dir=str(source_path.parent), |
| 239 | target=str(benchmark_metadata.get("target", "")), |
| 240 | run_index=_int_or_none(payload.get("count")), |
| 241 | commands=_as_list(payload.get("commands")), |
| 242 | history=_as_list(payload.get("history")), |
| 243 | prompts=_as_dict(payload.get("prompts")), |
| 244 | benchmark=benchmark_metadata, |
| 245 | raw=payload, |
| 246 | ) |
| 247 | |
| 248 | |
| 249 | def _load_benchmark_index(benchmark_files: Sequence[str | Path]) -> dict[str, dict[str, Any]]: |
no test coverage detected