| from __future__ import annotations |
|
|
| import argparse |
| import contextlib |
| import csv |
| import json |
| import logging |
| from collections import Counter, defaultdict |
| from pathlib import Path |
|
|
| import numpy as np |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| def write_validation_results_csv( |
| output_path: Path, |
| pairs: list[dict], |
| reviews_by_reviewer: dict[str, list[dict]], |
| exclusions: dict, |
| ) -> None: |
| pair_strata = {idx: classify_transformation(pair) for idx, pair in enumerate(pairs)} |
| implausible_any = set(exclusions["implausible_any_rater"]) |
| implausible_majority = set(exclusions["implausible_majority"]) |
| disputed_quality = set(exclusions["disputed_quality"]) |
| final_excluded = set(exclusions["all_excluded"]) |
|
|
| fieldnames = [ |
| "pair_index", |
| "reviewer_id", |
| "clinically_plausible", |
| "pathology_preserved", |
| "quality_score", |
| "comments", |
| "timestamp", |
| "transform_type", |
| "excluded_implausible_any", |
| "excluded_implausible_majority", |
| "excluded_disputed_quality", |
| "excluded_final", |
| ] |
|
|
| output_path.parent.mkdir(parents=True, exist_ok=True) |
| with open(output_path, "w", newline="") as f: |
| writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") |
| writer.writeheader() |
| for reviewer, rows in sorted(reviews_by_reviewer.items()): |
| for row in sorted(rows, key=lambda r: int(r["pair_index"])): |
| pair_index = int(row["pair_index"]) |
| writer.writerow( |
| { |
| "pair_index": pair_index, |
| "reviewer_id": row.get("reviewer_id", reviewer), |
| "clinically_plausible": row.get("clinically_plausible", ""), |
| "pathology_preserved": row.get("pathology_preserved", ""), |
| "quality_score": row.get("quality_score", ""), |
| "comments": row.get("comments", ""), |
| "timestamp": row.get("timestamp", ""), |
| "transform_type": pair_strata.get(pair_index, "unknown"), |
| "excluded_implausible_any": int(pair_index in implausible_any), |
| "excluded_implausible_majority": int(pair_index in implausible_majority), |
| "excluded_disputed_quality": int(pair_index in disputed_quality), |
| "excluded_final": int(pair_index in final_excluded), |
| } |
| ) |
|
|
|
|
| def load_all_reviews(results_dir: Path) -> dict[str, list[dict]]: |
| reviews_by_reviewer: dict[str, list[dict]] = {} |
| for path in sorted(results_dir.glob("reviews_*.csv")): |
| reviewer = path.stem.replace("reviews_", "") |
| rows: list[dict] = [] |
| with open(path) as f: |
| for row in csv.DictReader(f): |
| rows.append(row) |
| if rows: |
| reviews_by_reviewer[reviewer] = rows |
| return reviews_by_reviewer |
|
|
|
|
| def classify_transformation(pair: dict) -> str: |
| age_changed = pair.get("source_age") != pair.get("target_age") |
| sex_changed = pair.get("source_sex") != pair.get("target_sex") |
| if pair.get("_transform_type"): |
| return str(pair["_transform_type"]) |
| if age_changed and sex_changed: |
| return "intersectional" |
| if age_changed: |
| return "age_only" |
| if sex_changed: |
| return "sex_only" |
| return "none" |
|
|
|
|
| def build_rating_matrix( |
| reviews_by_reviewer: dict[str, list[dict]], |
| field: str, |
| num_pairs: int, |
| value_map: dict[str, float] | None = None, |
| ) -> np.ndarray: |
| reviewers = sorted(reviews_by_reviewer.keys()) |
| matrix = np.full((num_pairs, len(reviewers)), np.nan) |
| for j, reviewer in enumerate(reviewers): |
| for row in reviews_by_reviewer[reviewer]: |
| idx = int(row["pair_index"]) |
| if idx < num_pairs and row.get(field): |
| val = row[field] |
| if value_map and val in value_map: |
| matrix[idx, j] = value_map[val] |
| else: |
| with contextlib.suppress(ValueError, TypeError): |
| matrix[idx, j] = float(val) |
| return matrix |
|
|
|
|
| def fleiss_kappa(matrix: np.ndarray, categories: list[int | float]) -> float: |
| valid_rows = ~np.any(np.isnan(matrix), axis=1) |
| data = matrix[valid_rows] |
| n_subjects = data.shape[0] |
| n_raters = data.shape[1] |
|
|
| if n_subjects == 0 or n_raters < 2: |
| return float("nan") |
|
|
| counts = np.zeros((n_subjects, len(categories))) |
| for k_idx, k in enumerate(categories): |
| counts[:, k_idx] = np.sum(data == k, axis=1) |
|
|
| p_j = np.sum(counts, axis=0) / (n_subjects * n_raters) |
| p_i = (np.sum(counts**2, axis=1) - n_raters) / (n_raters * (n_raters - 1)) |
|
|
| p_bar = np.mean(p_i) |
| p_e = np.sum(p_j**2) |
|
|
| if abs(1.0 - p_e) < 1e-10: |
| return 1.0 |
|
|
| return float((p_bar - p_e) / (1.0 - p_e)) |
|
|
|
|
| def build_stratum_rating_matrix( |
| reviews_by_reviewer: dict[str, list[dict]], |
| field: str, |
| indices: list[int], |
| value_map: dict[str, float] | None = None, |
| ) -> np.ndarray: |
| index_set = set(indices) |
| reviewers = sorted(reviews_by_reviewer.keys()) |
| idx_to_row = {idx: row_num for row_num, idx in enumerate(sorted(indices))} |
| matrix = np.full((len(indices), len(reviewers)), np.nan) |
| for j, reviewer in enumerate(reviewers): |
| for row in reviews_by_reviewer[reviewer]: |
| pair_idx = int(row["pair_index"]) |
| if pair_idx in index_set and row.get(field): |
| val = row[field] |
| matrix_row = idx_to_row[pair_idx] |
| if value_map and val in value_map: |
| matrix[matrix_row, j] = value_map[val] |
| else: |
| with contextlib.suppress(ValueError, TypeError): |
| matrix[matrix_row, j] = float(val) |
| return matrix |
|
|
|
|
| def compute_stratum_kappa( |
| pairs: list[dict], |
| reviews_by_reviewer: dict[str, list[dict]], |
| ) -> dict[str, dict[str, float | None]]: |
| pair_strata: dict[int, str] = {} |
| for i, pair in enumerate(pairs): |
| pair_strata[i] = classify_transformation(pair) |
|
|
| strata = ["age_only", "sex_only", "intersectional"] |
| result: dict[str, dict[str, float | None]] = {} |
|
|
| for stratum in strata: |
| indices = [i for i, s in pair_strata.items() if s == stratum] |
| if not indices: |
| result[stratum] = { |
| "fleiss_kappa_quality": None, |
| "fleiss_kappa_plausibility": None, |
| "fleiss_kappa_pathology_preservation": None, |
| } |
| continue |
|
|
| q_matrix = build_stratum_rating_matrix(reviews_by_reviewer, "quality_score", indices) |
| p_matrix = build_stratum_rating_matrix( |
| reviews_by_reviewer, "clinically_plausible", indices, |
| value_map={"Yes": 1.0, "No": 0.0}, |
| ) |
| pp_matrix = build_stratum_rating_matrix( |
| reviews_by_reviewer, "pathology_preserved", indices, |
| value_map={"Yes": 1.0, "No": 0.0, "Uncertain": 0.5}, |
| ) |
|
|
| kq = fleiss_kappa(q_matrix, categories=[1, 2, 3, 4, 5]) |
| kp = fleiss_kappa(p_matrix, categories=[0, 1]) |
| kpp = fleiss_kappa(pp_matrix, categories=[0, 0.5, 1]) |
|
|
| result[stratum] = { |
| "fleiss_kappa_quality": round(kq, 4) if not np.isnan(kq) else None, |
| "fleiss_kappa_plausibility": round(kp, 4) if not np.isnan(kp) else None, |
| "fleiss_kappa_pathology_preservation": round(kpp, 4) if not np.isnan(kpp) else None, |
| } |
|
|
| return result |
|
|
|
|
| def compute_exclusions( |
| reviews_by_reviewer: dict[str, list[dict]], |
| num_pairs: int, |
| ) -> dict: |
| plausible_counts: dict[int, dict[str, int]] = defaultdict(lambda: {"Yes": 0, "No": 0}) |
| quality_scores: dict[int, list[int]] = defaultdict(list) |
| reviewer_counts: dict[int, int] = defaultdict(int) |
|
|
| for reviews in reviews_by_reviewer.values(): |
| for row in reviews: |
| idx = int(row["pair_index"]) |
| reviewer_counts[idx] += 1 |
| p = row.get("clinically_plausible", "Yes") |
| if p in ("Yes", "No"): |
| plausible_counts[idx][p] += 1 |
| if row.get("quality_score"): |
| quality_scores[idx].append(int(row["quality_score"])) |
|
|
| excluded_implausible: list[int] = [] |
| excluded_majority: list[int] = [] |
| excluded_disputed: list[int] = [] |
|
|
| for idx in range(num_pairs): |
| if reviewer_counts[idx] == 0: |
| continue |
|
|
| no_votes = plausible_counts[idx]["No"] |
| if no_votes > 0: |
| excluded_implausible.append(idx) |
|
|
| total = reviewer_counts[idx] |
| if no_votes > total / 2: |
| excluded_majority.append(idx) |
|
|
| scores = quality_scores.get(idx, []) |
| if len(scores) >= 3 and (max(scores) - min(scores)) >= 3 and idx not in excluded_implausible: |
| score_counts = Counter(scores) |
| has_majority = any(c > len(scores) / 2 for c in score_counts.values()) |
| if not has_majority: |
| excluded_disputed.append(idx) |
|
|
| all_excluded = sorted(set(excluded_implausible + excluded_disputed)) |
|
|
| return { |
| "implausible_any_rater": excluded_implausible, |
| "implausible_majority": excluded_majority, |
| "disputed_quality": excluded_disputed, |
| "all_excluded": all_excluded, |
| } |
|
|
|
|
| def compute_stratum_stats( |
| pairs: list[dict], |
| reviews_by_reviewer: dict[str, list[dict]], |
| exclusions: dict, |
| ) -> dict: |
| pair_strata: dict[int, str] = {} |
| for i, pair in enumerate(pairs): |
| pair_strata[i] = classify_transformation(pair) |
|
|
| all_reviews_by_pair: dict[int, list[dict]] = defaultdict(list) |
| for reviews in reviews_by_reviewer.values(): |
| for row in reviews: |
| idx = int(row["pair_index"]) |
| all_reviews_by_pair[idx].append(row) |
|
|
| strata = ["age_only", "sex_only", "intersectional"] |
| stats: dict[str, dict] = {} |
|
|
| excluded_set = set(exclusions["all_excluded"]) |
|
|
| for stratum in strata: |
| indices = [i for i, s in pair_strata.items() if s == stratum] |
| if not indices: |
| stats[stratum] = {"total": 0} |
| continue |
|
|
| n_excluded = sum(1 for i in indices if i in excluded_set) |
| n_passed = len(indices) - n_excluded |
|
|
| plausible_yes = 0 |
| plausible_total = 0 |
| preserved_yes = 0 |
| preserved_total = 0 |
| quality_vals: list[int] = [] |
|
|
| for idx in indices: |
| for row in all_reviews_by_pair.get(idx, []): |
| if row.get("clinically_plausible") in ("Yes", "No"): |
| plausible_total += 1 |
| if row["clinically_plausible"] == "Yes": |
| plausible_yes += 1 |
| if row.get("pathology_preserved") in ("Yes", "No"): |
| preserved_total += 1 |
| if row["pathology_preserved"] == "Yes": |
| preserved_yes += 1 |
| if row.get("quality_score"): |
| quality_vals.append(int(row["quality_score"])) |
|
|
| stats[stratum] = { |
| "total": len(indices), |
| "passed": n_passed, |
| "excluded": n_excluded, |
| "pass_rate": round(n_passed / len(indices), 3) if indices else 0, |
| "plausibility_rate": round(plausible_yes / plausible_total, 3) if plausible_total else None, |
| "pathology_preservation_rate": round(preserved_yes / preserved_total, 3) if preserved_total else None, |
| "mean_quality": round(float(np.mean(quality_vals)), 2) if quality_vals else None, |
| "quality_distribution": { |
| str(k): int(np.sum(np.array(quality_vals) == k)) |
| for k in range(1, 6) |
| } if quality_vals else {}, |
| } |
|
|
| return stats |
|
|
|
|
| def main() -> None: |
| logging.basicConfig(level=logging.INFO) |
| parser = argparse.ArgumentParser(description="Aggregate radiologist reviews") |
| parser.add_argument("--pairs", type=str, required=True) |
| parser.add_argument("--results-dir", type=str, default="results") |
| parser.add_argument("--output", type=str, default="validation_report.json") |
| parser.add_argument("--csv-output", type=str, default=None) |
| args = parser.parse_args() |
|
|
| with open(args.pairs) as f: |
| pairs = json.load(f) |
| num_pairs = len(pairs) |
|
|
| reviews_by_reviewer = load_all_reviews(Path(args.results_dir)) |
| reviewers = sorted(reviews_by_reviewer.keys()) |
|
|
| if not reviewers: |
| logger.warning("No review files found.") |
| return |
|
|
| quality_matrix = build_rating_matrix( |
| reviews_by_reviewer, "quality_score", num_pairs, |
| ) |
| plausible_matrix = build_rating_matrix( |
| reviews_by_reviewer, "clinically_plausible", num_pairs, |
| value_map={"Yes": 1.0, "No": 0.0}, |
| ) |
| preserved_matrix = build_rating_matrix( |
| reviews_by_reviewer, "pathology_preserved", num_pairs, |
| value_map={"Yes": 1.0, "No": 0.0, "Uncertain": 0.5}, |
| ) |
|
|
| kappa_quality = fleiss_kappa(quality_matrix, categories=[1, 2, 3, 4, 5]) |
| kappa_plausible = fleiss_kappa(plausible_matrix, categories=[0, 1]) |
| kappa_preserved = fleiss_kappa(preserved_matrix, categories=[0, 0.5, 1]) |
|
|
| exclusions = compute_exclusions(reviews_by_reviewer, num_pairs) |
| stratum_stats = compute_stratum_stats(pairs, reviews_by_reviewer, exclusions) |
| stratum_kappa = compute_stratum_kappa(pairs, reviews_by_reviewer) |
|
|
| total_reviewed = sum(len(rows) for rows in reviews_by_reviewer.values()) |
| per_reviewer: dict[str, dict] = {} |
| all_quality: list[int] = [] |
| for reviewer, rows in reviews_by_reviewer.items(): |
| scores = [int(r["quality_score"]) for r in rows if r.get("quality_score")] |
| n_implausible = sum(1 for r in rows if r.get("clinically_plausible") == "No") |
| n_not_preserved = sum(1 for r in rows if r.get("pathology_preserved") == "No") |
| per_reviewer[reviewer] = { |
| "reviewed": len(rows), |
| "mean_quality": round(float(np.mean(scores)), 2) if scores else None, |
| "flagged_implausible": n_implausible, |
| "flagged_not_preserved": n_not_preserved, |
| } |
| all_quality.extend(scores) |
|
|
| quality_dist = {str(k): int(np.sum(np.array(all_quality) == k)) for k in range(1, 6)} if all_quality else {} |
|
|
| valid_quality = quality_matrix[~np.all(np.isnan(quality_matrix), axis=1)] |
| mean_quality = float(np.nanmean(valid_quality)) if valid_quality.size > 0 else None |
|
|
| report = { |
| "num_pairs": num_pairs, |
| "num_reviewers": len(reviewers), |
| "reviewers": reviewers, |
| "total_reviews": total_reviewed, |
| "inter_rater_reliability": { |
| "overall": { |
| "fleiss_kappa_quality": round(kappa_quality, 4) if not np.isnan(kappa_quality) else None, |
| "fleiss_kappa_plausibility": round(kappa_plausible, 4) if not np.isnan(kappa_plausible) else None, |
| "fleiss_kappa_pathology_preservation": round(kappa_preserved, 4) if not np.isnan(kappa_preserved) else None, |
| }, |
| "by_transformation_type": stratum_kappa, |
| }, |
| "overall": { |
| "mean_quality_score": round(mean_quality, 2) if mean_quality else None, |
| "quality_distribution": quality_dist, |
| "pass_rate": round( |
| (num_pairs - len(exclusions["all_excluded"])) / num_pairs, 3 |
| ) if num_pairs > 0 else 0, |
| "plausibility_rate": round( |
| float(np.nanmean(plausible_matrix[~np.all(np.isnan(plausible_matrix), axis=1)])), 3 |
| ) if plausible_matrix.size > 0 else None, |
| "pathology_preservation_rate": round( |
| float(np.nanmean(preserved_matrix[~np.all(np.isnan(preserved_matrix), axis=1)])), 3 |
| ) if preserved_matrix.size > 0 else None, |
| }, |
| "by_transformation_type": stratum_stats, |
| "exclusions": { |
| "implausible_any_rater": len(exclusions["implausible_any_rater"]), |
| "implausible_majority_vote": len(exclusions["implausible_majority"]), |
| "disputed_quality": len(exclusions["disputed_quality"]), |
| "total_excluded": len(exclusions["all_excluded"]), |
| "excluded_pair_indices": exclusions["all_excluded"], |
| }, |
| "per_reviewer": per_reviewer, |
| } |
|
|
| output_path = Path(args.output) |
| output_path.parent.mkdir(parents=True, exist_ok=True) |
| with open(output_path, "w") as f: |
| json.dump(report, f, indent=2) |
|
|
| exclusion_path = output_path.parent / "exclusion_list.json" |
| with open(exclusion_path, "w") as f: |
| json.dump({"excluded_pair_indices": exclusions["all_excluded"]}, f, indent=2) |
|
|
| csv_output_path = ( |
| Path(args.csv_output) |
| if args.csv_output |
| else output_path.parent / "validation_results.csv" |
| ) |
| write_validation_results_csv(csv_output_path, pairs, reviews_by_reviewer, exclusions) |
|
|
| logger.info("Reviewers: %d", len(reviewers)) |
| logger.info("Total reviews: %d", total_reviewed) |
| irr = report["inter_rater_reliability"]["overall"] |
| logger.info("Fleiss kappa (quality): %s", irr["fleiss_kappa_quality"]) |
| logger.info("Fleiss kappa (plausibility): %s", irr["fleiss_kappa_plausibility"]) |
| logger.info("Fleiss kappa (pathology): %s", irr["fleiss_kappa_pathology_preservation"]) |
| for stratum, sk in stratum_kappa.items(): |
| logger.info( |
| " %s kappa: quality=%s plausibility=%s pathology=%s", |
| stratum, sk["fleiss_kappa_quality"], sk["fleiss_kappa_plausibility"], |
| sk["fleiss_kappa_pathology_preservation"], |
| ) |
| logger.info("Overall pass rate: %s", report["overall"]["pass_rate"]) |
| logger.info("Total excluded: %d", report["exclusions"]["total_excluded"]) |
| for stratum, stats in stratum_stats.items(): |
| if stats.get("total", 0) > 0: |
| logger.info(" %s: %d/%d passed (%s)", stratum, stats["passed"], stats["total"], stats["pass_rate"]) |
| logger.info("Report: %s", output_path) |
| logger.info("Exclusion list: %s", exclusion_path) |
| logger.info("Validation CSV: %s", csv_output_path) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
