code: def get_WPD_filter_with_rtf(
      psd_observed_bar: Union[torch.Tensor, ComplexTensor],
      psd_speech: Union[torch.Tensor, ComplexTensor],
      psd_noise: Union[torch.Tensor, ComplexTensor],
      iterations: int = 3,
      reference_vector: Union[int, torch.Tensor, None] = None,
      diagonal_loading: bool = True,
      diag_eps: float = 1e-7,
      eps: float = 1e-15,
  ) -> Union[torch.Tensor, ComplexTensor]:
      """Return the WPD vector calculated with RTF.
  
          WPD is the Weighted Power minimization Distortionless response
          convolutional beamformer. As follows:
  
          h = (Rf^-1 @ vbar) / (vbar^H @ R^-1 @ vbar)
  
      Reference:
          T. Nakatani and K. Kinoshita, "A Unified Convolutional Beamformer
          for Simultaneous Denoising and Dereverberation," in IEEE Signal
          Processing Letters, vol. 26, no. 6, pp. 903-907, June 2019, doi:
          10.1109/LSP.2019.2911179.
          https://ieeexplore.ieee.org/document/8691481
  
      Args:
          psd_observed_bar (torch.complex64/ComplexTensor):
              stacked observation covariance matrix
          psd_speech (torch.complex64/ComplexTensor):
              speech covariance matrix (..., F, C, C)
          psd_noise (torch.complex64/ComplexTensor):
              noise covariance matrix (..., F, C, C)
          iterations (int): number of iterations in power method
          reference_vector (torch.Tensor or int): (..., C) or scalar
          diagonal_loading (bool):
              Whether to add a tiny term to the diagonal of psd_n
          diag_eps (float):
          eps (float):
      Returns:
          beamform_vector (torch.complex64/ComplexTensor)r: (..., F, C)
      """
      if isinstance(psd_speech, ComplexTensor):
          pad_func = FC.pad
      elif is_torch_complex_tensor(psd_speech):
          pad_func = torch.nn.functional.pad
      else:
          raise ValueError(
              "Please update your PyTorch version to 1.9+ for complex support."
          )
  
      C = psd_noise.size(-1)
      if diagonal_loading:
          psd_noise = tik_reg(psd_noise, reg=diag_eps, eps=eps)
  
      # (B, F, C, 1)
      rtf = get_rtf(
          psd_speech,
          psd_noise,
          mode="power",
          reference_vector=reference_vector,
          iterations=iterations,
      )
  
      # (B, F, (K+1)*C, 1)
      rtf = pad_func(
          rtf, (0, 0, 0, psd_observed_bar.shape[-1] - C), "constant", 0
      )
      # numerator: (..., C_1, C_2) x (..., C_2, 1) -> (..., C_1)
      numerator = solve(rtf, psd_observed_bar).squeeze(-1)
      denominator = einsum(
          "...d,...d->...", rtf.squeeze(-1).conj(), numerator
      )
      if reference_vector is not None:
          if isinstance(reference_vector, int):
              scale = rtf.squeeze(-1)[
                  ..., reference_vector, None
              ].conj()
          else:
              scale = (
                  rtf.squeeze(-1)[:, :, :C].conj()
                  * reference_vector[..., None, :]
              ).sum(dim=-1, keepdim=True)
          beamforming_vector = (
              numerator * scale / (denominator.real.unsqueeze(-1) + eps)
          )
      else:
          beamforming_vector = numerator / (
              denominator.real.unsqueeze(-1) + eps
          )
      return beamforming_vector
  
imports: from typing import List, Union
  import torch
  from packaging.version import parse
  from torch_complex import functional
  from torch_complex.tensor import ComplexTensor
  from espnet2.enh.layers.complex_utils import cat, complex_norm, einsum, inverse, is_complex, is_torch_complex_tensor, matmul, reverse, solve, to_double
label_desc: Use image node_id 20 for calling a global function with example usage: get_WPD_filter_with_rtf(psd_observed_bar, psd_speech, psd_noise, iterations, reference_vector, diagonal_loading, diag_eps, eps) and returns: beamforming_vector