CS-Flow

CS-Flow#

Architecture#

CS-Flow Architecture CS-Flow Coupling Block Architecture CS-Flow Cross-Scale Architecture

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) – Number of coupling blocks in the model. Defaults to 4.

  • cross_conv_hidden_channels (int) – Number of hidden channels in the cross convolution layer. Defaults to 1024.

  • clamp (int) – Clamping value for the affine coupling layers in the Glow model. Defaults to 3.

  • num_channels (int) – Number of input image channels. Defaults to 3 for RGB images.

  • pre_processor (PreProcessor | bool) – Preprocessing module or flag to enable default preprocessing. Defaults to True.

  • post_processor (PostProcessor | bool) – Post-processing module or flag to enable default post-processing. Defaults to True.

  • evaluator (Evaluator | bool) – Evaluation module or flag to enable default evaluation. Defaults to True.

  • visualizer (Visualizer | bool) – 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:

Optimizer

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

property trainer_arguments: dict[str, Any]#

Get CS-Flow-specific trainer arguments.

Returns:

Dictionary containing trainer arguments:

  • gradient_clip_val: Maximum gradient norm for clipping

  • num_sanity_val_steps: Number of validation steps to run before training

Return type:

dict[str, Any]

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:

Union[Tensor, Mapping[str, Any], None]

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:

Union[Tensor, Mapping[str, Any], None]

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:

  • input_size (tuple[int, int]) – Input image size (H, W).

  • cross_conv_hidden_channels (int) – Hidden channels in cross convolutions.

  • n_coupling_blocks (int) – Number of coupling blocks. Defaults to 4.

  • clamp (int) – Clamping value for coupling layers. Defaults to 3.

  • num_channels (int) – 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 (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:

tuple[Tensor, Tensor] | InferenceBatch

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[Tensor]) – List of latent space tensors from each scale of the normalizing flow. Each tensor has shape (batch_size, channels, height, width).

  • jacobians (Tensor) – Log-determinant of the Jacobian matrices from the normalizing flows. Shape: (batch_size,).

Returns:

Scalar loss value averaged over the batch.

Return type:

Tensor

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:

  1. ALL - Combines anomaly scores from all scales (default)

  2. 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) – 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 (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:

Tensor

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(*values)#

Bases: str, Enum

Mode for generating anomaly maps.

The mode determines how the anomaly scores from different scales are combined:

  • ALL: Combines scores from all scales by multiplication

  • MAX: Uses only the score from the largest scale