class_docstr: The encoder for ASR model of Conformer.
  Based on this paper:
  'Conformer: Convolution-augmented Transformer for Speech Recognition' by Anmol Gulati et al.
  https://arxiv.org/abs/2005.08100
  
  Args:
      feat_in (int): the size of feature channels
      n_layers (int): number of layers of ConformerBlock
      d_model (int): the hidden size of the model
      feat_out (int): the size of the output features
          Defaults to -1 (means feat_out is d_model)
      subsampling (str): the method of subsampling, choices=['vggnet', 'striding', 'dw-striding', 'stacking', 'stacking_norm']
          Defaults to striding.
      subsampling_factor (int): the subsampling factor which should be power of 2
          Defaults to 4.
      subsampling_conv_chunking_factor(int): optionally, force chunk inputs (helpful for large inputs)
          Should be power of 2, 1 (auto-chunking, default), or -1 (no chunking)
      subsampling_conv_channels (int): the size of the convolutions in the subsampling module
          Defaults to -1 which would set it to d_model.
      reduction (str, Optional): the method of reduction, choices=['pooling', 'striding']. If no value
          is passed, then no reduction is performed and the models runs with the original 4x subsampling.
      reduction_position (int, Optional): the index of the layer to apply reduction. If -1, apply reduction
          at the end.
      reduction_factor (int): the reduction factor which should be either 1 or a power of 2
          Defaults to 1.
      ff_expansion_factor (int): the expansion factor in feed forward layers
          Defaults to 4.
      self_attention_model (str): type of the attention layer and positional encoding
          'rel_pos': relative positional embedding and Transformer-XL
          'rel_pos_local_attn': relative positional embedding and Transformer-XL with local attention using
              overlapping chunks. Attention context is determined by att_context_size parameter.
          'abs_pos': absolute positional embedding and Transformer
          Default is rel_pos.
      pos_emb_max_len (int): the maximum length of positional embeddings
          Defaults to 5000
      n_heads (int): number of heads in multi-headed attention layers
          Defaults to 4.
      att_context_size (List[Union[List[int],int]]): specifies the context sizes on each side. Each context size should be a list of two integers like [100,100].
          A list of context sizes like [[100,100],[100,50]] can also be passed. -1 means unlimited context.
          Defaults to [-1,-1]
      att_context_probs (List[float]): a list of probabilities of each one of the att_context_size when a list of them is passed. If not specified, uniform distribution is being used.
          Defaults to None
      att_context_style (str): 'regular' or 'chunked_limited'.
          Defaults to 'regular'
      xscaling (bool): enables scaling the inputs to the multi-headed attention layers by sqrt(d_model)
          Defaults to True.
      untie_biases (bool): whether to not share (untie) the bias weights between layers of Transformer-XL
          Defaults to True.
      conv_kernel_size (int): the size of the convolutions in the convolutional modules
          Defaults to 31.
      conv_norm_type (str): the type of the normalization in the convolutional modules
          Defaults to 'batch_norm'.
      conv_context_size (list): it can be"causal" or a list of two integers while conv_context_size[0]+conv_context_size[1]+1==conv_kernel_size.
          None means [(conv_kernel_size-1)//2, (conv_kernel_size-1)//2], and 'causal' means [(conv_kernel_size-1), 0].
          Defaults to None.
      conv_dual_mode (bool): specifies if convolution should be dual mode when dual_offline mode is being used. When enables, the left half of the convolution kernel would get masked in streaming cases.
          Defaults to False
      dropout (float): the dropout rate used in all layers except the attention layers
          Defaults to 0.1.
      dropout_pre_encoder (float): the dropout rate used before the encoder
          Defaults to 0.1.
      dropout_emb (float): the dropout rate used for the positional embeddings
          Defaults to 0.1.
      dropout_att (float): the dropout rate used for the attention layer
          Defaults to 0.0.
      stochastic_depth_drop_prob (float): if non-zero, will randomly drop
          layers during training. The higher this value, the more often layers
          are dropped. Defaults to 0.0.
      stochastic_depth_mode (str): can be either "linear" or "uniform". If
          set to "uniform", all layers have the same probability of drop. If
          set to "linear", the drop probability grows linearly from 0 for the
          first layer to the desired value for the final layer. Defaults to
          "linear".
      stochastic_depth_start_layer (int): starting layer for stochastic depth.
          All layers before this will never be dropped. Note that drop
          probability will be adjusted accordingly if mode is "linear" when
          start layer is > 1. Defaults to 1.
      global_tokens (int): number of tokens to be used for global attention.
          Only relevant if self_attention_model is 'rel_pos_local_attn'.
          Defaults to 0.
      global_tokens_spacing (int): how far apart the global tokens are
          Defaults to 1.
      global_attn_separate (bool): whether the q, k, v layers used for global tokens should be separate.
          Defaults to False.
code: def forward_internal(
      self,
      audio_signal,
      length,
      cache_last_channel=None,
      cache_last_time=None,
      cache_last_channel_len=None,
  ):
      self.update_max_seq_length(
          seq_length=audio_signal.size(2), device=audio_signal.device
      )
  
      if length is None:
          length = audio_signal.new_full(
              (audio_signal.size(0),),
              audio_signal.size(-1),
              dtype=torch.int64,
              device=audio_signal.device,
          )
  
      # select a random att_context_size with the distribution specified by att_context_probs during training
      # for non-validation cases like test, validation or inference, it uses the first mode in self.att_context_size
      if self.training and len(self.att_context_size_all) > 1:
          cur_att_context_size = random.choices(
              self.att_context_size_all, weights=self.att_context_probs
          )[0]
      else:
          cur_att_context_size = self.att_context_size
  
      audio_signal = torch.transpose(audio_signal, 1, 2)
  
      if isinstance(self.pre_encode, nn.Linear):
          audio_signal = self.pre_encode(audio_signal)
      else:
          audio_signal, length = self.pre_encode(
              x=audio_signal, lengths=length
          )
          length = length.to(torch.int64)
          # self.streaming_cfg is set by setup_streaming_cfg(), called in the init
          if (
              self.streaming_cfg.drop_extra_pre_encoded > 0
              and cache_last_channel is not None
          ):
              audio_signal = audio_signal[
                  :, self.streaming_cfg.drop_extra_pre_encoded :, :
              ]
              length = (
                  length - self.streaming_cfg.drop_extra_pre_encoded
              ).clamp(min=0)
  
      if (
          self.reduction_position is not None
          and cache_last_channel is not None
      ):
          raise ValueError(
              "Caching with reduction feature is not supported yet!"
          )
  
      max_audio_length = audio_signal.size(1)
      if cache_last_channel is not None:
          cache_len = self.streaming_cfg.last_channel_cache_size
          cache_keep_size = (
              max_audio_length - self.streaming_cfg.cache_drop_size
          )
          max_audio_length = max_audio_length + cache_len
          padding_length = length + cache_len
          offset = torch.neg(cache_last_channel_len) + cache_len
      else:
          padding_length = length
          cache_last_channel_next = None
          cache_len = 0
          offset = None
  
      audio_signal, pos_emb = self.pos_enc(
          x=audio_signal, cache_len=cache_len
      )
  
      # Create the self-attention and padding masks
      pad_mask, att_mask = self._create_masks(
          att_context_size=cur_att_context_size,
          padding_length=padding_length,
          max_audio_length=max_audio_length,
          offset=offset,
          device=audio_signal.device,
      )
  
      if cache_last_channel is not None:
          pad_mask = pad_mask[:, cache_len:]
          if att_mask is not None:
              att_mask = att_mask[:, cache_len:]
          # Convert caches from the tensor to list
          cache_last_time_next = []
          cache_last_channel_next = []
  
      for lth, (drop_prob, layer) in enumerate(
          zip(self.layer_drop_probs, self.layers)
      ):
          original_signal = audio_signal
          if cache_last_channel is not None:
              cache_last_channel_cur = cache_last_channel[lth]
              cache_last_time_cur = cache_last_time[lth]
          else:
              cache_last_channel_cur = None
              cache_last_time_cur = None
          audio_signal = layer(
              x=audio_signal,
              att_mask=att_mask,
              pos_emb=pos_emb,
              pad_mask=pad_mask,
              cache_last_channel=cache_last_channel_cur,
              cache_last_time=cache_last_time_cur,
          )
  
          if cache_last_channel_cur is not None:
              (
                  audio_signal,
                  cache_last_channel_cur,
                  cache_last_time_cur,
              ) = audio_signal
              cache_last_channel_next.append(cache_last_channel_cur)
              cache_last_time_next.append(cache_last_time_cur)
  
          # applying stochastic depth logic from https://arxiv.org/abs/2102.03216
          if self.training and drop_prob > 0.0:
              should_drop = torch.rand(1) < drop_prob
              # adjusting to match expectation
              if should_drop:
                  # that's not efficient, but it's hard to implement distributed
                  # version of dropping layers without deadlock or random seed meddling
                  # so multiplying the signal by 0 to ensure all weights get gradients
                  audio_signal = audio_signal * 0.0 + original_signal
              else:
                  # not doing this operation if drop prob is 0 as it's identity in that case
                  audio_signal = (audio_signal - original_signal) / (
                      1.0 - drop_prob
                  ) + original_signal
  
          if self.reduction_position == lth:
              audio_signal, length = self.reduction_subsampling(
                  x=audio_signal, lengths=length
              )
              max_audio_length = audio_signal.size(1)
              # Don't update the audio_signal here because then it will again scale the audio_signal
              # and cause an increase in the WER
              _, pos_emb = self.pos_enc(
                  x=audio_signal, cache_len=cache_len
              )
              pad_mask, att_mask = self._create_masks(
                  att_context_size=cur_att_context_size,
                  padding_length=length,
                  max_audio_length=max_audio_length,
                  offset=offset,
                  device=audio_signal.device,
              )
  
          # saving tensors if required for interctc loss
          if self.is_access_enabled():
              if self.interctc_capture_at_layers is None:
                  self.interctc_capture_at_layers = self.access_cfg.get(
                      "interctc", {}
                  ).get("capture_layers", [])
              if lth in self.interctc_capture_at_layers:
                  lth_audio_signal = audio_signal
                  if self.out_proj is not None:
                      lth_audio_signal = self.out_proj(audio_signal)
                  # shape is the same as the shape of audio_signal output, i.e. [B, D, T]
                  self.register_accessible_tensor(
                      name=f"interctc/layer_output_{lth}",
                      tensor=torch.transpose(lth_audio_signal, 1, 2),
                  )
                  self.register_accessible_tensor(
                      name=f"interctc/layer_length_{lth}", tensor=length
                  )
  
      if self.out_proj is not None:
          audio_signal = self.out_proj(audio_signal)
  
      # Reduction
      if self.reduction_position == -1:
          audio_signal, length = self.reduction_subsampling(
              x=audio_signal, lengths=length
          )
  
      audio_signal = torch.transpose(audio_signal, 1, 2)
      length = length.to(dtype=torch.int64)
  
      if cache_last_channel is not None:
          cache_last_channel_next = torch.stack(
              cache_last_channel_next, dim=0
          )
          cache_last_time_next = torch.stack(
              cache_last_time_next, dim=0
          )
          return (
              audio_signal,
              length,
              cache_last_channel_next,
              cache_last_time_next,
              torch.clamp(
                  cache_last_channel_len + cache_keep_size,
                  max=cache_len,
              ),
          )
      else:
          return audio_signal, length
  
imports: import math
  import random
  from collections import OrderedDict
  from dataclasses import dataclass
  from typing import List, Optional, Set
  import torch
  import torch.distributed
  import torch.nn
  from omegaconf import DictConfig, ListConfig, open_dict
  from nemo.collections.asr.models.configs import CacheAwareStreamingConfig
  from nemo.collections.asr.parts.mixins.streaming import StreamingEncoder
  from nemo.collections.asr.parts.submodules.causal_convs import CausalConv1D
  from nemo.collections.asr.parts.submodules.conformer_modules import ConformerLayer
  from nemo.collections.asr.parts.submodules.multi_head_attention import LocalAttRelPositionalEncoding, MultiHeadAttention, PositionalEncoding, RelPositionalEncoding, RelPositionMultiHeadAttention, RelPositionMultiHeadAttentionLongformer
  from nemo.collections.asr.parts.submodules.subsampling import ConvSubsampling, StackingSubsampling, SubsamplingReductionModule
  from nemo.collections.asr.parts.utils import adapter_utils
  from nemo.collections.asr.parts.utils.regularization_utils import compute_stochastic_depth_drop_probs
  from nemo.core.classes.common import typecheck
  from nemo.core.classes.exportable import Exportable
  from nemo.core.classes.mixins import AccessMixin, adapter_mixins
  from nemo.core.classes.module import NeuralModule
  from nemo.core.neural_types import AcousticEncodedRepresentation, ChannelType, LengthsType, NeuralType, SpectrogramType
  from nemo.utils import logging
label_desc: Use image node_id 12 for calling the ConformerEncoder obj's underlying member method code with example usage: obj.forward_internal(audio_signal, length, cache_last_channel, cache_last_time, cache_last_channel_len) and returns: audio_signal,  length,  cache_last_channel_next,  cache_last_time_next,  torch, audio_signal,  length