(
meta_json: Path,
quality_report: Path,
*,
device: str,
jitter_threshold: float,
max_samples: Optional[int],
gemini_api_key: Optional[str],
video_field: str,
eval_field: str,
num_threads: int,
sampling_fps: float,
)
| 258 | |
| 259 | |
| 260 | def apply_quality_gate( |
| 261 | meta_json: Path, |
| 262 | quality_report: Path, |
| 263 | *, |
| 264 | device: str, |
| 265 | jitter_threshold: float, |
| 266 | max_samples: Optional[int], |
| 267 | gemini_api_key: Optional[str], |
| 268 | video_field: str, |
| 269 | eval_field: str, |
| 270 | num_threads: int, |
| 271 | sampling_fps: float, |
| 272 | ) -> None: |
| 273 | entries = json.loads(meta_json.read_text()) |
| 274 | if not isinstance(entries, list): |
| 275 | raise ValueError(f"Expected list in {meta_json}") |
| 276 | |
| 277 | if max_samples: |
| 278 | entries = entries[:max_samples] |
| 279 | |
| 280 | samples = list_motion_samples(entries, meta_json, eval_field) |
| 281 | motion_metrics = compute_motion_quality_metrics(samples, device) |
| 282 | sample_ids = [motion_id for motion_id, _ in samples] |
| 283 | |
| 284 | if not gemini_api_key: |
| 285 | raise RuntimeError("Gemini API key is required for VLM gating.") |
| 286 | |
| 287 | sampling_fps_int = max(1, int(round(sampling_fps))) |
| 288 | vlm_results = run_alignment_checks(entries, meta_json, video_field, gemini_api_key, num_threads, sampling_fps_int) |
| 289 | |
| 290 | final_quality = decide_quality(motion_metrics, vlm_results, jitter_threshold) |
| 291 | |
| 292 | updated_entries = [] |
| 293 | for entry in entries: |
| 294 | motion_id = entry.get("global_id", entry.get("id")) |
| 295 | use_ref = bool(final_quality.get(motion_id, False)) |
| 296 | new_entry = dict(entry) |
| 297 | new_entry["use_ref_motion"] = use_ref |
| 298 | updated_entries.append(new_entry) |
| 299 | |
| 300 | meta_json.write_text(json.dumps(updated_entries, indent=2)) |
| 301 | |
| 302 | quality_records = [] |
| 303 | for motion_id in sample_ids: |
| 304 | vlm_entry = vlm_results.get(motion_id, {}) |
| 305 | quality_records.append( |
| 306 | { |
| 307 | "global_id": motion_id, |
| 308 | "motion_metrics": motion_metrics.get(motion_id, {}), |
| 309 | "vlm_analysis": vlm_entry.get("analysis"), |
| 310 | "vlm_raw": vlm_entry.get("raw"), |
| 311 | "vlm_matches": vlm_entry.get("matches", False), |
| 312 | "final_quality": final_quality.get(motion_id, False), |
| 313 | } |
| 314 | ) |
| 315 | quality_report.parent.mkdir(parents=True, exist_ok=True) |
| 316 | quality_report.write_text( |
| 317 | json.dumps({"source_meta": str(meta_json), "records": quality_records}, indent=2) |
no test coverage detected