code: def convolution_invariant_target(
      estimate: torch.Tensor,
      target: torch.Tensor,
      input_length: Optional[torch.Tensor] = None,
      mask: Optional[torch.Tensor] = None,
      filter_length: int = 512,
      diag_reg: float = 1e-6,
      eps: float = 1e-8,
  ) -> torch.Tensor:
      """Calculate optimal convolution-invariant target for a given estimate.
      Assumes time dimension is the last dimension in the array.
  
      Calculate target filtered with a linear f obtained by solving
  
          min_filter || conv(filter, target) - estimate ||^2
  
      for each example in batch and each channel (b, c).
  
      Args:
          estimate: tensor, shape (B, C, T)
          target: tensor, shape (B, C, T)
          input_length: optional, length of valid samples, shape (B,)
          mask: optional, mask for input samples, shape (B, T)
          filter_length: length of the (convolutional) filter for target
          diag_reg: relative diagonal regularization for the linear system
          eps: absolute regularization for the diagonal
  
      Returns:
          Filtered target, shape (B, C, T)
  
      Reference:
          C. Boeddeker et al., Convolutive Transfer Function Invariant SDR training criteria for Multi-Channel Reverberant Speech Separation, 2021
      """
      if input_length is not None:
          if mask is not None:
              raise RuntimeError(
                  "Argument `input_length` is mutually exclusive with `mask`. Both cannot be used at the same time."
              )
  
          if torch.min(input_length) < filter_length:
              logging.warning(
                  "Current min input_length (%d) is smaller than filter_length (%d). This will result in a singular linear system.",
                  torch.min(input_length),
                  filter_length,
              )
  
          # Construct a binary mask
          mask = make_seq_mask_like(
              lengths=input_length,
              like=estimate,
              time_dim=-1,
              valid_ones=True,
          ).squeeze(1)
  
      # Apply a mask, if available
      if mask is not None:
          estimate = mask.unsqueeze(1) * estimate
          target = mask.unsqueeze(1) * target
  
      # Calculate filtered target
      input_shape = estimate.shape
      estimate = estimate.view(-1, input_shape[-1])
      target = target.view(-1, input_shape[-1])
  
      n_fft = 2 ** math.ceil(math.log2(2 * input_shape[-1] - 1))
  
      T = torch.fft.rfft(target, n=n_fft)
      E = torch.fft.rfft(estimate, n=n_fft)
  
      # Target autocorrelation
      tt_corr = torch.fft.irfft(torch.abs(T) ** 2, n=n_fft)
      # Target-estimate crosscorrelation
      te_corr = torch.fft.irfft(T.conj() * E, n=n_fft)
  
      # Use only filter_length
      tt_corr = tt_corr[..., :filter_length]
      te_corr = te_corr[..., :filter_length]
  
      # Diagonal regularization
      if diag_reg is not None:
          tt_corr[..., 0] += diag_reg * tt_corr[..., 0] + eps
  
      # Construct the Toeplitz system matrix
      TT = toeplitz(tt_corr)
  
      # Solve the linear system for the optimal filter
      filt = torch.linalg.solve(TT, te_corr)
  
      # Calculate filtered target
      T_filt = T * torch.fft.rfft(filt, n=n_fft)
      target_filt = torch.fft.irfft(T_filt, n=n_fft)
  
      # Reshape to the original format
      target_filt = target_filt[..., : input_shape[-1]].view(
          *input_shape
      )
  
      # Mask to keep only the valid samples
      if mask is not None:
          target_filt = mask.unsqueeze(1) * target_filt
  
      return target_filt
  
file_path: NeMo/nemo/collections/asr/losses/audio_losses.py
imports: import math
  from typing import List, Optional
  import numpy
  import torch
  from nemo.collections.asr.parts.preprocessing.features import make_seq_mask_like
  from nemo.collections.asr.parts.utils.audio_utils import toeplitz
  from nemo.core.classes import Loss, Typing, typecheck
  from nemo.core.neural_types import AudioSignal, LengthsType, LossType, MaskType, NeuralType
  from nemo.utils import logging
label_desc: Use image node_id 3 for calling a global function with example usage: convolution_invariant_target(estimate, target, input_length, mask, filter_length, diag_reg, eps) and returns: target_filt