class_docstr: NaiveRNNDP-SVS module.
  
  This is an implementation of naive RNN with duration prediction
  for singing voice synthesis
  The features are processed directly over time-domain from music score and
  predict the singing voice features
class_docstr_tok: ["NaiveRNNDP-SVS","module",".","This","is","an","implementation","of","naive","RNN","with","duration","prediction","for","singing","voice","synthesis","The","features","are","processed","directly","over","time-domain","from","music","score","and","predict","the","singing","voice","features"]
docstr: Calculate forward propagation.
  
  Args:
      text (LongTensor): Batch of padded character ids (B, Tmax).
      text_lengths (LongTensor): Batch of lengths of each input batch (B,).
      feats (Tensor): Batch of padded target features (B, Lmax, odim).
      feats_lengths (LongTensor): Batch of the lengths of each target (B,).
      label (Optional[Dict]): key is "lab" or "score";
          value (LongTensor): Batch of padded label ids (B, Tmax).
      label_lengths (Optional[Dict]): key is "lab" or "score";
          value (LongTensor): Batch of the lengths of padded label ids (B, ).
      melody (Optional[Dict]): key is "lab" or "score";
          value (LongTensor): Batch of padded melody (B, Tmax).
      melody_lengths (Optional[Dict]): key is "lab" or "score";
          value (LongTensor): Batch of the lengths of padded melody (B, ).
      pitch (FloatTensor): Batch of padded f0 (B, Tmax).
      pitch_lengths (LongTensor): Batch of the lengths of padded f0 (B, ).
      duration (Optional[Dict]): key is "lab", "score_phn" or "score_syb";
          value (LongTensor): Batch of padded duration (B, Tmax).
      duration_length (Optional[Dict]): key is "lab", "score_phn" or "score_syb";
          value (LongTensor): Batch of the lengths of padded duration (B, ).
      slur (LongTensor): Batch of padded slur (B, Tmax).
      slur_lengths (LongTensor): Batch of the lengths of padded slur (B, ).
      spembs (Optional[Tensor]): Batch of speaker embeddings (B, spk_embed_dim).
      sids (Optional[Tensor]): Batch of speaker IDs (B, 1).
      lids (Optional[Tensor]): Batch of language IDs (B, 1).
      joint_training (bool): Whether to perform joint training with vocoder.
  
  GS Fix:
      arguements from forward func. V.S. **batch from espnet_model.py
      label == durations | phone sequence
      melody -> pitch sequence
  
  Returns:
      Tensor: Loss scalar value.
      Dict: Statistics to be monitored.
      Tensor: Weight value if not joint training else model outputs.
docstr_tok: ["Calculate","forward","propagation",".","Args",":","text","(","LongTensor",")",":","Batch","of","padded","character","ids","(","B",",","Tmax",")",".","text_lengths","(","LongTensor",")",":","Batch","of","lengths","of","each","input","batch","(","B",",",")",".","feats","(","Tensor",")",":","Batch","of","padded","target","features","(","B",",","Lmax",",","odim",")",".","feats_lengths","(","LongTensor",")",":","Batch","of","the","lengths","of","each","target","(","B",",",")",".","label","(","Optional","[","Dict","]",")",":","key","is","``","lab","''","or","``","score","''",";","value","(","LongTensor",")",":","Batch","of","padded","label","ids","(","B",",","Tmax",")",".","label_lengths","(","Optional","[","Dict","]",")",":","key","is","``","lab","''","or","``","score","''",";","value","(","LongTensor",")",":","Batch","of","the","lengths","of","padded","label","ids","(","B",",",")",".","melody","(","Optional","[","Dict","]",")",":","key","is","``","lab","''","or","``","score","''",";","value","(","LongTensor",")",":","Batch","of","padded","melody","(","B",",","Tmax",")",".","melody_lengths","(","Optional","[","Dict","]",")",":","key","is","``","lab","''","or","``","score","''",";","value","(","LongTensor",")",":","Batch","of","the","lengths","of","padded","melody","(","B",",",")",".","pitch","(","FloatTensor",")",":","Batch","of","padded","f0","(","B",",","Tmax",")",".","pitch_lengths","(","LongTensor",")",":","Batch","of","the","lengths","of","padded","f0","(","B",",",")",".","duration","(","Optional","[","Dict","]",")",":","key","is","``","lab","''",",","``","score_phn","''","or","``","score_syb","''",";","value","(","LongTensor",")",":","Batch","of","padded","duration","(","B",",","Tmax",")",".","duration_length","(","Optional","[","Dict","]",")",":","key","is","``","lab","''",",","``","score_phn","''","or","``","score_syb","''",";","value","(","LongTensor",")",":","Batch","of","the","lengths","of","padded","duration","(","B",",",")",".","slur","(","LongTensor",")",":","Batch","of","padded","slur","(","B",",","Tmax",")",".","slur_lengths","(","LongTensor",")",":","Batch","of","the","lengths","of","padded","slur","(","B",",",")",".","spembs","(","Optional","[","Tensor","]",")",":","Batch","of","speaker","embeddings","(","B",",","spk_embed_dim",")",".","sids","(","Optional","[","Tensor","]",")",":","Batch","of","speaker","IDs","(","B",",","1",")",".","lids","(","Optional","[","Tensor","]",")",":","Batch","of","language","IDs","(","B",",","1",")",".","joint_training","(","bool",")",":","Whether","to","perform","joint","training","with","vocoder",".","GS","Fix",":","arguements","from","forward","func",".","V.S",".","*","*","batch","from","espnet_model.py","label","==","durations","|","phone","sequence","melody","-",">","pitch","sequence","Returns",":","Tensor",":","Loss","scalar","value",".","Dict",":","Statistics","to","be","monitored",".","Tensor",":","Weight","value","if","not","joint","training","else","model","outputs","."]
code: def forward(
      self,
      text: torch.Tensor,
      text_lengths: torch.Tensor,
      feats: torch.Tensor,
      feats_lengths: torch.Tensor,
      label: Optional[Dict[str, torch.Tensor]] = None,
      label_lengths: Optional[Dict[str, torch.Tensor]] = None,
      melody: Optional[Dict[str, torch.Tensor]] = None,
      melody_lengths: Optional[Dict[str, torch.Tensor]] = None,
      pitch: Optional[torch.Tensor] = None,
      pitch_lengths: Optional[torch.Tensor] = None,
      duration: Optional[Dict[str, torch.Tensor]] = None,
      duration_lengths: Optional[Dict[str, torch.Tensor]] = None,
      slur: torch.LongTensor = None,
      slur_lengths: torch.Tensor = None,
      spembs: Optional[torch.Tensor] = None,
      sids: Optional[torch.Tensor] = None,
      lids: Optional[torch.Tensor] = None,
      joint_training: bool = False,
      flag_IsValid=False,
  ) -> Tuple[torch.Tensor, Dict[str, torch.Tensor], torch.Tensor]:
      """Calculate forward propagation.
  
      Args:
          text (LongTensor): Batch of padded character ids (B, Tmax).
          text_lengths (LongTensor): Batch of lengths of each input batch (B,).
          feats (Tensor): Batch of padded target features (B, Lmax, odim).
          feats_lengths (LongTensor): Batch of the lengths of each target (B,).
          label (Optional[Dict]): key is "lab" or "score";
              value (LongTensor): Batch of padded label ids (B, Tmax).
          label_lengths (Optional[Dict]): key is "lab" or "score";
              value (LongTensor): Batch of the lengths of padded label ids (B, ).
          melody (Optional[Dict]): key is "lab" or "score";
              value (LongTensor): Batch of padded melody (B, Tmax).
          melody_lengths (Optional[Dict]): key is "lab" or "score";
              value (LongTensor): Batch of the lengths of padded melody (B, ).
          pitch (FloatTensor): Batch of padded f0 (B, Tmax).
          pitch_lengths (LongTensor): Batch of the lengths of padded f0 (B, ).
          duration (Optional[Dict]): key is "lab", "score_phn" or "score_syb";
              value (LongTensor): Batch of padded duration (B, Tmax).
          duration_length (Optional[Dict]): key is "lab", "score_phn" or "score_syb";
              value (LongTensor): Batch of the lengths of padded duration (B, ).
          slur (LongTensor): Batch of padded slur (B, Tmax).
          slur_lengths (LongTensor): Batch of the lengths of padded slur (B, ).
          spembs (Optional[Tensor]): Batch of speaker embeddings (B, spk_embed_dim).
          sids (Optional[Tensor]): Batch of speaker IDs (B, 1).
          lids (Optional[Tensor]): Batch of language IDs (B, 1).
          joint_training (bool): Whether to perform joint training with vocoder.
  
      GS Fix:
          arguements from forward func. V.S. **batch from espnet_model.py
          label == durations | phone sequence
          melody -> pitch sequence
  
      Returns:
          Tensor: Loss scalar value.
          Dict: Statistics to be monitored.
          Tensor: Weight value if not joint training else model outputs.
      """
      if joint_training:
          label = label
          midi = melody
          label_lengths = label_lengths
          midi_lengths = melody_lengths
          duration_ = duration
          ds = duration
      else:
          label = label["score"]
          midi = melody["score"]
          duration_ = duration["score_phn"]
          label_lengths = label_lengths["score"]
          midi_lengths = melody_lengths["score"]
          duration_lengths = duration_lengths["score_phn"]
          ds = duration["lab"]
  
      feats = feats[:, : feats_lengths.max()]  # for data-parallel
      midi = midi[:, : midi_lengths.max()]  # for data-parallel
      label = label[:, : label_lengths.max()]  # for data-parallel
      duration_ = duration_[
          :, : duration_lengths.max()
      ]  # for data-parallel
      batch_size = feats.size(0)
  
      label_emb = self.encoder_input_layer(
          label
      )  # FIX ME: label Float to Int
      midi_emb = self.midi_encoder_input_layer(midi)
      duration_emb = self.duration_encoder_input_layer(duration_)
  
      label_emb = torch.nn.utils.rnn.pack_padded_sequence(
          label_emb,
          label_lengths.to("cpu"),
          batch_first=True,
          enforce_sorted=False,
      )
      midi_emb = torch.nn.utils.rnn.pack_padded_sequence(
          midi_emb,
          midi_lengths.to("cpu"),
          batch_first=True,
          enforce_sorted=False,
      )
      duration_emb = torch.nn.utils.rnn.pack_padded_sequence(
          duration_emb,
          duration_lengths.to("cpu"),
          batch_first=True,
          enforce_sorted=False,
      )
  
      hs_label, (_, _) = self.encoder(label_emb)
      hs_midi, (_, _) = self.midi_encoder(midi_emb)
      hs_duration, (_, _) = self.duration_encoder(duration_emb)
  
      hs_label, _ = torch.nn.utils.rnn.pad_packed_sequence(
          hs_label, batch_first=True
      )
      hs_midi, _ = torch.nn.utils.rnn.pad_packed_sequence(
          hs_midi, batch_first=True
      )
      hs_duration, _ = torch.nn.utils.rnn.pad_packed_sequence(
          hs_duration, batch_first=True
      )
  
      if self.midi_embed_integration_type == "add":
          hs = hs_label + hs_midi + hs_duration
          hs = F.leaky_relu(self.midi_projection(hs))
      else:
          hs = torch.cat((hs_label, hs_midi, hs_duration), dim=-1)
          hs = F.leaky_relu(self.midi_projection(hs))
      # integrate spk & lang embeddings
      if self.spks is not None:
          sid_embs = self.sid_emb(sids.view(-1))
          hs = hs + sid_embs.unsqueeze(1)
      if self.langs is not None:
          lid_embs = self.lid_emb(lids.view(-1))
          hs = hs + lid_embs.unsqueeze(1)
  
      # integrate speaker embedding
      if self.spk_embed_dim is not None:
          hs = self._integrate_with_spk_embed(hs, spembs)
  
      # forward duration predictor and length regulator
      d_masks = make_pad_mask(label_lengths).to(hs.device)
      d_outs = self.duration_predictor(hs, d_masks)  # (B, T_text)
      hs = self.length_regulator(hs, ds)  # (B, seq_len, eunits)
  
      olens = feats_lengths
      if self.reduction_factor > 1:
          olens_in = olens.new(
              [
                  torch.div(
                      olen, self.reduction_factor, rounding_mode="trunc"
                  )
                  for olen in olens
              ]
          )
      else:
          olens_in = olens
  
      hs_emb = torch.nn.utils.rnn.pack_padded_sequence(
          hs, olens_in.to("cpu"), batch_first=True, enforce_sorted=False
      )
  
      zs, (_, _) = self.decoder(hs_emb)
      zs, _ = torch.nn.utils.rnn.pad_packed_sequence(
          zs, batch_first=True
      )
  
      # feat_out: (B, T_feats//r, dunits * dim_direction) -> (B, T_feats//r, odim * r)
      # view: (B, T_feats//r, odim * r) -> (B, T_feats//r * r, odim)
      before_outs = F.leaky_relu(
          self.feat_out(zs).view(zs.size(0), -1, self.odim)
      )
  
      # postnet -> (B, T_feats//r * r, odim)
      if self.postnet is None:
          after_outs = before_outs
      else:
          after_outs = before_outs + self.postnet(
              before_outs.transpose(1, 2)
          ).transpose(1, 2)
  
      # modifiy mod part of groundtruth
      if self.reduction_factor > 1:
          assert feats_lengths.ge(
              self.reduction_factor
          ).all(), "Output length must be greater than or equal to reduction factor."
          olens = feats_lengths.new(
              [
                  olen - olen % self.reduction_factor
                  for olen in feats_lengths
              ]
          )
          max_olen = max(olens)
          ys = feats[:, :max_olen]
      else:
          ys = feats
          olens = feats_lengths
  
      # calculate loss values
      ilens = label_lengths
      l1_loss, duration_loss = self.criterion(
          after_outs, before_outs, d_outs, ys, ds, ilens, olens
      )
      loss = l1_loss + duration_loss
  
      stats = dict(
          loss=loss.item(),
          l1_loss=l1_loss.item(),
          duration_loss=duration_loss.item(),
      )
  
      loss, stats, weight = force_gatherable(
          (loss, stats, batch_size), loss.device
      )
  
      if joint_training:
          return (
              loss,
              stats,
              after_outs if after_outs is not None else before_outs,
          )
      else:
          if flag_IsValid is False:
              # training stage
              return loss, stats, weight
          else:
              # validation stage
              return (
                  loss,
                  stats,
                  weight,
                  after_outs[:, : olens.max()],
                  ys,
                  olens,
              )
  
file_path: espnet/espnet2/svs/naive_rnn/naive_rnn_dp.py
imports: from typing import Dict, Optional, Tuple
  import torch
  import torch.nn.functional
  from typeguard import check_argument_types
  from espnet2.svs.abs_svs import AbsSVS
  from espnet2.torch_utils.device_funcs import force_gatherable
  from espnet2.torch_utils.initialize import initialize
  from espnet.nets.pytorch_backend.e2e_tts_fastspeech import FeedForwardTransformerLoss
  from espnet.nets.pytorch_backend.fastspeech.duration_predictor import DurationPredictor
  from espnet.nets.pytorch_backend.fastspeech.length_regulator import LengthRegulator
  from espnet.nets.pytorch_backend.nets_utils import make_pad_mask
  from espnet.nets.pytorch_backend.tacotron2.decoder import Postnet
  from espnet.nets.pytorch_backend.tacotron2.encoder import Encoder
label_desc: Use image node_id 3 for calling the NaiveRNNDP obj's underlying member method code with example usage: obj.forward(text, text_lengths, feats, feats_lengths, label, label_lengths, melody, melody_lengths, pitch, pitch_lengths, duration, duration_lengths, slur, slur_lengths, spembs, sids, lids, joint_training, flag_IsValid) and returns: loss,  stats,  unknown, loss,  stats,  weight, loss,  stats,  weight,  after_outs,  ys,  olens