Registry and method contract¶

The registry resolves a configuration's method name to a BaseMethod subclass. BaseMethod defines the lifecycle used by both training and evaluation.

methods ¶

Method registry.

The unified train.py calls get_method(name)(cfg) to obtain a BaseMethod instance. Adding a new entry below is enough to expose it on the command line.

get_method ¶

get_method(name: str) -> Type['BaseMethod']

Resolve a method name or supported alias to its adapter class.

Imports are intentionally lazy so registry inspection does not initialize foundation models or require every method's optional dependencies.

Parameters:

Name	Type	Description	Default
`name`	`str`	Canonical method name or a documented command-line alias.	required

Returns:

Type	Description
`Type['BaseMethod']`	The matching :class:`methods.base.BaseMethod` subclass.

Raises:

Type	Description
`KeyError`	If `name` is not registered.

Source code in methods/__init__.py

def get_method(name: str) -> Type["BaseMethod"]:
    """Resolve a method name or supported alias to its adapter class.

    Imports are intentionally lazy so registry inspection does not initialize
    foundation models or require every method's optional dependencies.

    Args:
        name: Canonical method name or a documented command-line alias.

    Returns:
        The matching :class:`methods.base.BaseMethod` subclass.

    Raises:
        KeyError: If ``name`` is not registered.
    """
    name = canonical_method_name(name)

    if name == "composite":
        from .composite.adapter import CompositeMethod
        return CompositeMethod

    if name == "focus":
        from .focus.adapter import FOCUSMethod
        return FOCUSMethod

    if name == "vila_mil":
        from .vila_mil.adapter import ViLaMILMethod
        return ViLaMILMethod

    if name == "cod_mil":
        from .cod_mil.adapter import CoDMILMethod
        return CoDMILMethod

    if name == "maple":
        from .maple.adapter import MAPLEMethod
        return MAPLEMethod

    if name == "mscpt":
        from .mscpt.adapter import MSCPTMethod
        return MSCPTMethod

    if name == "pathpt":
        from .pathpt.adapter import PathPTMethod
        return PathPTMethod

    if name == "top":
        from .top.adapter import TOPMethod
        return TOPMethod

    if name == "slip":
        from .slip.adapter import SLIPMethod
        return SLIPMethod

    if name == "wsi_five":
        from .wsi_five.adapter import WSIFiVEMethod
        return WSIFiVEMethod

    if name == "muse":
        from .muse.adapter import MUSEMethod
        return MUSEMethod

    if name == "convlm":
        from .convlm.adapter import ConVLMMethod
        return ConVLMMethod

    if name == "sldpc":
        from .sldpc.adapter import SLDPCMethod
        return SLDPCMethod

    if name == "hive_mil":
        from .hive_mil.adapter import HiVEMILMethod
        return HiVEMILMethod

    if name == "mi_visionshot":
        from .mi_visionshot.adapter import MIVisionShotMethod
        return MIVisionShotMethod

    if name == "libra_mil":
        from .libra_mil.adapter import LibraMILMethod
        return LibraMILMethod

    if name == "dyko":
        from .dyko.adapter import DyKoMethod
        return DyKoMethod

    if name == "mgpath":
        from .mgpath.adapter import MGPathMethod
        return MGPathMethod

    if name == "hipss":
        from .hipss.adapter import HIPSSMethod
        return HIPSSMethod

    raise AssertionError(f"Method registry is incomplete for {name!r}")

list_methods ¶

list_methods() -> list[str]

Return canonical method names in stable registry order.

Source code in methods/__init__.py

def list_methods() -> list[str]:
    """Return canonical method names in stable registry order."""
    return list(_CANONICAL_METHODS)

get_backbone_contracts ¶

get_backbone_contracts() -> dict[
    str, "MethodBackboneContract"
]

Return every adapter's declared encoder contract, keyed by method.

Returns:

Type	Description
`dict[str, 'MethodBackboneContract']`	A mapping from canonical registry name to its immutable
`dict[str, 'MethodBackboneContract']`	class:`common.backbones.MethodBackboneContract`.

Source code in methods/__init__.py

def get_backbone_contracts() -> dict[str, "MethodBackboneContract"]:
    """Return every adapter's declared encoder contract, keyed by method.

    Returns:
        A mapping from canonical registry name to its immutable
        :class:`common.backbones.MethodBackboneContract`.
    """
    return {name: get_method(name).get_backbone_contract()
            for name in list_methods()}

base ¶

Method registry.

Every paper in this codebase is exposed through a BaseMethod adapter that has a uniform interface. The unified train.py then dispatches to the right method by looking it up in the registry below.

Adding a new method¶

Create a folder methods/<my_method>/ with a __init__.py.
Put the method's unique model file(s) there. Re-use anything you can from common/.
Subclass BaseMethod (file methods/<my_method>/adapter.py) and implement at minimum:

build_model(self, cfg) -> nn.Module train_step(self, batch, model, optimizer, loss_fn) -> dict eval_step(self, batch, model, loss_fn) -> dict

Many methods can simply inherit CLAMScaffoldMethod (defined below) which already wires up the FOCUS/ViLa-MIL training loop. 4. Register the adapter in methods/__init__.py METHODS = {...}.

That's it -- train.py --method <my_method> --config configs/<my>.yaml will pick it up.

BaseMethod ¶

BaseMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: ABC

Define the uniform lifecycle implemented by every method adapter.

A method adapter holds the recipe for a paper: - how to instantiate the model from a config dict - how a single train step looks - how a single eval step looks

State that survives across steps (e.g. running EMA, prototypes, pseudo labels) should live as attributes on self.

Parameters:

Name	Type	Description	Default
`cfg`	`Dict[str, Any]`	Validated run configuration. Each adapter receives a private copy so method-specific defaults cannot alter the persisted run identity or leak into another cross-validation fold.	required
`device`	`str`	PyTorch device used for model parameters and input batches.	`'cuda'`

Source code in methods/base.py

def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

get_backbone_contract `classmethod` ¶

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type	Description
`NotImplementedError`	If the adapter omitted its required contract.

Source code in methods/base.py

@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder ¶

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name	Type	Description	Default
`weights_path`	`Optional[str]`	Optional local checkpoint or model identifier. When omitted, `cfg['backbone_weights']` is used.	`None`
`**loader_options`	`Any`	Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.	`{}`

Returns:

Type	Description
`EncoderBundle`	A validated capability-aware encoder bundle.

Raises:

Type	Description
`RuntimeError`	If the adapter has no runtime backbone contract.
`BackboneCompatibilityError`	If the bundle violates the method contract or declared feature provenance.

Source code in methods/base.py

def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_model `abstractmethod` ¶

build_model() -> Module

Instantiate the trainable model and move it to self.device.

Source code in methods/base.py

@abstractmethod
def build_model(self) -> nn.Module:
    """Instantiate the trainable model and move it to ``self.device``."""

train_step `abstractmethod` ¶

train_step(
    batch,
    model: Module,
    optimizer: Optimizer,
    loss_fn: Module,
) -> Dict[str, float]

Run one optimization step.

Returns:

Type	Description
`Dict[str, float]`	A mapping containing at least `loss`, `logits`, and `label`.

Source code in methods/base.py

@abstractmethod
def train_step(self, batch, model: nn.Module,
               optimizer: torch.optim.Optimizer,
               loss_fn: nn.Module) -> Dict[str, float]:
    """Run one optimization step.

    Returns:
        A mapping containing at least ``loss``, ``logits``, and ``label``.
    """

eval_step `abstractmethod` ¶

eval_step(
    batch, model: Module, loss_fn: Optional[Module] = None
) -> Dict[str, float]

Run a forward-only validation or test step.

Returns:

Type	Description
`Dict[str, float]`	A mapping containing at least `loss`, `logits`, and `label`.

Source code in methods/base.py

@abstractmethod
def eval_step(self, batch, model: nn.Module,
              loss_fn: Optional[nn.Module] = None
              ) -> Dict[str, float]:
    """Run a forward-only validation or test step.

    Returns:
        A mapping containing at least ``loss``, ``logits``, and ``label``.
    """

build_optimizer ¶

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py

def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

build_scheduler ¶

build_scheduler(optimizer)

Default: ReduceLROnPlateau(min, factor=0.1, patience=10).

Adapters override this whenever the released executable recipe uses a different scheduler or constructs one without ever stepping it.

Source code in methods/base.py

def build_scheduler(self, optimizer):
    """Default: ReduceLROnPlateau(min, factor=0.1, patience=10).

    Adapters override this whenever the released executable recipe uses a
    different scheduler or constructs one without ever stepping it.
    """
    return torch.optim.lr_scheduler.ReduceLROnPlateau(
        optimizer, "min", factor=0.1, patience=10)

validation_monitor ¶

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py

def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold ¶

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py

def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start ¶

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py

def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end ¶

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py

def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end ¶

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py

def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end ¶

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py

def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded ¶

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py

def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step ¶

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name	Type	Description	Default
`batch`	`Any`	One loader batch for a single slide.	required
`model`	`Module`	Loaded fold checkpoint in evaluation mode.	required
`target_class`	`int \| None`	Optional class index for class-specific evidence.	`None`

Raises:

Type	Description
`NotImplementedError`	If the method has no audited patch-evidence mapping in PGVL-Gym.

Source code in methods/base.py

def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

Registry and method contract¶

methods ¶

get_method ¶

list_methods ¶

get_backbone_contracts ¶

base ¶

Adding a new method¶

BaseMethod ¶

get_backbone_contract classmethod ¶

load_encoder ¶

build_model abstractmethod ¶

train_step abstractmethod ¶

eval_step abstractmethod ¶

build_optimizer ¶

build_scheduler ¶

validation_monitor ¶

prepare_fold ¶

on_train_epoch_start ¶

on_epoch_end ¶

on_validation_end ¶

on_fold_end ¶

on_checkpoint_loaded ¶

interpret_step ¶

get_backbone_contract `classmethod` ¶

build_model `abstractmethod` ¶

train_step `abstractmethod` ¶

eval_step `abstractmethod` ¶