ahcore.callbacks package

Submodules

ahcore.callbacks.h5_callback module

class ahcore.callbacks.h5_callback.WriteH5Callback(max_queue_size: int, max_concurrent_writers: int, dump_dir: Path, normalization_type: str = NormalizationType.LOGITS, precision: str = InferencePrecision.FP32)

Bases: Callback

Callback to write predictions to H5 files. This callback is used to write whole-slide predictions to single H5 files in a separate thread.

TODO:

• Add support for distributed data parallel

Parameters:

max_queue_sizeint

The maximum number of items to store in the queue (i.e. tiles).

max_concurrent_writersint

The maximum number of concurrent writers.

dump_dirpathlib.Path

The directory to dump the H5 files to.

normalization_typestr

The normalization type to use for the predictions. One of “sigmoid”, “softmax” or “logits”.

precisionstr

The precision to use for the predictions. One of “float16”, “float32” or “uint8”.

property dump_dir: Path

static generator(queue: Queue[GenericArray | None]) → Generator[GenericArray, None, None]

on_predict_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the predict batch ends.

on_predict_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the predict epoch ends.

on_validation_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the validation batch ends.

on_validation_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch ends.

property writers: dict[str, ahcore.callbacks.h5_callback._WriterMessage]

ahcore.callbacks.tiff_callback module

class ahcore.callbacks.tiff_callback.WriteTiffCallback(max_concurrent_writers: int, tile_size: tuple[int, int] = (1024, 1024), colormap: dict[int, str] | None = None)

Bases: Callback

property dump_dir: Path | None

on_predict_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the predict batch ends.

on_predict_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the predict epoch ends.

on_validation_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the validation batch ends.

on_validation_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch ends.

setup(trainer: Trainer, pl_module: LightningModule, stage: str | None = None) → None

Called when fit, validate, test, predict, or tune begins.

ahcore.callbacks.wsi_metric_callback module

class ahcore.callbacks.wsi_metric_callback.ComputeWsiMetricsCallback(max_processes: int = 10, save_per_image: bool = True)

Bases: Callback

Callback to compute metrics on whole-slide images. This callback is used to compute metrics on whole-slide images in separate processes.

Parameters:

max_processesint

The maximum number of concurrent processes.

compute_metrics(trainer: Trainer, pl_module: LightningModule) → list[list[dict[str, dict[str, float]]]]

on_validation_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the validation batch ends.

on_validation_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch ends.

on_validation_epoch_start(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch begins.

setup(trainer: Trainer, pl_module: LightningModule, stage: str | None = None) → None

Called when fit, validate, test, predict, or tune begins.

class ahcore.callbacks.wsi_metric_callback.TaskData(filename, h5_filename, metadata, mask, annotations)

Bases: tuple

Create new instance of TaskData(filename, h5_filename, metadata, mask, annotations)

annotations

Alias for field number 4

filename

Alias for field number 0

h5_filename

Alias for field number 1

mask

Alias for field number 3

metadata

Alias for field number 2

ahcore.callbacks.wsi_metric_callback.compute_metrics_for_case(task_data: TaskData, class_names: dict[int, str], data_description: DataDescription, wsi_metrics: WSIMetricFactory, save_per_image: bool) → list[dict[str, Any]]

ahcore.callbacks.wsi_metric_callback.prepare_task_data(filename: Path, dump_dir: Path, pl_module: LightningModule, data_description: DataDescription, data_manager: DataManager) → TaskData

ahcore.callbacks.wsi_metric_callback.schedule_task(task_data: TaskData, pool: Pool, results_dict: dict[Any, str], class_names: dict[int, str], data_description: DataDescription, wsi_metrics: WSIMetricFactory, save_per_image: bool) → None

Module contents

Ahcore’s callbacks

class ahcore.callbacks.ComputeWsiMetricsCallback(max_processes: int = 10, save_per_image: bool = True)

Bases: Callback

Callback to compute metrics on whole-slide images. This callback is used to compute metrics on whole-slide images in separate processes.

Parameters:

max_processesint

The maximum number of concurrent processes.

compute_metrics(trainer: Trainer, pl_module: LightningModule) → list[list[dict[str, dict[str, float]]]]

on_validation_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the validation batch ends.

on_validation_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch ends.

on_validation_epoch_start(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch begins.

setup(trainer: Trainer, pl_module: LightningModule, stage: str | None = None) → None

Called when fit, validate, test, predict, or tune begins.

class ahcore.callbacks.WriteH5Callback(max_queue_size: int, max_concurrent_writers: int, dump_dir: Path, normalization_type: str = NormalizationType.LOGITS, precision: str = InferencePrecision.FP32)

Bases: Callback

Callback to write predictions to H5 files. This callback is used to write whole-slide predictions to single H5 files in a separate thread.

TODO:

• Add support for distributed data parallel

Parameters:

max_queue_sizeint

The maximum number of items to store in the queue (i.e. tiles).

max_concurrent_writersint

The maximum number of concurrent writers.

dump_dirpathlib.Path

The directory to dump the H5 files to.

normalization_typestr

The normalization type to use for the predictions. One of “sigmoid”, “softmax” or “logits”.

precisionstr

The precision to use for the predictions. One of “float16”, “float32” or “uint8”.

property dump_dir: Path

static generator(queue: Queue[GenericArray | None]) → Generator[GenericArray, None, None]

on_predict_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the predict batch ends.

on_predict_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the predict epoch ends.

on_validation_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the validation batch ends.

on_validation_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch ends.

property writers: dict[str, ahcore.callbacks.h5_callback._WriterMessage]

class ahcore.callbacks.WriteTiffCallback(max_concurrent_writers: int, tile_size: tuple[int, int] = (1024, 1024), colormap: dict[int, str] | None = None)

Bases: Callback

property dump_dir: Path | None

on_predict_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the predict batch ends.

on_predict_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the predict epoch ends.

on_validation_batch_end(trainer: Trainer, pl_module: LightningModule, outputs: Any, batch: Any, batch_idx: int, dataloader_idx: int = 0) → None

Called when the validation batch ends.

on_validation_epoch_end(trainer: Trainer, pl_module: LightningModule) → None

Called when the val epoch ends.

setup(trainer: Trainer, pl_module: LightningModule, stage: str | None = None) → None

Called when fit, validate, test, predict, or tune begins.