class_docstr: This is an implementation of Conformer encoder layer.
  
  Arguments
  ----------
  d_model : int
      The expected size of the input embedding.
  d_ffn : int
      Hidden size of self-attention Feed Forward layer.
  nhead : int
      Number of attention heads.
  kernel_size : int, optional
      Kernel size of convolution model.
  kdim : int, optional
      Dimension of the key.
  vdim : int, optional
      Dimension of the value.
  activation: torch.nn.Module
       Activation function used in each Conformer layer.
  bias : bool, optional
      Whether  convolution module.
  dropout : int, optional
      Dropout for the encoder.
  causal: bool, optional
      Whether the convolutions should be causal or not.
  attention_type: str, optional
      type of attention layer, e.g. regulaMHA for regular MultiHeadAttention.
  
  Example
  -------
  >>> import torch
  >>> x = torch.rand((8, 60, 512))
  >>> pos_embs = torch.rand((1, 2*60-1, 512))
  >>> net = ConformerEncoderLayer(d_ffn=512, nhead=8, d_model=512, kernel_size=3)
  >>> output = net(x, pos_embs=pos_embs)
  >>> output[0].shape
  torch.Size([8, 60, 512])
docstr: Conformer layer streaming forward (typically for
  DynamicChunkTraining-trained models), which is to be used at inference
  time. Relies on a mutable context object as initialized by
  `make_streaming_context` that should be used across chunks.
  Invoked by `ConformerEncoder.forward_streaming`.
  
  Arguments
  ---------
  x : torch.Tensor
      Input tensor for this layer. Batching is supported as long as you
      keep the context consistent.
  context: ConformerEncoderStreamingContext
      Mutable streaming context; the same object should be passed across
      calls.
  pos_embs: torch.Tensor, optional
      Positional embeddings, if used.
docstr_tok: ["Conformer","layer","streaming","forward","(","typically","for","DynamicChunkTraining-trained","models",")",",","which","is","to","be","used","at","inference","time",".","Relies","on","a","mutable","context","object","as","initialized","by","`","make_streaming_context","`","that","should","be","used","across","chunks",".","Invoked","by","`","ConformerEncoder.forward_streaming","`",".","Arguments","--","--","--","--","-","x",":","torch.Tensor","Input","tensor","for","this","layer",".","Batching","is","supported","as","long","as","you","keep","the","context","consistent",".","context",":","ConformerEncoderStreamingContext","Mutable","streaming","context",";","the","same","object","should","be","passed","across","calls",".","pos_embs",":","torch.Tensor",",","optional","Positional","embeddings",",","if","used","."]
code: def forward_streaming(
      self,
      x,
      context: ConformerEncoderLayerStreamingContext,
      pos_embs: torch.Tensor = None,
  ):
      """Conformer layer streaming forward (typically for
      DynamicChunkTraining-trained models), which is to be used at inference
      time. Relies on a mutable context object as initialized by
      `make_streaming_context` that should be used across chunks.
      Invoked by `ConformerEncoder.forward_streaming`.
  
      Arguments
      ---------
      x : torch.Tensor
          Input tensor for this layer. Batching is supported as long as you
          keep the context consistent.
      context: ConformerEncoderStreamingContext
          Mutable streaming context; the same object should be passed across
          calls.
      pos_embs: torch.Tensor, optional
          Positional embeddings, if used."""
  
      orig_len = x.shape[-2]
      # ffn module
      x = x + 0.5 * self.ffn_module1(x)
  
      # TODO: make the approach for MHA left context more efficient.
      # currently, this saves the inputs to the MHA.
      # the naive approach is suboptimal in a few ways, namely that the
      # outputs for this left padding is being re-computed even though we
      # discard them immediately after.
  
      # left pad `x` with our MHA left context
      if context.mha_left_context is not None:
          x = torch.cat((context.mha_left_context, x), dim=1)
  
      # compute new MHA left context for the next call to our function
      if context.mha_left_context_size > 0:
          context.mha_left_context = x[
              ..., -context.mha_left_context_size :, :
          ]
  
      # multi-head attention module
      skip = x
      x = self.norm1(x)
  
      x, self_attn = self.mha_layer(
          x,
          x,
          x,
          attn_mask=None,
          key_padding_mask=None,
          pos_embs=pos_embs,
      )
      x = x + skip
  
      # truncate outputs corresponding to the MHA left context (we only care
      # about our chunk's outputs); see above to-do
      x = x[..., -orig_len:, :]
  
      if context.dcconv_left_context is not None:
          x = torch.cat((context.dcconv_left_context, x), dim=1)
  
      # compute new DCConv left context for the next call to our function
      context.dcconv_left_context = x[
          ..., -self.convolution_module.padding :, :
      ]
  
      # convolution module
      x = x + self.convolution_module(x)
  
      # truncate outputs corresponding to the DCConv left context
      x = x[..., -orig_len:, :]
  
      # ffn module
      x = self.norm2(x + 0.5 * self.ffn_module2(x))
      return x, self_attn
  
imports: from dataclasses import dataclass
  import torch
  import torch.nn
  import torch.nn.functional
  from typing import Optional, List
  import speechbrain
  import warnings
  from speechbrain.nnet.attention import RelPosMHAXL, MultiheadAttention, PositionalwiseFeedForward
  from speechbrain.utils.dynamic_chunk_training import DynChunkTrainConfig
  from speechbrain.lobes.models.transformer.hypermixing import HyperMixing
  from speechbrain.nnet.normalization import LayerNorm
  from speechbrain.nnet.activations import Swish
label_desc: Use image node_id 3 for calling the ConformerEncoderLayer obj's underlying member method code with example usage: obj.forward_streaming(x, context, pos_embs) and returns: x,  self_attn