Spaces:

ProCreations
/

repro-human-ai-collaborative-uncertainty-quantification-native

Running

repro-human-ai-collaborative-uncertainty-quantification-native / code /offline_helpers.py

Upgrade own low-score logbook with native ImageNet-16H execution

002f2e6 verified 3 months ago

22.5 kB

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional, Sequence, List, Tuple, Any

import json
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split



LABELS: List[str] = [
    "knife", "keyboard", "elephant", "bicycle", "airplane", "clock", "oven", "chair",
    "bear", "boat", "cat", "bottle", "truck", "car", "bird", "dog",
]

NOISE_LEVELS: List[int] = [80, 95, 110, 125]
MODEL_NAMES: List[str] = ["alexnet", "vgg19", "densenet161", "googlenet", "resnet152"]


#path config
@dataclass
class Imagenet16HPaths:
    """
    Centralized paths for the ImageNet16H experiments.
    """
    data_root: Path
    raw_model_csv: Path
    human_alone_csv: Path
    models_dir: Path
    users_dir: Path
    meta_dir: Path
    true_labels_csv: Path
    all_expert_csv: Path

    @classmethod
    def from_data_root(cls, data_root: Path) -> "Imagenet16HPaths":
        base = data_root / "imagenet16h"
        models_dir = data_root / "models"
        users_dir = data_root / "users"
        meta_dir = base / "metadata"
        models_dir.mkdir(parents=True, exist_ok=True)
        users_dir.mkdir(parents=True, exist_ok=True)
        meta_dir.mkdir(parents=True, exist_ok=True)

        return cls(
            data_root=data_root,
            raw_model_csv=base / "hai_epoch10_model_preds_max_normalized.csv",
            human_alone_csv=base / "human_only_classification_6per_img_export.csv",
            models_dir=models_dir,
            users_dir=users_dir,
            meta_dir=meta_dir,
            true_labels_csv=data_root / "imagenet16H.csv",
            all_expert_csv=base / "human_only_classification_6per_img_export.csv",
        )


#data prep utilities

def _ensure_exists(p: Path, hint: str = "") -> None:
    if not p.exists():
        raise FileNotFoundError(f"Missing: {p}\n{hint}")


def build_labels_from_raw_csv(paths: Imagenet16HPaths) -> None:
    """
    Create imagenet16H.csv = mapping image_name -> category (one row per image),
    and classes.json describing the 16 classes.
    """
    _ensure_exists(paths.raw_model_csv, "Put the provided model-output CSV here.")

    raw = pd.read_csv(paths.raw_model_csv, dtype=str)
    base = raw[
        (raw["model_name"] == "vgg19") & (raw["noise_level"] == "80")
    ][["image_name", "category"]].drop_duplicates()
    base = base.set_index("image_name").sort_index()

    paths.true_labels_csv.parent.mkdir(parents=True, exist_ok=True)
    base.to_csv(paths.true_labels_csv)

    classes_json = {i: {"name": l} for i, l in enumerate(LABELS)}
    (paths.meta_dir / "classes.json").write_text(
        json.dumps(classes_json, indent=2)
    )

    print(f"[labels] wrote {paths.true_labels_csv} ({len(base)} rows)")
    print(f"[labels] wrote {paths.meta_dir / 'classes.json'}")


def split_model_predictions(paths: Imagenet16HPaths, model_name: str) -> None:
    """
    Save per-noise CSV: models/noise{nl}/{model_name}.csv
    columns = probs for 16 labels + 'correct', index=image_name.
    """
    _ensure_exists(paths.raw_model_csv)

    dtypes = {
        "image_name": str,
        "correct": int,
        "noise_level": str,
        "model_name": str,
        "category": str,
    }
    for l in LABELS:
        dtypes[l] = float

    raw = pd.read_csv(paths.raw_model_csv, dtype=dtypes)
    raw = raw[raw["model_name"] == model_name]

    for nl in NOISE_LEVELS:
        sub = raw[raw["noise_level"] == str(nl)]
        cols = ["image_name"] + LABELS + ["correct"]
        df = sub[cols].drop_duplicates().set_index("image_name")
        out_dir = paths.models_dir / f"noise{nl}"
        out_dir.mkdir(parents=True, exist_ok=True)
        out = out_dir / f"{model_name}.csv"
        df.to_csv(out)
        print(f"[model] {model_name} ω={nl}: {len(df)} rows -> {out}")


def sort_model_predictions(paths: Imagenet16HPaths, model_name: str) -> None:
    """
    Write models/noise{nl}/{model_name}_sorted.csv = rank-ordered labels per image
    (columns '1'..'16').
    """
    _ensure_exists(paths.raw_model_csv)

    dtypes = {"image_name": str, "model_name": str, "noise_level": str}
    for l in LABELS:
        dtypes[l] = float

    raw = pd.read_csv(paths.raw_model_csv, dtype=dtypes)
    raw = raw[raw["model_name"] == model_name]

    labels_arr = np.array(LABELS)
    out_cols = [str(i) for i in range(1, 17)]

    for nl in NOISE_LEVELS:
        sub = raw[raw["noise_level"] == str(nl)]
        probs = sub[["image_name"] + LABELS].drop_duplicates().set_index("image_name")

        row_sums = probs.sum(axis=1).replace(0, np.nan)
        probs = probs.div(row_sums, axis=0).fillna(1.0 / len(LABELS))

        order = np.argsort(probs.values, axis=1)[:, ::-1]
        sorted_labels = labels_arr[order]

        sorted_df = pd.DataFrame(sorted_labels, index=probs.index, columns=out_cols)
        out_dir = paths.models_dir / f"noise{nl}"
        out_dir.mkdir(parents=True, exist_ok=True)
        out_path = out_dir / f"{model_name}_sorted.csv"
        sorted_df.to_csv(out_path)
        print(f"[sorted] {model_name} ω={nl}: {len(sorted_df)} rows -> {out_path}")


def prep_human_tables(paths: Imagenet16HPaths) -> None:
    """
    From the human-alone CSV, write:
        users/noise{nl}/success.csv
        users/noise{nl}/predictions.csv
    """
    _ensure_exists(paths.human_alone_csv)

    raw = pd.read_csv(paths.human_alone_csv, dtype=str)
    for nl in NOISE_LEVELS:
        sub = raw[raw["noise_level"] == str(nl)]
        success = sub[["image_name", "participant_id", "correct"]].set_index("image_name")
        preds = sub[["image_name", "participant_id", "participant_classification"]].set_index("image_name")
        out_dir = paths.users_dir / f"noise{nl}"
        out_dir.mkdir(parents=True, exist_ok=True)
        success.to_csv(out_dir / "success.csv")
        preds.to_csv(out_dir / "predictions.csv")
        print(f"[human] ω={nl}: success({len(success)}), predictions({len(preds)}) -> {out_dir}")


# expert prob

def _canon(s: str) -> str:
    return str(s).strip().lower()


def _label_mapper() -> Dict[str, str]:
    m = {l: l for l in LABELS}
    m.update({
        "aeroplane": "airplane", "plane": "airplane",
        "bike": "bicycle",
        "kitty": "cat", "cat.": "cat",
        "puppy": "dog",
        "sofa": "chair", "couch": "chair",
        "jug": "bottle", "bottle.": "bottle",
        "truck.": "truck", "car.": "car",
        "bird.": "bird", "dog.": "dog",
    })
    return m


def _map_to_vocab(raw: str, mapper: Dict[str, str]) -> Optional[str]:
    r = _canon(raw)
    if r in mapper:
        return mapper[r]
    if r.endswith("s") and r[:-1] in mapper:
        return mapper[r[:-1]]
    return None


def build_expert_freq_probs(
    paths: Imagenet16HPaths,
    noise_level: int,
    labels: Sequence[str] = LABELS,
) -> pd.DataFrame:
    """
    Build p(y|x) from expert classifications; write:
        users/noise{nl}/all_expert_pyx_strict.csv
    """
    _ensure_exists(paths.all_expert_csv)

    df = pd.read_csv(paths.all_expert_csv, dtype=str)
    req = {"image_name", "participant_classification", "noise_level"}
    if not req.issubset(df.columns):
        raise ValueError(f"Expert CSV must contain columns {req}, got {set(df.columns)}")

    df_noise = df[df["noise_level"] == str(noise_level)].copy()
    mapper = _label_mapper()
    df_noise["mapped"] = df_noise["participant_classification"].map(
        lambda s: _map_to_vocab(s, mapper)
    )
    df_noise = df_noise[df_noise["mapped"].isin(labels)].copy()

    counts = df_noise.groupby(["image_name", "mapped"]).size().unstack(fill_value=0)
    counts = counts.reindex(columns=list(labels), fill_value=0)

    row_sums = counts.sum(axis=1)
    images_with_votes = row_sums[row_sums > 0].index
    counts = counts.loc[images_with_votes]
    row_sums = row_sums.loc[images_with_votes]

    pyx = counts.div(row_sums, axis=0)
    pyx.index.name = "image_name"

    out_dir = paths.users_dir / f"noise{noise_level}"
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / "all_expert_pyx_strict.csv"
    pyx.to_csv(out_path)
    print(f"[all expert p(y|x)] strict ω={noise_level}: {len(pyx)} images -> {out_path}")
    return pyx


# human expert and ai model class

class HumanExpert:
    """
    Represents a human expert providing p(y|x) and prediction sets.
    Expects users/noise{nl}/all_expert_pyx_strict.csv 
    """
    def __init__(self, paths: Imagenet16HPaths, noise_level: int, labels: Sequence[str] = LABELS):
        self.paths = paths
        self.noise_level = noise_level
        self.labels = list(labels)

        path = paths.users_dir / f"noise{noise_level}" / "all_expert_pyx_strict.csv"
        if not path.exists():
            print(f"[HumanExpert] {path} not found. Building from ALL_EXPERT_CSV.")
            build_expert_freq_probs(paths, noise_level)
        if not path.exists():
            raise FileNotFoundError(f"Failed to build {path}")

        self.p = pd.read_csv(path, index_col="image_name").reindex(columns=self.labels, fill_value=0.0)
        row_sums = self.p.sum(axis=1).replace(0, np.nan)
        self.p = self.p.div(row_sums, axis=0).fillna(1.0 / len(self.labels))

    def available_images(self) -> List[str]:
        return self.p.index.tolist()

    def get_prob(self, image_name: str) -> pd.Series:
        if image_name not in self.p.index:
            raise ValueError(f"Image {image_name} not found in human expert data for noise level {self.noise_level}.")
        return self.p.loc[image_name].astype(float)

    def predict_set(
        self,
        image_name: str,
        strategy: str = "top1",
        k: Optional[int] = None,
        mass: Optional[float] = None,
        threshold: Optional[float] = None,
    ) -> List[str]:
        """
        Build a prediction set from the human probability vector.
        Supported strategies:
          - "top1"
          - "empty"
          - "topk"      -> uses k argument or defaults to 2
          - "topk5"     -> parses k=5 from the strategy name (any k in [1, num_labels])
          - "mass"
          - "threshold"
        """
        strat = strategy.lower()
        p = self.get_prob(image_name).sort_values(ascending=False)

        # top-1
        if strat == "top1":
            return [p.index[0]]

        # empty set (AI-alone baseline)
        if strat == "empty":
            return []

        # top-k: accept "topk", "topk5", etc.
        if strat.startswith("topk"):
            parsed_k: Optional[int] = None
            suffix = strat[4:]  # part after "topk"
            if suffix:
                if suffix.isdigit():
                    parsed_k = int(suffix)
                else:
                    raise ValueError(f"Unrecognized topk strategy '{strategy}'")

        
            if k is not None:
                parsed_k = int(k)

         
            if parsed_k is None:
                parsed_k = 2


            parsed_k = max(1, min(parsed_k, len(p)))
            return list(p.index[:parsed_k])

        # cumulative mass strategy
        if strat == "mass":
            if not (isinstance(mass, (int, float)) and 0 < mass <= 1):
                raise ValueError("mass in (0,1] required for 'mass' strategy")
            cs = p.cumsum().values
            cut_idx = int(np.searchsorted(cs, float(mass), side="left"))
            cut_idx = min(cut_idx, len(p) - 1)
            return list(p.index[: cut_idx + 1])

        # probability threshold strategy
        if strat == "threshold":
            if not (isinstance(threshold, (int, float)) and 0 < threshold <= 1):
                raise ValueError("threshold in (0,1] required for 'threshold' strategy")
            chosen = [lbl for lbl, val in p.items() if val >= float(threshold)]
            return chosen or [p.index[0]]

        raise ValueError(f"Unknown strategy: {strategy}")



class AIModel:
    """
    Represents an AI model providing p(y|x).
    Expects models/noise{nl}/{model_name}.csv.
    """
    def __init__(self, paths: Imagenet16HPaths, noise_level: int, model_name: str, labels: Sequence[str] = LABELS):
        self.paths = paths
        self.noise_level = noise_level
        self.model_name = model_name
        self.labels = list(labels)

        path = paths.models_dir / f"noise{noise_level}" / f"{model_name}.csv"
        if not path.exists():
            raise FileNotFoundError(
                f"Missing model probabilities file: {path}. "
                f"Ensure split_model_predictions(paths, '{model_name}') was run for ω={noise_level}."
            )

        try:
            self.p = pd.read_csv(path, index_col="image_name")
        except ValueError:
            self.p = pd.read_csv(path, index_col=0)

        self.p = (
            self.p.reindex(columns=self.labels)
            .apply(pd.to_numeric, errors="coerce")
            .fillna(0.0)
        )
        row_sums = self.p.sum(axis=1)
        row_sums[row_sums == 0] = 1.0
        self.p = self.p.div(row_sums, axis=0)

    def get_prob(self, image_name: str) -> pd.Series:
        if image_name not in self.p.index:
            raise ValueError(
                f"Image {image_name} not found in {self.model_name} data for noise level {self.noise_level}."
            )
        return self.p.loc[image_name].astype(float)


# CP utilities


def get_common_image_names(
    ai_model: AIModel,
    human_expert: HumanExpert,
    true_labels_df: pd.DataFrame,
) -> List[str]:
    hs = set(human_expert.available_images())
    ms = set(ai_model.p.index)
    ts = set(true_labels_df.index)
    return sorted(hs & ms & ts)


def calculate_thresholds(
    cal_images: List[str],
    ai_model: AIModel,
    human_expert: HumanExpert,
    true_labels_df: pd.DataFrame,
    epsilon: float,
    delta: float,
    human_strategy: str,
) -> Tuple[float, float]:
    scores_in_h = []
    scores_not_in_h = []

    for image_name in cal_images:
        if image_name not in true_labels_df.index:
            continue
        try:
            ai_probs = ai_model.get_prob(image_name)
            human_set = human_expert.predict_set(image_name, strategy=human_strategy)
        except ValueError:
            continue

        true_label = true_labels_df.loc[image_name, "category"]
        score = 1.0 - ai_probs.get(true_label, 0.0)

        if true_label in human_set:
            scores_in_h.append(score)
        else:
            scores_not_in_h.append(score)

    if scores_in_h:
        b = np.quantile(scores_in_h, 1.0 - epsilon, method="higher")
    else:
        b = np.inf

    if scores_not_in_h:
        a = np.quantile(scores_not_in_h, delta, method="higher")
    else:
        a = np.inf

    return a, b


def generate_prediction_sets(
    test_images: List[str],
    ai_model: AIModel,
    human_expert: HumanExpert,
    a_threshold: float,
    b_threshold: float,
    human_strategy: str,
) -> Dict[str, Dict[str, Any]]:
    predictions: Dict[str, Dict[str, Any]] = {}

    for image_name in test_images:
        try:
            ai_probs = ai_model.get_prob(image_name)
            human_set = human_expert.predict_set(image_name, strategy=human_strategy)
        except ValueError:
            predictions[image_name] = {"prediction_set": [], "set_size": 0}
            continue

        pred_set: List[str] = []
        for label in ai_model.labels:
            score = 1.0 - ai_probs.get(label, 0.0)
            threshold = b_threshold if label in human_set else a_threshold
            if score <= threshold:
                pred_set.append(label)

        predictions[image_name] = {"prediction_set": pred_set, "set_size": len(pred_set)}

    return predictions


def calculate_coverage(
    predictions: Dict[str, Dict[str, Any]],
    true_labels_df: pd.DataFrame,
) -> float:
    if not predictions:
        return 0.0

    correct = 0
    total = 0

    for image_name, data in predictions.items():
        if image_name not in true_labels_df.index:
            continue
        total += 1
        true_label = true_labels_df.loc[image_name, "category"]
        if true_label in data.get("prediction_set", []):
            correct += 1

    if total == 0:
        return 0.0
    return correct / total


#sing run of experiment

def run_single_split(
    paths: Imagenet16HPaths,
    noise_level: int,
    model_name: str,
    human_strategy: str,
    epsilon: float,
    delta: float,
    test_size: float = 0.5,
    random_state: int = 123,
) -> Dict[str, float]:
    true_labels_df = pd.read_csv(paths.true_labels_csv, index_col="image_name")

    ai_model = AIModel(paths, noise_level=noise_level, model_name=model_name)
    human_expert = HumanExpert(paths, noise_level=noise_level)

    all_images = get_common_image_names(ai_model, human_expert, true_labels_df)
    if not all_images:
        raise RuntimeError(f"No common images for ω={noise_level}, model={model_name}.")

    cal_images, test_images = train_test_split(
        all_images,
        test_size=test_size,
        random_state=random_state,
    )

    # Human-alone on test
    human_preds_test: Dict[str, Dict[str, Any]] = {}
    for img in test_images:
        hset = human_expert.predict_set(img, strategy=human_strategy)
        human_preds_test[img] = {"prediction_set": hset, "set_size": len(hset)}

    human_cov = calculate_coverage(human_preds_test, true_labels_df)
    human_sizes = [d["set_size"] for d in human_preds_test.values()]
    human_size = float(np.mean(human_sizes)) if human_sizes else 0.0

    # Collaborative method
    a, b = calculate_thresholds(
        cal_images,
        ai_model,
        human_expert,
        true_labels_df,
        epsilon=epsilon,
        delta=delta,
        human_strategy=human_strategy,
    )

    preds_test = generate_prediction_sets(
        test_images,
        ai_model,
        human_expert,
        a_threshold=a,
        b_threshold=b,
        human_strategy=human_strategy,
    )

    method_cov = calculate_coverage(preds_test, true_labels_df)
    method_sizes = [d["set_size"] for d in preds_test.values()]
    method_size = float(np.mean(method_sizes)) if method_sizes else 0.0


    true_labels_dict = true_labels_df.loc[test_images, "category"].to_dict()

    count_Y_not_in_H = 0
    count_Y_in_C_and_Y_not_in_H = 0
    count_Y_in_H = 0
    count_Y_not_in_C_and_Y_in_H = 0

    for img in test_images:
        if img not in true_labels_dict:
            continue
        y = true_labels_dict[img]
        hset = human_preds_test.get(img, {}).get("prediction_set", [])
        cset = preds_test.get(img, {}).get("prediction_set", [])

        if y not in hset:
            count_Y_not_in_H += 1
            if y in cset:
                count_Y_in_C_and_Y_not_in_H += 1
        else:
            count_Y_in_H += 1
            if y not in cset:
                count_Y_not_in_C_and_Y_in_H += 1

    cond_cov_not_in_H = (
        count_Y_in_C_and_Y_not_in_H / count_Y_not_in_H
        if count_Y_not_in_H > 0 else np.nan
    )
    cond_err_in_H = (
        count_Y_not_in_C_and_Y_in_H / count_Y_in_H
        if count_Y_in_H > 0 else np.nan
    )

    return {
        "human_coverage": float(human_cov),
        "human_set_size": float(human_size),
        "method_coverage": float(method_cov),
        "method_set_size": float(method_size),
        "cond_coverage_not_in_H": float(cond_cov_not_in_H),
        "cond_error_in_H": float(cond_err_in_H),
    }



# Sweeps over strategies / (ε, δ)


def sweep_strategies_eps_delta(
    paths: Imagenet16HPaths,
    noise_level: int,
    model_name: str,
    strategies: Sequence[str],
    deltas: Sequence[float],
    epsilons: Sequence[float],
    num_splits: int = 10,
    test_size: float = 0.5,
    base_seed: int = 123,
) -> pd.DataFrame:
    rows: List[Dict[str, Any]] = []

    for strategy in strategies:
        for delta in deltas:
            for eps in epsilons:
                metrics: Dict[str, List[float]] = {
                    "human_coverage": [],
                    "human_set_size": [],
                    "method_coverage": [],
                    "method_set_size": [],
                    "cond_coverage_not_in_H": [],
                    "cond_error_in_H": [],
                }

                for split_idx in range(num_splits):
                    seed = base_seed + split_idx
                    try:
                        out = run_single_split(
                            paths=paths,
                            noise_level=noise_level,
                            model_name=model_name,
                            human_strategy=strategy,
                            epsilon=eps,
                            delta=delta,
                            test_size=test_size,
                            random_state=seed,
                        )
                    except Exception as e:
                        print(f"[WARN] split {split_idx} failed for strategy={strategy}, δ={delta}, ε={eps}: {e}")
                        out = {k: np.nan for k in metrics.keys()}

                    for k in metrics:
                        metrics[k].append(out[k])

                row: Dict[str, Any] = {
                    "strategy": strategy,
                    "delta": float(delta),
                    "epsilon": float(eps),
                }
                for k, vals in metrics.items():
                    arr = np.array(vals, dtype=float)
                    valid = arr[~np.isnan(arr)]
                    if valid.size == 0:
                        row[f"{k}_mean"] = np.nan
                        row[f"{k}_std"] = np.nan
                    else:
                        row[f"{k}_mean"] = float(valid.mean())
                        row[f"{k}_std"] = float(valid.std())

                rows.append(row)

    return pd.DataFrame(rows)


def sweep_ai_alone(
    paths: Imagenet16HPaths,
    noise_level: int,
    model_name: str,
    deltas: Sequence[float],
    epsilon: float,
    num_splits: int = 10,
    test_size: float = 0.5,
    base_seed: int = 123,
) -> pd.DataFrame:
    """
    AI-alone baseline: H(x) = ∅, i.e. strategy="empty".
    """
    return sweep_strategies_eps_delta(
        paths=paths,
        noise_level=noise_level,
        model_name=model_name,
        strategies=["empty"],
        deltas=deltas,
        epsilons=[epsilon],
        num_splits=num_splits,
        test_size=test_size,
        base_seed=base_seed,
    )


	from __future__ import annotations

	from dataclasses import dataclass
	from pathlib import Path
	from typing import Dict, Optional, Sequence, List, Tuple, Any

	import json
	import numpy as np
	import pandas as pd
	from sklearn.model_selection import train_test_split



	LABELS: List[str] = [
	"knife", "keyboard", "elephant", "bicycle", "airplane", "clock", "oven", "chair",
	"bear", "boat", "cat", "bottle", "truck", "car", "bird", "dog",
	]

	NOISE_LEVELS: List[int] = [80, 95, 110, 125]
	MODEL_NAMES: List[str] = ["alexnet", "vgg19", "densenet161", "googlenet", "resnet152"]


	#path config
	@dataclass
	class Imagenet16HPaths:
	"""
	Centralized paths for the ImageNet16H experiments.
	"""
	data_root: Path
	raw_model_csv: Path
	human_alone_csv: Path
	models_dir: Path
	users_dir: Path
	meta_dir: Path
	true_labels_csv: Path
	all_expert_csv: Path

	@classmethod
	def from_data_root(cls, data_root: Path) -> "Imagenet16HPaths":
	base = data_root / "imagenet16h"
	models_dir = data_root / "models"
	users_dir = data_root / "users"
	meta_dir = base / "metadata"
	models_dir.mkdir(parents=True, exist_ok=True)
	users_dir.mkdir(parents=True, exist_ok=True)
	meta_dir.mkdir(parents=True, exist_ok=True)

	return cls(
	data_root=data_root,
	raw_model_csv=base / "hai_epoch10_model_preds_max_normalized.csv",
	human_alone_csv=base / "human_only_classification_6per_img_export.csv",
	models_dir=models_dir,
	users_dir=users_dir,
	meta_dir=meta_dir,
	true_labels_csv=data_root / "imagenet16H.csv",
	all_expert_csv=base / "human_only_classification_6per_img_export.csv",
	)


	#data prep utilities

	def _ensure_exists(p: Path, hint: str = "") -> None:
	if not p.exists():
	raise FileNotFoundError(f"Missing: {p}\n{hint}")


	def build_labels_from_raw_csv(paths: Imagenet16HPaths) -> None:
	"""
	Create imagenet16H.csv = mapping image_name -> category (one row per image),
	and classes.json describing the 16 classes.
	"""
	_ensure_exists(paths.raw_model_csv, "Put the provided model-output CSV here.")

	raw = pd.read_csv(paths.raw_model_csv, dtype=str)
	base = raw[
	(raw["model_name"] == "vgg19") & (raw["noise_level"] == "80")
	][["image_name", "category"]].drop_duplicates()
	base = base.set_index("image_name").sort_index()

	paths.true_labels_csv.parent.mkdir(parents=True, exist_ok=True)
	base.to_csv(paths.true_labels_csv)

	classes_json = {i: {"name": l} for i, l in enumerate(LABELS)}
	(paths.meta_dir / "classes.json").write_text(
	json.dumps(classes_json, indent=2)
	)

	print(f"[labels] wrote {paths.true_labels_csv} ({len(base)} rows)")
	print(f"[labels] wrote {paths.meta_dir / 'classes.json'}")


	def split_model_predictions(paths: Imagenet16HPaths, model_name: str) -> None:
	"""
	Save per-noise CSV: models/noise{nl}/{model_name}.csv
	columns = probs for 16 labels + 'correct', index=image_name.
	"""
	_ensure_exists(paths.raw_model_csv)

	dtypes = {
	"image_name": str,
	"correct": int,
	"noise_level": str,
	"model_name": str,
	"category": str,
	}
	for l in LABELS:
	dtypes[l] = float

	raw = pd.read_csv(paths.raw_model_csv, dtype=dtypes)
	raw = raw[raw["model_name"] == model_name]

	for nl in NOISE_LEVELS:
	sub = raw[raw["noise_level"] == str(nl)]
	cols = ["image_name"] + LABELS + ["correct"]
	df = sub[cols].drop_duplicates().set_index("image_name")
	out_dir = paths.models_dir / f"noise{nl}"
	out_dir.mkdir(parents=True, exist_ok=True)
	out = out_dir / f"{model_name}.csv"
	df.to_csv(out)
	print(f"[model] {model_name} ω={nl}: {len(df)} rows -> {out}")


	def sort_model_predictions(paths: Imagenet16HPaths, model_name: str) -> None:
	"""
	Write models/noise{nl}/{model_name}_sorted.csv = rank-ordered labels per image
	(columns '1'..'16').
	"""
	_ensure_exists(paths.raw_model_csv)

	dtypes = {"image_name": str, "model_name": str, "noise_level": str}
	for l in LABELS:
	dtypes[l] = float

	raw = pd.read_csv(paths.raw_model_csv, dtype=dtypes)
	raw = raw[raw["model_name"] == model_name]

	labels_arr = np.array(LABELS)
	out_cols = [str(i) for i in range(1, 17)]

	for nl in NOISE_LEVELS:
	sub = raw[raw["noise_level"] == str(nl)]
	probs = sub[["image_name"] + LABELS].drop_duplicates().set_index("image_name")

	row_sums = probs.sum(axis=1).replace(0, np.nan)
	probs = probs.div(row_sums, axis=0).fillna(1.0 / len(LABELS))

	order = np.argsort(probs.values, axis=1)[:, ::-1]
	sorted_labels = labels_arr[order]

	sorted_df = pd.DataFrame(sorted_labels, index=probs.index, columns=out_cols)
	out_dir = paths.models_dir / f"noise{nl}"
	out_dir.mkdir(parents=True, exist_ok=True)
	out_path = out_dir / f"{model_name}_sorted.csv"
	sorted_df.to_csv(out_path)
	print(f"[sorted] {model_name} ω={nl}: {len(sorted_df)} rows -> {out_path}")


	def prep_human_tables(paths: Imagenet16HPaths) -> None:
	"""
	From the human-alone CSV, write:
	users/noise{nl}/success.csv
	users/noise{nl}/predictions.csv
	"""
	_ensure_exists(paths.human_alone_csv)

	raw = pd.read_csv(paths.human_alone_csv, dtype=str)
	for nl in NOISE_LEVELS:
	sub = raw[raw["noise_level"] == str(nl)]
	success = sub[["image_name", "participant_id", "correct"]].set_index("image_name")
	preds = sub[["image_name", "participant_id", "participant_classification"]].set_index("image_name")
	out_dir = paths.users_dir / f"noise{nl}"
	out_dir.mkdir(parents=True, exist_ok=True)
	success.to_csv(out_dir / "success.csv")
	preds.to_csv(out_dir / "predictions.csv")
	print(f"[human] ω={nl}: success({len(success)}), predictions({len(preds)}) -> {out_dir}")


	# expert prob

	def _canon(s: str) -> str:
	return str(s).strip().lower()


	def _label_mapper() -> Dict[str, str]:
	m = {l: l for l in LABELS}
	m.update({
	"aeroplane": "airplane", "plane": "airplane",
	"bike": "bicycle",
	"kitty": "cat", "cat.": "cat",
	"puppy": "dog",
	"sofa": "chair", "couch": "chair",
	"jug": "bottle", "bottle.": "bottle",
	"truck.": "truck", "car.": "car",
	"bird.": "bird", "dog.": "dog",
	})
	return m


	def _map_to_vocab(raw: str, mapper: Dict[str, str]) -> Optional[str]:
	r = _canon(raw)
	if r in mapper:
	return mapper[r]
	if r.endswith("s") and r[:-1] in mapper:
	return mapper[r[:-1]]
	return None


	def build_expert_freq_probs(
	paths: Imagenet16HPaths,
	noise_level: int,
	labels: Sequence[str] = LABELS,
	) -> pd.DataFrame:
	"""
	Build p(y\|x) from expert classifications; write:
	users/noise{nl}/all_expert_pyx_strict.csv
	"""
	_ensure_exists(paths.all_expert_csv)

	df = pd.read_csv(paths.all_expert_csv, dtype=str)
	req = {"image_name", "participant_classification", "noise_level"}
	if not req.issubset(df.columns):
	raise ValueError(f"Expert CSV must contain columns {req}, got {set(df.columns)}")

	df_noise = df[df["noise_level"] == str(noise_level)].copy()
	mapper = _label_mapper()
	df_noise["mapped"] = df_noise["participant_classification"].map(
	lambda s: _map_to_vocab(s, mapper)
	)
	df_noise = df_noise[df_noise["mapped"].isin(labels)].copy()

	counts = df_noise.groupby(["image_name", "mapped"]).size().unstack(fill_value=0)
	counts = counts.reindex(columns=list(labels), fill_value=0)

	row_sums = counts.sum(axis=1)
	images_with_votes = row_sums[row_sums > 0].index
	counts = counts.loc[images_with_votes]
	row_sums = row_sums.loc[images_with_votes]

	pyx = counts.div(row_sums, axis=0)
	pyx.index.name = "image_name"

	out_dir = paths.users_dir / f"noise{noise_level}"
	out_dir.mkdir(parents=True, exist_ok=True)
	out_path = out_dir / "all_expert_pyx_strict.csv"
	pyx.to_csv(out_path)
	print(f"[all expert p(y\|x)] strict ω={noise_level}: {len(pyx)} images -> {out_path}")
	return pyx


	# human expert and ai model class

	class HumanExpert:
	"""
	Represents a human expert providing p(y\|x) and prediction sets.
	Expects users/noise{nl}/all_expert_pyx_strict.csv
	"""
	def __init__(self, paths: Imagenet16HPaths, noise_level: int, labels: Sequence[str] = LABELS):
	self.paths = paths
	self.noise_level = noise_level
	self.labels = list(labels)

	path = paths.users_dir / f"noise{noise_level}" / "all_expert_pyx_strict.csv"
	if not path.exists():
	print(f"[HumanExpert] {path} not found. Building from ALL_EXPERT_CSV.")
	build_expert_freq_probs(paths, noise_level)
	if not path.exists():
	raise FileNotFoundError(f"Failed to build {path}")

	self.p = pd.read_csv(path, index_col="image_name").reindex(columns=self.labels, fill_value=0.0)
	row_sums = self.p.sum(axis=1).replace(0, np.nan)
	self.p = self.p.div(row_sums, axis=0).fillna(1.0 / len(self.labels))

	def available_images(self) -> List[str]:
	return self.p.index.tolist()

	def get_prob(self, image_name: str) -> pd.Series:
	if image_name not in self.p.index:
	raise ValueError(f"Image {image_name} not found in human expert data for noise level {self.noise_level}.")
	return self.p.loc[image_name].astype(float)

	def predict_set(
	self,
	image_name: str,
	strategy: str = "top1",
	k: Optional[int] = None,
	mass: Optional[float] = None,
	threshold: Optional[float] = None,
	) -> List[str]:
	"""
	Build a prediction set from the human probability vector.

	Supported strategies:
	- "top1"
	- "empty"
	- "topk" -> uses k argument or defaults to 2
	- "topk5" -> parses k=5 from the strategy name (any k in [1, num_labels])
	- "mass"
	- "threshold"
	"""
	strat = strategy.lower()
	p = self.get_prob(image_name).sort_values(ascending=False)

	# top-1
	if strat == "top1":
	return [p.index[0]]

	# empty set (AI-alone baseline)
	if strat == "empty":
	return []

	# top-k: accept "topk", "topk5", etc.
	if strat.startswith("topk"):
	parsed_k: Optional[int] = None
	suffix = strat[4:] # part after "topk"
	if suffix:
	if suffix.isdigit():
	parsed_k = int(suffix)
	else:
	raise ValueError(f"Unrecognized topk strategy '{strategy}'")


	if k is not None:
	parsed_k = int(k)


	if parsed_k is None:
	parsed_k = 2


	parsed_k = max(1, min(parsed_k, len(p)))
	return list(p.index[:parsed_k])

	# cumulative mass strategy
	if strat == "mass":
	if not (isinstance(mass, (int, float)) and 0 < mass <= 1):
	raise ValueError("mass in (0,1] required for 'mass' strategy")
	cs = p.cumsum().values
	cut_idx = int(np.searchsorted(cs, float(mass), side="left"))
	cut_idx = min(cut_idx, len(p) - 1)
	return list(p.index[: cut_idx + 1])

	# probability threshold strategy
	if strat == "threshold":
	if not (isinstance(threshold, (int, float)) and 0 < threshold <= 1):
	raise ValueError("threshold in (0,1] required for 'threshold' strategy")
	chosen = [lbl for lbl, val in p.items() if val >= float(threshold)]
	return chosen or [p.index[0]]

	raise ValueError(f"Unknown strategy: {strategy}")



	class AIModel:
	"""
	Represents an AI model providing p(y\|x).
	Expects models/noise{nl}/{model_name}.csv.
	"""
	def __init__(self, paths: Imagenet16HPaths, noise_level: int, model_name: str, labels: Sequence[str] = LABELS):
	self.paths = paths
	self.noise_level = noise_level
	self.model_name = model_name
	self.labels = list(labels)

	path = paths.models_dir / f"noise{noise_level}" / f"{model_name}.csv"
	if not path.exists():
	raise FileNotFoundError(
	f"Missing model probabilities file: {path}. "
	f"Ensure split_model_predictions(paths, '{model_name}') was run for ω={noise_level}."
	)

	try:
	self.p = pd.read_csv(path, index_col="image_name")
	except ValueError:
	self.p = pd.read_csv(path, index_col=0)

	self.p = (
	self.p.reindex(columns=self.labels)
	.apply(pd.to_numeric, errors="coerce")
	.fillna(0.0)
	)
	row_sums = self.p.sum(axis=1)
	row_sums[row_sums == 0] = 1.0
	self.p = self.p.div(row_sums, axis=0)

	def get_prob(self, image_name: str) -> pd.Series:
	if image_name not in self.p.index:
	raise ValueError(
	f"Image {image_name} not found in {self.model_name} data for noise level {self.noise_level}."
	)
	return self.p.loc[image_name].astype(float)


	# CP utilities


	def get_common_image_names(
	ai_model: AIModel,
	human_expert: HumanExpert,
	true_labels_df: pd.DataFrame,
	) -> List[str]:
	hs = set(human_expert.available_images())
	ms = set(ai_model.p.index)
	ts = set(true_labels_df.index)
	return sorted(hs & ms & ts)


	def calculate_thresholds(
	cal_images: List[str],
	ai_model: AIModel,
	human_expert: HumanExpert,
	true_labels_df: pd.DataFrame,
	epsilon: float,
	delta: float,
	human_strategy: str,
	) -> Tuple[float, float]:
	scores_in_h = []
	scores_not_in_h = []

	for image_name in cal_images:
	if image_name not in true_labels_df.index:
	continue
	try:
	ai_probs = ai_model.get_prob(image_name)
	human_set = human_expert.predict_set(image_name, strategy=human_strategy)
	except ValueError:
	continue

	true_label = true_labels_df.loc[image_name, "category"]
	score = 1.0 - ai_probs.get(true_label, 0.0)

	if true_label in human_set:
	scores_in_h.append(score)
	else:
	scores_not_in_h.append(score)

	if scores_in_h:
	b = np.quantile(scores_in_h, 1.0 - epsilon, method="higher")
	else:
	b = np.inf

	if scores_not_in_h:
	a = np.quantile(scores_not_in_h, delta, method="higher")
	else:
	a = np.inf

	return a, b


	def generate_prediction_sets(
	test_images: List[str],
	ai_model: AIModel,
	human_expert: HumanExpert,
	a_threshold: float,
	b_threshold: float,
	human_strategy: str,
	) -> Dict[str, Dict[str, Any]]:
	predictions: Dict[str, Dict[str, Any]] = {}

	for image_name in test_images:
	try:
	ai_probs = ai_model.get_prob(image_name)
	human_set = human_expert.predict_set(image_name, strategy=human_strategy)
	except ValueError:
	predictions[image_name] = {"prediction_set": [], "set_size": 0}
	continue

	pred_set: List[str] = []
	for label in ai_model.labels:
	score = 1.0 - ai_probs.get(label, 0.0)
	threshold = b_threshold if label in human_set else a_threshold
	if score <= threshold:
	pred_set.append(label)

	predictions[image_name] = {"prediction_set": pred_set, "set_size": len(pred_set)}

	return predictions


	def calculate_coverage(
	predictions: Dict[str, Dict[str, Any]],
	true_labels_df: pd.DataFrame,
	) -> float:
	if not predictions:
	return 0.0

	correct = 0
	total = 0

	for image_name, data in predictions.items():
	if image_name not in true_labels_df.index:
	continue
	total += 1
	true_label = true_labels_df.loc[image_name, "category"]
	if true_label in data.get("prediction_set", []):
	correct += 1

	if total == 0:
	return 0.0
	return correct / total


	#sing run of experiment

	def run_single_split(
	paths: Imagenet16HPaths,
	noise_level: int,
	model_name: str,
	human_strategy: str,
	epsilon: float,
	delta: float,
	test_size: float = 0.5,
	random_state: int = 123,
	) -> Dict[str, float]:
	true_labels_df = pd.read_csv(paths.true_labels_csv, index_col="image_name")

	ai_model = AIModel(paths, noise_level=noise_level, model_name=model_name)
	human_expert = HumanExpert(paths, noise_level=noise_level)

	all_images = get_common_image_names(ai_model, human_expert, true_labels_df)
	if not all_images:
	raise RuntimeError(f"No common images for ω={noise_level}, model={model_name}.")

	cal_images, test_images = train_test_split(
	all_images,
	test_size=test_size,
	random_state=random_state,
	)

	# Human-alone on test
	human_preds_test: Dict[str, Dict[str, Any]] = {}
	for img in test_images:
	hset = human_expert.predict_set(img, strategy=human_strategy)
	human_preds_test[img] = {"prediction_set": hset, "set_size": len(hset)}

	human_cov = calculate_coverage(human_preds_test, true_labels_df)
	human_sizes = [d["set_size"] for d in human_preds_test.values()]
	human_size = float(np.mean(human_sizes)) if human_sizes else 0.0

	# Collaborative method
	a, b = calculate_thresholds(
	cal_images,
	ai_model,
	human_expert,
	true_labels_df,
	epsilon=epsilon,
	delta=delta,
	human_strategy=human_strategy,
	)

	preds_test = generate_prediction_sets(
	test_images,
	ai_model,
	human_expert,
	a_threshold=a,
	b_threshold=b,
	human_strategy=human_strategy,
	)

	method_cov = calculate_coverage(preds_test, true_labels_df)
	method_sizes = [d["set_size"] for d in preds_test.values()]
	method_size = float(np.mean(method_sizes)) if method_sizes else 0.0


	true_labels_dict = true_labels_df.loc[test_images, "category"].to_dict()

	count_Y_not_in_H = 0
	count_Y_in_C_and_Y_not_in_H = 0
	count_Y_in_H = 0
	count_Y_not_in_C_and_Y_in_H = 0

	for img in test_images:
	if img not in true_labels_dict:
	continue
	y = true_labels_dict[img]
	hset = human_preds_test.get(img, {}).get("prediction_set", [])
	cset = preds_test.get(img, {}).get("prediction_set", [])

	if y not in hset:
	count_Y_not_in_H += 1
	if y in cset:
	count_Y_in_C_and_Y_not_in_H += 1
	else:
	count_Y_in_H += 1
	if y not in cset:
	count_Y_not_in_C_and_Y_in_H += 1

	cond_cov_not_in_H = (
	count_Y_in_C_and_Y_not_in_H / count_Y_not_in_H
	if count_Y_not_in_H > 0 else np.nan
	)
	cond_err_in_H = (
	count_Y_not_in_C_and_Y_in_H / count_Y_in_H
	if count_Y_in_H > 0 else np.nan
	)

	return {
	"human_coverage": float(human_cov),
	"human_set_size": float(human_size),
	"method_coverage": float(method_cov),
	"method_set_size": float(method_size),
	"cond_coverage_not_in_H": float(cond_cov_not_in_H),
	"cond_error_in_H": float(cond_err_in_H),
	}



	# Sweeps over strategies / (ε, δ)


	def sweep_strategies_eps_delta(
	paths: Imagenet16HPaths,
	noise_level: int,
	model_name: str,
	strategies: Sequence[str],
	deltas: Sequence[float],
	epsilons: Sequence[float],
	num_splits: int = 10,
	test_size: float = 0.5,
	base_seed: int = 123,
	) -> pd.DataFrame:
	rows: List[Dict[str, Any]] = []

	for strategy in strategies:
	for delta in deltas:
	for eps in epsilons:
	metrics: Dict[str, List[float]] = {
	"human_coverage": [],
	"human_set_size": [],
	"method_coverage": [],
	"method_set_size": [],
	"cond_coverage_not_in_H": [],
	"cond_error_in_H": [],
	}

	for split_idx in range(num_splits):
	seed = base_seed + split_idx
	try:
	out = run_single_split(
	paths=paths,
	noise_level=noise_level,
	model_name=model_name,
	human_strategy=strategy,
	epsilon=eps,
	delta=delta,
	test_size=test_size,
	random_state=seed,
	)
	except Exception as e:
	print(f"[WARN] split {split_idx} failed for strategy={strategy}, δ={delta}, ε={eps}: {e}")
	out = {k: np.nan for k in metrics.keys()}

	for k in metrics:
	metrics[k].append(out[k])

	row: Dict[str, Any] = {
	"strategy": strategy,
	"delta": float(delta),
	"epsilon": float(eps),
	}
	for k, vals in metrics.items():
	arr = np.array(vals, dtype=float)
	valid = arr[~np.isnan(arr)]
	if valid.size == 0:
	row[f"{k}_mean"] = np.nan
	row[f"{k}_std"] = np.nan
	else:
	row[f"{k}_mean"] = float(valid.mean())
	row[f"{k}_std"] = float(valid.std())

	rows.append(row)

	return pd.DataFrame(rows)


	def sweep_ai_alone(
	paths: Imagenet16HPaths,
	noise_level: int,
	model_name: str,
	deltas: Sequence[float],
	epsilon: float,
	num_splits: int = 10,
	test_size: float = 0.5,
	base_seed: int = 123,
	) -> pd.DataFrame:
	"""
	AI-alone baseline: H(x) = ∅, i.e. strategy="empty".
	"""
	return sweep_strategies_eps_delta(
	paths=paths,
	noise_level=noise_level,
	model_name=model_name,
	strategies=["empty"],
	deltas=deltas,
	epsilons=[epsilon],
	num_splits=num_splits,
	test_size=test_size,
	base_seed=base_seed,
	)