Function evaluate_entry

mbench/motion_generalizability.py:65–99

(prompt_dict)

    per_motion_metrics = []

    def evaluate_entry(prompt_dict):
        video_label = prompt_dict.get('prompt', 'unknown')
        video_path = prompt_dict.get('evaluation_file', '')
        motion_id = prompt_dict.get('id')

        if not os.path.exists(video_path):
            return {"skip": True, "id": motion_id, "error": f"Missing video {video_path}"}

        try:
            # Read video bytes directly - fps is now handled by Gemini API
            with open(video_path, 'rb') as f:
                video_bytes = f.read()
        except Exception as exc:
            return {"skip": True, "id": motion_id, "error": f"Video read error: {exc}"}


        try:
            response = call_gemini_api(video_bytes, prompt + f'"{video_label}".', api_key, fps=vlm_fps)

        except Exception as exc:
            return {"skip": True, "id": motion_id, "error": f"Gemini error: {exc}"}

        response = response.encode('utf-8').decode('utf-8').strip()
        decision_text = response.split("Answer:", 1)[-1].strip()
        is_correct = "yes" in decision_text.lower()

        return {
            "skip": False,
            "id": motion_id,
            "is_correct": is_correct,
            "video_label": video_label,
            "video_path": video_path,
            "decision_text": decision_text,
            "raw_response": response,
        }

    with ThreadPoolExecutor(max_workers=gemini_workers) as executor:
        futures = {executor.submit(evaluate_entry, prompt_dict): prompt_dict for prompt_dict in prompt_dict_ls}