Pre-processing

Pre-processing module for anomaly detection pipelines.

This module provides functionality for pre-processing data before model training and inference through the PreProcessor class.

The pre-processor handles:

• Applying transforms to data during different pipeline stages

• Managing stage-specific transforms (train/val/test)

• Integrating with both PyTorch and Lightning workflows

Example

>>> from anomalib.pre_processing import PreProcessor
>>> from torchvision.transforms.v2 import Resize
>>> pre_processor = PreProcessor(transform=Resize(size=(256, 256)))
>>> transformed_batch = pre_processor(batch)

The pre-processor is implemented as both a torch.nn.Module and lightning.pytorch.Callback to support both inference and training workflows.

class anomalib.pre_processing.PreProcessor(transform=None)

Bases: Module, Callback

Anomalib pre-processor.

This class serves as both a PyTorch module and a Lightning callback, handling the application of transforms to data batches as a pre-processing step.

Parameters:

transform (Transform | None) – Transform to apply to the data before passing it to the model.

Note

.transform may be any torchvision Transform, but only a closed set of deterministic transforms (see anomalib.pre_processing.utils._transform_registry) can be persisted in a checkpoint. Randomized transforms such as RandomHorizontalFlip or RandomRotation belong in dataset augmentations (train_augmentations etc.), not in a model’s pre-processor; see the Transforms guide for why mixing the two is a common pitfall. An unsupported transform does not raise at construction; instead, checkpoint_config degrades gracefully (with a warning) so training is not interrupted, but the transform itself is not restored on reload.

Example

>>> from torchvision.transforms.v2 import Compose, Resize, ToTensor
>>> from anomalib.pre_processing import PreProcessor

>>> # Define a custom set of transforms
>>> transform = Compose([Resize((224, 224)), Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])])

>>> # Pass the custom set of transforms to a model
>>> pre_processor = PreProcessor(transform=transform)
>>> model = MyModel(pre_processor=pre_processor)

>>> # Advanced use: configure the default pre-processing behaviour of a Lightning module
>>> class MyModel(LightningModule):
...     def __init__(self):
...         super().__init__()
...         ...
...
...     def configure_pre_processor(self):
...         transform = Compose([
...             Resize((224, 224)),
...             Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
...         ])
...         return PreProcessor(transform)
...

checkpoint_config()

Get plain-data configuration to persist in a checkpoint.

Uses getattr rather than self.transform directly so that a subclass which does not call super().__init__() (and therefore has no transform attribute) can still be checkpointed safely, instead of raising AttributeError during on_save_checkpoint.

.transform may hold a transform outside the safe-serialization registry (e.g. RandomHorizontalFlip, which anomalib’s own docs warn against using as a model-specific transform rather than a dataset augmentation). Rather than aborting the whole checkpoint save, this degrades gracefully: the transform is not persisted, a warning is logged, and the checkpoint still saves. The transform is also not restored on reload; see load_checkpoint_config().

Override this method (together with load_checkpoint_config()) in a subclass that manages additional or different transform state, for example separate per-stage transforms.

Returns:

Plain-data configuration.

Return type:

dict[str, Any]

forward(batch)

Apply transforms to the batch of tensors for inference.

This forward-pass is only used after the model is exported. Within the Lightning training/validation/testing loops, the transforms are applied in the on_*_batch_start methods.

Parameters:

batch (Tensor) – Input batch to transform.

Returns:

Transformed batch.

Return type:

Tensor

load_checkpoint_config(config)

Restore configuration previously returned by checkpoint_config().

Parameters:

config (dict[str, Any]) – Plain-data configuration to restore.

Return type:

None

on_predict_batch_start(trainer, pl_module, batch, batch_idx, dataloader_idx=0)

Apply transforms to the batch of tensors during prediction.

Return type:

None

on_test_batch_start(trainer, pl_module, batch, batch_idx, dataloader_idx=0)

Apply transforms to the batch of tensors during testing.

Return type:

None

on_train_batch_start(trainer, pl_module, batch, batch_idx)

Apply transforms to the batch of tensors during training.

Return type:

None

on_validation_batch_start(trainer, pl_module, batch, batch_idx)

Apply transforms to the batch of tensors during validation.

Return type:

None