C-Flow#
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, optional) – Name of the backbone CNN network. Defaults to
"wide_resnet50_2"
.layers (Sequence[str], optional) – List of layer names to extract features from. Defaults to
("layer2", "layer3", "layer4")
.pre_trained (bool, optional) – If True, use pre-trained weights for the backbone. Defaults to
True
.fiber_batch_size (int, optional) – Batch size for processing individual fibers. Defaults to
64
.decoder (str, optional) – Type of normalizing flow decoder to use. Defaults to
"freia-cflow"
.condition_vector (int, optional) – Dimension of the condition vector. Defaults to
128
.coupling_blocks (int, optional) – Number of coupling blocks in the flow. Defaults to
8
.clamp_alpha (float, optional) – Clamping value for the alpha parameter in flows. Defaults to
1.9
.permute_soft (bool, optional) – If True, use soft permutation in flows. Defaults to
False
.lr (float, optional) – Learning rate for the optimizer. Defaults to
0.0001
.pre_processor (PreProcessor | bool, optional) – Pre-processing module. Defaults to
True
.post_processor (PostProcessor | bool, optional) – Post-processing module. Defaults to
True
.evaluator (Evaluator | bool, optional) – Evaluation module. Defaults to
True
.visualizer (Visualizer | bool, optional) – 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
- 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:
STEP_OUTPUT
- 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:
STEP_OUTPUT
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, optional) – Whether to use pre-trained weights for the backbone. Defaults to
True
.fiber_batch_size (int, optional) – Batch size for processing feature fibers. Defaults to
64
.decoder (str, optional) – Type of decoder architecture to use. Defaults to
"freia-cflow"
.condition_vector (int, optional) – Size of the condition vector for the normalizing flows. Defaults to
128
.coupling_blocks (int, optional) – Number of coupling blocks in the normalizing flows. Defaults to
8
.clamp_alpha (float, optional) – Clamping value for the alpha parameter in the flows. Defaults to
1.9
.permute_soft (bool, optional) – 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 (torch.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:
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[torch.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] | torch.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:
- 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: