Feature-loader API

Patch bags

bag_features

Single-scale bag-features dataset.

Used by SLIP, TOP, and any method that takes one tensor of patch features per slide.

CSV format

slide_id, label

BagFeaturesDataset

BagFeaturesDataset(
    csv_path: str | DataFrame,
    feature_root: str,
    label_dict: dict,
    max_patches: int | None = None,
    ext: str = ".pt",
    feature_path_column: str | None = None,
    feature_key: str = "features",
    feature_dim: int | None = None,
    include_metadata: bool = False,
    random_subsampling: bool = True,
)

Bases: Dataset

Load one variable-length patch-feature bag per slide.

Parameters:

Name	Type	Description	Default
csv_path	str | DataFrame	Manifest containing at least slide_id and label.	required
feature_root	str	Root used when no explicit feature-path column is set.	required
label_dict	dict	Mapping from string labels to integer class indices.	required
max_patches	int | None	Optional cap applied after loading a bag.	None
ext	str	Per-slide file suffix used with feature_root.	'.pt'
feature_path_column	str | None	Optional manifest column containing exact paths.	None
feature_key	str	Tensor key used for mapping or HDF5 payloads.	'features'
feature_dim	int | None	Expected patch width. A mismatch raises immediately.	None
include_metadata	bool	Include slide/case identifiers in returned samples.	False

Each sample returns (features, label) or (features, metadata, label). Features have shape [patches, dim]; batching is normally restricted to one slide because bag lengths vary.

Source code in common/datasets/bag_features.py

def __init__(self, csv_path: str | pd.DataFrame, feature_root: str,
             label_dict: dict, max_patches: int | None = None,
             ext: str = ".pt", feature_path_column: str | None = None,
             feature_key: str = "features",
             feature_dim: int | None = None,
             include_metadata: bool = False,
             random_subsampling: bool = True):
    self.df = (csv_path.copy() if isinstance(csv_path, pd.DataFrame)
               else pd.read_csv(csv_path))
    self.feature_root = feature_root
    self.label_dict = label_dict
    self.max_patches = max_patches
    self.ext = ext
    self.feature_path_column = feature_path_column
    self.feature_key = feature_key
    self.feature_dim = int(feature_dim) if feature_dim is not None else None
    self.include_metadata = include_metadata
    self.random_subsampling = bool(random_subsampling)

build_bag_loader

build_bag_loader(
    cfg: Mapping[str, Any],
    split: str = "train",
    shuffle: bool = True,
    fold: int | None = None,
) -> DataLoader

Construct a single-scale patch-bag loader from a run config.

Parameters:

Name	Type	Description	Default
cfg	Mapping[str, Any]	Run configuration containing split and feature-location fields.	required
split	str	Split filename stem, such as train, val, or test.	'train'
shuffle	bool	Request shuffling; only the training split is shuffled.	True
fold	int | None	Fold index used for nested or wide split resolution. When omitted, the private _fold_index dispatch field defaults to zero.	None

Returns:

Type	Description
DataLoader	A PyTorch data loader yielding the dataset's bag tuples.

Source code in common/datasets/bag_features.py

def build_bag_loader(
    cfg: Mapping[str, Any], split: str = "train", shuffle: bool = True,
    fold: int | None = None,
) -> DataLoader:
    """Construct a single-scale patch-bag loader from a run config.

    Args:
        cfg: Run configuration containing split and feature-location fields.
        split: Split filename stem, such as ``train``, ``val``, or ``test``.
        shuffle: Request shuffling; only the training split is shuffled.
        fold: Fold index used for nested or wide split resolution. When omitted,
            the private ``_fold_index`` dispatch field defaults to zero.

    Returns:
        A PyTorch data loader yielding the dataset's bag tuples.
    """
    from common.datasets.split_tables import load_phase_table

    fold = cfg.get("_fold_index", 0) if fold is None else fold
    phase_table = load_phase_table(cfg, split, fold)
    feature_path_column = cfg.get("feature_path_column")
    feature_root = cfg.get("data_folder_s")
    if not feature_path_column and not feature_root:
        raise KeyError(
            "A patch-bag loader requires 'feature_path_column' or "
            "'data_folder_s'.")
    ds = BagFeaturesDataset(
        csv_path=phase_table,
        feature_root=str(feature_root or ""),
        label_dict=cfg["label_dict"],
        max_patches=cfg.get("max_patches"),
        ext=cfg.get("feature_ext", ".pt"),
        feature_path_column=feature_path_column,
        feature_key=cfg.get("feature_key", "features"),
        feature_dim=cfg.get("feature_dim"),
        include_metadata=cfg.get("include_metadata", False),
        random_subsampling=split == "train")
    return DataLoader(ds, batch_size=cfg.get("batch_size", 1),
                      shuffle=shuffle and split == "train",
                      num_workers=cfg.get("num_workers", 4))

Slide embeddings

slide_embeddings

Shared one-vector-per-slide feature loading.

This module is method-agnostic. Any adapter whose backbone contract declares FeatureLevel.SLIDE_EMBEDDING receives the same exact-key, exact-width, provenance-aware HDF5/torch/pickle input behavior from the unified trainer. Patch bags are intentionally handled elsewhere.

SlideEmbeddingSource dataclass

SlideEmbeddingSource(
    source_type: str,
    features_path: Path,
    feature_path_column: str | None,
    feature_key: str,
    feature_dim: int,
    slide_id_key: str,
)

Normalize runtime fields shared by slide-vector consumers.

The source separates offline slide encoder provenance from any runtime prompt encoder selected by methods such as SLDPC.

from_config classmethod

from_config(
    cfg: Mapping[str, Any],
) -> "SlideEmbeddingSource"

Validate and construct a source from a generated run config.

Source code in common/datasets/slide_embeddings.py

@classmethod
def from_config(cls, cfg: Mapping[str, Any]) -> "SlideEmbeddingSource":
    """Validate and construct a source from a generated run config."""
    feature_dim = int(cfg.get("feature_dim", 0))
    if feature_dim <= 0:
        raise ValueError("Slide-embedding configs require positive feature_dim")
    feature_key = str(cfg.get("feature_key", "")).strip()
    if not feature_key:
        raise ValueError("Slide-embedding configs require feature_key")
    source_type = str(cfg.get("source_type", "")).strip()
    if not source_type:
        source_type = infer_slide_embedding_source_type(
            cfg.get("slide_features"), cfg.get("storage"))
    if source_type == "per_slide_pth":
        source_type = "per_slide_torch"
    if source_type not in {"pkl", "per_slide_h5", "per_slide_torch"}:
        raise ValueError(
            "Slide-embedding source_type must be pkl, per_slide_h5, or "
            f"per_slide_torch, got {source_type!r}")
    root = cfg.get("slide_features")
    if not root:
        raise ValueError("Slide-embedding configs require slide_features")
    return cls(
        source_type=source_type,
        features_path=Path(str(root)).expanduser(),
        feature_path_column=cfg.get("feature_path_column"),
        feature_key=feature_key,
        feature_dim=feature_dim,
        slide_id_key=str(cfg.get("slide_id_key", "filenames")),
    )

SlideEmbeddingDataset

SlideEmbeddingDataset(
    source_type: str,
    features_path: str | Path,
    csv_path: str | Path | DataFrame,
    label_dict: Mapping[str, int],
    feature_path_column: str | None = None,
    feature_key: str = "features",
    feature_dim: int | None = None,
    slide_id_key: str = "filenames",
)

Bases: Dataset

Load one exact-width vector and label for every split slide.

Parameters:

Name	Type	Description	Default
source_type	str	pkl, per_slide_h5, or per_slide_torch.	required
features_path	str | Path	Shared pickle file or root containing per-slide files.	required
csv_path	str | Path | DataFrame	Split CSV with slide IDs and labels.	required
label_dict	Mapping[str, int]	Mapping from string labels to integer class indices.	required
feature_path_column	str | None	Optional CSV column with exact per-slide paths.	None
feature_key	str	Exact tensor key inside HDF5 or mapping payloads.	'features'
feature_dim	int | None	Expected flattened vector width.	None
slide_id_key	str	Identifier key used by a shared pickle payload.	'filenames'

Samples are dictionaries with feat, label, slide_id, and case_id. Every declared split row must match exactly one feature; missing or ambiguous IDs are fatal rather than silently changing a split.

Source code in common/datasets/slide_embeddings.py

def __init__(
    self,
    source_type: str,
    features_path: str | Path,
    csv_path: str | Path | pd.DataFrame,
    label_dict: Mapping[str, int],
    feature_path_column: str | None = None,
    feature_key: str = "features",
    feature_dim: int | None = None,
    slide_id_key: str = "filenames",
):
    self.source_type = (
        "per_slide_torch" if source_type == "per_slide_pth" else source_type)
    self.features_path = Path(expand_path(features_path))
    self.label_dict = dict(label_dict)
    self.feature_path_column = feature_path_column
    self.feature_key = feature_key
    self.slide_id_key = slide_id_key
    self.feature_dim = int(feature_dim) if feature_dim is not None else None
    self.entries = self._read_entries(csv_path)
    if not self.entries:
        raise ValueError(f"slide-embedding split has no rows: {csv_path}")

    if self.source_type == "pkl":
        with self.features_path.open("rb") as handle:
            payload = pickle.load(handle)
        if not isinstance(payload, Mapping):
            raise TypeError("pickle slide-embedding payload must be a mapping")
        missing_keys = [
            key for key in (self.feature_key, self.slide_id_key)
            if key not in payload]
        if missing_keys:
            raise KeyError(
                "pickle slide-embedding payload is missing keys "
                f"{missing_keys}; available keys: {list(payload.keys())}")
        self.embeddings = torch.as_tensor(
            np.asarray(payload[self.feature_key])).float()
        if self.embeddings.ndim != 2 or self.embeddings.shape[0] == 0:
            raise ValueError(
                "pickle slide embeddings must have shape [slides, dimension], "
                f"got {tuple(self.embeddings.shape)}")
        ids = payload[self.slide_id_key]
        if isinstance(ids, (str, bytes)):
            raise TypeError(
                f"pickle {self.slide_id_key!r} must be a sequence of IDs")
        try:
            ids = list(ids)
        except TypeError as error:
            raise TypeError(
                f"pickle {self.slide_id_key!r} must be a sequence of IDs") \
                from error
        if len(ids) != len(self.embeddings):
            raise ValueError(
                f"pickle has {len(ids)} slide IDs for "
                f"{len(self.embeddings)} embeddings")
        indexed = [
            (normalise_slide_id(slide_id), index)
            for index, slide_id in enumerate(ids)]
    elif self.source_type == "per_slide_torch":
        self.embeddings = None
        paths = (() if all(entry[2] for entry in self.entries) else (
            path for pattern in ("*.pt", "*.pth")
            for path in self.features_path.rglob(pattern)))
        indexed = [
            (normalise_slide_id(path.name), path) for path in paths]
    elif self.source_type == "per_slide_h5":
        self.embeddings = None
        paths = (() if all(entry[2] for entry in self.entries) else (
            path for pattern in ("*.h5", "*.hdf5")
            for path in self.features_path.rglob(pattern)))
        indexed = [
            (normalise_slide_id(path.name), path) for path in paths]
    else:
        raise ValueError(
            "source_type must be pkl, per_slide_h5, or per_slide_torch")

    self.paths = {}
    duplicate_sources: dict[str, list[str]] = {}
    for slide_id, pointer in indexed:
        if slide_id in self.paths:
            duplicate_sources.setdefault(
                slide_id, [str(self.paths[slide_id])]).append(str(pointer))
        else:
            self.paths[slide_id] = pointer
    if duplicate_sources:
        sample = "; ".join(
            f"{slide_id}: {paths}"
            for slide_id, paths in list(duplicate_sources.items())[:3])
        raise ValueError(
            "slide-embedding store has duplicate normalized slide IDs: "
            f"{sample}")

    missing: list[str] = []
    for slide_id, _label, explicit_path, _case_id in self.entries:
        if explicit_path and self.embeddings is None:
            try:
                available = Path(expand_path(explicit_path)).is_file()
            except ValueError:
                available = False
        else:
            available = normalise_slide_id(slide_id) in self.paths
        if not available:
            missing.append(slide_id)
    if missing:
        sample = ", ".join(missing[:5])
        suffix = " ..." if len(missing) > 5 else ""
        raise FileNotFoundError(
            f"{len(missing)} split slide embeddings are missing: "
            f"{sample}{suffix}")

normalise_slide_id

normalise_slide_id(value: Any) -> str

Decode an identifier and remove only known slide/feature suffixes.

NumPy and Python pickle producers commonly persist identifier arrays as fixed-width byte strings. Calling str on those values produces text such as "b'slide-a.svs'", which can never match a manifest slide ID. Decode byte-like values explicitly and reject blank identifiers at the source boundary instead.

Source code in common/datasets/slide_embeddings.py

def normalise_slide_id(value: Any) -> str:
    """Decode an identifier and remove only known slide/feature suffixes.

    NumPy and Python pickle producers commonly persist identifier arrays as
    fixed-width byte strings. Calling ``str`` on those values produces text
    such as ``"b'slide-a.svs'"``, which can never match a manifest slide ID.
    Decode byte-like values explicitly and reject blank identifiers at the
    source boundary instead.
    """
    if isinstance(value, (bytes, bytearray, memoryview)):
        text = bytes(value).decode("utf-8")
    else:
        text = str(value)
    text = text.strip()
    if not text:
        raise ValueError("slide ID must not be blank")
    path = Path(text)
    return path.stem if path.suffix.lower() in _FILE_SUFFIXES else path.name

infer_slide_embedding_source_type

infer_slide_embedding_source_type(
    path_template: str | Path | None = None,
    storage: str | None = None,
) -> str

Map a registered storage declaration to the shared loader layout.

Parameters:

Name	Type	Description	Default
path_template	str | Path | None	Representative source path or template.	None
storage	str | None	Explicit storage name such as h5, torch, or pkl.	None

Returns:

Type	Description
str	One of per_slide_h5, per_slide_torch, or pkl.

Raises:

Type	Description
ValueError	If neither the declaration nor suffix identifies a layout.

Source code in common/datasets/slide_embeddings.py

def infer_slide_embedding_source_type(
    path_template: str | Path | None = None,
    storage: str | None = None,
) -> str:
    """Map a registered storage declaration to the shared loader layout.

    Args:
        path_template: Representative source path or template.
        storage: Explicit storage name such as ``h5``, ``torch``, or ``pkl``.

    Returns:
        One of ``per_slide_h5``, ``per_slide_torch``, or ``pkl``.

    Raises:
        ValueError: If neither the declaration nor suffix identifies a layout.
    """
    storage_name = str(storage or "").strip().lower()
    if storage_name in {"h5", "hdf5"}:
        return "per_slide_h5"
    if storage_name in {"pt", "pth", "torch"}:
        return "per_slide_torch"
    if storage_name in {"pkl", "pickle"}:
        return "pkl"
    suffix = Path(str(path_template or "")).suffix.lower()
    if suffix in {".h5", ".hdf5"}:
        return "per_slide_h5"
    if suffix in {".pt", ".pth"}:
        return "per_slide_torch"
    if suffix in {".pkl", ".pickle"}:
        return "pkl"
    raise ValueError(
        "Slide-embedding sources require storage h5/torch/pkl or a matching "
        f"path suffix, got path={path_template!r}, storage={storage!r}")

split_csv

split_csv(
    cfg: Mapping[str, Any], split: str, fold: int
) -> Path

Resolve a fold-specific or shared split CSV.

Raises:

Type	Description
FileNotFoundError	If neither supported split layout exists.

Source code in common/datasets/slide_embeddings.py

def split_csv(cfg: Mapping[str, Any], split: str, fold: int) -> Path:
    """Resolve a fold-specific or shared split CSV.

    Raises:
        FileNotFoundError: If neither supported split layout exists.
    """
    root = Path(str(cfg["split_dir"]))
    candidates = (
        root / f"fold{fold}" / f"{split}.csv", root / f"{split}.csv")
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    rendered = "\n  ".join(str(path) for path in candidates)
    raise FileNotFoundError(
        f"No slide-embedding {split} split found. Tried:\n  {rendered}")

build_slide_embedding_loader

build_slide_embedding_loader(
    cfg: Mapping[str, Any],
    split: str,
    fold: int,
    shuffle: bool = True,
) -> DataLoader

Build the shared slide-embedding loader for a run and fold.

Parameters:

Name	Type	Description	Default
cfg	Mapping[str, Any]	Generated run configuration.	required
split	str	Split name such as train, val, or test.	required
fold	int	Outer fold index used to resolve the split path.	required
shuffle	bool	Request shuffling; only training data is shuffled.	True

Returns:

Type	Description
DataLoader	A data loader yielding slide-vector dictionaries.

Source code in common/datasets/slide_embeddings.py

def build_slide_embedding_loader(
    cfg: Mapping[str, Any],
    split: str,
    fold: int,
    shuffle: bool = True,
) -> DataLoader:
    """Build the shared slide-embedding loader for a run and fold.

    Args:
        cfg: Generated run configuration.
        split: Split name such as ``train``, ``val``, or ``test``.
        fold: Outer fold index used to resolve the split path.
        shuffle: Request shuffling; only training data is shuffled.

    Returns:
        A data loader yielding slide-vector dictionaries.
    """
    from common.datasets.split_tables import load_phase_table

    source = SlideEmbeddingSource.from_config(cfg)
    dataset = SlideEmbeddingDataset(
        source.source_type,
        source.features_path,
        load_phase_table(cfg, split, fold),
        cfg["label_dict"],
        feature_path_column=source.feature_path_column,
        feature_key=source.feature_key,
        feature_dim=source.feature_dim,
        slide_id_key=source.slide_id_key,
    )
    return DataLoader(
        dataset,
        batch_size=int(cfg.get("batch_size", 4)),
        shuffle=shuffle and split == "train",
        num_workers=int(cfg.get("num_workers", 0)),
    )