C-Flow

C-Flow#

Architecture#

C-Flow Architecture

CFLOW - Real-Time Unsupervised Anomaly Detection via Conditional Normalizing Flows.

This module implements the CFLOW model for anomaly detection. CFLOW uses conditional normalizing flows to model the distribution of normal data and detect anomalies in real-time.

The model consists of:

  • A CNN backbone encoder to extract features

  • Multiple decoders using normalizing flows to model feature distributions

  • Positional encoding to capture spatial information

Paper: Real-Time Unsupervised Anomaly Detection via Conditional Normalizing Flows

class anomalib.models.image.cflow.lightning_model.Cflow(backbone='wide_resnet50_2', layers=('layer2', 'layer3', 'layer4'), pre_trained=True, fiber_batch_size=64, decoder='freia-cflow', condition_vector=128, coupling_blocks=8, clamp_alpha=1.9, permute_soft=False, lr=0.0001, pre_processor=True, post_processor=True, evaluator=True, visualizer=True)#

Bases: AnomalibModule

PyTorch Lightning implementation of the CFLOW model.

The model uses a pre-trained CNN backbone to extract features, followed by conditional normalizing flow decoders to model the distribution of normal data.

Parameters:

  • backbone (str) – Name of the backbone CNN network. Defaults to "wide_resnet50_2".

  • layers (Sequence[str]) – List of layer names to extract features from. Defaults to ("layer2", "layer3", "layer4").

  • pre_trained (bool) – If True, use pre-trained weights for the backbone. Defaults to True.

  • fiber_batch_size (int) – Batch size for processing individual fibers. Defaults to 64.

  • decoder (str) – Type of normalizing flow decoder to use. Defaults to "freia-cflow".

  • condition_vector (int) – Dimension of the condition vector. Defaults to 128.

  • coupling_blocks (int) – Number of coupling blocks in the flow. Defaults to 8.

  • clamp_alpha (float) – Clamping value for the alpha parameter in flows. Defaults to 1.9.

  • permute_soft (bool) – If True, use soft permutation in flows. Defaults to False.

  • lr (float) – Learning rate for the optimizer. Defaults to 0.0001.

  • pre_processor (PreProcessor | bool) – Pre-processing module. Defaults to True.

  • post_processor (PostProcessor | bool) – Post-processing module. Defaults to True.

  • evaluator (Evaluator | bool) – Evaluation module. Defaults to True.

  • visualizer (Visualizer | bool) – Visualization module. Defaults to True.

configure_optimizers()#

Configure optimizers for each decoder.

Creates an Adam optimizer for all decoder parameters with the specified learning rate.

Returns:

Adam optimizer instance configured for the decoders.

Return type:

Optimizer

property learning_type: LearningType#

Get the learning type of the model.

Returns:

ONE_CLASS learning type

Return type:

LearningType

property trainer_arguments: dict[str, Any]#

Get CFLOW-specific trainer arguments.

Returns:

Dictionary containing trainer arguments:

  • gradient_clip_val: 0

  • num_sanity_val_steps: 0

Return type:

dict[str, Any]

training_step(batch, *args, **kwargs)#

Perform a training step of the CFLOW model.

The training process involves: 1. Extract features using the encoder 2. Process features in fiber batches 3. Apply positional encoding 4. Train decoders using normalizing flows

Parameters:

  • batch (Batch) – Input batch containing images

  • *args – Additional arguments (unused)

  • **kwargs – Additional keyword arguments (unused)

Returns:

Dictionary containing the average loss for the batch

Return type:

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

Raises:

ValueError – If the fiber batch size is too large for the input size

validation_step(batch, *args, **kwargs)#

Perform a validation step of the CFLOW model.

The validation process: 1. Extracts features using the encoder 2. Computes anomaly maps using the trained decoders 3. Updates the batch with predictions

Parameters:

  • batch (Batch) – Input batch containing images

  • *args – Additional arguments (unused)

  • **kwargs – Additional keyword arguments (unused)

Returns:

Batch updated with model predictions

Return type:

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

PyTorch model for the CFLOW anomaly detection model.

This module provides the PyTorch implementation of the CFLOW model for anomaly detection. The model uses conditional normalizing flows to model the distribution of normal data in the feature space.

The model consists of:

  • A CNN backbone encoder to extract features

  • Multiple decoders using normalizing flows to model feature distributions

  • Positional encoding to capture spatial information

Example

>>> import torch
>>> from anomalib.models.image.cflow.torch_model import CflowModel
>>> # Initialize the model
>>> model = CflowModel(
...     backbone="resnet18",
...     layers=["layer1", "layer2", "layer3"],
...     fiber_batch_size=64,
...     decoder="freia-cflow",
...     condition_vector=128,
...     coupling_blocks=8,
...     clamp_alpha=1.9,
...     permute_soft=False
... )
>>> # Forward pass
>>> x = torch.randn(32, 3, 256, 256)
>>> predictions = model(x)
class anomalib.models.image.cflow.torch_model.CflowModel(backbone, layers, pre_trained=True, fiber_batch_size=64, decoder='freia-cflow', condition_vector=128, coupling_blocks=8, clamp_alpha=1.9, permute_soft=False)#

Bases: Module

CFLOW: Conditional Normalizing Flows.

Parameters:

  • backbone (str) – Name of the backbone CNN network to use as feature extractor.

  • layers (Sequence[str]) – Names of layers from which to extract features.

  • pre_trained (bool) – Whether to use pre-trained weights for the backbone. Defaults to True.

  • fiber_batch_size (int) – Batch size for processing feature fibers. Defaults to 64.

  • decoder (str) – Type of decoder architecture to use. Defaults to "freia-cflow".

  • condition_vector (int) – Size of the condition vector for the normalizing flows. Defaults to 128.

  • coupling_blocks (int) – Number of coupling blocks in the normalizing flows. Defaults to 8.

  • clamp_alpha (float) – Clamping value for the alpha parameter in the flows. Defaults to 1.9.

  • permute_soft (bool) – Whether to use soft permutation in the flows. Defaults to False.

Example

>>> model = CflowModel(
...     backbone="resnet18",
...     layers=["layer1", "layer2", "layer3"]
... )
>>> x = torch.randn(32, 3, 256, 256)
>>> predictions = model(x)
forward(images)#

Forward pass through the model.

The method extracts features using the encoder, processes them through normalizing flows, and generates anomaly predictions.

Parameters:

images (Tensor) – Input images of shape (batch_size, channels, height, width).

Returns:

Batch containing predicted anomaly scores and maps.

The anomaly maps have shape (batch_size, 1, height, width).

Return type:

InferenceBatch

Example

>>> x = torch.randn(32, 3, 256, 256)
>>> model = CflowModel(backbone="resnet18", layers=["layer1"])
>>> predictions = model(x)
>>> predictions.anomaly_map.shape
torch.Size([32, 1, 256, 256])

Anomaly Map Generator for CFlow model implementation.

This module provides the anomaly map generation functionality for the CFlow model. The generator takes feature distributions from multiple layers and combines them into a single anomaly heatmap.

Example

>>> from anomalib.models.image.cflow.anomaly_map import AnomalyMapGenerator
>>> import torch
>>> # Initialize generator
>>> pool_layers = ["layer1", "layer2", "layer3"]
>>> generator = AnomalyMapGenerator(pool_layers=pool_layers)
>>> # Generate anomaly map
>>> distribution = [torch.randn(32, 64) for _ in range(3)]
>>> height = [32, 16, 8]
>>> width = [32, 16, 8]
>>> anomaly_map = generator(
...     distribution=distribution,
...     height=height,
...     width=width
... )
class anomalib.models.image.cflow.anomaly_map.AnomalyMapGenerator(pool_layers)#

Bases: Module

Generate anomaly heatmap from layer-wise feature distributions.

The generator combines likelihood estimations from multiple feature layers into a single anomaly heatmap by upsampling and aggregating the scores.

Parameters:

pool_layers (Sequence[str]) – Names of pooling layers from which to extract features.

Example

>>> pool_layers = ["layer1", "layer2", "layer3"]
>>> generator = AnomalyMapGenerator(pool_layers=pool_layers)
>>> distribution = [torch.randn(32, 64) for _ in range(3)]
>>> height = [32, 16, 8]
>>> width = [32, 16, 8]
>>> anomaly_map = generator(
...     distribution=distribution,
...     height=height,
...     width=width
... )
compute_anomaly_map(distribution, height, width, image_size)#

Compute anomaly map from layer-wise likelihood distributions.

The method normalizes likelihood scores from each layer, upsamples them to a common size, and combines them into a final anomaly map.

Parameters:

  • distribution (list[Tensor]) – List of likelihood distributions for each layer.

  • height (list[int]) – List of feature map heights for each layer.

  • width (list[int]) – List of feature map widths for each layer.

  • image_size (tuple[int, int] | Size | None) – Target size for the output anomaly map. If None, keeps the original size.

Returns:

Anomaly map tensor where higher values indicate higher

likelihood of anomaly.

Return type:

Tensor

forward(**kwargs)#

Generate anomaly map from input feature distributions.

The method expects keyword arguments containing the feature distributions and corresponding spatial dimensions.

Parameters:

**kwargs (list[Tensor] | list[int] | list[list]) –

Keyword arguments containing: - distribution (list[torch.Tensor]): Feature distributions - height (list[int]): Feature map heights - width (list[int]): Feature map widths - image_size (tuple[int, int] | torch.Size | None, optional):

Target output size

Example

>>> generator = AnomalyMapGenerator(pool_layers=["layer1", "layer2"])
>>> distribution = [torch.randn(32, 64) for _ in range(2)]
>>> height = [32, 16]
>>> width = [32, 16]
>>> anomaly_map = generator(
...     distribution=distribution,
...     height=height,
...     width=width
... )
Raises:

KeyError – If required arguments distribution, height or width are missing.

Returns:

Generated anomaly map.

Return type:

Tensor