from __future__ import annotations from typing import Any import pandas as pd from evaluation.subject_figure_data import ( best_image_aggregation_method, best_patient_aggregation_method, hypothesis_slug, load_best_baseline_subject_frame, load_h2_subject_frame, load_h3_subject_frames, load_h4_subject_bundle, load_h5_subject_frame, load_object_roi_frame, ) VARIANT_COUNT = 5 OBJECT_KEY_COLUMNS = ["patient_id", "image_id", "object_id_within_image"] AGGREGATION_METHOD_ORDER = [ "max_object_score", "mean_top3_scores", "count_objects_above_threshold", "soft_noisy_or", ] def _base_spec( *, article_id: str, figure_id: str, figure_basename: str, figure_kind: str, hypothesis: str, title: str, variant_index: int, ) -> dict[str, Any]: return { "article_id": article_id, "figure_id": figure_id, "figure_variant_id": f"{figure_id}_v{variant_index:02d}", "figure_basename": figure_basename, "figure_kind": figure_kind, "hypothesis": hypothesis, "hypothesis_slug": hypothesis_slug(hypothesis), "title": title, "variant_index": variant_index, } def _panel_row( *, panel_id: str, role: str, source_table: str, source_path: str, patient_id: str | None = None, image_id: str | None = None, object_id_within_image: int | None = None, class_name: str | None = None, target_binary: int | None = None, split_name: str | None = None, notes: str = "", ) -> dict[str, Any]: return { "panel_id": panel_id, "role": role, "patient_id": patient_id, "image_id": image_id, "object_id_within_image": object_id_within_image, "class_name": class_name, "target_binary": target_binary, "split_name": split_name, "source_table": source_table, "source_path": source_path, "notes": notes, } def _rotate_frame(frame: pd.DataFrame, variant_index: int) -> pd.DataFrame: ordered = frame.reset_index(drop=True) if ordered.empty: return ordered.copy() offset = (variant_index - 1) % len(ordered) if offset == 0: return ordered.copy() return pd.concat([ordered.iloc[offset:], ordered.iloc[:offset]], ignore_index=True) def _take_rows( frame: pd.DataFrame, *, n_items: int, variant_index: int, sort_columns: list[str], ascending: bool | list[bool], unique_columns: tuple[str, ...] | None = ("patient_id", "image_id"), extra_distinct_column: str | None = None, ) -> pd.DataFrame: if frame.empty or n_items <= 0: return frame.head(0).copy() ordered = frame.drop_duplicates(subset=OBJECT_KEY_COLUMNS).sort_values(sort_columns, ascending=ascending).reset_index(drop=True) rotated = _rotate_frame(ordered, variant_index) selected: list[dict[str, Any]] = [] seen_objects: set[tuple[Any, ...]] = set() seen_unique: set[tuple[Any, ...]] = set() seen_extra: set[Any] = set() def consider(row: pd.Series, enforce_unique: bool, enforce_extra: bool) -> bool: object_key = tuple(row[column] for column in OBJECT_KEY_COLUMNS) if object_key in seen_objects: return False if enforce_unique and unique_columns is not None: unique_key = tuple(row[column] for column in unique_columns) if unique_key in seen_unique: return False if enforce_extra and extra_distinct_column is not None: extra_value = row[extra_distinct_column] if extra_value in seen_extra: return False selected.append(row.to_dict()) seen_objects.add(object_key) if unique_columns is not None: seen_unique.add(tuple(row[column] for column in unique_columns)) if extra_distinct_column is not None: seen_extra.add(row[extra_distinct_column]) return len(selected) >= n_items for enforce_unique, enforce_extra in [(True, True), (True, False), (False, False)]: for _, row in rotated.iterrows(): if consider(row, enforce_unique=enforce_unique, enforce_extra=enforce_extra): return pd.DataFrame(selected) return pd.DataFrame(selected) def _take_group_rows( frame: pd.DataFrame, *, n_items: int, variant_index: int, sort_columns: list[str], ascending: bool | list[bool], dedupe_columns: list[str], ) -> pd.DataFrame: if frame.empty or n_items <= 0: return frame.head(0).copy() ordered = frame.sort_values(sort_columns, ascending=ascending).drop_duplicates(subset=dedupe_columns).reset_index(drop=True) rotated = _rotate_frame(ordered, variant_index) return rotated.head(n_items).copy() def _pick_single_row( frame: pd.DataFrame, *, variant_index: int, sort_columns: list[str], ascending: bool | list[bool], dedupe_columns: list[str], ) -> dict[str, Any]: selected = _take_group_rows( frame, n_items=1, variant_index=variant_index, sort_columns=sort_columns, ascending=ascending, dedupe_columns=dedupe_columns, ) if selected.empty: raise ValueError("Expected at least one image candidate.") return selected.iloc[0].to_dict() def _target_binary(row: pd.Series | dict[str, Any]) -> int | None: value = row.get("target_binary") if isinstance(row, dict) else row.get("target_binary") if value is None and "y_true" in row: value = row.get("y_true") if value is None and "is_blast" in row: value = row.get("is_blast") if pd.isna(value): return None return int(value) def _note_score(score: float | None, prefix: str = "score") -> str: if score is None or pd.isna(score): return prefix return f"{prefix}={float(score):.3f}" def _group_rows(image_rows: pd.DataFrame, top_n: int) -> list[dict[str, Any]]: return image_rows.sort_values(["y_score", "visual_priority"], ascending=[False, False]).head(top_n).to_dict("records") def _select_smear_story_objects(image_objects: pd.DataFrame, variant_index: int) -> list[dict[str, Any]]: positive = _take_rows( image_objects.loc[image_objects["target_binary"].eq(1)].copy(), n_items=min(2, int(image_objects["target_binary"].sum())), variant_index=variant_index, sort_columns=["visual_priority", "bbox_area_px", "object_id_within_image"], ascending=[False, False, True], unique_columns=None, ) negative = _take_rows( image_objects.loc[image_objects["target_binary"].eq(0)].copy(), n_items=min(4, max(0, len(image_objects) - len(positive))), variant_index=variant_index + 1, sort_columns=["visual_priority", "bbox_area_px", "object_id_within_image"], ascending=[False, False, True], unique_columns=None, extra_distinct_column="class_name", ) selected = pd.concat([positive, negative], ignore_index=True) if len(selected) < min(6, len(image_objects)): remaining = image_objects.loc[ ~image_objects["object_id_within_image"].isin(selected["object_id_within_image"].tolist()) ].copy() filler = _take_rows( remaining, n_items=min(6, len(image_objects)) - len(selected), variant_index=variant_index, sort_columns=["visual_priority", "bbox_area_px", "object_id_within_image"], ascending=[False, False, True], unique_columns=None, ) selected = pd.concat([selected, filler], ignore_index=True) selected = selected.head(min(6, len(image_objects))).copy() return selected.to_dict("records") def _build_a01_f1_variants(variant_count: int) -> list[dict[str, Any]]: object_roi_df = load_object_roi_frame() class_plan = [ "blast", "Lymphocyte", "Monocyte", "Myelocyte", "Segmentonuclear neutrophil", "normoblasts: oxyfilny" if int(object_roi_df["class_name"].eq("normoblasts: oxyfilny").sum()) >= 4 else "Normoblasts", ] variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): panel_rows: list[dict[str, Any]] = [] class_rows: list[dict[str, Any]] = [] for row_index, class_name in enumerate(class_plan, start=1): selected = _take_rows( object_roi_df.loc[object_roi_df["class_name"].eq(class_name)].copy(), n_items=4, variant_index=variant_index + row_index - 1, sort_columns=["visual_priority", "bbox_area_px", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), ) records = selected.to_dict("records") class_rows.append({"class_name": class_name, "rows": records}) for col_index, row in enumerate(records, start=1): panel_rows.append( _panel_row( panel_id=f"r{row_index:02d}c{col_index:02d}", role="bbox_roi", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="roi_manifest.parquet", source_path=str(row["bbox_crop_path"]), notes="class gallery bbox crop", ) ) variants.append( { **_base_spec( article_id="A01", figure_id="A01-F1", figure_basename="a01_subject_classes_gallery", figure_kind="a01_f1_class_gallery", hypothesis="H1", title="A01-F1. Object classes in the blast versus non-blast task", variant_index=variant_index, ), "class_rows": class_rows, "panel_rows": panel_rows, } ) return variants def _build_a01_f2_variants(variant_count: int) -> list[dict[str, Any]]: object_roi_df = load_object_roi_frame() image_candidates = ( object_roi_df.groupby(["patient_id", "image_id", "image_path"], as_index=False) .agg( n_objects=("object_id_within_image", "size"), n_blasts=("target_binary", "sum"), class_diversity=("class_name", "nunique"), mean_priority=("visual_priority", "mean"), ) ) image_candidates["n_non_blasts"] = image_candidates["n_objects"] - image_candidates["n_blasts"] image_candidates["readability_penalty"] = (image_candidates["n_objects"] - 8).abs() image_candidates = image_candidates.loc[ image_candidates["n_objects"].between(6, 18) & image_candidates["n_blasts"].ge(1) & image_candidates["n_non_blasts"].ge(2) ].copy() variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): image_row = _pick_single_row( image_candidates, variant_index=variant_index, sort_columns=["class_diversity", "readability_penalty", "mean_priority", "patient_id", "image_id"], ascending=[False, True, False, True, True], dedupe_columns=["patient_id", "image_id"], ) image_objects = object_roi_df.loc[ object_roi_df["patient_id"].eq(image_row["patient_id"]) & object_roi_df["image_id"].eq(image_row["image_id"]) ].copy() selected_objects = _select_smear_story_objects(image_objects, variant_index=variant_index) panel_rows = [ _panel_row( panel_id="A", role="smear_image", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), target_binary=None, split_name="", source_table="image_index.parquet", source_path=str(image_row["image_path"]), notes="original smear image", ), _panel_row( panel_id="B", role="smear_with_boxes", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), target_binary=None, split_name="", source_table="roi_manifest.parquet", source_path=str(image_row["image_path"]), notes="smear with selected bounding boxes", ), ] for slot, row in enumerate(selected_objects, start=1): panel_rows.append( _panel_row( panel_id=f"C{slot}", role="bbox_roi", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="roi_manifest.parquet", source_path=str(row["bbox_crop_path"]), notes=f"selected ROI #{slot}", ) ) variants.append( { **_base_spec( article_id="A01", figure_id="A01-F2", figure_basename="a01_from_smear_to_objects", figure_kind="a01_f2_smear_to_objects", hypothesis="H1", title="A01-F2. From a smear image to object-level cells", variant_index=variant_index, ), "image_row": image_row, "object_rows": selected_objects, "panel_rows": panel_rows, } ) return variants def _build_a01_f3_variants(variant_count: int) -> list[dict[str, Any]]: h2_df = load_h2_subject_frame(split_name="test") full_df = h2_df.loc[h2_df["task_name"].eq("a43_full_realistic_binary")].copy() pivot_df = full_df.pivot_table( index=[ "patient_id", "image_id", "object_id_within_image", "class_name", "split_name", "target_binary", "bbox_crop_path", "mask_crop_path", "mask_tight_crop_path", "bbox_area_px", "visual_priority", ], columns="label", values=["y_score", "y_pred"], aggfunc="first", ) pivot_df.columns = [f"{label}_{value_name}" for value_name, label in pivot_df.columns] pivot_df = pivot_df.reset_index() pivot_df = pivot_df.dropna(subset=["bbox_crop_path", "mask_crop_path", "mask_tight_crop_path"]).copy() pivot_df["score_spread"] = pivot_df[ ["bbox_only_y_score", "mask_only_y_score", "bbox_plus_mask_y_score"] ].max(axis=1) - pivot_df[ ["bbox_only_y_score", "mask_only_y_score", "bbox_plus_mask_y_score"] ].min(axis=1) pivot_df["combined_gain"] = pivot_df["bbox_plus_mask_y_score"] - pivot_df["bbox_only_y_score"] variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): positive = _take_rows( pivot_df.loc[pivot_df["target_binary"].eq(1)].copy(), n_items=3, variant_index=variant_index, sort_columns=["score_spread", "combined_gain", "visual_priority", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, False, True, True, True], unique_columns=("patient_id", "image_id"), ) negative = _take_rows( pivot_df.loc[pivot_df["target_binary"].eq(0)].copy(), n_items=3, variant_index=variant_index + 1, sort_columns=["score_spread", "combined_gain", "visual_priority", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, False, True, True, True], unique_columns=("patient_id", "image_id"), extra_distinct_column="class_name", ) selected = pd.concat([positive, negative], ignore_index=True) records = selected.to_dict("records") panel_rows: list[dict[str, Any]] = [] for row_index, row in enumerate(records, start=1): for label, source_column in [ ("bbox", "bbox_crop_path"), ("mask", "mask_crop_path"), ("mask_tight", "mask_tight_crop_path"), ]: panel_rows.append( _panel_row( panel_id=f"r{row_index:02d}_{label}", role=label, patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="h2_mask_usage_predictions.parquet", source_path=str(row[source_column]), notes=( f"bbox={float(row['bbox_only_y_score']):.3f}; " f"mask={float(row['mask_only_y_score']):.3f}; " f"combined={float(row['bbox_plus_mask_y_score']):.3f}" ), ) ) variants.append( { **_base_spec( article_id="A01", figure_id="A01-F3", figure_basename="a01_mask_representations_comparison", figure_kind="a01_f3_mask_representations", hypothesis="H2", title="A01-F3. Bounding-box, mask, and mask-tight cell views", variant_index=variant_index, ), "object_rows": records, "panel_rows": panel_rows, } ) return variants def _build_a01_f4_variants(variant_count: int) -> list[dict[str, Any]]: subset_frames = load_h3_subject_frames() clean_df = subset_frames["clean_cell"].copy() strict_df = subset_frames["strict_morphology"].copy() full_df = subset_frames["full_realistic"].copy() strict_keys = set(map(tuple, strict_df[OBJECT_KEY_COLUMNS].itertuples(index=False, name=None))) clean_only_df = clean_df.loc[ [tuple(values) not in strict_keys for values in clean_df[OBJECT_KEY_COLUMNS].itertuples(index=False, name=None)] ].copy() gumpricht_df = clean_only_df.loc[clean_only_df["class_name"].eq("gumpricht shadows")].copy() variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): clean_reference_rows: list[dict[str, Any]] = [] for offset, class_name in enumerate(["blast", "Lymphocyte", "Monocyte", "Myelocyte"], start=0): selected = _take_rows( clean_df.loc[clean_df["class_name"].eq(class_name)].copy(), n_items=1, variant_index=variant_index + offset, sort_columns=["visual_priority", "bbox_area_px", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), ) clean_reference_rows.extend(selected.to_dict("records")) gumpricht_rows = _take_rows( gumpricht_df, n_items=4, variant_index=variant_index, sort_columns=["visual_priority", "bbox_area_px", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), ).to_dict("records") noise_rows = _take_rows( full_df.loc[full_df["class_name"].isin(["Bad cells", "artifacts"])].copy(), n_items=2, variant_index=variant_index, sort_columns=["visual_priority", "bbox_area_px", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), ).to_dict("records") hard_non_blast_rows = _take_rows( full_df.loc[ full_df["class_name"].isin( [ "Lymphocyte", "Monocyte", "Myelocyte", "Segmentonuclear neutrophil", "Normoblasts", "normoblasts: oxyfilny", ] ) ].copy(), n_items=2, variant_index=variant_index + 2, sort_columns=["visual_priority", "bbox_area_px", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), extra_distinct_column="class_name", ).to_dict("records") full_problem_rows = (noise_rows + hard_non_blast_rows)[:4] grid_rows = [ { "clean": clean_reference_rows[index], "strict_removed": gumpricht_rows[index], "full_problem": full_problem_rows[index], } for index in range(4) ] panel_rows: list[dict[str, Any]] = [] for row_index, row_group in enumerate(grid_rows, start=1): for column_index, (role, row) in enumerate(row_group.items(), start=1): panel_rows.append( _panel_row( panel_id=f"r{row_index:02d}c{column_index:02d}", role=role, patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="roi_manifest.parquet", source_path=str(row["bbox_crop_path"]), notes=role.replace("_", " "), ) ) variants.append( { **_base_spec( article_id="A01", figure_id="A01-F4", figure_basename="a01_dataset_regimes_subject_gallery", figure_kind="a01_f4_dataset_regimes", hypothesis="H3", title="A01-F4. Subject differences between clean, strict, and full-realistic corpora", variant_index=variant_index, ), "grid_rows": grid_rows, "panel_rows": panel_rows, } ) return variants def _build_a01_f5_variants(variant_count: int) -> list[dict[str, Any]]: _, baseline_df = load_best_baseline_subject_frame(split_name="test") baseline_df = baseline_df.copy() baseline_df["decision_margin"] = (baseline_df["y_score"] - baseline_df["threshold"]).abs() outcome_sort = { "TP": (["decision_margin", "y_score", "visual_priority", "patient_id", "image_id", "object_id_within_image"], [False, False, False, True, True, True]), "TN": (["decision_margin", "y_score", "visual_priority", "patient_id", "image_id", "object_id_within_image"], [False, True, False, True, True, True]), "FP": (["decision_margin", "y_score", "visual_priority", "patient_id", "image_id", "object_id_within_image"], [False, False, False, True, True, True]), "FN": (["decision_margin", "y_score", "visual_priority", "patient_id", "image_id", "object_id_within_image"], [False, True, False, True, True, True]), } variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): outcome_rows: dict[str, list[dict[str, Any]]] = {} panel_rows: list[dict[str, Any]] = [] for row_index, outcome_label in enumerate(["TP", "TN", "FP", "FN"], start=1): sort_columns, ascending = outcome_sort[outcome_label] selected = _take_rows( baseline_df.loc[baseline_df["outcome_label"].eq(outcome_label)].copy(), n_items=4, variant_index=variant_index + row_index - 1, sort_columns=sort_columns, ascending=ascending, unique_columns=("patient_id", "image_id"), extra_distinct_column="class_name" if outcome_label in {"TN", "FP"} else None, ) records = selected.to_dict("records") outcome_rows[outcome_label] = records for col_index, row in enumerate(records, start=1): panel_rows.append( _panel_row( panel_id=f"{outcome_label}{col_index}", role=outcome_label, patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=( f"true={row['true_label']}; pred={row['predicted_label']}; " f"score={float(row['y_score']):.3f}" ), ) ) variants.append( { **_base_spec( article_id="A01", figure_id="A01-F5", figure_basename="a01_best_handcrafted_tp_tn_fp_fn", figure_kind="a01_f5_confusion_gallery", hypothesis="H1", title="A01-F5. True and false decisions of the best handcrafted model", variant_index=variant_index, ), "outcome_rows": outcome_rows, "panel_rows": panel_rows, } ) return variants def _build_a01_f6_variants(variant_count: int) -> list[dict[str, Any]]: frame = load_h5_subject_frame(split_name="test").copy() score_columns = [ "handcrafted_y_score", "cnn_bbox_y_score", "cnn_mask_y_score", "cnn_late_fusion_y_score", ] pred_columns = [ "handcrafted_y_pred", "cnn_bbox_y_pred", "cnn_mask_y_pred", "cnn_late_fusion_y_pred", ] frame["score_spread"] = frame[score_columns].max(axis=1) - frame[score_columns].min(axis=1) frame["prediction_disagreement"] = frame[pred_columns].nunique(axis=1) frame["handcrafted_margin"] = (frame["handcrafted_y_score"] - 0.5).abs() variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): confident_positive = _take_rows( frame.loc[frame["y_true"].eq(1) & frame["handcrafted_y_pred"].eq(frame["y_true"])].copy(), n_items=2, variant_index=variant_index, sort_columns=["handcrafted_margin", "visual_priority", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), ) confident_negative = _take_rows( frame.loc[frame["y_true"].eq(0) & frame["handcrafted_y_pred"].eq(frame["y_true"])].copy(), n_items=2, variant_index=variant_index + 1, sort_columns=["handcrafted_margin", "visual_priority", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, True, True, True], unique_columns=("patient_id", "image_id"), extra_distinct_column="class_name", ) disagreement_positive = _take_rows( frame.loc[frame["y_true"].eq(1) & frame["prediction_disagreement"].gt(1)].copy(), n_items=2, variant_index=variant_index, sort_columns=["prediction_disagreement", "score_spread", "visual_priority", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, False, True, True, True], unique_columns=("patient_id", "image_id"), ) disagreement_negative = _take_rows( frame.loc[frame["y_true"].eq(0) & frame["prediction_disagreement"].gt(1)].copy(), n_items=2, variant_index=variant_index + 2, sort_columns=["prediction_disagreement", "score_spread", "visual_priority", "patient_id", "image_id", "object_id_within_image"], ascending=[False, False, False, True, True, True], unique_columns=("patient_id", "image_id"), extra_distinct_column="class_name", ) confident_rows = pd.concat([confident_positive, confident_negative], ignore_index=True).to_dict("records") disagreement_rows = pd.concat([disagreement_positive, disagreement_negative], ignore_index=True).to_dict("records") panel_rows: list[dict[str, Any]] = [] for row_index, row in enumerate(confident_rows, start=1): panel_rows.append( _panel_row( panel_id=f"confident_{row_index}", role="confident", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=( f"hcf={float(row['handcrafted_y_score']):.3f}; " f"bbox={float(row['cnn_bbox_y_score']):.3f}; " f"mask={float(row['cnn_mask_y_score']):.3f}; " f"fusion={float(row['cnn_late_fusion_y_score']):.3f}" ), ) ) for row_index, row in enumerate(disagreement_rows, start=1): panel_rows.append( _panel_row( panel_id=f"disagreement_{row_index}", role="disagreement", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or ""), source_table="object_level_deep_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=( f"hcf={float(row['handcrafted_y_score']):.3f}; " f"bbox={float(row['cnn_bbox_y_score']):.3f}; " f"mask={float(row['cnn_mask_y_score']):.3f}; " f"fusion={float(row['cnn_late_fusion_y_score']):.3f}" ), ) ) variants.append( { **_base_spec( article_id="A01", figure_id="A01-F6", figure_basename="a01_handcrafted_vs_deep_same_cells", figure_kind="a01_f6_same_cells_model_comparison", hypothesis="H5", title="A01-F6. Handcrafted and deep scores on the same test cells", variant_index=variant_index, ), "confident_rows": confident_rows, "disagreement_rows": disagreement_rows, "panel_rows": panel_rows, } ) return variants def _build_a02_f1_variants(variant_count: int) -> list[dict[str, Any]]: bundle = load_h4_subject_bundle() image_scores_df = bundle["image_scores_df"].copy() best_prediction_frame = bundle["best_prediction_frame"].copy() pivot = image_scores_df.pivot_table(index=["patient_id", "image_id"], columns="aggregation_method", values="score").reset_index() image_meta = image_scores_df[ ["patient_id", "image_id", "image_path", "blast_object_count", "non_blast_object_count", "n_objects"] ].drop_duplicates() pivot = pivot.merge(image_meta, on=["patient_id", "image_id"], how="left", validate="one_to_one") pivot["method_spread"] = pivot[AGGREGATION_METHOD_ORDER].max(axis=1) - pivot[AGGREGATION_METHOD_ORDER].min(axis=1) pivot = pivot.loc[ pivot["n_objects"].between(6, 18) & pivot["blast_object_count"].ge(1) & pivot["non_blast_object_count"].ge(1) ].copy() variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): image_row = _pick_single_row( pivot, variant_index=variant_index, sort_columns=["method_spread", "n_objects", "max_object_score", "patient_id", "image_id"], ascending=[False, True, False, True, True], dedupe_columns=["patient_id", "image_id"], ) top_object_rows = _group_rows( best_prediction_frame.loc[ best_prediction_frame["patient_id"].eq(image_row["patient_id"]) & best_prediction_frame["image_id"].eq(image_row["image_id"]) ].copy(), top_n=6, ) method_rows = ( image_scores_df.loc[ image_scores_df["patient_id"].eq(image_row["patient_id"]) & image_scores_df["image_id"].eq(image_row["image_id"]) ] .copy() .set_index("aggregation_method") .reindex(AGGREGATION_METHOD_ORDER) .reset_index() .to_dict("records") ) panel_rows = [ _panel_row( panel_id="A", role="smear_image", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), split_name="test", source_table="image_index.parquet", source_path=str(image_row["image_path"]), notes="original smear image", ), _panel_row( panel_id="B", role="smear_with_boxes", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), split_name="test", source_table="object_level_baseline_predictions.parquet", source_path=str(image_row["image_path"]), notes="test image with top scored boxes", ), ] for slot, row in enumerate(top_object_rows, start=1): panel_rows.append( _panel_row( panel_id=f"C{slot}", role="top_object", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or "test"), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=_note_score(float(row["y_score"])), ) ) for method_row in method_rows: panel_rows.append( _panel_row( panel_id=f"D_{method_row['aggregation_method']}", role="aggregation_score", patient_id=str(method_row["patient_id"]), image_id=str(method_row["image_id"]), split_name="test", source_table="image_level_scores.parquet", source_path="image_level_scores.parquet", notes=f"{method_row['aggregation_method']}={float(method_row['score']):.3f}", ) ) variants.append( { **_base_spec( article_id="A02", figure_id="A02-F1", figure_basename="a02_from_cells_to_image_score", figure_kind="a02_f1_cells_to_image_score", hypothesis="H4", title="A02-F1. From cell scores to an image-level score", variant_index=variant_index, ), "image_row": image_row, "top_object_rows": top_object_rows, "method_rows": method_rows, "panel_rows": panel_rows, } ) return variants def _image_case_outcomes(image_scores_df: pd.DataFrame, best_method: str) -> pd.DataFrame: frame = image_scores_df.loc[image_scores_df["aggregation_method"].eq(best_method)].copy() frame["y_true"] = frame["image_contains_blast"].astype(int) frame["y_pred"] = frame["score"].ge(0.5).astype(int) mapping = {(1, 1): "TP", (0, 0): "TN", (0, 1): "FP", (1, 0): "FN"} frame["outcome_label"] = frame.apply(lambda row: mapping[(int(row["y_true"]), int(row["y_pred"]))], axis=1) return frame def _build_a02_f2_variants(variant_count: int) -> list[dict[str, Any]]: bundle = load_h4_subject_bundle() image_scores_df = bundle["image_scores_df"].copy() best_prediction_frame = bundle["best_prediction_frame"].copy() best_method = best_image_aggregation_method() case_df = _image_case_outcomes(image_scores_df, best_method).copy() outcome_sort = { "TP": (["score", "n_objects", "patient_id", "image_id"], [False, True, True, True]), "TN": (["score", "n_objects", "patient_id", "image_id"], [True, True, True, True]), "FP": (["score", "n_objects", "patient_id", "image_id"], [False, True, True, True]), "FN": (["score", "n_objects", "patient_id", "image_id"], [True, True, True, True]), } variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): cases: list[dict[str, Any]] = [] panel_rows: list[dict[str, Any]] = [] for outcome_label in ["TP", "TN", "FP", "FN"]: sort_columns, ascending = outcome_sort[outcome_label] selected_case = _pick_single_row( case_df.loc[case_df["outcome_label"].eq(outcome_label)].copy(), variant_index=variant_index, sort_columns=sort_columns, ascending=ascending, dedupe_columns=["patient_id", "image_id"], ) top_objects = _group_rows( best_prediction_frame.loc[ best_prediction_frame["patient_id"].eq(selected_case["patient_id"]) & best_prediction_frame["image_id"].eq(selected_case["image_id"]) ].copy(), top_n=3, ) cases.append({"outcome_label": outcome_label, "image_row": selected_case, "top_object_rows": top_objects}) panel_rows.append( _panel_row( panel_id=f"{outcome_label}_image", role="case_image", patient_id=str(selected_case["patient_id"]), image_id=str(selected_case["image_id"]), target_binary=int(selected_case["image_contains_blast"]), split_name="test", source_table="image_level_scores.parquet", source_path=str(selected_case["image_path"]), notes=f"{outcome_label}; {best_method}={float(selected_case['score']):.3f}", ) ) for slot, row in enumerate(top_objects, start=1): panel_rows.append( _panel_row( panel_id=f"{outcome_label}_roi_{slot}", role="case_object", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or "test"), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=_note_score(float(row["y_score"])), ) ) variants.append( { **_base_spec( article_id="A02", figure_id="A02-F2", figure_basename="a02_image_level_case_studies", figure_kind="a02_f2_image_case_studies", hypothesis="H4", title="A02-F2. Image-level true and false cases", variant_index=variant_index, ), "best_method": best_method, "cases": cases, "panel_rows": panel_rows, } ) return variants def _build_a02_f3_variants(variant_count: int) -> list[dict[str, Any]]: bundle = load_h4_subject_bundle() image_scores_df = bundle["image_scores_df"].copy() best_prediction_frame = bundle["best_prediction_frame"].copy() pivot = image_scores_df.pivot_table(index=["patient_id", "image_id"], columns="aggregation_method", values="score").reset_index() image_meta = image_scores_df[ ["patient_id", "image_id", "image_path", "blast_object_count", "non_blast_object_count", "n_objects"] ].drop_duplicates() pivot = pivot.merge(image_meta, on=["patient_id", "image_id"], how="left", validate="one_to_one") pivot["method_spread"] = pivot[AGGREGATION_METHOD_ORDER].max(axis=1) - pivot[AGGREGATION_METHOD_ORDER].min(axis=1) pivot = pivot.loc[pivot["n_objects"].between(6, 18)].copy() variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): image_row = _pick_single_row( pivot, variant_index=variant_index + 1, sort_columns=["method_spread", "n_objects", "patient_id", "image_id"], ascending=[False, True, True, True], dedupe_columns=["patient_id", "image_id"], ) top_object_rows = _group_rows( best_prediction_frame.loc[ best_prediction_frame["patient_id"].eq(image_row["patient_id"]) & best_prediction_frame["image_id"].eq(image_row["image_id"]) ].copy(), top_n=4, ) method_rows = ( image_scores_df.loc[ image_scores_df["patient_id"].eq(image_row["patient_id"]) & image_scores_df["image_id"].eq(image_row["image_id"]) ] .copy() .set_index("aggregation_method") .reindex(AGGREGATION_METHOD_ORDER) .reset_index() .to_dict("records") ) panel_rows = [ _panel_row( panel_id="A", role="smear_with_boxes", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), split_name="test", source_table="object_level_baseline_predictions.parquet", source_path=str(image_row["image_path"]), notes="same image for four aggregation methods", ) ] for slot, row in enumerate(top_object_rows, start=1): panel_rows.append( _panel_row( panel_id=f"C{slot}", role="top_object", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or "test"), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=_note_score(float(row["y_score"])), ) ) for method_row in method_rows: panel_rows.append( _panel_row( panel_id=f"D_{method_row['aggregation_method']}", role="aggregation_score", patient_id=str(method_row["patient_id"]), image_id=str(method_row["image_id"]), split_name="test", source_table="image_level_scores.parquet", source_path="image_level_scores.parquet", notes=f"{method_row['aggregation_method']}={float(method_row['score']):.3f}", ) ) variants.append( { **_base_spec( article_id="A02", figure_id="A02-F3", figure_basename="a02_four_aggregation_methods_same_image", figure_kind="a02_f3_same_image_four_aggregators", hypothesis="H4", title="A02-F3. One image under four aggregation methods", variant_index=variant_index, ), "image_row": image_row, "top_object_rows": top_object_rows, "method_rows": method_rows, "panel_rows": panel_rows, } ) return variants def _best_image_method_rows(bundle: dict[str, Any], patient_id: str) -> pd.DataFrame: best_method = best_image_aggregation_method() image_scores_df = bundle["image_scores_df"].copy() return image_scores_df.loc[ image_scores_df["patient_id"].eq(patient_id) & image_scores_df["aggregation_method"].eq(best_method) ].copy() def _top_patient_images(bundle: dict[str, Any], patient_id: str, max_images: int) -> list[dict[str, Any]]: patient_images = _best_image_method_rows(bundle, patient_id).sort_values( ["score", "n_objects", "image_id"], ascending=[False, False, True] ) if len(patient_images) >= 4: selected = patient_images.head(min(max_images, 4)) else: selected = patient_images.head(max_images) return selected.to_dict("records") def _build_a02_f4_variants(variant_count: int) -> list[dict[str, Any]]: bundle = load_h4_subject_bundle() patient_scores_df = bundle["patient_scores_df"].copy() best_prediction_frame = bundle["best_prediction_frame"].copy() best_patient_method = best_patient_aggregation_method() patient_candidates = patient_scores_df.loc[ patient_scores_df["aggregation_method"].eq(best_patient_method) & patient_scores_df["patient_contains_blast"].eq(1) ].copy() best_image_method = best_image_aggregation_method() image_counts = ( bundle["image_scores_df"] .loc[bundle["image_scores_df"]["aggregation_method"].eq(best_image_method)] .groupby("patient_id", as_index=False) .agg(n_images=("image_id", "size"), max_image_score=("score", "max")) ) patient_candidates = patient_candidates.merge(image_counts, on="patient_id", how="left", validate="one_to_one") patient_candidates = patient_candidates.loc[patient_candidates["n_images"].between(3, 25)].copy() variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): patient_row = _pick_single_row( patient_candidates, variant_index=variant_index, sort_columns=["score", "n_images", "patient_id"], ascending=[False, True, True], dedupe_columns=["patient_id"], ) image_rows = _top_patient_images(bundle, str(patient_row["patient_id"]), max_images=4) image_panels: list[dict[str, Any]] = [] panel_rows: list[dict[str, Any]] = [] for image_slot, image_row in enumerate(image_rows, start=1): top_objects = _group_rows( best_prediction_frame.loc[ best_prediction_frame["patient_id"].eq(image_row["patient_id"]) & best_prediction_frame["image_id"].eq(image_row["image_id"]) ].copy(), top_n=2, ) image_panels.append({"image_row": image_row, "top_object_rows": top_objects}) panel_rows.append( _panel_row( panel_id=f"image_{image_slot}", role="patient_image", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), target_binary=int(image_row["image_contains_blast"]), split_name="test", source_table="image_level_scores.parquet", source_path=str(image_row["image_path"]), notes=f"{best_image_method}={float(image_row['score']):.3f}", ) ) for roi_slot, row in enumerate(top_objects, start=1): panel_rows.append( _panel_row( panel_id=f"image_{image_slot}_roi_{roi_slot}", role="patient_image_object", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or "test"), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=_note_score(float(row["y_score"])), ) ) panel_rows.append( _panel_row( panel_id="patient_score", role="patient_score", patient_id=str(patient_row["patient_id"]), split_name="test", source_table="patient_level_scores.parquet", source_path="patient_level_scores.parquet", notes=f"{best_patient_method}={float(patient_row['score']):.3f}", ) ) variants.append( { **_base_spec( article_id="A02", figure_id="A02-F4", figure_basename="a02_from_images_to_patient_score", figure_kind="a02_f4_images_to_patient_score", hypothesis="H4", title="A02-F4. From image-level scores to a patient-level score", variant_index=variant_index, ), "patient_row": patient_row, "best_image_method": best_image_method, "best_patient_method": best_patient_method, "image_panels": image_panels, "panel_rows": panel_rows, } ) return variants def _patient_case_outcomes(patient_scores_df: pd.DataFrame, best_method: str) -> pd.DataFrame: frame = patient_scores_df.loc[patient_scores_df["aggregation_method"].eq(best_method)].copy() frame["y_true"] = frame["patient_contains_blast"].astype(int) frame["y_pred"] = frame["score"].ge(0.5).astype(int) mapping = {(1, 1): "TP", (0, 0): "TN", (0, 1): "FP", (1, 0): "FN"} frame["outcome_label"] = frame.apply(lambda row: mapping[(int(row["y_true"]), int(row["y_pred"]))], axis=1) return frame def _build_a02_f5_variants(variant_count: int) -> list[dict[str, Any]]: bundle = load_h4_subject_bundle() patient_scores_df = bundle["patient_scores_df"].copy() best_prediction_frame = bundle["best_prediction_frame"].copy() best_patient_method = best_patient_aggregation_method() best_image_method = best_image_aggregation_method() patient_cases = _patient_case_outcomes(patient_scores_df, best_patient_method) variants: list[dict[str, Any]] = [] for variant_index in range(1, variant_count + 1): tp_row = _pick_single_row( patient_cases.loc[patient_cases["outcome_label"].eq("TP")].copy(), variant_index=variant_index, sort_columns=["score", "n_objects", "patient_id"], ascending=[False, False, True], dedupe_columns=["patient_id"], ) fp_row = _pick_single_row( patient_cases.loc[patient_cases["outcome_label"].eq("FP")].copy(), variant_index=variant_index, sort_columns=["score", "n_objects", "patient_id"], ascending=[False, False, True], dedupe_columns=["patient_id"], ) case_panels: list[dict[str, Any]] = [] panel_rows: list[dict[str, Any]] = [] for case_label, patient_row in [("TP", tp_row), ("FP", fp_row)]: image_rows = _top_patient_images(bundle, str(patient_row["patient_id"]), max_images=3) image_panels: list[dict[str, Any]] = [] for image_slot, image_row in enumerate(image_rows, start=1): top_objects = _group_rows( best_prediction_frame.loc[ best_prediction_frame["patient_id"].eq(image_row["patient_id"]) & best_prediction_frame["image_id"].eq(image_row["image_id"]) ].copy(), top_n=2, ) image_panels.append({"image_row": image_row, "top_object_rows": top_objects}) panel_rows.append( _panel_row( panel_id=f"{case_label}_image_{image_slot}", role=f"{case_label.lower()}_patient_image", patient_id=str(image_row["patient_id"]), image_id=str(image_row["image_id"]), target_binary=int(image_row["image_contains_blast"]), split_name="test", source_table="image_level_scores.parquet", source_path=str(image_row["image_path"]), notes=f"{best_image_method}={float(image_row['score']):.3f}", ) ) for roi_slot, row in enumerate(top_objects, start=1): panel_rows.append( _panel_row( panel_id=f"{case_label}_image_{image_slot}_roi_{roi_slot}", role=f"{case_label.lower()}_patient_object", patient_id=str(row["patient_id"]), image_id=str(row["image_id"]), object_id_within_image=int(row["object_id_within_image"]), class_name=str(row["class_name"]), target_binary=_target_binary(row), split_name=str(row.get("split_name") or "test"), source_table="object_level_baseline_predictions.parquet", source_path=str(row["bbox_crop_path"]), notes=_note_score(float(row["y_score"])), ) ) panel_rows.append( _panel_row( panel_id=f"{case_label}_patient_score", role=f"{case_label.lower()}_patient_score", patient_id=str(patient_row["patient_id"]), split_name="test", source_table="patient_level_scores.parquet", source_path="patient_level_scores.parquet", notes=f"{best_patient_method}={float(patient_row['score']):.3f}", ) ) case_panels.append({"case_label": case_label, "patient_row": patient_row, "image_panels": image_panels}) variants.append( { **_base_spec( article_id="A02", figure_id="A02-F5", figure_basename="a02_patient_level_case_studies", figure_kind="a02_f5_patient_case_studies", hypothesis="H4", title="A02-F5. Patient-level true-positive and false-positive case studies", variant_index=variant_index, ), "best_image_method": best_image_method, "best_patient_method": best_patient_method, "case_panels": case_panels, "panel_rows": panel_rows, } ) return variants def build_subject_figure_specs(variant_count: int = VARIANT_COUNT) -> list[dict[str, Any]]: specs: list[dict[str, Any]] = [] specs.extend(_build_a01_f1_variants(variant_count)) specs.extend(_build_a01_f2_variants(variant_count)) specs.extend(_build_a01_f3_variants(variant_count)) specs.extend(_build_a01_f4_variants(variant_count)) specs.extend(_build_a01_f5_variants(variant_count)) specs.extend(_build_a01_f6_variants(variant_count)) specs.extend(_build_a02_f1_variants(variant_count)) specs.extend(_build_a02_f2_variants(variant_count)) specs.extend(_build_a02_f3_variants(variant_count)) specs.extend(_build_a02_f4_variants(variant_count)) specs.extend(_build_a02_f5_variants(variant_count)) return specs def flatten_subject_specs(specs: list[dict[str, Any]]) -> pd.DataFrame: rows: list[dict[str, Any]] = [] for spec in specs: for panel_row in spec.get("panel_rows", []): rows.append( { "article_id": spec["article_id"], "figure_id": spec["figure_id"], "figure_variant_id": spec["figure_variant_id"], "figure_basename": spec["figure_basename"], "hypothesis": spec["hypothesis"], "hypothesis_slug": spec["hypothesis_slug"], "variant_index": spec["variant_index"], "figure_kind": spec["figure_kind"], "title": spec["title"], **panel_row, } ) return pd.DataFrame(rows)