QWENVL-PRIVATE/analysis/failure_case_analysis.py

import json
import os


def load_jsonl(path):
    """
    Loads a JSONL file into a dictionary keyed by 'doc_id'.
    """
    data = {}
    try:
        with open(path, 'r', encoding='utf-8') as f:
            for i, line in enumerate(f):
                try:
                    record = json.loads(line)
                    # Use 'doc_id' as the unique key for comparison
                    if 'doc_id' not in record:
                        print(f"Warning: Missing 'doc_id' in {path} at line {i + 1}. Skipping.")
                        continue
                    data[record['doc_id']] = record
                except json.JSONDecodeError:
                    print(f"Error: Could not parse JSON in {path} at line {i + 1}. Skipping.")
    except FileNotFoundError:
        print(f"Error: File not found at {path}")
        return None
    except Exception as e:
        print(f"Error opening file {path}: {e}")
        return None

    print(f"Successfully loaded {len(data)} records from {path}.")
    return data


def find_failure_cases_textvqa(ours_output_path, baseline_output_path, score_key="exact_match"):
    """
    Compares two output files and finds items where 'ours' is wrong (0.0)
    and 'baseline' is right (1.0).

    Args:
        ours_output_path (str): Path to your model's output JSONL.
        baseline_output_path (str): Path to the baseline model's output JSONL.
        score_key (str): The key in the JSON to find the score dictionary.
                         Default is "vstar_overall_acc".

    Returns:
        list: A list of dictionaries, each containing details of a failure case.
    """
    failure_cases = []

    # Load the baseline results into a dictionary for fast lookup
    baseline_results = load_jsonl(baseline_output_path)
    if baseline_results is None:
        return []

    print(f"\nProcessing {ours_output_path} against baseline...")

    try:
        with open(ours_output_path, 'r', encoding='utf-8') as f:
            for i, line in enumerate(f):
                try:
                    our_record = json.loads(line)
                    if 'doc_id' not in our_record:
                        print(f"Warning: Missing 'doc_id' in 'ours' file at line {i + 1}. Skipping.")
                        continue
                    doc_id = our_record['doc_id']

                    # Check if this doc_id exists in the baseline
                    if doc_id not in baseline_results:
                        print(f"Warning: doc_id {doc_id} from 'ours' file not found in baseline. Skipping.")
                        continue
                    baseline_record = baseline_results[doc_id]
                    # Check if score keys exist before accessing
                    if score_key not in our_record :
                        print(f"Warning: Missing score key '{score_key}' in 'ours' file for doc_id {doc_id}. Skipping.")
                        continue
                    if score_key not in baseline_record:
                        print(
                            f"Warning: Missing score key '{score_key}' in baseline file for doc_id {doc_id}. Skipping.")
                        continue
                    # Get the scores
                    if type(our_record[score_key]) is not dict or type(baseline_record[score_key]) is not dict:
                        our_score = our_record[score_key]
                        baseline_score = baseline_record[score_key]
                    else:
                        our_score = our_record[score_key]['score']
                        baseline_score = baseline_record[score_key]['score']

                    # The core logic: Find where we failed and baseline succeeded
                    if our_score < 0.3 and baseline_score > 0.7:
                        # Format a clean dictionary for case study
                        failure_info = {
                            "doc_id": doc_id,
                            "question": our_record['input'],
                            "ground_truth": baseline_record['filtered_resps'][0],
                            "our_prediction": our_record['filtered_resps'][0],#[score_key].get('prediction', 'N/A'),
                            "baseline_prediction": baseline_record['filtered_resps'][0],
                            "full_our_record": our_record  # Optional: include for full context
                        }
                        failure_cases.append(failure_info)

                except json.JSONDecodeError:
                    print(f"Error: Could not parse JSON in 'ours' file at line {i + 1}. Skipping.")
                except KeyError:
                    # This catches if a sub-key like 'score' is missing
                    print(f"Error: Missing expected key structure in 'ours' file at line {i + 1}. Skipping.")

    except FileNotFoundError:
        print(f"Error: File not found at {ours_output_path}")
        return []
    except Exception as e:
        print(f"Error opening file {ours_output_path}: {e}")
        return []

    print(f"\n--- Analysis Complete ---")
    print(f"Found {len(failure_cases)} failure cases where 'ours' was wrong and 'baseline' was right.")
    return failure_cases


def find_failure_cases_vstar(ours_output_path, baseline_output_path, score_key="vstar_overall_acc"):
    """
    Compares two output files and finds items where 'ours' is wrong (0.0)
    and 'baseline' is right (1.0).

    Args:
        ours_output_path (str): Path to your model's output JSONL.
        baseline_output_path (str): Path to the baseline model's output JSONL.
        score_key (str): The key in the JSON to find the score dictionary.
                         Default is "vstar_overall_acc".

    Returns:
        list: A list of dictionaries, each containing details of a failure case.
    """
    failure_cases = []

    # Load the baseline results into a dictionary for fast lookup
    baseline_results = load_jsonl(baseline_output_path)
    if baseline_results is None:
        return []

    print(f"\nProcessing {ours_output_path} against baseline...")

    try:
        with open(ours_output_path, 'r', encoding='utf-8') as f:
            for i, line in enumerate(f):
                try:
                    our_record = json.loads(line)

                    if 'doc_id' not in our_record:
                        print(f"Warning: Missing 'doc_id' in 'ours' file at line {i + 1}. Skipping.")
                        continue

                    doc_id = our_record['doc_id']

                    # Check if this doc_id exists in the baseline
                    if doc_id not in baseline_results:
                        print(f"Warning: doc_id {doc_id} from 'ours' file not found in baseline. Skipping.")
                        continue

                    baseline_record = baseline_results[doc_id]

                    # Check if score keys exist before accessing
                    if score_key not in our_record or 'score' not in our_record[score_key]:
                        print(f"Warning: Missing score key '{score_key}' in 'ours' file for doc_id {doc_id}. Skipping.")
                        continue

                    if score_key not in baseline_record or 'score' not in baseline_record[score_key]:
                        print(
                            f"Warning: Missing score key '{score_key}' in baseline file for doc_id {doc_id}. Skipping.")
                        continue

                    # Get the scores
                    our_score = our_record[score_key]['score']
                    baseline_score = baseline_record[score_key]['score']

                    # The core logic: Find where we failed and baseline succeeded
                    if our_score == 0.0 and baseline_score == 1.0:
                        # Format a clean dictionary for case study
                        failure_info = {
                            "doc_id": doc_id,
                            "question": our_record.get('doc', {}).get('text', 'N/A'),
                            "ground_truth": our_record.get('target', 'N/A'),
                            "our_prediction": our_record[score_key].get('prediction', 'N/A'),
                            "baseline_prediction": baseline_record[score_key].get('prediction', 'N/A'),
                            "full_our_record": our_record  # Optional: include for full context
                        }
                        failure_cases.append(failure_info)

                except json.JSONDecodeError:
                    print(f"Error: Could not parse JSON in 'ours' file at line {i + 1}. Skipping.")
                except KeyError:
                    # This catches if a sub-key like 'score' is missing
                    print(f"Error: Missing expected key structure in 'ours' file at line {i + 1}. Skipping.")

    except FileNotFoundError:
        print(f"Error: File not found at {ours_output_path}")
        return []
    except Exception as e:
        print(f"Error opening file {ours_output_path}: {e}")
        return []

    print(f"\n--- Analysis Complete ---")
    print(f"Found {len(failure_cases)} failure cases where 'ours' was wrong and 'baseline' was right.")
    return failure_cases

# --- Example Usage ---
if __name__ == "__main__":

    # --- 1. Create dummy data files for the example ---

    # This example has 3 cases:
    # 1. We fail (B), baseline succeeds (A). -> This should be found.
    # 2. We succeed (A), baseline fails (C). -> This should be ignored.
    # 3. Both succeed (A). -> This should be ignored.



    # # Define file paths
    # baseline_file = "./logs/liuhaotian__llava-v1.5-7b/20250902_190647_samples_textvqa_val.jsonl"
    # ours_file = "./logs/checkpoints__llava-v1.5-7b-roi-K15T3-152k-v1bf16Mheads-bg0p05fg0p2-filled/20251113_113923_samples_textvqa_val.jsonl"
    #
    # #failures = find_failure_cases_textvqa(ours_file, baseline_file, 'exact_match')
    # failures = find_failure_cases_textvqa(baseline_file, ours_file , 'exact_match')


    #baseline_file = "./logs/liuhaotian__llava-v1.5-7b/20250902_201850_samples_vstar_bench.jsonl"
    #ours_file = "./logs/checkpoints__llava-v1.5-7b-roi-K15T3-152k-v1bf16Mheads-bg0p05fg0p2-filled/20251113_130922_samples_vstar_bench.jsonl"

    # baseline_file = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251113_135313_samples_vstar_bench.jsonl"
    # ours_file = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K16T3-152k-v1bf16Mheads-twiginit-filled/20251113_134244_samples_vstar_bench.jsonl"
    #
    # failures = find_failure_cases_vstar(baseline_file, ours_file)
    # failure_ids = [case['doc_id'] for case in failures]

    ###### TextVQA Failure Case Analysis
    #baseline_file_textvqa = "logs/output__qwen2_5vl-7b-roi-K18T3-185k-v1bf16-denoisedv2-twiginit-filled/20251209_181751_samples_textvqa_val.jsonl"
    baseline_file_textvqa = "logs/Qwen__Qwen3-VL-4B-Instruct/20260122_115842_samples_textvqa_val.jsonl"

    ours_file_textvqa = "logs/output__qwen3vl-4b-roi-K24T3-185k-v1bf16-twiginit/20260122_120855_samples_textvqa_val.jsonl"
    failures_baseline_textvqa = find_failure_cases_textvqa(baseline_file_textvqa, ours_file_textvqa, 'exact_match')
    failure_ids_baseline_textvqa = [case['doc_id'] for case in failures_baseline_textvqa]

    failures_ours_textvqa = find_failure_cases_textvqa(ours_file_textvqa, baseline_file_textvqa, 'exact_match')
    failure_ids_ours_textvqa = [case['doc_id'] for case in failures_ours_textvqa]
    failure_ids_ours_textvqa_dict = {case['doc_id']: case for case in failures_ours_textvqa}
    ### print some results
    for i, case in enumerate(failures_ours_textvqa[:10]):
        print(f"\nCase {i + 1} (doc_id: {case['doc_id']})")
        print(f"  Question: {case['question']}")
        print(f"  Ground Truth: {case['ground_truth']}")
        print(f"  Our (Wrong) Prediction: {case['our_prediction']}")
        print(f"  Baseline (Correct) Prediction: {case['baseline_prediction']}")



    # ###### InfoVQA Failure Case Analysis
    # baseline_file_infovqa = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_093838_samples_infovqa_val.jsonl"
    # ours_file_infovqa = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-185k-v1bf16-denoisedv2-twiginit-filled/20251215_131620_samples_infovqa_val.jsonl"
    #
    # failures_baseline_infovqa = find_failure_cases_textvqa(baseline_file_infovqa, ours_file_infovqa, 'anls')
    # failure_ids_baseline_infovqa = [case['doc_id'] for case in failures_baseline_infovqa]
    #
    # failures_ours_infovqa = find_failure_cases_textvqa(ours_file_infovqa, baseline_file_infovqa, 'anls')
    # failure_ids_ours_infovqa = [case['doc_id'] for case in failures_ours_infovqa]
    #
    # ##### DocVQA Failure Case Analysis
    # baseline_file_docvqa = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_095001_samples_docvqa_val.jsonl"
    # ours_file_docvqa = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-134k-gatedv2PB-wGTwRoI-twiginit-filled/20251210_143202_samples_docvqa_val.jsonl"
    #
    # failures_baseline_docvqa = find_failure_cases_textvqa(baseline_file_docvqa, ours_file_docvqa, 'anls')
    # failure_ids_baseline_docvqa = [case['doc_id'] for case in failures_baseline_docvqa]
    # failures_ours_docvqa = find_failure_cases_textvqa(ours_file_docvqa, baseline_file_docvqa, 'anls')
    # failure_ids_ours_docvqa = [case['doc_id'] for case in failures_ours_docvqa]
    #
    # ##### ChartQA Failure Case Analysis
    # baseline_file_chartqa = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_094359_samples_chartqa.jsonl"
    # ours_file_chartqa = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-134k-gatedv2PB-wGTwRoI-twiginit-filled/20251210_142453_samples_chartqa.jsonl"
    #
    # failures_baseline_chartqa = find_failure_cases_textvqa(baseline_file_chartqa, ours_file_chartqa, 'relaxed_overall')
    # failure_ids_baseline_chartqa = [case['doc_id'] for case in failures_baseline_chartqa]
    # failures_ours_chartqa = find_failure_cases_textvqa(ours_file_chartqa, baseline_file_chartqa, 'relaxed_overall')
    # failure_ids_ours_chartqa = [case['doc_id'] for case in failures_ours_chartqa]
    #
    #
    # ###### OCRBench Failure Case Analysis
    # baseline_file_ocrbench = "../Qwen2.5-VL/logs/Qwen__Qwen2.5-VL-7B-Instruct/20251204_094800_samples_ocrbench.jsonl"
    # ours_file_ocrbench = "../Qwen2.5-VL/logs/output__qwen2_5vl-7b-roi-K18T3-134k-gatedv2PB-wGTwRoI-twiginit-filled/20251210_142936_samples_ocrbench.jsonl"
    # failures_baseline_ocrbench = find_failure_cases_textvqa(baseline_file_ocrbench, ours_file_ocrbench, 'ocrbench_accuracy')
    # failure_ids_baseline_ocrbench = [case['doc_id'] for case in failures_baseline_ocrbench]
    # failures_ours_ocrbench = find_failure_cases_textvqa(ours_file_ocrbench, baseline_file_ocrbench, 'ocrbench_accuracy')
    # failure_ids_ours_ocrbench = [case['doc_id'] for case in failures_ours_ocrbench]
    #

    pass


    # if failure_ids_ours:
    #     print("\n--- Failure Cases ---")
    #     for i, case in enumerate(failure_ids_ours):
    #         print(f"\nCase {i + 1} (doc_id: {case['doc_id']})")
    #         print(f"  Question: {case['question']}")
    #         print(f"  Ground Truth: {case['ground_truth']}")
    #         print(f"  Our (Wrong) Prediction: {case['our_prediction']}")
    #         print(f"  Baseline (Correct) Prediction: {case['baseline_prediction']}")