CS-Flow#
Fully Convolutional Cross-Scale-Flows for Image-based Defect Detection.
Paper: https://arxiv.org/pdf/2110.02855.pdf
This module provides the CS-Flow model implementation for anomaly detection. CS-Flow uses normalizing flows across multiple scales to model the distribution of normal images and detect anomalies.
- class anomalib.models.image.csflow.lightning_model.Csflow(cross_conv_hidden_channels=1024, n_coupling_blocks=4, clamp=3, num_channels=3, pre_processor=True, post_processor=True, evaluator=True, visualizer=True)#
Bases:
AnomalibModule
CS-Flow Lightning Model for anomaly detection.
CS-Flow uses normalizing flows across multiple scales to model the distribution of normal images. During inference, it assigns anomaly scores based on the likelihood of test samples under the learned distribution.
- Parameters:
n_coupling_blocks (int, optional) – Number of coupling blocks in the model. Defaults to
4
.cross_conv_hidden_channels (int, optional) – Number of hidden channels in the cross convolution layer. Defaults to
1024
.clamp (int, optional) – Clamping value for the affine coupling layers in the Glow model. Defaults to
3
.num_channels (int, optional) – Number of input image channels. Defaults to
3
for RGB images.pre_processor (PreProcessor | bool, optional) – Preprocessing module or flag to enable default preprocessing. Defaults to
True
.post_processor (PostProcessor | bool, optional) – Post-processing module or flag to enable default post-processing. Defaults to
True
.evaluator (Evaluator | bool, optional) – Evaluation module or flag to enable default evaluation. Defaults to
True
.visualizer (Visualizer | bool, optional) – Visualization module or flag to enable default visualization. Defaults to
True
.
- Raises:
ValueError – If
input_size
is not provided during initialization.
Example
>>> from anomalib.models.image.csflow import Csflow >>> model = Csflow( ... n_coupling_blocks=4, ... cross_conv_hidden_channels=1024, ... clamp=3, ... num_channels=3 ... )
- configure_optimizers()#
Configure the Adam optimizer for CS-Flow.
- Returns:
- Configured Adam optimizer with specific
hyperparameters
- Return type:
Example
>>> model = Csflow() >>> optimizer = model.configure_optimizers()
- property learning_type: LearningType#
Get the learning type of the model.
- Returns:
The learning type, which is ONE_CLASS for CS-Flow
- Return type:
LearningType
- training_step(batch, *args, **kwargs)#
Perform a training step of CS-Flow model.
- Parameters:
batch (Batch) – Input batch containing images and targets
*args – Additional positional arguments (unused)
**kwargs – Additional keyword arguments (unused)
- Returns:
Dictionary containing the loss value
- Return type:
STEP_OUTPUT
Example
>>> batch = Batch(image=torch.randn(32, 3, 256, 256)) >>> model = Csflow() >>> output = model.training_step(batch) >>> output["loss"] tensor(...)
- validation_step(batch, *args, **kwargs)#
Perform a validation step of CS-Flow model.
- Parameters:
batch (Batch) – Input batch containing images and targets
*args – Additional positional arguments (unused)
**kwargs – Additional keyword arguments (unused)
- Returns:
- Dictionary containing predictions including anomaly maps
and scores
- Return type:
STEP_OUTPUT
Example
>>> batch = Batch(image=torch.randn(32, 3, 256, 256)) >>> model = Csflow() >>> predictions = model.validation_step(batch)
PyTorch model for CS-Flow implementation.
This module contains the PyTorch implementation of CS-Flow model for anomaly detection. The model uses cross-scale coupling layers to learn the distribution of normal images and detect anomalies based on the likelihood of test images under this distribution.
- The implementation is based on the paper:
CS-Flow: Learning Cross-Scale Semantic Flow for Unsupervised Anomaly Detection Marco Rudolph, Tom Wehrbein, Bodo Rosenhahn, Bastian Wandt https://arxiv.org/abs/2110.02855
- class anomalib.models.image.csflow.torch_model.CsFlowModel(input_size, cross_conv_hidden_channels, n_coupling_blocks=4, clamp=3, num_channels=3)#
Bases:
Module
CS-Flow model for anomaly detection.
This module implements the complete CS-Flow model that learns the distribution of normal images using cross-scale coupling layers.
- Parameters:
cross_conv_hidden_channels (int) – Hidden channels in cross convolutions.
n_coupling_blocks (int, optional) – Number of coupling blocks. Defaults to
4
.clamp (int, optional) – Clamping value for coupling layers. Defaults to
3
.num_channels (int, optional) – Number of input image channels. Defaults to
3
.
Example
>>> model = CsFlowModel((256, 256), 64) >>> x = torch.randn(1, 3, 256, 256) >>> output = model(x) >>> isinstance(output, InferenceBatch) True
- forward(images)#
Forward method of the model.
- Parameters:
images (torch.Tensor) – Input images.
- Returns:
- During training: tuple containing the z_distribution for three scales
and the sum of log determinant of the Jacobian. During evaluation: tuple containing anomaly maps and anomaly scores
- Return type:
Loss function for the CS-Flow Model Implementation.
This module implements the loss function used in the CS-Flow model for anomaly detection. The loss combines the squared L2 norm of the latent space representations with the log-determinant of the Jacobian from the normalizing flows.
Example
>>> import torch
>>> from anomalib.models.image.csflow.loss import CsFlowLoss
>>> criterion = CsFlowLoss()
>>> z_dist = [torch.randn(2, 64, 32, 32) for _ in range(3)]
>>> jacobians = torch.randn(2)
>>> loss = criterion(z_dist, jacobians)
- class anomalib.models.image.csflow.loss.CsFlowLoss(*args, **kwargs)#
Bases:
Module
Loss function for the CS-Flow model.
The loss is computed as the mean of the squared L2 norm of the latent space representations minus the log-determinant of the Jacobian, normalized by the dimensionality of the latent space.
- static forward(z_dist, jacobians)#
Compute the CS-Flow loss.
- Parameters:
z_dist (list[torch.Tensor]) – List of latent space tensors from each scale of the normalizing flow. Each tensor has shape
(batch_size, channels, height, width)
.jacobians (torch.Tensor) – Log-determinant of the Jacobian matrices from the normalizing flows. Shape:
(batch_size,)
.
- Returns:
Scalar loss value averaged over the batch.
- Return type:
Example
>>> z_dist = [torch.randn(2, 64, 32, 32) for _ in range(3)] >>> jacobians = torch.randn(2) >>> loss = CsFlowLoss.forward(z_dist, jacobians)
Anomaly Map Generator for CS-Flow model.
This module provides functionality to generate anomaly maps from the CS-Flow model’s outputs. The generator can operate in two modes:
ALL
- Combines anomaly scores from all scales (default)MAX
- Uses only the largest scale as mentioned in the paper
The anomaly maps are generated by computing the mean of squared z-scores across channels and upsampling to the input dimensions.
Example
>>> import torch
>>> generator = AnomalyMapGenerator(input_dims=(3, 256, 256))
>>> z_dist = [torch.randn(2, 64, 32, 32) for _ in range(3)]
>>> anomaly_map = generator(z_dist)
>>> anomaly_map.shape
torch.Size([2, 1, 256, 256])
- class anomalib.models.image.csflow.anomaly_map.AnomalyMapGenerator(input_dims, mode=AnomalyMapMode.ALL)#
Bases:
Module
Generate anomaly maps from CS-Flow model outputs.
- Parameters:
input_dims (tuple[int, int, int]) – Input dimensions in the format
(channels, height, width)
.mode (AnomalyMapMode, optional) – Mode for generating anomaly maps. Defaults to
AnomalyMapMode.ALL
.
Example
>>> generator = AnomalyMapGenerator((3, 256, 256)) >>> z_dist = [torch.randn(1, 64, 32, 32) for _ in range(3)] >>> anomaly_map = generator(z_dist)
- forward(inputs)#
Generate anomaly maps from z-distributions.
- Parameters:
inputs (torch.Tensor) – List of z-distributions from different scales, where each element has shape
(batch_size, channels, height, width)
.- Returns:
- Anomaly maps with shape ``(batch_size, 1, height,
width)``, where height and width match the input dimensions.
- Return type:
Example
>>> z_dist = [torch.randn(2, 64, 32, 32) for _ in range(3)] >>> generator = AnomalyMapGenerator((3, 256, 256)) >>> maps = generator(z_dist) >>> maps.shape torch.Size([2, 1, 256, 256])
- class anomalib.models.image.csflow.anomaly_map.AnomalyMapMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)#
-
Mode for generating anomaly maps.
The mode determines how the anomaly scores from different scales are combined:
ALL
: Combines scores from all scales by multiplicationMAX
: Uses only the score from the largest scale