class_docstr: CFLOW: Conditional Normalizing Flows.
docstr: Forward-pass images into the network to extract encoder features and compute probability.
  
  Args:
    images: Batch of images.
  
  Returns:
    Predicted anomaly maps.
code: def forward(self, images) -> Any:
      """Forward-pass images into the network to extract encoder features and compute probability.
  
      Args:
        images: Batch of images.
  
      Returns:
        Predicted anomaly maps.
  
      """
  
      self.encoder.eval()
      self.decoders.eval()
      with torch.no_grad():
          activation = self.encoder(images)
  
      distribution = [
          torch.Tensor(0).to(images.device) for _ in self.pool_layers
      ]
  
      height: list[int] = []
      width: list[int] = []
      for layer_idx, layer in enumerate(self.pool_layers):
          encoder_activations = activation[layer]  # BxCxHxW
  
          (
              batch_size,
              dim_feature_vector,
              im_height,
              im_width,
          ) = encoder_activations.size()
          image_size = im_height * im_width
          embedding_length = (
              batch_size * image_size
          )  # number of rows in the conditional vector
  
          height.append(im_height)
          width.append(im_width)
          # repeats positional encoding for the entire batch 1 C H W to B C H W
          pos_encoding = einops.repeat(
              positional_encoding_2d(
                  self.condition_vector, im_height, im_width
              ).unsqueeze(0),
              "b c h w-> (tile b) c h w",
              tile=batch_size,
          ).to(images.device)
          c_r = einops.rearrange(
              pos_encoding, "b c h w -> (b h w) c"
          )  # BHWxP
          e_r = einops.rearrange(
              encoder_activations, "b c h w -> (b h w) c"
          )  # BHWxC
          decoder = self.decoders[layer_idx].to(images.device)
  
          # Sometimes during validation, the last batch E / N is not a whole number. Hence we need to add 1.
          # It is assumed that during training that E / N is a whole number as no errors were discovered during
          # testing. In case it is observed in the future, we can use only this line and ensure that FIB is at
          # least 1 or set `drop_last` in the dataloader to drop the last non-full batch.
          fiber_batches = (
              embedding_length // self.fiber_batch_size
              + int(embedding_length % self.fiber_batch_size > 0)
          )
  
          for batch_num in range(fiber_batches):  # per-fiber processing
              if batch_num < (fiber_batches - 1):
                  idx = torch.arange(
                      batch_num * self.fiber_batch_size,
                      (batch_num + 1) * self.fiber_batch_size,
                  )
              else:  # When non-full batch is encountered batch_num+1 * N will go out of bounds
                  idx = torch.arange(
                      batch_num * self.fiber_batch_size,
                      embedding_length,
                  )
              c_p = c_r[idx]  # NxP
              e_p = e_r[idx]  # NxC
              # decoder returns the transformed variable z and the log Jacobian determinant
              with torch.no_grad():
                  p_u, log_jac_det = decoder(e_p, [c_p])
              #
              decoder_log_prob = get_logp(
                  dim_feature_vector, p_u, log_jac_det
              )
              log_prob = (
                  decoder_log_prob / dim_feature_vector
              )  # likelihood per dim
              distribution[layer_idx] = torch.cat(
                  (distribution[layer_idx], log_prob)
              )
  
      output = self.anomaly_map_generator(
          distribution=distribution, height=height, width=width
      )
      self.decoders.train()
  
      return output.to(images.device)
  
file_path: anomalib/src/anomalib/models/cflow/torch_model.py
imports: from __future__ import annotations
  from typing import Any
  import einops
  import torch
  from torch import nn
  from anomalib.models.cflow.anomaly_map import AnomalyMapGenerator
  from anomalib.models.cflow.utils import cflow_head, get_logp, positional_encoding_2d
  from anomalib.models.components import FeatureExtractor
label_desc: Use image node_id 2 for calling the CflowModel obj's underlying member method code with example usage: obj.forward(images) and returns: output