(
entries: List[Dict[str, Any]],
meta_path: Path,
video_field: str,
api_key: str,
num_threads: int,
sampling_fps: int,
)
| 205 | return {"analysis": analysis, "matches": matches, "raw": raw_text} |
| 206 | |
| 207 | def run_alignment_checks( |
| 208 | entries: List[Dict[str, Any]], |
| 209 | meta_path: Path, |
| 210 | video_field: str, |
| 211 | api_key: str, |
| 212 | num_threads: int, |
| 213 | sampling_fps: int, |
| 214 | ) -> Dict[int, Dict[str, Any]]: |
| 215 | client = genai.Client(api_key=api_key) |
| 216 | tasks: List[Tuple[int, Path, str]] = [] |
| 217 | for entry in entries: |
| 218 | # Use global_id as the unique identifier |
| 219 | motion_id = entry.get("global_id", entry.get("id")) |
| 220 | if motion_id is None: |
| 221 | continue |
| 222 | raw_video_path = entry.get(video_field) |
| 223 | if raw_video_path is None: |
| 224 | continue |
| 225 | video_path = resolve_repo_path(meta_path, str(raw_video_path)) |
| 226 | if not video_path.exists(): |
| 227 | continue |
| 228 | prompt = PROMPT_TEMPLATE.format(description=entry.get("prompt", "")) |
| 229 | tasks.append((int(motion_id), video_path, prompt)) |
| 230 | |
| 231 | results: Dict[int, Dict[str, Any]] = {} |
| 232 | with ThreadPoolExecutor(max_workers=num_threads) as executor: |
| 233 | future_to_id = { |
| 234 | executor.submit(call_gemini, client, video_path, prompt, sampling_fps): motion_id |
| 235 | for motion_id, video_path, prompt in tasks |
| 236 | } |
| 237 | for future in tqdm(as_completed(future_to_id), total=len(future_to_id), desc="Gemini"): |
| 238 | motion_id = future_to_id[future] |
| 239 | try: |
| 240 | results[motion_id] = future.result() |
| 241 | except Exception as exc: |
| 242 | results[motion_id] = {"matches": False, "analysis": str(exc), "raw": ""} |
| 243 | return results |
| 244 | |
| 245 | |
| 246 | def decide_quality( |
no test coverage detected