class_docstr: DNN mask based Beamformer.
  
  Citation:
      Multichannel End-to-end Speech Recognition; T. Ochiai et al., 2017;
      http://proceedings.mlr.press/v70/ochiai17a/ochiai17a.pdf
class_docstr_tok: ["DNN","mask","based","Beamformer",".","Citation",":","Multichannel","End-to-end","Speech","Recognition",";","T.","Ochiai","et","al.",",","2017",";","http",":","\/\/proceedings.mlr.press\/v70\/ochiai17a\/ochiai17a.pdf"]
docstr: DNN_Beamformer forward function.
  
  Notation:
      B: Batch
      C: Channel
      T: Time or Sequence length
      F: Freq
  
  Args:
      data (torch.complex64/ComplexTensor): (B, T, C, F)
      ilens (torch.Tensor): (B,)
      powers (List[torch.Tensor] or None): used for wMPDR or WPD (B, F, T)
      oracle_masks (List[torch.Tensor] or None): oracle masks (B, F, C, T)
          if not None, oracle_masks will be used instead of self.mask
  Returns:
      enhanced (torch.complex64/ComplexTensor): (B, T, F)
      ilens (torch.Tensor): (B,)
      masks (torch.Tensor): (B, T, C, F)
docstr_tok: ["DNN_Beamformer","forward","function",".","Notation",":","B",":","Batch","C",":","Channel","T",":","Time","or","Sequence","length","F",":","Freq","Args",":","data","(","torch.complex64\/ComplexTensor",")",":","(","B",",","T",",","C",",","F",")","ilens","(","torch.Tensor",")",":","(","B",",",")","powers","(","List","[","torch.Tensor","]","or","None",")",":","used","for","wMPDR","or","WPD","(","B",",","F",",","T",")","oracle_masks","(","List","[","torch.Tensor","]","or","None",")",":","oracle","masks","(","B",",","F",",","C",",","T",")","if","not","None",",","oracle_masks","will","be","used","instead","of","self.mask","Returns",":","enhanced","(","torch.complex64\/ComplexTensor",")",":","(","B",",","T",",","F",")","ilens","(","torch.Tensor",")",":","(","B",",",")","masks","(","torch.Tensor",")",":","(","B",",","T",",","C",",","F",")"]
code: def forward(
      self,
      data: Union[torch.Tensor, ComplexTensor],
      ilens: torch.LongTensor,
      powers: Optional[List[torch.Tensor]] = None,
      oracle_masks: Optional[List[torch.Tensor]] = None,
  ) -> Tuple[
      Union[torch.Tensor, ComplexTensor], torch.LongTensor, torch.Tensor
  ]:
      """DNN_Beamformer forward function.
  
      Notation:
          B: Batch
          C: Channel
          T: Time or Sequence length
          F: Freq
  
      Args:
          data (torch.complex64/ComplexTensor): (B, T, C, F)
          ilens (torch.Tensor): (B,)
          powers (List[torch.Tensor] or None): used for wMPDR or WPD (B, F, T)
          oracle_masks (List[torch.Tensor] or None): oracle masks (B, F, C, T)
              if not None, oracle_masks will be used instead of self.mask
      Returns:
          enhanced (torch.complex64/ComplexTensor): (B, T, F)
          ilens (torch.Tensor): (B,)
          masks (torch.Tensor): (B, T, C, F)
      """
      # data (B, T, C, F) -> (B, F, C, T)
      data = data.permute(0, 3, 2, 1)
      data_d = to_double(data)
  
      # mask: [(B, F, C, T)]
      if oracle_masks is not None:
          masks = oracle_masks
      else:
          masks, _ = self.mask(data, ilens)
      assert self.nmask == len(masks), len(masks)
      # floor masks to increase numerical stability
      if self.mask_flooring:
          masks = [
              torch.clamp(m, min=self.flooring_thres) for m in masks
          ]
  
      if self.num_spk == 1:  # single-speaker case
          if self.use_noise_mask:
              # (mask_speech, mask_noise)
              mask_speech, mask_noise = masks
          else:
              # (mask_speech,)
              mask_speech = masks[0]
              mask_noise = 1 - mask_speech
  
          if self.beamformer_type in ("lcmv", "lcmp", "wlcmp"):
              raise NotImplementedError(
                  "Single source is not supported yet"
              )
          beamformer_stats = self.bf_func.prepare_beamformer_stats(
              data_d,
              [mask_speech],
              mask_noise,
              powers=powers,
              beamformer_type=self.beamformer_type,
              bdelay=self.bdelay,
              btaps=self.btaps,
              eps=self.eps,
          )
  
          if self.beamformer_type in ("mvdr", "mpdr", "wmpdr", "wpd"):
              enhanced, ws = self.apply_beamforming(
                  data,
                  ilens,
                  beamformer_stats["psd_n"],
                  beamformer_stats["psd_speech"],
                  psd_distortion=beamformer_stats["psd_distortion"],
              )
          elif (
              self.beamformer_type.endswith("_souden")
              or self.beamformer_type == "mwf"
              or self.beamformer_type == "wmwf"
              or self.beamformer_type == "sdw_mwf"
              or self.beamformer_type == "r1mwf"
              or self.beamformer_type.startswith("gev")
          ):
              enhanced, ws = self.apply_beamforming(
                  data,
                  ilens,
                  beamformer_stats["psd_n"],
                  beamformer_stats["psd_speech"],
              )
          else:
              raise ValueError(
                  "Not supporting beamformer_type={}".format(
                      self.beamformer_type
                  )
              )
  
          # (..., F, T) -> (..., T, F)
          enhanced = enhanced.transpose(-1, -2)
      else:  # multi-speaker case
          if self.use_noise_mask:
              # (mask_speech1, ..., mask_noise)
              mask_speech = list(masks[:-1])
              mask_noise = masks[-1]
          else:
              # (mask_speech1, ..., mask_speechX)
              mask_speech = list(masks)
              mask_noise = None
  
          beamformer_stats = self.bf_func.prepare_beamformer_stats(
              data_d,
              mask_speech,
              mask_noise,
              powers=powers,
              beamformer_type=self.beamformer_type,
              bdelay=self.bdelay,
              btaps=self.btaps,
              eps=self.eps,
          )
          if self.beamformer_type in ("lcmv", "lcmp", "wlcmp"):
              rtf_mat = self.bf_func.get_rtf_matrix(
                  beamformer_stats["psd_speech"],
                  beamformer_stats["psd_distortion"],
                  diagonal_loading=self.diagonal_loading,
                  ref_channel=self.ref_channel,
                  rtf_iterations=self.rtf_iterations,
                  diag_eps=self.diag_eps,
              )
  
          enhanced, ws = [], []
          for i in range(self.num_spk):
              # treat all other speakers' psd_speech as noises
              if self.beamformer_type in (
                  "mvdr",
                  "mvdr_tfs",
                  "wmpdr",
                  "wpd",
              ):
                  enh, w = self.apply_beamforming(
                      data,
                      ilens,
                      beamformer_stats["psd_n"][i],
                      beamformer_stats["psd_speech"][i],
                      psd_distortion=beamformer_stats["psd_distortion"][
                          i
                      ],
                  )
              elif self.beamformer_type in (
                  "mvdr_souden",
                  "mvdr_tfs_souden",
                  "wmpdr_souden",
                  "wpd_souden",
                  "wmwf",
                  "sdw_mwf",
                  "r1mwf",
                  "gev",
                  "gev_ban",
              ):
                  enh, w = self.apply_beamforming(
                      data,
                      ilens,
                      beamformer_stats["psd_n"][i],
                      beamformer_stats["psd_speech"][i],
                  )
              elif self.beamformer_type == "mpdr":
                  enh, w = self.apply_beamforming(
                      data,
                      ilens,
                      beamformer_stats["psd_n"],
                      beamformer_stats["psd_speech"][i],
                      psd_distortion=beamformer_stats["psd_distortion"][
                          i
                      ],
                  )
              elif self.beamformer_type in ("mpdr_souden", "mwf"):
                  enh, w = self.apply_beamforming(
                      data,
                      ilens,
                      beamformer_stats["psd_n"],
                      beamformer_stats["psd_speech"][i],
                  )
              elif self.beamformer_type == "lcmp":
                  enh, w = self.apply_beamforming(
                      data,
                      ilens,
                      beamformer_stats["psd_n"],
                      beamformer_stats["psd_speech"][i],
                      rtf_mat=rtf_mat,
                      spk=i,
                  )
              elif self.beamformer_type in ("lcmv", "wlcmp"):
                  enh, w = self.apply_beamforming(
                      data,
                      ilens,
                      beamformer_stats["psd_n"][i],
                      beamformer_stats["psd_speech"][i],
                      rtf_mat=rtf_mat,
                      spk=i,
                  )
              else:
                  raise ValueError(
                      "Not supporting beamformer_type={}".format(
                          self.beamformer_type
                      )
                  )
  
              # (..., F, T) -> (..., T, F)
              enh = enh.transpose(-1, -2)
              enhanced.append(enh)
              ws.append(w)
  
      # (..., F, C, T) -> (..., T, C, F)
      masks = [m.transpose(-1, -3) for m in masks]
      return enhanced, ilens, masks
  
file_path: espnet/espnet2/enh/layers/dnn_beamformer.py
imports: import logging
  from typing import List, Optional, Tuple, Union
  import torch
  from packaging.version import parse
  from torch.nn import functional
  from torch_complex.tensor import ComplexTensor
  import espnet2.enh.layers.beamformer
  import espnet2.enh.layers.beamformer_th
  from espnet2.enh.layers.complex_utils import stack, to_double, to_float
  from espnet2.enh.layers.mask_estimator import MaskEstimator
label_desc: Use image node_id 2 for calling the DNN_Beamformer obj's underlying member method code with example usage: obj.forward(data, ilens, powers, oracle_masks) and returns: enhanced,  ilens,  masks