pathforge.adapters¶

Concrete integrations with optional third-party packages. All adapter imports are isolated here and in pathforge.utils.optional. The rest of PathForge is import-safe without these packages.

TorchMIL¶

Backend model adapter¶

class pathforge.adapters.torchmil.backend.MILBackendModelProtocol[source]¶

Bases: Protocol

Protocol implemented by backend MIL model adapters.

Parameters:

bag – Feature tensor shaped [B, N, D] with finite floating values.
mask – Optional boolean tensor shaped [B, N] where true means real instance.
coords – Optional coordinate tensor shaped [B, N, 2].

Returns:

Tensor or mapping containing task predictions. Classification outputs are logits shaped [B, C]; survival outputs are risk [B] or discrete hazards [B, T].

Parameters:

bag (Tensor)
mask (Tensor | None)
coords (Tensor | None)
label (Tensor | None)
loss_fn (Module | None)

Return type:

Tensor | dict[str, Any]

__init__(*args, **kwargs)¶

class pathforge.adapters.torchmil.backend.TorchMILModelSpec[source]¶

Bases: object

Factory contract for one TorchMIL model name.

Variables:

name (str) – Class or factory name in torchmil.models.
task_types (tuple[str, ...]) – PathForge tasks supported by this model specification.
required_keys (tuple[str, ...]) – Batch keys routed to the backend model. X is always required by the canonical schema.
build_kwargs (dict[str, Any]) – Default constructor kwargs merged before config kwargs.

Example

spec = TorchMILModelSpec(name="ABMIL", task_types=("classification",))
model = build_torchmil_model(spec, {"in_shape": (1024,), "out_shape": 2})

name: str¶

task_types: tuple[str, ...] = ('classification',)¶

required_keys: tuple[str, ...] = ('X',)¶

build_kwargs: dict[str, Any]¶

__init__(name: str, task_types: tuple[str, ...]=('classification', ), required_keys: tuple[str, ...]=('X', ), build_kwargs: dict[str, ~typing.Any]=<factory>) → None¶

Parameters:

name (str)
task_types (tuple[str, ...])
required_keys (tuple[str, ...])
build_kwargs (dict[str, Any])

Return type:

None

pathforge.adapters.torchmil.backend.resolve_torchmil_model_spec(model_name: str) → TorchMILModelSpec[source]¶

Resolve a TorchMIL model specification by name.

Unknown names are allowed as generic TorchMIL model class names so long as the installed torchmil.models module exposes them. This preserves the single-adapter design without requiring one PathForge class per model.

Raises:

RuntimeError – If TorchMIL is not installed.
ValueError – If the model name is empty.

Parameters:

model_name (str)

Return type:

TorchMILModelSpec

pathforge.adapters.torchmil.backend.build_torchmil_model(spec: TorchMILModelSpec, config_kwargs: dict[str, Any]) → Module[source]¶

Instantiate a TorchMIL model from one generic factory path.

Parameters:

spec (TorchMILModelSpec) – Resolved model specification.
config_kwargs (dict[str, Any]) – User-provided constructor kwargs from config.

Returns:

Instantiated TorchMIL model.

Return type:

nn.Module

Raises:

ValueError – If spec.name is not present in torchmil.models.
RuntimeError – If TorchMIL cannot be imported.

class pathforge.adapters.torchmil.backend.TorchMILBackendModel[source]¶

Bases: MILModelBase

Generic PathForge adapter for TorchMIL MIL models.

Parameters:

torchmil_model – Name of the TorchMIL model class in torchmil.models.
task – PathForge task name. classification expects logits, survival expects one risk score per bag, and survival_discrete expects per-bin outputs.
torchmil_model_kwargs – Constructor kwargs forwarded to the TorchMIL model factory. Common examples include in_shape=(1024,) and task-specific output dimensions.

Input:: forward_bag consumes bag shaped [B, N, D] with dtype float32 or another floating dtype. Optional mask is shaped [B, N] and optional coords is shaped [B, N, 2].
Output:: Tensor normalized by task: classification logits [B, C], continuous survival risk [B], or discrete survival logits/hazards [B, T].

Example

model = TorchMILBackendModel(
    torchmil_model="ABMIL",
    task="classification",
    torchmil_model_kwargs={"in_shape": (1024,), "out_shape": 2},
)
logits = model.forward_bag(torch.zeros(2, 8, 1024), mask=torch.ones(2, 8, dtype=torch.bool))

Raises:

RuntimeError – If TorchMIL is not installed.
ValueError – If the selected model/task combination is unsupported.

__init__(*, torchmil_model: str, task: str = 'classification', torchmil_model_kwargs: dict[str, Any] | None = None, **_: Any) → None[source]¶

Parameters:

torchmil_model (str)
task (str)
torchmil_model_kwargs (dict[str, Any] | None)
_ (Any)

Return type:

None

property bag_size: int | None¶: Returns fixed bag size (int) or None for variable sizes.

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).
adj (Tensor | None)

Returns:

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

Return type:

Tensor | dict[str, Any]

get_learnable_parameters() → Iterable[Parameter][source]¶

Yield all gradient-enabled parameters.

Return type:: Iterable[Parameter]

pathforge.adapters.torchmil.backend.register_torchmil_backend() → None[source]¶

Return type:: None

Collate function¶

pathforge.adapters.torchmil.collate.torchmil_or_pathforge_collate(batch: list[Any], *, use_torchmil: bool = True) → BagBatch[source]¶

Collate variable-length MIL bags with TorchMIL when available.

Parameters:

batch (list[Any]) – List of canonical single-bag dictionaries. Each X is shaped [N_i, D].
use_torchmil (bool) – If true and TorchMIL is installed, dispatch to torchmil.data.collate_fn. Otherwise use PathForge’s fallback implementation with the same padded X and mask semantics.

Returns:

Padded batch with X shaped [B, N_max, D], Y stacked along [B] where possible, and mask shaped [B, N_max] where real instances are true.

Return type:

BagBatch

Example

torch.manual_seed(7)
batch = [
    {"X": torch.ones(2, 4), "Y": torch.tensor(0)},
    {"X": torch.ones(3, 4), "Y": torch.tensor(1)},
]
out = torchmil_or_pathforge_collate(batch, use_torchmil=False)
assert out["X"].shape == (2, 3, 4)
assert out["mask"].tolist() == [[True, True, False], [True, True, True]]

pathforge.adapters.torchmil.collate.pathforge_collate(batch: list[BagBatch]) → BagBatch[source]¶

Fallback collate implementation matching TorchMIL padding semantics.

Parameters:: batch (list[BagBatch]) – Canonical single-bag dictionaries with X shaped [N_i, D].
Returns:: Padded batch. Padding values are zero, X is float32, and mask is boolean.
Return type:: BagBatch
Raises:: AssertionError – If feature dimensions differ or optional tensor shapes are incompatible.

Heatmap explainer¶

class pathforge.adapters.torchmil.heatmap_explainer.HeatMap[source]¶

Bases: object

Prediction heatmap payload produced from instance-level scores.

Variables:

coords (torch.Tensor) – Tensor shaped [N, 2] with x/y tile coordinates in pixels.
scores (torch.Tensor) – Tensor shaped [N] with finite normalized importance values in [0, 1].

coords: Tensor¶

scores: Tensor¶

__init__(coords: Tensor, scores: Tensor) → None¶

Parameters:

coords (Tensor)
scores (Tensor)

Return type:

None

class pathforge.adapters.torchmil.heatmap_explainer.TorchMILHeatmapExplainer[source]¶

Bases: ExplainerBase

Explainer adapter for TorchMIL instance scores.

Input:: explain expects a dictionary with coords shaped [N, 2] and either instance_scores, attention, or scores shaped [N]. Optional mask shaped [N] removes padded instances.
Output:: HeatMap with finite scores min-max normalized to [0, 1]. The caller can persist this under a prediction heatmap H5 namespace.

Example

explainer = TorchMILHeatmapExplainer()
heatmap = explainer.explain({
    "coords": torch.tensor([[0, 0], [256, 0]]),
    "instance_scores": torch.tensor([0.2, 0.8]),
})
assert heatmap.scores.tolist() == [0.0, 1.0]

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

Initialize the explainer from runtime configuration.

Parameters:: config (dict[str, Any])
Return type:: None

explain(input: Any) → HeatMap[source]¶

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

Parameters:: input (Any)
Return type:: HeatMap

pathforge.adapters.torchmil.heatmap_explainer.register_torchmil_heatmap_explainer() → None[source]¶

Return type:: None

Task output¶

pathforge.adapters.torchmil.task_output.normalize_torchmil_output(output: Any, *, task: Literal['classification', 'regression', 'survival', 'survival_discrete'] | str) → Tensor[source]¶

Normalize backend model outputs to PathForge loss contracts.

Parameters:

output (Any) – Tensor or mapping emitted by a TorchMIL model. Supported mapping keys are checked in order: logits, Y_hat, pred, prediction, risk, hazards, and survival.
task (Literal['classification', 'regression', 'survival', 'survival_discrete'] | str) – Current PathForge task. Classification expects logits shaped [B, C] or [B]. Continuous survival expects risk/log-hazard shaped [B] or [B, 1]. Discrete survival expects hazards or logits shaped [B, T].

Returns:

Normalized finite tensor matching PathForge task losses.

Return type:

torch.Tensor

Example

logits = normalize_torchmil_output({"logits": torch.zeros(2, 3)}, task="classification")
assert logits.shape == (2, 3)

Raises:

TypeError – If no tensor output can be extracted.
AssertionError – If shape or finite-value checks fail.

Parameters:

output (Any)
task (Literal['classification', 'regression', 'survival', 'survival_discrete'] | str)

Return type:

Tensor

Losses¶

Torch / TorchSurv loss adapters¶

Backend-specific loss adapters and registry helpers.

This module contains concrete PathForge loss adapters backed by optional or framework-provided dependencies. The domain layer should depend only on BaseLoss and the registry interface, while concrete implementations are registered from this adapter layer.

class pathforge.adapters.losses.TorchClassificationLoss[source]¶

Bases: BaseLoss

Classification loss adapter backed by one torch.nn loss module.

Parameters:

loss_ctor – torch.nn loss constructor.
target_dtype – Target dtype expected by the wrapped loss. Use "long" for class-index targets and "float" for binary/soft targets.
squeeze_binary – Whether to squeeze [B, 1] logits and targets to [B] before evaluation.
**loss_kwargs – Forwarded to loss_ctor.

Example

loss = TorchClassificationLoss(nn.CrossEntropyLoss, target_dtype="long")
value = loss(torch.zeros(2, 3), torch.tensor([0, 1]))
assert value.ndim == 0

__init__(loss_ctor: Callable[[...], Module], *, target_dtype: str, squeeze_binary: bool = False, **loss_kwargs: Any) → None[source]¶

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

Parameters:

loss_ctor (Callable[[...], Module])
target_dtype (str)
squeeze_binary (bool)
loss_kwargs (Any)

Return type:

None

forward(preds: Tensor, target: Tensor, **_: Any) → Tensor[source]¶

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)
_ (Any)

Return type:

Tensor

class pathforge.adapters.losses.TorchRegressionLoss[source]¶

Bases: BaseLoss

Regression loss adapter backed by one torch.nn loss module.

Parameters:

loss_ctor – torch.nn loss constructor.
**loss_kwargs – Forwarded to loss_ctor.

__init__(loss_ctor: Callable[[...], Module], **loss_kwargs: Any) → None[source]¶

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

Parameters:

loss_ctor (Callable[[...], Module])
loss_kwargs (Any)

Return type:

None

forward(preds: Tensor, target: Tensor, **_: Any) → Tensor[source]¶

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Parameters:

preds (Tensor)
target (Tensor)
_ (Any)

Return type:

Tensor

class pathforge.adapters.losses.TorchSurvivalLoss[source]¶

Bases: BaseLoss

Survival loss adapter backed by one torchsurv.loss function.

Parameters:

loss_fn – TorchSurv loss callable.
task_type – One of "survival" or "survival_discrete".
uses_eval_time – Whether the wrapped loss requires an eval_time grid.

Example

from torchsurv.loss.cox import neg_partial_log_likelihood

loss = TorchSurvivalLoss(neg_partial_log_likelihood, task_type="survival")
target = {"time": torch.tensor([1.0]), "event": torch.tensor([1.0])}
value = loss(torch.tensor([[0.2]]), target)
assert torch.isfinite(value)

__init__(loss_fn: Callable[[...], Tensor], *, task_type: str, uses_eval_time: bool = False) → None[source]¶

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

Parameters:

loss_fn (Callable[[...], Tensor])
task_type (str)
uses_eval_time (bool)

Return type:

None

forward(preds: Tensor, target: dict[str, Tensor], **kwargs: Any) → Tensor[source]¶

Define the computation performed at every call.

Should be overridden by all subclasses.

Note

Parameters:

preds (Tensor)
target (dict[str, Tensor])
kwargs (Any)

Return type:

Tensor

pathforge.adapters.losses.register_builtin_loss_factories(registry: Any) → None[source]¶

Populate one registry with supported torch.nn and torchsurv losses.

Parameters:: registry (Any) – PathForge registry instance used for loss resolution.
Return type:: None

Example

from pathforge.utils.registry import Registry

registry = Registry()
register_builtin_loss_factories(registry)
assert registry.get("CrossEntropyLoss") is not None

Metrics¶

Classification metrics (TorchMetrics)¶

class pathforge.adapters.metrics.classification.ClassificationMetricsBackend[source]¶

Bases: Protocol

Backend protocol for classification metrics.

Parameters:

logits – Tensor shaped [B, C] or binary logits shaped [B].
target – Integer class labels shaped [B].

Returns:

Dictionary of scalar finite metric tensors.

compute(logits: Tensor, target: Tensor) → dict[str, Tensor][source]¶

Parameters:

logits (Tensor)
target (Tensor)

Return type:

dict[str, Tensor]

__init__(*args, **kwargs)¶

class pathforge.adapters.metrics.classification.TorchMetricsClassificationBackend[source]¶

Bases: object

TorchMetrics-backed classification metrics adapter.

Example

backend = TorchMetricsClassificationBackend(num_classes=2)
metrics = backend.compute(torch.zeros(4, 2), torch.tensor([0, 1, 0, 1]))
assert "accuracy" in metrics

__init__(*, num_classes: int, task: str | None = None) → None[source]¶

Parameters:

num_classes (int)
task (str | None)

Return type:

None

compute(logits: Tensor, target: Tensor) → dict[str, Tensor][source]¶

Parameters:

logits (Tensor)
target (Tensor)

Return type:

dict[str, Tensor]

Survival metrics (TorchSurv)¶

class pathforge.adapters.metrics.survival.SurvivalMetricsBackend[source]¶

Bases: Protocol

Backend protocol for continuous survival metrics.

Parameters:

risk – Finite risk/log-hazard tensor shaped [B].
time – Event/censoring time tensor shaped [B] in study time units.
event – Binary event indicator shaped [B] where one means observed event and zero means censored.

concordance_index(risk: Tensor, time: Tensor, event: Tensor) → Tensor[source]¶

Parameters:

risk (Tensor)
time (Tensor)
event (Tensor)

Return type:

Tensor

__init__(*args, **kwargs)¶

class pathforge.adapters.metrics.survival.SurvivalLossBackend[source]¶

Bases: Protocol

Backend protocol for continuous survival losses.

neg_partial_log_likelihood(risk: Tensor, time: Tensor, event: Tensor) → Tensor[source]¶

Parameters:

risk (Tensor)
time (Tensor)
event (Tensor)

Return type:

Tensor

__init__(*args, **kwargs)¶

class pathforge.adapters.metrics.survival.TorchSurvBackend[source]¶

Bases: object

TorchSurv-backed survival metrics and losses adapter.

Example

backend = TorchSurvBackend()
risk = torch.tensor([0.1, -0.2])
time = torch.tensor([5.0, 8.0])
event = torch.tensor([1.0, 0.0])
loss = backend.neg_partial_log_likelihood(risk, time, event)

__init__() → None[source]¶

Return type:: None

concordance_index(risk: Tensor, time: Tensor, event: Tensor) → Tensor[source]¶

Parameters:

risk (Tensor)
time (Tensor)
event (Tensor)

Return type:

Tensor

neg_partial_log_likelihood(risk: Tensor, time: Tensor, event: Tensor) → Tensor[source]¶

Parameters:

risk (Tensor)
time (Tensor)
event (Tensor)

Return type:

Tensor

MIL-Lab¶

class pathforge.adapters.mil_lab.backend.MILLabModelSpec[source]¶

Bases: object

Factory contract for one MIL-Lab model name.

name: str¶

task_types: tuple[str, ...] = ('classification',)¶

build_kwargs: dict[str, Any]¶

__init__(name: str, task_types: tuple[str, ...]=('classification', ), build_kwargs: dict[str, ~typing.Any]=<factory>) → None¶

Parameters:

name (str)
task_types (tuple[str, ...])
build_kwargs (dict[str, Any])

Return type:

None

pathforge.adapters.mil_lab.backend.build_mil_lab_model(spec: MILLabModelSpec, config_kwargs: dict[str, Any], *, from_pretrained: bool = False) → Module[source]¶

Instantiate a MIL-Lab model from one generic factory path.

Parameters:

spec (MILLabModelSpec)
config_kwargs (dict[str, Any])
from_pretrained (bool)

Return type:

Module

pathforge.adapters.mil_lab.backend.normalize_mil_lab_output(output: Any) → Tensor | dict[str, Any][source]¶

Normalize MIL-Lab forward output to PathForge expectations.

Parameters:: output (Any)
Return type:: Tensor | dict[str, Any]

class pathforge.adapters.mil_lab.backend.MILLabBackendModel[source]¶

Bases: MILModelBase

Generic PathForge adapter for MIL-Lab MIL models.

__init__(*, mil_lab_model: str, task: str = 'classification', mil_lab_model_kwargs: dict[str, Any] | None = None, mil_lab_from_pretrained: bool = False, **_: Any) → None[source]¶

Parameters:

mil_lab_model (str)
task (str)
mil_lab_model_kwargs (dict[str, Any] | None)
mil_lab_from_pretrained (bool)
_ (Any)

Return type:

None

property bag_size: int | None¶: Returns fixed bag size (int) or None for variable sizes.

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).
adj (Tensor | None)

Returns:

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

Return type:

Tensor | dict[str, Any]

get_learnable_parameters() → Iterable[Parameter][source]¶

Yield all gradient-enabled parameters.

Return type:: Iterable[Parameter]

pathforge.adapters.mil_lab.backend.register_mil_lab_backend() → None[source]¶

Return type:: None

pathforge.adapters.mil_lab.backend.register_torchmil_fallback_aliases() → None[source]¶

Return type:: None

TCGA Tools¶

pathforge.adapters.tcga_tools.resolve_external_dataset_sources(config_data: dict[str, Any], *, config_path: Path) → dict[str, Any][source]¶

Resolve TCGA/TCIA dataset declarations into local PathForge datasets.

Parameters:

config_data (dict[str, Any]) – Mutable YAML payload parsed from a PathForge config file.
config_path (Path) – Absolute path to the YAML file being loaded.

Returns:

Config data with external dataset declarations replaced by concrete local dataset entries and a generated annotation file path.

Return type:

dict[str, Any]