pathforge.core

Stable domain abstractions and task orchestration shared by the other layers.

Common contracts

class pathforge.core.base.RegistryBase

Bases: ABC

Plugin registry base class for managing different types of plugins.

abstractmethod register(key: str, obj: Any) → None

Register a plugin with a given key.

Parameters:

• key (str)

• obj (Any)

Return type:

None

abstractmethod get(key: str) → Any

Retrieve a plugin by its key.

Parameters:

key (str)

Return type:

Any

abstractmethod list_plugins() → Sequence[str]

List all registered plugin keys.

Return type:

Sequence[str]

abstractmethod is_available(key: str) → bool

Check if a plugin is available.

Parameters:

key (str)

Return type:

bool

class pathforge.core.base.CoreRegistries

Bases: object

Container for core plugin registries.

datasets: RegistryBase

models: RegistryBase

losses: RegistryBase

tasks: RegistryBase

explainers: RegistryBase

feature_extractors: RegistryBase

normalizers: RegistryBase

augmentation_methods: RegistryBase

__init__(datasets: RegistryBase, models: RegistryBase, losses: RegistryBase, tasks: RegistryBase, explainers: RegistryBase, feature_extractors: RegistryBase, normalizers: RegistryBase, augmentation_methods: RegistryBase) → None

Parameters:

• datasets (RegistryBase)

• models (RegistryBase)

• losses (RegistryBase)

• tasks (RegistryBase)

• explainers (RegistryBase)

• feature_extractors (RegistryBase)

• normalizers (RegistryBase)

• augmentation_methods (RegistryBase)

Return type:

None

Models

class pathforge.core.models.base.ModelBase

Bases: ABC

Root model abstraction for PathForge.

This class is framework-agnostic. Implementations could be: - PyTorch models (nn.Module) - Scikit-learn estimators - XGBoost/LightGBM boosters

abstractmethod initialize(config: Dict[str, Any] | None = None) → None

Initialize the model. For PyTorch: Load weights, reset parameters. For Sklearn: Configure hyperparameters.

Parameters:

config (Dict[str, Any] | None)

Return type:

None

abstractmethod save(path: str) → None

Persist the model to disk.

Parameters:

path (str)

Return type:

None

abstractmethod load(path: str) → None

Load the model from disk.

Parameters:

path (str)

Return type:

None

get_learnable_parameters() → Iterable[Any]

Return parameters for optimization. Returns empty iterator for non-gradient models (e.g. Random Forest).

Return type:

Iterable[Any]

class pathforge.core.models.base.TorchModelBase

Bases: ModelBase, Module

Canonical PyTorch implementation of the PathForge model interface.

This centralizes the shared initialize/save/load/parameter access behavior so MIL and slide-level model bases do not re-implement the same framework plumbing.

__init__() → None

Return type:

None

initialize(config: Dict[str, Any] | None = None) → None

Initialize the model from an optional config dictionary.

The default PyTorch implementation resets any submodule exposing a reset_parameters method. This keeps initialize() meaningful instead of leaving a silent no-op in the concrete torch-backed base.

Parameters:

config (Dict[str, Any] | None)

Return type:

None

save(path: str) → None

Persist the model state_dict to disk.

Parameters:

path (str)

Return type:

None

load(path: str) → None

Load the model state_dict from disk onto CPU memory.

Parameters:

path (str)

Return type:

None

get_learnable_parameters() → Iterable[Parameter]

Yield all gradient-enabled parameters.

Return type:

Iterable[Parameter]

class pathforge.core.models.base.ScikitBase

Bases: ModelBase

Abstract base for scikit-learn / scikit-survival slide-level estimators.

Concrete subclasses wrap a fitted sklearn estimator and expose a task-specific predict_as_tensor method so the shared save_task_evaluation_artifacts path can evaluate them without touching PyTorch training infrastructure.

abstractmethod fit(X: Any, y: Any) → ScikitBase

Fit the estimator on numpy feature matrix X and targets y.

Parameters:

• X (Any)

• y (Any)

Return type:

ScikitBase

abstractmethod predict_as_tensor(X: Any) → Any

Return predictions as a torch.Tensor compatible with metrics helpers.

Parameters:

X (Any)

Return type:

Any

save(path: str) → None

Persist the estimator to disk via pickle.

Parameters:

path (str)

Return type:

None

load(path: str) → None

Replace this instance’s state from a pickle file.

Parameters:

path (str)

Return type:

None

class pathforge.core.models.mil_base.MILModelBase

Bases: TorchModelBase

Base class for Deep MIL models. Expects input: (Batch, Bags, Dim).

__init__(*args: Any, **kwargs: Any)

Parameters:

• args (Any)

• kwargs (Any)

abstract property bag_size: int | None

Returns fixed bag size (int) or None for variable sizes.

abstractmethod forward_bag(bag: Tensor, mask: Tensor | None = None, coords: Tensor | None = None, label: Tensor | None = None, loss_fn: Module | None = None) → Tensor | Dict[str, Any]

Core MIL logic.

Parameters:

• bag (Tensor) – (B, N, D) features.

• mask (Tensor | None) – (B, N) mask.

• coords (Tensor | None) – (B, N, 2) spatial coordinates.

• label (Tensor | None) – (B,) Ground truth labels (optional, for internal loss calc).

• loss_fn (Module | None) – Loss function module (optional, for internal loss calc).

Returns:

logits (Tensor) OR Dict containing ‘logits’ and ‘loss’.

Return type:

Tensor | Dict[str, Any]

forward(bag: Tensor, *args, **kwargs) → Tensor | Dict[str, Any]

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

bag (Tensor)

Return type:

Tensor | Dict[str, Any]

instance_scores(bag: Tensor, *, mask: Tensor | None = None, coords: Tensor | None = None) → Tensor

Return one instance score per bag element for heatmap inference.

Parameters:

• bag (Tensor) – Bag feature tensor shaped [B, N, D].

• mask (Tensor | None) – Optional boolean padding mask shaped [B, N].

• coords (Tensor | None) – Optional coordinates shaped [B, N, 2].

Returns:

Instance score tensor shaped [B, N].

Return type:

torch.Tensor

Raises:

• AttributeError – If the model does not expose attention-like outputs that can be reduced to one score per instance.

• ValueError – If the returned attention tensor cannot be aligned to the bag’s [B, N] instance axis.

class pathforge.core.models.slide_base.SlideLevelModel

Bases: TorchModelBase

Base class for models that operate on pre-aggregated slide vectors. Expects input: (Batch, Dim).

__init__(**kwargs)

abstractmethod forward_slide(x: Tensor, label: Tensor | None = None, loss_fn: Module | None = None) → Tensor | Dict[str, Any]

Core logic for vector-based models. :param x: (B, Input_Dim) feature vector.

Parameters:

• x (Tensor)

• label (Tensor | None)

• loss_fn (Module | None)

Return type:

Tensor | Dict[str, Any]

forward(x: Tensor, *args, **kwargs) → Tensor | Dict[str, Any]

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

x (Tensor)

Return type:

Tensor | Dict[str, Any]

Native MIL Models

pathforge.core.models.layers.create_mlp(in_dim: int = 768, hid_dims: list[int] | tuple[int, ...] | None = None, out_dim: int = 512, act: Module | None = None, dropout: float = 0.0, end_with_fc: bool = True, end_with_dropout: bool = False, bias: bool = True) → Module

Create one configurable MLP block.

Parameters:

• in_dim (int) – Input feature dimension D_in.

• hid_dims (list[int] | tuple[int, ...] | None) – Hidden layer widths in order. None defaults to [512, 512]. An empty sequence produces a single Linear(in_dim, out_dim) layer.

• out_dim (int) – Output feature dimension D_out.

• act (Module | None) – Activation module inserted after each hidden linear layer and, optionally, after the final layer when end_with_fc is false.

• dropout (float) – Dropout probability applied after hidden activations and, optionally, after the output layer.

• end_with_fc (bool) – When true, end with the final linear layer only. When false, append the activation after the output layer as well.

• end_with_dropout (bool) – Whether to append dropout after the output layer.

• bias (bool) – Whether linear layers use a bias term.

Returns:

nn.Sequential MLP module or nn.Identity when the hidden-dimension contract is invalid.

Return type:

nn.Module

Example

mlp = create_mlp(in_dim=16, hid_dims=[8], out_dim=4, dropout=0.1)
assert isinstance(mlp, nn.Module)

class pathforge.core.models.layers.GlobalAttention

Bases: Module

Attention Network without Gating (2 fc layers)

__init__(L=1024, D=256, dropout=0., num_classes=1)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pathforge.core.models.layers.GlobalGatedAttention

Bases: Module

Attention Network with Sigmoid Gating (3 fc layers)

__init__(L=1024, D=256, dropout=0., num_classes=1)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

pathforge.core.models.layers.Attn_Net_Gated

alias of GlobalGatedAttention

class pathforge.core.models.layers.StandardTransformerBlock

Bases: Module

Standard Self-Attention Block for Transformer MIL.

__init__(dim, heads=8, dropout=0.1)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(x, mask=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pathforge.core.models.layers.NystromAttention

Bases: Module

Native PyTorch implementation of Nystrom Attention (O(N) complexity). Used when nystrom-attention library is not installed.

__init__(dim, head=8, num_landmarks=64, dropout=0.1)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(x, mask=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pathforge.core.models.layers.TransLayer

Bases: Module

Transformer Layer using Nystrom Attention.

__init__(dim, head=8, dropout=0.1)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(x, mask=None)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pathforge.core.models.layers.PPEG

Bases: Module

Pyramid Position Encoding Generator (TransMIL).

__init__(dim=512)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(x)

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pathforge.core.models.perceiver_mil.PerceiverMIL

Bases: MILModelBase

Perceiver-based MIL.

Uses a fixed set of latent query vectors to attend to the variable-sized bag input via Cross-Attention, mapping $O(N)$ complexity to $O(1)$ fixed latents.

Mathematical Formulation: Latents $L in mathbb{R}^{M times D}$. Bag $H in mathbb{R}^{N times D}$. $$ O = text{CrossAttn}(Q=L, K=H, V=H) $$ $$ z = text{Mean}(O) $$

__init__(input_dim=1024, num_latents=32, latent_dim=256, output_dim=2)

property bag_size

Returns fixed bag size (int) or None for variable sizes.

forward_bag(bag: Tensor, mask: Tensor | None = None, coords: Tensor | None = None, label=None, loss_fn=None, return_attention=False) → Tensor | Dict

Core MIL logic.

Parameters:

• bag (Tensor) – (B, N, D) features.

• mask (Tensor | None) – (B, N) mask.

• coords (Tensor | None) – (B, N, 2) spatial coordinates.

• label – (B,) Ground truth labels (optional, for internal loss calc).

• loss_fn – Loss function module (optional, for internal loss calc).

Returns:

logits (Tensor) OR Dict containing ‘logits’ and ‘loss’.

Return type:

Tensor | Dict

class pathforge.core.models.prototype_mil.PrototypeMIL

Bases: MILModelBase

Prototype-based MIL.

Learns a set of global prototypes. The bag representation is the similarity vector indicating the presence of each prototype in the bag.

Mathematical Formulation: Prototypes $P = {p_1, dots, p_k}$. Similarity: $s_{i,j} = exp(-||h_i - p_j||_2)$. Bag Feature: $z_j = max_{i} s_{i,j}$ (presence of prototype $j$).

__init__(input_dim=1024, num_prototypes=8, output_dim=2)

property bag_size

Returns fixed bag size (int) or None for variable sizes.

forward_bag(bag: Tensor, mask: Tensor | None = None, coords: Tensor | None = None, label=None, loss_fn=None, return_attention=False) → Tensor | Dict

Core MIL logic.

Parameters:

• bag (Tensor) – (B, N, D) features.

• mask (Tensor | None) – (B, N) mask.

• coords (Tensor | None) – (B, N, 2) spatial coordinates.

• label – (B,) Ground truth labels (optional, for internal loss calc).

• loss_fn – Loss function module (optional, for internal loss calc).

Returns:

logits (Tensor) OR Dict containing ‘logits’ and ‘loss’.

Return type:

Tensor | Dict

class pathforge.core.models.var_mil.VarMIL

Bases: MILModelBase

Variance-aware Multiple Instance Learning.

Aggregates statistics using both the weighted mean and the weighted variance of the bag.

Mathematical Formulation: Mean: $mu = frac{sum a_i h_i}{sum a_i}$ Variance: $sigma^2 = frac{sum a_i (h_i - mu)^2}{sum a_i}$ Output: $text{Classifier}([mu, sigma^2])$

__init__(input_dim=1024, hidden_dim=256, output_dim=2)

property bag_size

Returns fixed bag size (int) or None for variable sizes.

forward_bag(bag: Tensor, mask: Tensor | None = None, coords: Tensor | None = None, label=None, loss_fn=None, return_attention=False) → Tensor | Dict

Core MIL logic.

Parameters:

• bag (Tensor) – (B, N, D) features.

• mask (Tensor | None) – (B, N) mask.

• coords (Tensor | None) – (B, N, 2) spatial coordinates.

• label – (B,) Ground truth labels (optional, for internal loss calc).

• loss_fn – Loss function module (optional, for internal loss calc).

Returns:

logits (Tensor) OR Dict containing ‘logits’ and ‘loss’.

Return type:

Tensor | Dict

class pathforge.core.models.mil_ens.EnsembleMILModel

Bases: MILModelBase

Average the predictions from multiple MIL members.

This lightweight utility model is intentionally not registry-exposed for benchmark selection. It remains available as an importable composition helper and as interface coverage for the shared MIL base classes.

__init__(members: Sequence[MILModelBase])

Parameters:

members (Sequence[MILModelBase])

property bag_size: int | None

Returns fixed bag size (int) or None for variable sizes.

forward_bag(bag: Tensor, *args: Any, **kwargs: Any) → Tensor

Core MIL logic.

Parameters:

• bag (Tensor) – (B, N, D) features.

• mask – (B, N) mask.

• coords – (B, N, 2) spatial coordinates.

• label – (B,) Ground truth labels (optional, for internal loss calc).

• loss_fn – Loss function module (optional, for internal loss calc).

• args (Any)

• kwargs (Any)

Returns:

logits (Tensor) OR Dict containing ‘logits’ and ‘loss’.

Return type:

Tensor

class pathforge.core.models.mil_graph.GraphMILModel

Bases: MILModelBase

Minimal graph-aware MIL placeholder with masked mean pooling.

This module is intentionally lightweight and import-safe without graph extras. It is not currently registered as a benchmark-selectable model, but it preserves the graph-MIL interface surface for incremental extension.

__init__(embed_dim: int = 256, lr: float = 1e-3)

Parameters:

• embed_dim (int)

• lr (float)

property bag_size: int | None

Returns fixed bag size (int) or None for variable sizes.

forward_bag(bag: torch.Tensor, mask: torch.Tensor | None = None, **_: object) → torch.Tensor

Core MIL logic.

Parameters:

• bag (torch.Tensor) – (B, N, D) features.

• mask (torch.Tensor | None) – (B, N) mask.

• coords – (B, N, 2) spatial coordinates.

• label – (B,) Ground truth labels (optional, for internal loss calc).

• loss_fn – Loss function module (optional, for internal loss calc).

• _ (object)

Returns:

logits (Tensor) OR Dict containing ‘logits’ and ‘loss’.

Return type:

torch.Tensor

class pathforge.core.models.slide_mlp.SlideVectorMLP

Bases: SlideLevelModel, MILModelBase

MLP applied to mean-pooled slide-level feature vectors.

Inherits from both SlideLevelModel and MILModelBase so it can be trained through the standard LightningTrainer pipeline. forward_bag mean-pools the bag (B, N, D) → (B, D) before applying the MLP, making it usable on any bag regardless of bag size.

Parameters:

• input_dim – Feature dimension per instance.

• hidden_dim – Hidden layer width.

• output_dim – Number of output logits (classes / time bins / 1 for regression and continuous survival).

__init__(input_dim: int = 1024, hidden_dim: int = 256, output_dim: int = 2) → None

Parameters:

• input_dim (int)

• hidden_dim (int)

• output_dim (int)

Return type:

None

property bag_size: None

Returns fixed bag size (int) or None for variable sizes.

forward_bag(bag: Tensor, mask: Tensor | None = None, coords: Tensor | None = None, label: Tensor | None = None, loss_fn: Module | None = None, **kwargs: Any) → Tensor | Dict[str, Any]

Mean-pool the bag then forward through the MLP.

Parameters:

• bag (Tensor) – (B, N, D) feature bag.

• mask (Tensor | None) – Ignored (variable-length bags are mean-pooled anyway).

• coords (Tensor | None) – Ignored.

• label (Tensor | None) – Optional target for internal loss computation.

• loss_fn (Module | None) – Optional loss module for internal loss computation.

• kwargs (Any)

Returns:

Logits tensor (B, output_dim) or dict with logits and loss when both label and loss_fn are provided.

Return type:

Tensor | Dict[str, Any]

forward_slide(x: Tensor, label: Tensor | None = None, loss_fn: Module | None = None) → Tensor | Dict[str, Any]

Core logic for vector-based models. :param x: (B, Input_Dim) feature vector.

Parameters:

• x (Tensor)

• label (Tensor | None)

• loss_fn (Module | None)

Return type:

Tensor | Dict[str, Any]

Scikit-learn and scikit-survival based slide-level predictors.

All classes inherit from ScikitBase (which in turn inherits from ModelBase) so they participate in the shared PathForge model hierarchy without requiring PyTorch or the Lightning training stack.

These models are designed exclusively for slide-level vectors — either precomputed aggregated feature vectors or mean/max-pooled tile features. They are trained via SklearnSlideTrainer rather than LightningTrainer.

Feature normalization

All base wrapper classes (classifiers, regressors, survival) fit a sklearn.preprocessing.StandardScaler on the training features by default (normalize=True). The scaler is stored on the instance so that the same transformation is applied during prediction. A structured log message at INFO level is emitted whenever normalization is performed, making it easy to verify that scaling occurred. Pass normalize=False to disable.

Survival models

sksurv models require scikit-survival (optional). All sksurv classes guard their import inside __init__ via _require_sksurv() so the rest of the module remains importable when the package is absent.

Dynamic catalog

SKLEARN_ESTIMATOR_CATALOG maps PathForge model names to (sklearn_module, class_name, task, fixed_kwargs). Use make_sklearn_slide_model() to instantiate any catalog entry and list_sklearn_slide_models() to enumerate those whose package is installed.

ivar SKLEARN_ESTIMATOR_CATALOG:

Full model catalog.

vartype SKLEARN_ESTIMATOR_CATALOG:

dict[str, tuple[str, str, str, dict[str, Any]]]

ivar SKLEARN_SLIDE_MODEL_NAMES:

Frozenset derived from catalog keys.

vartype SKLEARN_SLIDE_MODEL_NAMES:

frozenset[str]

ivar SLIDE_LEVEL_MODEL_NAMES:

Union of sklearn names and SlideVectorMLP.

vartype SLIDE_LEVEL_MODEL_NAMES:

frozenset[str]

pathforge.core.models.sklearn_slide.list_sklearn_slide_models(task: str | None = None) → list[str]

Return catalog names whose backing package is installed.

Parameters:

task (str | None) – Optional filter — "classification", "regression", or "survival". None returns all available models.

Returns:

Sorted list of PathForge model names whose underlying sklearn/sksurv package can be imported.

Return type:

list[str]

pathforge.core.models.sklearn_slide.make_sklearn_slide_model(name: str, normalize: bool = True, **kwargs: Any) → ScikitBase

Instantiate a slide-level sklearn/sksurv model by catalog name.

Parameters:

• name (str) – One of the keys in SKLEARN_ESTIMATOR_CATALOG.

• normalize (bool) – When True (default), a StandardScaler is fitted on training data in fit() and applied during prediction.

• kwargs (Any) – Forwarded to the underlying sklearn estimator constructor, overriding catalog defaults.

Returns:

A fitted-ready ScikitBase instance.

Raises:

• ValueError – If name is not in the catalog.

• ImportError – If the backing package (e.g. sksurv) is not installed.

Return type:

ScikitBase

class pathforge.core.models.sklearn_slide.SklearnSlideClassifier

Bases: ScikitBase

Wrapper around any scikit-learn classifier for slide-level prediction.

Features are standardized with StandardScaler before fitting and prediction when normalize=True (default). A log message is emitted whenever normalization is applied.

Parameters:

• estimator – A fitted or unfitted sklearn classifier implementing fit and predict_proba.

• normalize – Standardize features with StandardScaler (default True).

__init__(estimator: Any, normalize: bool = True) → None

Parameters:

• estimator (Any)

• normalize (bool)

Return type:

None

initialize(config: dict[str, Any] | None = None) → None

Initialize the model. For PyTorch: Load weights, reset parameters. For Sklearn: Configure hyperparameters.

Parameters:

config (dict[str, Any] | None)

Return type:

None

fit(X: ndarray, y: ndarray) → SklearnSlideClassifier

Fit the estimator on numpy feature matrix X and targets y.

Parameters:

• X (ndarray)

• y (ndarray)

Return type:

SklearnSlideClassifier

predict(X: ndarray) → ndarray

Parameters:

X (ndarray)

Return type:

ndarray

predict_proba(X: ndarray) → ndarray

Parameters:

X (ndarray)

Return type:

ndarray

predict_as_tensor(X: ndarray) → Any

Return log-probability tensor (N, C) compatible with metrics helpers.

Parameters:

X (ndarray)

Return type:

Any

get_learnable_parameters() → list

Return parameters for optimization. Returns empty iterator for non-gradient models (e.g. Random Forest).

Return type:

list

class pathforge.core.models.sklearn_slide.SklearnLogisticRegressionClassifier

Bases: SklearnSlideClassifier

Logistic regression slide-level classifier.

Parameters:

• C – Inverse regularisation strength.

• max_iter – Maximum solver iterations.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.linear_model.LogisticRegression.

__init__(C: float = 1.0, max_iter: int = 1000, normalize: bool = True, **kwargs: Any) → None

Parameters:

• C (float)

• max_iter (int)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnRandomForestClassifier

Bases: SklearnSlideClassifier

Random-forest slide-level classifier.

Parameters:

• n_estimators – Number of trees.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.ensemble.RandomForestClassifier.

__init__(n_estimators: int = 100, normalize: bool = True, **kwargs: Any) → None

Parameters:

• n_estimators (int)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnGradientBoostingClassifier

Bases: SklearnSlideClassifier

Gradient boosted trees slide-level classifier.

Parameters:

• n_estimators – Number of boosting rounds.

• learning_rate – Shrinkage factor applied to each tree.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.ensemble.GradientBoostingClassifier.

__init__(n_estimators: int = 100, learning_rate: float = 0.1, normalize: bool = True, **kwargs: Any) → None

Parameters:

• n_estimators (int)

• learning_rate (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnSVMClassifier

Bases: SklearnSlideClassifier

SVM slide-level classifier with probability calibration.

Parameters:

• C – Regularisation parameter.

• kernel – SVM kernel ("rbf", "linear", …).

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.svm.SVC.

__init__(C: float = 1.0, kernel: str = 'rbf', normalize: bool = True, **kwargs: Any) → None

Parameters:

• C (float)

• kernel (str)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnSlideRegressor

Bases: ScikitBase

Wrapper around any scikit-learn regressor for slide-level prediction.

Features are standardized with StandardScaler before fitting and prediction when normalize=True (default).

Parameters:

• estimator – A fitted or unfitted sklearn regressor implementing fit and predict.

• normalize – Standardize features with StandardScaler (default True).

__init__(estimator: Any, normalize: bool = True) → None

Parameters:

• estimator (Any)

• normalize (bool)

Return type:

None

initialize(config: dict[str, Any] | None = None) → None

Initialize the model. For PyTorch: Load weights, reset parameters. For Sklearn: Configure hyperparameters.

Parameters:

config (dict[str, Any] | None)

Return type:

None

fit(X: ndarray, y: ndarray) → SklearnSlideRegressor

Fit the estimator on numpy feature matrix X and targets y.

Parameters:

• X (ndarray)

• y (ndarray)

Return type:

SklearnSlideRegressor

predict(X: ndarray) → ndarray

Parameters:

X (ndarray)

Return type:

ndarray

predict_as_tensor(X: ndarray) → Any

Return predictions as a float tensor shaped (N,).

Parameters:

X (ndarray)

Return type:

Any

get_learnable_parameters() → list

Return parameters for optimization. Returns empty iterator for non-gradient models (e.g. Random Forest).

Return type:

list

class pathforge.core.models.sklearn_slide.SklearnLinearRegressor

Bases: SklearnSlideRegressor

Ordinary least-squares linear regression.

Parameters:

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.linear_model.LinearRegression.

__init__(normalize: bool = True, **kwargs: Any) → None

Parameters:

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnRidgeRegressor

Bases: SklearnSlideRegressor

Ridge regression slide-level regressor.

Parameters:

• alpha – Regularisation strength.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.linear_model.Ridge.

__init__(alpha: float = 1.0, normalize: bool = True, **kwargs: Any) → None

Parameters:

• alpha (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnElasticNetRegressor

Bases: SklearnSlideRegressor

ElasticNet slide-level regressor.

Parameters:

• alpha – Overall regularisation strength.

• l1_ratio – Mix between L1 (1.0) and L2 (0.0) penalties.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.linear_model.ElasticNet.

__init__(alpha: float = 1.0, l1_ratio: float = 0.5, normalize: bool = True, **kwargs: Any) → None

Parameters:

• alpha (float)

• l1_ratio (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnRandomForestRegressor

Bases: SklearnSlideRegressor

Random-forest slide-level regressor.

Parameters:

• n_estimators – Number of trees.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.ensemble.RandomForestRegressor.

__init__(n_estimators: int = 100, normalize: bool = True, **kwargs: Any) → None

Parameters:

• n_estimators (int)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnGradientBoostingRegressor

Bases: SklearnSlideRegressor

Gradient boosted trees slide-level regressor.

Parameters:

• n_estimators – Number of boosting rounds.

• learning_rate – Shrinkage factor applied to each tree.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.ensemble.GradientBoostingRegressor.

__init__(n_estimators: int = 100, learning_rate: float = 0.1, normalize: bool = True, **kwargs: Any) → None

Parameters:

• n_estimators (int)

• learning_rate (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnSVMRegressor

Bases: SklearnSlideRegressor

Support vector regression slide-level regressor.

Parameters:

• C – Regularisation parameter.

• kernel – SVR kernel ("rbf", "linear", …).

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sklearn.svm.SVR.

__init__(C: float = 1.0, kernel: str = 'rbf', normalize: bool = True, **kwargs: Any) → None

Parameters:

• C (float)

• kernel (str)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnSlideSurvival

Bases: ScikitBase

Wrapper around scikit-survival estimators for slide-level survival prediction.

Features are standardized with StandardScaler before fitting and prediction when normalize=True (default).

Parameters:

• estimator – A fitted or unfitted scikit-survival estimator.

• normalize – Standardize features with StandardScaler (default True).

__init__(estimator: Any, normalize: bool = True) → None

Parameters:

• estimator (Any)

• normalize (bool)

Return type:

None

initialize(config: dict[str, Any] | None = None) → None

Initialize the model. For PyTorch: Load weights, reset parameters. For Sklearn: Configure hyperparameters.

Parameters:

config (dict[str, Any] | None)

Return type:

None

fit(X: ndarray, y: ndarray) → SklearnSlideSurvival

Fit the survival estimator.

Parameters:

• X (ndarray) – Feature matrix shaped [N, D].

• y (ndarray) – Structured array with dtype [('event', bool), ('time', float)] as expected by scikit-survival.

Return type:

SklearnSlideSurvival

predict(X: ndarray) → ndarray

Parameters:

X (ndarray)

Return type:

ndarray

predict_as_tensor(X: ndarray) → Any

Return risk scores as a float tensor shaped (N,).

Parameters:

X (ndarray)

Return type:

Any

get_learnable_parameters() → list

Return parameters for optimization. Returns empty iterator for non-gradient models (e.g. Random Forest).

Return type:

list

class pathforge.core.models.sklearn_slide.SklearnCoxPH

Bases: SklearnSlideSurvival

Cox proportional-hazards model (L2 penalty) from scikit-survival.

Parameters:

• alpha – Ridge penalty strength.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.linear_model.CoxPHSurvivalAnalysis.

__init__(alpha: float = 0.1, normalize: bool = True, **kwargs: Any) → None

Parameters:

• alpha (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnCoxnet

Bases: SklearnSlideSurvival

Elastic-net penalized Cox proportional-hazards model from scikit-survival.

Combines L1 (Lasso) and L2 (Ridge) penalties — useful for high-dimensional feature vectors.

Parameters:

• l1_ratio – Mixing parameter between Ridge (0) and Lasso (1).

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.linear_model.CoxnetSurvivalAnalysis.

__init__(l1_ratio: float = 0.5, normalize: bool = True, **kwargs: Any) → None

Parameters:

• l1_ratio (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnIPCRidge

Bases: SklearnSlideSurvival

Inverse probability of censoring weighted Ridge regression for survival.

Parameters:

• alpha – Ridge penalty strength.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.linear_model.IPCRidge.

__init__(alpha: float = 1.0, normalize: bool = True, **kwargs: Any) → None

Parameters:

• alpha (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnHingeLossSurvivalSVM

Bases: SklearnSlideSurvival

Ranking SVM with hinge loss for survival from scikit-survival.

Parameters:

• alpha – Regularisation strength.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.svm.HingeLossSurvivalSVM.

__init__(alpha: float = 1.0, normalize: bool = True, **kwargs: Any) → None

Parameters:

• alpha (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnNaiveSurvivalSVM

Bases: SklearnSlideSurvival

Naive ranking SVM for survival from scikit-survival.

Parameters:

• alpha – Regularisation strength.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.svm.NaiveSurvivalSVM.

__init__(alpha: float = 1.0, normalize: bool = True, **kwargs: Any) → None

Parameters:

• alpha (float)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnSurvivalTree

Bases: SklearnSlideSurvival

Survival decision tree from scikit-survival.

Parameters:

• max_depth – Maximum depth of the tree.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.tree.SurvivalTree.

__init__(max_depth: int | None = None, normalize: bool = True, **kwargs: Any) → None

Parameters:

• max_depth (int | None)

• normalize (bool)

• kwargs (Any)

Return type:

None

class pathforge.core.models.sklearn_slide.SklearnRandomSurvivalForest

Bases: SklearnSlideSurvival

Random survival forest from scikit-survival.

Parameters:

• n_estimators – Number of trees.

• normalize – Standardize features before fitting (default True).

• kwargs – Forwarded to sksurv.ensemble.RandomSurvivalForest.

__init__(n_estimators: int = 100, normalize: bool = True, **kwargs: Any) → None

Parameters:

• n_estimators (int)

• normalize (bool)

• kwargs (Any)

Return type:

None

pathforge.core.models.sklearn_slide.make_survival_structured_array(time: ndarray, event: ndarray) → ndarray

Build the structured array expected by scikit-survival estimators.

Parameters:

• time (ndarray) – Survival times shaped [N].

• event (ndarray) – Event indicators (0/1) shaped [N].

Returns:

numpy structured array with dtype [('event', bool), ('time', float64)].

Return type:

ndarray

pathforge.core.models.utils.perform_kmeans(x: Tensor, num_clusters: int, n_iter: int = 10) → tuple[Tensor, Tensor]

Simple differentiable K-Means for RRT-MIL. x: (N, D) Returns: assignments (N,), centers (K, D)

Parameters:

• x (Tensor)

• num_clusters (int)

• n_iter (int)

Return type:

tuple[Tensor, Tensor]

Losses

class pathforge.core.losses.base.BaseLoss

Bases: Module, ABC

Root abstract base class for all PathForge losses.

__init__(task_type: str)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

Parameters:

task_type (str)

abstractmethod forward(preds: Tensor, target: Any, **kwargs: Any) → Tensor

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

• preds (Tensor)

• target (Any)

• kwargs (Any)

Return type:

Tensor

class pathforge.core.losses.base.ClassificationLoss

Bases: BaseLoss

Enforces standard classification inputs.

__init__()

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(preds: Tensor, target: Tensor, **kwargs: Any) → Tensor

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

• preds (Tensor)

• target (Tensor)

• kwargs (Any)

Return type:

Tensor

abstractmethod calculate_loss(preds: Tensor, target: Tensor, **kwargs: Any) → Tensor

Implement the actual loss calculation.

Parameters:

• preds (Tensor)

• target (Tensor)

• kwargs (Any)

Return type:

Tensor

class pathforge.core.losses.base.RegressionLoss

Bases: BaseLoss

Enforces inputs are Floats.

__init__()

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(preds: Tensor, target: Tensor, **kwargs: Any) → Tensor

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

• preds (Tensor)

• target (Tensor)

• kwargs (Any)

Return type:

Tensor

abstractmethod calculate_loss(preds: Tensor, target: Tensor, **kwargs: Any) → Tensor

Parameters:

• preds (Tensor)

• target (Tensor)

• kwargs (Any)

Return type:

Tensor

class pathforge.core.losses.base.SurvivalContinuousLoss

Bases: BaseLoss

Enforces target is a Dict with ‘time’ and ‘event’, and all are Floats.

__init__()

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(preds: torch.Tensor, target: Dict[str, torch.Tensor], **kwargs: Any) → torch.Tensor

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

• preds (torch.Tensor)

• target (Dict[str, torch.Tensor])

• kwargs (Any)

Return type:

torch.Tensor

abstractmethod calculate_loss(preds: Tensor, time: Tensor, event: Tensor, **kwargs: Any) → Tensor

Parameters:

• preds (Tensor)

• time (Tensor)

• event (Tensor)

• kwargs (Any)

Return type:

Tensor

class pathforge.core.losses.base.SurvivalDiscreteLoss

Bases: BaseLoss

Enforces target is a Dict with ‘time’ (Long/Int) and ‘event’ (Float/Binary).

__init__()

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(preds: torch.Tensor, target: Dict[str, torch.Tensor], **kwargs: Any) → torch.Tensor

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Parameters:

• preds (torch.Tensor)

• target (Dict[str, torch.Tensor])

• kwargs (Any)

Return type:

torch.Tensor

abstractmethod calculate_loss(preds: Tensor, time: Tensor, event: Tensor, **kwargs: Any) → Tensor

Parameters:

• preds (Tensor)

• time (Tensor)

• event (Tensor)

• kwargs (Any)

Return type:

Tensor

Annotations

class pathforge.core.annotations.base.AnnotationsBase

Bases: ABC

Abstraction over any annotation storage backend.

abstractmethod load_annotations(id: str) → Any

Load annotations from a given identifier.

Parameters:

id (str)

Return type:

Any

abstractmethod save_annotations(id: str, annotations: Any) → None

Save annotations to a given identifier.

Parameters:

• id (str)

• annotations (Any)

Return type:

None

abstractmethod validate_annotations(annotations: Any) → bool

Validate the given annotations.

Parameters:

annotations (Any)

Return type:

bool

abstractmethod inspect_annotations(annotations: Any) → None

Print a summary of the annotations.

Parameters:

annotations (Any)

Return type:

None

pathforge.core.annotations.binning.bin_times(times: Sequence[float], n_bins: int = 3, method: str = 'quantile') → list[int]

Bin continuous survival times into integer classes for discrete survival tasks.

Parameters:

• times (Sequence[float])

• n_bins (int)

• method (str)

Return type:

list[int]

class pathforge.core.annotations.csv.CSVAnnotations

Bases: AnnotationsBase

CSV-backed annotation adapter.

The adapter loads a CSV file into a column-oriented dictionary where each key is a column name and each value is a list of row values for that column.

Example

>>> annotations = CSVAnnotations("annotations.csv")
>>> sorted(annotations.annotations)
['category', 'dataset', 'patient', 'slide']

__init__(path_to_csv: str)

Parameters:

path_to_csv (str)

load_annotations(id: str) → dict[str, list[Any]]

Load annotations from a CSV file.

Parameters:

id (str) – Path to a CSV file.

Returns:

Column-oriented annotation dictionary where each value list has shape [num_rows].

Return type:

dict[str, list[Any]]

save_annotations(id: str, annotations: dict[str, list[Any]]) → None

Persist annotations to a CSV file.

Parameters:

• id (str) – Output CSV path.

• annotations (dict[str, list[Any]]) – Column-oriented annotation dictionary.

Return type:

None

validate_annotations(annotations: dict[str, list[Any]]) → bool

Validate and normalize a PathForge annotation table.

Required columns:

• slide: slide identifier per row

• category: task label per row

Optional columns filled or tracked here:

• patient: defaults to slide when missing

• dataset: defaults to "default" when missing

• wsi_path: tracked for direct slide resolution when present

Parameters:

annotations (dict[str, list[Any]])

Return type:

bool

inspect_annotations(annotations: dict[str, list[Any]]) → None

Log a compact summary of the loaded annotations.

Parameters:

annotations (dict[str, list[Any]]) – Column-oriented annotation dictionary with lists shaped [num_rows].

Return type:

None

Datasets

class pathforge.core.datasets.base.DatasetBase

Bases: ABC

Generic dataset base class. Can represent tile-level, slide-level or other datasets.

abstract property name: str

abstract property num_samples: int

class pathforge.core.datasets.base.BagDatasetBase

Bases: DatasetBase

Dataset base class for Multiple Instance Learning (MIL) bags. Each item is a bag of instances with an associated label.

abstract property num_bags: int

property num_samples: int

class pathforge.core.datasets.base.TileDatasetBase

Bases: DatasetBase

Dataset base class for tile-level datasets. Each item is a single tile with an associated label.

abstract property num_tiles: int

class pathforge.core.datasets.wsi_dataset.WSI

Bases: object

One whole-slide sample with source path, artifact path, and cached backend object.

slide: str

patient: str

category: str

path: Path

artifact_path: Path

fallback_mpp: float | None

property is_loaded: bool

property obj: Any

__init__(slide: str, patient: str, category: str, path: Path, artifact_path: Path, fallback_mpp: float | None = None, _obj: Any | None = None) → None

Parameters:

• slide (str)

• patient (str)

• category (str)

• path (Path)

• artifact_path (Path)

• fallback_mpp (float | None)

• _obj (Any | None)

Return type:

None

class pathforge.core.datasets.wsi_dataset.WSIDataset

Bases: DatasetBase

One sample = one WSI. Builds samples from annotations_df rows where ann_df[‘dataset’] == config.name.

Artifacts are stored per-slide in:

artifacts_dir/{slide_id}.h5

Combos/tilings/features live inside that file (e.g. bags/{bag_id}/…), so this dataset does not track any active combo state.

__init__(ds_cfg: DatasetEntry, annotations_df: DataFrame)

Parameters:

• ds_cfg (DatasetEntry)

• annotations_df (DataFrame)

property name: str

property used_for: str

property num_samples: int

property slides_dir: Path

property artifacts_dir: Path

property tissue_annotations_dir: Path | None

slide_artifact_path(slide_id: str) → Path

Per-slide HDF5 file path: artifacts_dir/{slide_id}.h5

Parameters:

slide_id (str)

Return type:

Path

class pathforge.core.datasets.bag_dataset.BagSample

Bases: object

One logical bag unit, including its source slides and artifact locations.

sample_id: str

slide_ids: list[str]

artifact_paths: list[Path]

category: Any

patient_id: str | None

case_id: str | None

metadata: dict[str, Any]

__init__(sample_id: str, slide_ids: list[str], artifact_paths: list[~pathlib.Path], category: ~typing.Any, patient_id: str | None = None, case_id: str | None = None, metadata: dict[str, ~typing.Any] = <factory>) → None

Parameters:

• sample_id (str)

• slide_ids (list[str])

• artifact_paths (list[Path])

• category (Any)

• patient_id (str | None)

• case_id (str | None)

• metadata (dict[str, Any])

Return type:

None

class pathforge.core.datasets.bag_dataset.SlideRetrievalDatasetItem

Bases: object

One materialized retrieval item produced from a bag dataset.

index: int

sample: BagSample

inputs: dict[str, Any]

__init__(index: int, sample: BagSample, inputs: dict[str, ~typing.Any]=<factory>) → None

Parameters:

• index (int)

• sample (BagSample)

• inputs (dict[str, Any])

Return type:

None

class pathforge.core.datasets.bag_dataset.BagDataset

Bases: BagDatasetBase

Canonical bag dataset supporting artifact-backed and prepared-bag modes.

__init__(*args: Any, **kwargs: Any) → None

Parameters:

• args (Any)

• kwargs (Any)

Return type:

None

ds_cfg: DatasetEntry | None

annotations_df: DataFrame | None

combo_cfg: ComboConfig | None

aggregation_level: Literal['slide', 'case', 'patient']

target_column: str

task: str

slide_column: str | None

time_column: str | None

event_column: str | None

bag_size: int | None

samples: list[BagSample]

artifacts_dir: Path | None

feature_path: Path | None

tiling_id: str | None

extractor_name: str | None

property name: str

property num_bags: int

property feature_dim: int

output_dim() → int

Return type:

int

get_sample(index: int) → BagSample

Parameters:

index (int)

Return type:

BagSample

get_feature_level() → Literal['patch', 'slide', 'unknown', 'invalid']

Return type:

Literal[‘patch’, ‘slide’, ‘unknown’, ‘invalid’]

get_feature_level_reason() → str

Return type:

str

load_bag(index: int) → Tensor

Parameters:

index (int)

Return type:

Tensor

get_bag_sample(index: int) → tuple[Tensor, BagSample]

Parameters:

index (int)

Return type:

tuple[Tensor, BagSample]

class pathforge.core.datasets.bag_dataset.MILBagDataset

Bases: BagDataset

MIL dataset alias for the canonical bag schema.

class pathforge.core.datasets.bag_dataset.SlideRetrievalBagDataset

Bases: BagDataset

Bag dataset variant that binds retrieval-specific sample loaders.

__init__(*args: Any, **kwargs: Any) → None

Parameters:

• args (Any)

• kwargs (Any)

Return type:

None

sample_loader: Callable[[...], dict[str, Any]] | None

bind_sample_loader(sample_loader: Callable[[...], dict[str, Any]]) → None

Parameters:

sample_loader (Callable[[...], dict[str, Any]])

Return type:

None

clear_sample_loader() → None

Return type:

None

class pathforge.core.datasets.bag_schema.BagBatch

Bases: TypedDict

Canonical PathForge MIL batch schema.

X contains finite floating-point features shaped [B, N, D] for a padded batch or [N, D] for one unbatched bag. Y contains the bag-level target. Classification targets are normally integer class IDs; continuous-survival targets may contain time and event tensors.

Optional keys are mask (real-versus-padding flags shaped [B, N]), coords (x/y positions shaped [B, N, 2]), adj (dense adjacency shaped [B, N, N]), and y_inst (instance labels shaped [B, N]).

Example

import torch
from pathforge.core.datasets.bag_schema import assert_bag_schema

batch = {
    "X": torch.zeros(2, 4, 1024, dtype=torch.float32),
    "Y": torch.tensor([0, 1], dtype=torch.long),
    "mask": torch.tensor([[1, 1, 0, 0], [1, 1, 1, 1]], dtype=torch.bool),
}
assert_bag_schema(batch, batched=True)

Raises:

AssertionError – Validation helpers raise when required keys, ranks, dtypes, shapes, or finite-value contracts are violated.

X: Tensor

Y: Any

mask: NotRequired[Tensor]

coords: NotRequired[Tensor]

adj: NotRequired[Tensor]

y_inst: NotRequired[Tensor]

pathforge.core.datasets.bag_schema.as_bag_batch(batch: dict[str, Any]) → BagBatch

Cast a plain dictionary to BagBatch after runtime validation.

Parameters:

batch (dict[str, Any])

Return type:

BagBatch

pathforge.core.datasets.bag_schema.assert_bag_schema(batch: dict[str, Any], *, batched: bool | None = None, check_finite: bool = True) → None

Validate the canonical MIL bag schema at module boundaries.

Parameters:

• batch (dict[str, Any]) – Mapping containing at least X and Y.

• batched (bool | None) – True requires X rank 3, False requires rank 2, and None accepts either. Optional tensors are checked against the inferred rank.

• check_finite (bool) – When true, checks floating tensors for NaN/Inf.

Raises:

AssertionError – If the bag violates required shape, dtype, or finiteness invariants.

Return type:

None

pathforge.core.datasets.factory.build_wsi_dataset(ds_cfg: DatasetEntry, annotations_df: DataFrame, slide_ids: list[str] | None = None) → WSIDataset

Build a WSIDataset for one dataset, optionally restricted to specific slides.

Parameters:

• ds_cfg (DatasetEntry)

• annotations_df (DataFrame)

• slide_ids (list[str] | None)

Return type:

WSIDataset

pathforge.core.datasets.factory.build_wsi_datasets(cfg: Config, annotations_df: DataFrame) → list[WSIDataset]

Build WSIDataset objects for all non-ignored datasets.

Parameters:

• cfg (Config)

• annotations_df (DataFrame)

Return type:

list[WSIDataset]

pathforge.core.datasets.factory.build_bag_dataset(ds_cfg: DatasetEntry, annotations_df: DataFrame, combo_cfg: ComboConfig, aggregation_level: str, task: str, target_column: str | None = None, slide_ids: list[str] | None = None) → BagDataset

Build a BagDataset for one dataset, optionally restricted to specific slides.

Parameters:

• ds_cfg (DatasetEntry)

• annotations_df (DataFrame)

• combo_cfg (ComboConfig)

• aggregation_level (str)

• task (str)

• target_column (str | None)

• slide_ids (list[str] | None)

Return type:

BagDataset

pathforge.core.datasets.factory.build_bag_datasets(cfg: Config, annotations_df: DataFrame, combo_cfg: ComboConfig, task: str, target_column: str | None = None) → list[BagDataset]

Build BagDataset objects for all non-ignored datasets.

Parameters:

• cfg (Config)

• annotations_df (DataFrame)

• combo_cfg (ComboConfig)

• task (str)

• target_column (str | None)

Return type:

list[BagDataset]

pathforge.core.datasets.utils.group_datasets_by_use(bag_datasets: list) → dict[str, list]

Group datasets by their configured usage.

Parameters:

bag_datasets (list)

Return type:

dict[str, list]

Experiments

class pathforge.core.experiments.base.Experiment

Bases: object

Manage experiment project_root and experiment-level metadata files.

Creates/validates: - project.json - annotations.csv (copied into project_root)

Provides helpers to load annotations. Search-space materialization lives in pathforge.core.experiments.combinations and dataset construction lives in pathforge.core.datasets.factory so this layer stays focused on project-scoped metadata.

cfg: Config

project_root: str | None

load_annotations() → DataFrame

Load annotations CSV from project_root/annotations.csv.

Returns:

DataFrame with annotations.

Return type:

DataFrame

build_combinations(keys: list[str]) → list[ComboConfig]

Build all combinations of benchmark parameters for the given keys. :param keys: List of field names in cfg.benchmark_parameters to build combinations for.

Returns:

List of ComboConfig instances representing all combinations.

Parameters:

keys (list[str])

Return type:

list[ComboConfig]

__init__(cfg: Config, project_root: str | None = None) → None

Parameters:

• cfg (Config)

• project_root (str | None)

Return type:

None

class pathforge.core.experiments.base.ComboConfig

Bases: object

Generic, dynamically-populated benchmark combination configuration.

Inputs:

keyword arguments (object):

Benchmark parameter values keyed by parameter name. Each key becomes an attribute on the created object.

Outputs:

ComboConfig:

Configuration object exposing combo values and optional <key>_params hyperparameter dictionaries.

Semantic goal:

Represent one fully-materialized benchmark parameter combination in a lightweight attribute-based object.

Example

combo_cfg = ComboConfig.from_keys_values(
    keys=["feature_extraction", "tile_px"],
    values=[
        BenchmarkParamEntry(value="uni", hyperparams={"family": "foundation"}),
        256,
    ],
)
assert combo_cfg.feature_extraction == "uni"
assert combo_cfg.get_hyperparams("feature_extraction") == {
    "family": "foundation",
}

__init__(**kwargs: object) → None

Parameters:

kwargs (object)

Return type:

None

classmethod from_keys_values(keys: list[str], values: list[object]) → ComboConfig

Build one combo config from aligned key/value lists.

Inputs:

keys (list[str]):

Parameter names. Shape: (n_keys,).

values (list[object]):

Parameter values aligned with keys. Shape: (n_keys,).

Outputs:

ComboConfig:

Combo object containing one attribute per key and one <key>_params attribute per key.

Parameters:

• keys (list[str])

• values (list[object])

Return type:

ComboConfig

to_dict() → dict[str, object]

Return a shallow dictionary representation of the combo.

Return type:

dict[str, object]

get(key: str, default: object = None) → object

Return one combo value by key with a dict-like fallback default.

Parameters:

• key (str)

• default (object)

Return type:

object

get_hyperparams(key: str, default: dict[str, Any] | None = None) → dict[str, Any]

Return the hyperparameters attached to one combo value.

Parameters:

• key (str)

• default (dict[str, Any] | None)

Return type:

dict[str, Any]

class pathforge.core.experiments.combinations.ComboConfig

Bases: object

Generic, dynamically-populated benchmark combination configuration.

Inputs:

keyword arguments (object):

Benchmark parameter values keyed by parameter name. Each key becomes an attribute on the created object.

Outputs:

ComboConfig:

Configuration object exposing combo values and optional <key>_params hyperparameter dictionaries.

Semantic goal:

Represent one fully-materialized benchmark parameter combination in a lightweight attribute-based object.

Example

combo_cfg = ComboConfig.from_keys_values(
    keys=["feature_extraction", "tile_px"],
    values=[
        BenchmarkParamEntry(value="uni", hyperparams={"family": "foundation"}),
        256,
    ],
)
assert combo_cfg.feature_extraction == "uni"
assert combo_cfg.get_hyperparams("feature_extraction") == {
    "family": "foundation",
}

__init__(**kwargs: object) → None

Parameters:

kwargs (object)

Return type:

None

classmethod from_keys_values(keys: list[str], values: list[object]) → ComboConfig

Build one combo config from aligned key/value lists.

Inputs:

keys (list[str]):

Parameter names. Shape: (n_keys,).

values (list[object]):

Parameter values aligned with keys. Shape: (n_keys,).

Outputs:

ComboConfig:

Combo object containing one attribute per key and one <key>_params attribute per key.

Parameters:

• keys (list[str])

• values (list[object])

Return type:

ComboConfig

to_dict() → dict[str, object]

Return a shallow dictionary representation of the combo.

Return type:

dict[str, object]

get(key: str, default: object = None) → object

Return one combo value by key with a dict-like fallback default.

Parameters:

• key (str)

• default (object)

Return type:

object

get_hyperparams(key: str, default: dict[str, Any] | None = None) → dict[str, Any]

Return the hyperparameters attached to one combo value.

Parameters:

• key (str)

• default (dict[str, Any] | None)

Return type:

dict[str, Any]

pathforge.core.experiments.combinations.build_combinations(cfg: Config, keys: list[str]) → list[ComboConfig]

Materialize the benchmark search space for the requested parameter keys.

Inputs:

cfg (Config):

Validated application configuration containing benchmark_parameters.

keys (list[str]):

Parameter names to include in the grid. Shape: (n_keys,).

Outputs:

list[ComboConfig]:

List of materialized benchmark combinations. Shape: (n_combos,).

Semantic goal:

Convert the declarative benchmark parameter config into concrete combo objects that policies and tasks can execute.

Example

combos = build_combinations(
    cfg=cfg,
    keys=["feature_extraction", "tile_px", "tile_mpp"],
)

Parameters:

• cfg (Config)

• keys (list[str])

Return type:

list[ComboConfig]

pathforge.core.experiments.combo_ids.build_tiling_id(combo_cfg: ComboConfig) → str

Build the canonical tiling identifier from a combination config.

Parameters:

combo_cfg (ComboConfig)

Return type:

str

pathforge.core.experiments.combo_ids.build_bag_id(combo_cfg: ComboConfig) → str

Build the canonical bag identifier from a combination config.

Parameters:

combo_cfg (ComboConfig)

Return type:

str

pathforge.core.experiments.combo_ids.build_feature_name(combo_cfg: ComboConfig) → str

Build the canonical stored feature name from a combination config.

Parameters:

combo_cfg (ComboConfig)

Return type:

str

Feature helpers

pathforge.core.features.utils.find_slides_with_missing_features(ds_cfg: DatasetEntry, annotations_df: DataFrame, combo_cfg: ComboConfig) → list[str]

Return slide IDs for which the required features are missing.

Parameters:

• ds_cfg (DatasetEntry)

• annotations_df (DataFrame)

• combo_cfg (ComboConfig)

Return type:

list[str]

H5 I/O

class pathforge.core.io.h5.base.FileHandleH5

Bases: object

Context-managed HDF5 file wrapper used by PathForge artifact helpers.

path: Path

mode: str

property h5: File

__init__(path: Path, mode: str = 'a', _h5: File | None = None) → None

Parameters:

• path (Path)

• mode (str)

• _h5 (File | None)

Return type:

None

pathforge.core.io.h5.heatmaps.prediction_heatmap_exists(slide_artifact: FileHandleH5, bag_id: str, heatmap_name: str, *, layout: H5Layout = DEFAULT_LAYOUT) → bool

Return whether a prediction heatmap score dataset exists.

Parameters:

• slide_artifact (FileHandleH5)

• bag_id (str)

• heatmap_name (str)

• layout (H5Layout)

Return type:

bool

pathforge.core.io.h5.heatmaps.write_prediction_heatmap(slide_artifact: FileHandleH5, bag_id: str, heatmap_name: str, *, coords: ndarray, scores: ndarray, metadata: dict[str, Any] | None = None, layout: H5Layout = DEFAULT_LAYOUT) → None

Persist a prediction heatmap under the dedicated H5 prediction namespace.

Parameters:

• slide_artifact (FileHandleH5) – Open H5 handle for one slide artifact.

• bag_id (str) – Tiling/feature bag id, for example "256px_0.5mpp".

• heatmap_name (str) – Non-empty heatmap identifier without "/".

• coords (ndarray) – Coordinate array shaped (N, 2). Values are x/y tile coordinates in level-0 pixels or another coordinate space documented by metadata.

• scores (ndarray) – Finite normalized heatmap scores shaped (N,) with values in [0, 1].

• metadata (dict[str, Any] | None) – JSON-serializable metadata such as backend name, model path, source score file, and coordinate space.

• layout (H5Layout)

Raises:

ValueError – If shapes differ, values are non-finite, or scores are outside [0, 1].

Return type:

None

pathforge.core.io.h5.heatmaps.read_prediction_heatmap(slide_artifact: FileHandleH5, bag_id: str, heatmap_name: str, *, layout: H5Layout = DEFAULT_LAYOUT) → dict[str, Any]

Read a persisted prediction heatmap from an open slide H5 artifact.

Parameters:

• slide_artifact (FileHandleH5)

• bag_id (str)

• heatmap_name (str)

• layout (H5Layout)

Return type:

dict[str, Any]

class pathforge.core.io.h5.layout.H5Layout

Bases: object

Canonical HDF5 path layout for per-slide PathForge artifacts.

tissue_dataset: str

bags_group: str

coords_name: str

tiling_spec_name: str

tiles_overview_name: str

features_group_name: str

predictions_group_name: str

heatmaps_group_name: str

heatmap_coords_name: str

heatmap_scores_name: str

heatmap_metadata_name: str

bag_group(bag_id: str) → str

Parameters:

bag_id (str)

Return type:

str

coords_dataset(bag_id: str) → str

Parameters:

bag_id (str)

Return type:

str

tiling_spec_dataset(bag_id: str) → str

Parameters:

bag_id (str)

Return type:

str

tiles_overview_dataset(bag_id: str) → str

Parameters:

bag_id (str)

Return type:

str

features_group(bag_id: str) → str

Parameters:

bag_id (str)

Return type:

str

features_dataset(bag_id: str, extractor_name: str) → str

Parameters:

• bag_id (str)

• extractor_name (str)

Return type:

str

prediction_heatmaps_group(bag_id: str) → str

Parameters:

bag_id (str)

Return type:

str

prediction_heatmap_group(bag_id: str, heatmap_name: str) → str

Parameters:

• bag_id (str)

• heatmap_name (str)

Return type:

str

prediction_heatmap_coords_dataset(bag_id: str, heatmap_name: str) → str

Parameters:

• bag_id (str)

• heatmap_name (str)

Return type:

str

prediction_heatmap_scores_dataset(bag_id: str, heatmap_name: str) → str

Parameters:

• bag_id (str)

• heatmap_name (str)

Return type:

str

prediction_heatmap_metadata_dataset(bag_id: str, heatmap_name: str) → str

Parameters:

• bag_id (str)

• heatmap_name (str)

Return type:

str

__init__(tissue_dataset: str = 'annotations/tissue', bags_group: str = 'bags', coords_name: str = 'coords', tiling_spec_name: str = 'tiling_spec', tiles_overview_name: str = 'tiles_overview', features_group_name: str = 'features', predictions_group_name: str = 'predictions', heatmaps_group_name: str = 'heatmaps', heatmap_coords_name: str = 'coords', heatmap_scores_name: str = 'scores', heatmap_metadata_name: str = 'metadata') → None

Parameters:

• tissue_dataset (str)

• bags_group (str)

• coords_name (str)

• tiling_spec_name (str)

• tiles_overview_name (str)

• features_group_name (str)

• predictions_group_name (str)

• heatmaps_group_name (str)

• heatmap_coords_name (str)

• heatmap_scores_name (str)

• heatmap_metadata_name (str)

Return type:

None

Slide Processing

class pathforge.core.slide_processing.base.SlideProcessorBase

Bases: ABC

Base class for slide processing backends.

abstractmethod load_wsi(wsi: WSI) → None

Load / open the backend-native slide object and store it on the WSI.

Parameters:

wsi (WSI)

Return type:

None

close_wsi(wsi: WSI) → None

Close the backend-native slide object (if needed) and clear it from the WSI.

Parameters:

wsi (WSI)

Return type:

None

get_base_mpp(wsi: WSI) → float

Return the level-0 microns-per-pixel (MPP) as a scalar.

This should represent the slide’s base resolution used to convert physical tile size to level-0 pixels.

Parameters:

wsi (WSI)

Return type:

float

abstractmethod get_thumbnail(wsi: WSI, level: int = -1) → Tuple[Any, float, float]

Retrieve a thumbnail image for visualization and the downscale factors relative to level-0 (base resolution) coordinates.

Parameters:

• wsi (WSI) – Loaded WSI.

• level (int) – Pyramid level to use for thumbnail retrieval. By convention, level=-1 means the lowest-resolution level.

Returns:

(thumbnail_image, downscale_x, downscale_y)

• thumbnail_image: backend-defined image object (e.g. PIL image / ndarray)

• downscale_x: factor such that x_thumb = x_level0 / downscale_x

• downscale_y: factor such that y_thumb = y_level0 / downscale_y

Return type:

Tuple[Any, float, float]

abstractmethod segment_tissue(wsi: WSI, config: Dict[str, Any] | None = None) → Any

Segment tissue regions from the slide object.

Parameters:

• wsi (WSI)

• config (Dict[str, Any] | None)

Return type:

Any

abstractmethod extract_patches(wsi: WSI, tissues: List[ndarray], config: Dict[str, Any] | None = None) → Tuple[DataFrame, str]

Extract patches from the slide object.

Returns:

DataFrame with required columns: ‘tile_id’, ‘x’, ‘y’ tile_spec: JSON string (keys may vary; backend-defined)

Return type:

tiles_df

Parameters:

• wsi (WSI)

• tissues (List[ndarray])

• config (Dict[str, Any] | None)

abstractmethod validate_tile_spec(tile_spec: str | None, config: Dict[str, Any] | None = None) → bool

Validate whether the provided tile_spec (JSON string) is usable for this backend.

IMPORTANT: This MUST be implemented by downstream backends. There is no default permissive implementation, because the policy relies on this to decide whether cached tiles can be reused safely.

Parameters:

• tile_spec (str | None)

• config (Dict[str, Any] | None)

Return type:

bool

abstractmethod extract_features(wsi: WSI, tiles: DataFrame, tile_spec: str, config: Dict[str, Any] | None = None) → Any

Extract features from the slide object.

Parameters:

• tiles (DataFrame) – DataFrame with required columns: ‘tile_id’, ‘x’, ‘y’

• tile_spec (str) – JSON string returned by extract_patches() (backend-defined)

• wsi (WSI)

• config (Dict[str, Any] | None)

Return type:

Any

abstractmethod read_patch_region(wsi: WSI, x: int, y: int, width: int, height: int, level: int) → ndarray

Read one patch region from the source slide as an RGB image array.

Parameters:

• wsi (WSI) – Loaded slide wrapper.

• x (int) – Level-0 left coordinate.

• y (int) – Level-0 top coordinate.

• width (int) – Region width in pixels at level.

• height (int) – Region height in pixels at level.

• level (int) – Slide pyramid level used for the read.

Returns:

RGB uint8 array with shape (H, W, 3).

Return type:

ndarray

Example: .. code-block:: python

patch = processor.read_patch_region(wsi, 0, 0, 256, 256, 0)

abstractmethod extract_cells(wsi: WSI, config: Dict[str, Any] | None = None) → Any

Extract cells from the slide object.

Parameters:

• wsi (WSI)

• config (Dict[str, Any] | None)

Return type:

Any

abstractmethod inspect_slide(wsi: WSI) → None

Inspect the slide object for debugging or analysis.

Parameters:

wsi (WSI)

Return type:

None

class pathforge.core.slide_processing.lazyslide.LazySlideProcessor

Bases: SlideProcessorBase

Process slides with LazySlide while preserving PathForge H5 contracts.

Coordinates are stored as an (N, 5) int32 array containing x_level0, y_level0, read width, read height, and pyramid level. Features are returned as a row-aligned (N, D) float32 array. The backend-specific tile specification is reconstructed when features are extracted rather than persisted as PathForge’s source of truth.

BACKEND_NAME = 'lazyslide'

COORD_SPACE = 'level0'

__init__() → None

Return type:

None

load_wsi(wsi: WSI) → None

Load / open the backend-native slide object and store it on the WSI.

Parameters:

wsi (WSI)

Return type:

None

close_wsi(wsi: WSI) → None

Close the backend-native slide object (if needed) and clear it from the WSI.

Parameters:

wsi (WSI)

Return type:

None

get_base_mpp(wsi: WSI) → float

Return the base MPP for this slide.

Priority: 1. MPP from the loaded slide object (wsi.obj.properties.mpp) 2. fallback_mpp from the PathForge WSI dataclass

Parameters:

wsi (WSI)

Return type:

float

get_thumbnail(wsi: WSI, level: int = -1) → Tuple[Any, float, float]

Return a thumbnail image and downscale factors relative to level-0 coords.

Returns:

(thumbnail_image, downscale_x, downscale_y)

Parameters:

• wsi (WSI)

• level (int)

Return type:

Tuple[Any, float, float]

Notes: - Current implementation prefers LazySlide’s existing ‘wsi_thumbnail’. - level is accepted for API compatibility; for now we use the stored thumbnail.

segment_tissue(wsi: WSI, config: Dict[str, Any]) → list[list[list[list[float]]]]

Segment tissue regions from the slide object.

Parameters:

• wsi (WSI)

• config (Dict[str, Any])

Return type:

list[list[list[list[float]]]]

extract_patches(wsi: WSI, tissues: list[list[list[list[float]]]], config: Dict[str, Any]) → Tuple[ndarray, dict]

Produce: - coords: (N,5) int32 [x0,y0,read_w,read_h,level] - tiling_spec: dict written to H5 (backend-agnostic)

Parameters:

• wsi (WSI)

• tissues (list[list[list[list[float]]]])

• config (Dict[str, Any])

Return type:

Tuple[ndarray, dict]

validate_tile_spec(tiling_spec: dict | None, config: Dict[str, Any] | None = None) → bool

Validate whether the provided tile_spec (JSON string) is usable for this backend.

IMPORTANT: This MUST be implemented by downstream backends. There is no default permissive implementation, because the policy relies on this to decide whether cached tiles can be reused safely.

Parameters:

• tiling_spec (dict | None)

• config (Dict[str, Any] | None)

Return type:

bool

extract_features(wsi: WSI, coords: ndarray, tiling_spec: dict, config: Dict[str, Any]) → ndarray

Extract features from the slide object.

Parameters:

• tiles – DataFrame with required columns: ‘tile_id’, ‘x’, ‘y’

• tile_spec – JSON string returned by extract_patches() (backend-defined)

• wsi (WSI)

• coords (ndarray)

• tiling_spec (dict)

• config (Dict[str, Any])

Return type:

ndarray

read_patch_region(wsi: WSI, x: int, y: int, width: int, height: int, level: int) → ndarray

Read one patch region from the LazySlide backend as an RGB uint8 array.

Inputs: - wsi: loaded slide wrapper exposing wsi.obj.read_region(…). - x: level-0 left coordinate. - y: level-0 top coordinate. - width: region width in pixels at level. - height: region height in pixels at level. - level: slide pyramid level used for the read.

Returns: - np.ndarray[uint8] with shape (H, W, 3).

Parameters:

• wsi (WSI)

• x (int)

• y (int)

• width (int)

• height (int)

• level (int)

Return type:

ndarray

extract_cells(wsi: WSI, config: Dict[str, Any]) → Any

Extract cells from the slide object.

Parameters:

• wsi (WSI)

• config (Dict[str, Any])

Return type:

Any

inspect_slide(wsi: WSI) → None

Inspect the slide object for debugging or analysis.

Parameters:

wsi (WSI)

Return type:

None

Explainability

class pathforge.core.explainer_base.ExplainerBase

Bases: ABC

Base interface for prediction explainers.

Explainability backends consume model-specific payloads and return one inspectable explanation artifact, such as a heatmap.

abstractmethod explain(input: Any) → Any

Build one explanation artifact from a backend-specific input payload.

Parameters:

input (Any)

Return type:

Any

abstractmethod initialize(config: dict[str, Any]) → None

Initialize the explainer from runtime configuration.

Parameters:

config (dict[str, Any])

Return type:

None

Reports And Visualization

class pathforge.core.reports.base.Report

Bases: object

Base report payload wrapper used by generated PathForge artifacts.

payload: dict

__init__(payload: dict) → None

Parameters:

payload (dict)

Return type:

None

class pathforge.core.reports.base.ProcessingReport

Bases: Report

Report payload for slide-processing and feature-extraction workflows.

payload: dict

class pathforge.core.reports.base.DebugReport

Bases: Report

Report payload for debugging and inspection workflows.

payload: dict

class pathforge.core.reports.base.PredictionReport

Bases: Report

Report payload for inference or evaluation prediction outputs.

payload: dict

class pathforge.core.reports.tiles_report_pdf.TilesOverviewEntry

Bases: object

One slide-level tiles overview image and its derived metadata.

slide_id: str

artifact_path: Path

image_bytes: bytes

num_tiles: int | None

tiling_spec: dict[str, Any] | None

__init__(slide_id: str, artifact_path: Path, image_bytes: bytes, num_tiles: int | None, tiling_spec: dict[str, Any] | None) → None

Parameters:

• slide_id (str)

• artifact_path (Path)

• image_bytes (bytes)

• num_tiles (int | None)

• tiling_spec (dict[str, Any] | None)

Return type:

None

class pathforge.core.reports.tiles_report_pdf.TilesReportStats

Bases: object

Aggregate counters describing coverage and failures in a tiles report run.

total_slides_expected: int

included_slides: int

missing_overview: int

missing_coords: int

unreadable_h5: int

corrupt_overview: int

__init__(total_slides_expected: int, included_slides: int, missing_overview: int, missing_coords: int, unreadable_h5: int, corrupt_overview: int) → None

Parameters:

• total_slides_expected (int)

• included_slides (int)

• missing_overview (int)

• missing_coords (int)

• unreadable_h5 (int)

• corrupt_overview (int)

Return type:

None

class pathforge.core.reports.tiles_report_pdf.TilesReportCollection

Bases: object

Collected tiles-report entries plus their dataset-level summary statistics.

entries: list[TilesOverviewEntry]

stats: TilesReportStats

bag_id: str

representative_tiling_spec: dict[str, Any] | None

__init__(entries: list[TilesOverviewEntry], stats: TilesReportStats, bag_id: str, representative_tiling_spec: dict[str, Any] | None) → None

Parameters:

• entries (list[TilesOverviewEntry])

• stats (TilesReportStats)

• bag_id (str)

• representative_tiling_spec (dict[str, Any] | None)

Return type:

None

pathforge.core.reports.tiles_report_pdf.create_tiles_report_pdf(*, dataset: WSIDataset, bag_id: str, output_path: Path | None = None, timestamp: datetime | None = None, page_size: tuple[float, float] = A4) → Path

Create a timestamped tile extraction report PDF for one dataset and one bag_id.

The report is built from H5-stored tiles_overview images, so original WSI files are not required.

Parameters:

• dataset (WSIDataset) – PathForge WSIDataset (provides slide list/order and artifact paths).

• bag_id (str) – Tiling bag identifier, e.g. “256px_0.5mpp”.

• output_path (Path | None) – Optional explicit output PDF path. If None, a timestamped file is created in dataset.artifacts_dir.

• timestamp (datetime | None) – Optional datetime used for deterministic naming/testing.

• page_size (tuple[float, float]) – ReportLab page size (default A4 portrait).

Returns:

Path to the written PDF.

Raises:

RuntimeError – If no valid tiles_overview images were found for the dataset/bag.

Return type:

Path

pathforge.core.reports.tiles_report_pdf.collect_tiles_overview_entries(*, dataset: WSIDataset, bag_id: str) → TilesReportCollection

Collect reportable tiles_overview entries from dataset samples in dataset order.

Parameters:

• dataset (WSIDataset)

• bag_id (str)

Return type:

TilesReportCollection

class pathforge.core.visualization.tiles_overview.TilesOverviewRenderResult

Bases: object

JPEG overview image plus the thumbnail-space mapping used to render it.

Variables:

• image_bytes (bytes) – JPEG-encoded thumbnail with tile grid overlay.

• downscale_x (float) – Effective level-0-to-thumbnail x downscale.

• downscale_y (float) – Effective level-0-to-thumbnail y downscale.

• image_width_px (int) – Rendered overview width in pixels.

• image_height_px (int) – Rendered overview height in pixels.

image_bytes: bytes

downscale_x: float

downscale_y: float

image_width_px: int

image_height_px: int

__init__(image_bytes: bytes, downscale_x: float, downscale_y: float, image_width_px: int, image_height_px: int) → None

Parameters:

• image_bytes (bytes)

• downscale_x (float)

• downscale_y (float)

• image_width_px (int)

• image_height_px (int)

Return type:

None

pathforge.core.visualization.tiles_overview.render_tiles_overview_image(*, thumbnail_image: Any, coords_array: ndarray, downscale_x: float, downscale_y: float, slide_id: str, tiling_spec: dict[str, Any] | None = None, base_mpp: float | None = None, jpeg_quality: int = 65, max_long_side: int | None = 1200) → bytes

Backward-compatible wrapper returning only JPEG bytes.

Parameters:

• thumbnail_image (Any)

• coords_array (ndarray)

• downscale_x (float)

• downscale_y (float)

• slide_id (str)

• tiling_spec (dict[str, Any] | None)

• base_mpp (float | None)

• jpeg_quality (int)

• max_long_side (int | None)

Return type:

bytes

pathforge.core.visualization.tiles_overview.render_tiles_overview(*, thumbnail_image: Any, coords_array: ndarray, downscale_x: float, downscale_y: float, slide_id: str, tiling_spec: dict[str, Any] | None = None, base_mpp: float | None = None, jpeg_quality: int = 65, max_long_side: int | None = 1200) → TilesOverviewRenderResult

Render a tile overview image (thumbnail + tile grid overlay) and return JPEG bytes.

The tile overlay size is computed exactly from: - tile_px (output tile width in pixels at target mpp) - tile_mpp (target microns-per-pixel) - base_mpp (slide level-0 microns-per-pixel)

So:

tile_size_um = tile_px * tile_mpp tile_size_level0_px = tile_size_um / base_mpp

Notes: - coords[:, 0:2] are level-0 top-left coordinates. - read_w/read_h/read_level are not used for overlay sizing. - No title/text is drawn here (PDF draws text consistently).

Parameters:

• thumbnail_image (Any)

• coords_array (ndarray)

• downscale_x (float)

• downscale_y (float)

• slide_id (str)

• tiling_spec (dict[str, Any] | None)

• base_mpp (float | None)

• jpeg_quality (int)

• max_long_side (int | None)

Return type:

TilesOverviewRenderResult

Registry

Backward-compatible registry imports for legacy core modules.

Tasks

Task registry and base class for all benchmarking and retrieval tasks.

pathforge.core.tasks.registry.register_task(name: str) → Callable[[type['TaskBase']], type['TaskBase']]

Decorator to register a benchmarking task class.

Example

@register_task("classification")
class ClassificationTask(TaskBase):
    ...

Parameters:

name (str)

Return type:

Callable[[type[‘TaskBase’]], type[‘TaskBase’]]

pathforge.core.tasks.registry.build_task(name: str, experiment: Experiment) → TaskBase

Resolve and instantiate a registered task.

Parameters:

• name (str) – Task name.

• experiment (Experiment) – Experiment instance passed to the task constructor.

Returns:

Instantiated task object.

Return type:

TaskBase

pathforge.core.tasks.registry.get_task(name: str) → type['TaskBase']

Get a registered task class by name.

Parameters:

name (str)

Return type:

type[‘TaskBase’]

pathforge.core.tasks.registry.is_task_available(name: str) → bool

Check whether a task is registered.

Parameters:

name (str)

Return type:

bool

pathforge.core.tasks.registry.list_tasks() → list[str]

Return all registered task names in sorted order.

Return type:

list[str]

pathforge.core.tasks.registry.get_task_allowed_dataset_uses(name: str) → frozenset[str] | None

Return the allowed dataset-use semantics for one task.

Parameters:

name (str)

Return type:

frozenset[str] | None

pathforge.core.tasks.registry.import_task_modules(package_name: str = 'pathforge.core.tasks') → None

Import all modules inside the core.tasks package so that decorator-based registration is executed.

Call this once before get_task(…).

Parameters:

package_name (str)

Return type:

None

class pathforge.core.tasks.base.TaskBase

Bases: ABC

Base class for benchmarking tasks.

Each concrete task should define: - which benchmark grid keys it needs - how one combo is executed

grid_keys: list[str] = []

allowed_dataset_uses: frozenset[str] | None = None

inference_dataset_uses: frozenset[str] | None = None

inference_input_use: str = 'query'

__init__(experiment: Experiment) → None

Parameters:

experiment (Experiment)

Return type:

None

classmethod get_grid_keys() → list[str]

Return type:

list[str]

classmethod get_allowed_dataset_uses() → frozenset[str] | None

Return type:

frozenset[str] | None

classmethod get_inference_grid_keys() → list[str]

Return type:

list[str]

classmethod get_inference_dataset_uses() → frozenset[str] | None

Return type:

frozenset[str] | None

classmethod get_inference_input_use() → str

Return type:

str

abstractmethod execute(combo_cfg: ComboConfig, datasets_by_use: dict[str, list[BagDataset]]) → dict[str, Any]

Execute one benchmark run for one combo.

Parameters:

• combo_cfg (ComboConfig) – Active benchmark combination.

• datasets_by_use (dict[str, list[BagDataset]]) – dictionary of datasets grouped by their use case.

Returns:

Dictionary with results / metrics / metadata.

Return type:

dict[str, Any]

inference(combo_cfg: ComboConfig, datasets_by_use: dict[str, list[BagDataset]], inference_run_root: Path) → dict[str, Any]

Execute one inference run for one combo.

Parameters:

• combo_cfg (ComboConfig) – Active inference combination.

• datasets_by_use (dict[str, list[BagDataset]]) – dictionary of datasets grouped by task-specific use.

• inference_run_root (Path) – timestamped root folder for the current inference CLI invocation.

Returns:

Dictionary with output paths / counts / metadata.

Return type:

dict[str, Any]

MIL Tasks

class pathforge.core.tasks.mil.ClassificationMilTask

Bases: _BaseMilTask

Standard MIL classification benchmarking task.

class pathforge.core.tasks.mil.RegressionMilTask

Bases: _BaseMilTask

Standard MIL regression benchmarking task.

class pathforge.core.tasks.mil.SurvivalMilTask

Bases: _BaseMilTask

Standard MIL continuous-survival benchmarking task.

class pathforge.core.tasks.mil.SurvivalDiscreteMilTask

Bases: _BaseMilTask

Standard MIL discrete-survival benchmarking task.

Slide Retrieval Task

class pathforge.core.tasks.slide_retrieval.SlideRetrievalTask

Bases: TaskBase

Run slide retrieval from bag-level features.

grid_keys: list[str] = ['tile_px', 'tile_mpp', 'feature_extraction', 'color_norm', 'retrieval_representation', 'search_strategy']

allowed_dataset_uses: frozenset[str] | None = frozenset({'query', 'query_reference', 'reference'})

inference_dataset_uses: frozenset[str] | None = frozenset({'query_reference', 'reference'})

inference_input_use: str = 'query'

execute(combo_cfg: ComboConfig, datasets_by_use: dict[str, list[BagDataset]]) → dict[str, Any]

Execute one benchmark run for one combo.

Parameters:

• combo_cfg (ComboConfig) – Active benchmark combination.

• datasets_by_use (dict[str, list[BagDataset]]) – dictionary of datasets grouped by their use case.

Returns:

Dictionary with results / metrics / metadata.

Return type:

dict[str, Any]

inference(combo_cfg: ComboConfig, datasets_by_use: dict[str, list[BagDataset]], inference_run_root: Path) → dict[str, Any]

Execute one inference run for one combo.

Parameters:

• combo_cfg (ComboConfig) – Active inference combination.

• datasets_by_use (dict[str, list[BagDataset]]) – dictionary of datasets grouped by task-specific use.

• inference_run_root (Path) – timestamped root folder for the current inference CLI invocation.

Returns:

Dictionary with output paths / counts / metadata.

Return type:

dict[str, Any]

compute_retrieval_representations(*, bag_dataset: SlideRetrievalBagDataset, retrieval_loader: DataLoader[list[SlideRetrievalDatasetItem]], batch_thread_workers: int, combo_cfg: ComboConfig, representation_strategy: Any, representation_id: str, aggregation_level: str, exclusion_level: Literal['none', 'slide', 'case', 'patient']) → tuple[list[RetrievalRepresentation], dict[str, str]]

Create and persist retrieval representations for one retrieval dataset.

Parameters:

• bag_dataset (SlideRetrievalBagDataset) – Dataset that owns the target samples.

• retrieval_loader (DataLoader[list[SlideRetrievalDatasetItem]]) – Loader yielding retrieval batches to materialize.

• batch_thread_workers (int) – Threads used per retrieval batch.

• combo_cfg (ComboConfig) – Active representation configuration.

• representation_strategy (Any) – Instantiated representation strategy.

• representation_id (str) – Stable representation artifact key.

• aggregation_level (str) – Active experiment aggregation level.

• exclusion_level (Literal['none', 'slide', 'case', 'patient']) – Configured exclusion key level.

Returns:

Created representations and failure details keyed by sample ID.

Return type:

tuple[list[RetrievalRepresentation], dict[str, str]]