


# Copyright (c) 2021-, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

####################################################################################################

#
# Note: If when running this conversion script you're getting an exception:
#     ModuleNotFoundError: No module named 'megatron.model.enums'
# you need to tell python where to find the clone of Megatron-LM, e.g.:
#
# cd /tmp
# git clone https://github.com/NVIDIA/Megatron-LM
# PYTHONPATH=/tmp/Megatron-LM python src/transformers/models/megatron_bert/convert_megatron_bert_checkpoint.py ...
#
# if you already have it cloned elsewhere, simply adjust the path to the existing path
#
# If the training was done using a Megatron-LM fork, e.g.,
# https://github.com/microsoft/Megatron-DeepSpeed/ then chances are that you need to have that one
# in your path, i.e., /path/to/Megatron-DeepSpeed/
#

import argparse
import os
import re
import zipfile

import torch

from transformers import MegatronBertConfig




def 				lowercase_		(		_A      :   List[Any]   , _A      :   str   , _A      :   Dict=0						):

      """simple docstring"""




      if name is None:
            lowerCamelCase__		:    Union[str, Any]   =  None
      else:
            lowerCamelCase__		:    Dict   =  '.' * max(0   , spaces - 2						) + '# {:' + str(50 - spaces						) + 's}'
            lowerCamelCase__		:    List[str]   =  fmt.format(SCREAMING_SNAKE_CASE__						)

      # Print and recurse (if needed).
      if isinstance(SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__						):
            if msg is not None:
                  print(SCREAMING_SNAKE_CASE__						)
            for k in val.keys():
                  recursive_print(SCREAMING_SNAKE_CASE__   , val[k]   , spaces + 2						)
      elif isinstance(SCREAMING_SNAKE_CASE__   , torch.Tensor						):
            print(SCREAMING_SNAKE_CASE__   , ":"   , val.size()						)
      else:
            print(SCREAMING_SNAKE_CASE__   , ":"   , SCREAMING_SNAKE_CASE__						)




def 				lowercase_		(		_A      :   Tuple   , _A      :   str   , _A      :   str   , _A      :   Optional[int]   , _A      :   Union[str, Any]						):

      """simple docstring"""




      lowerCamelCase__		:    Tuple   =  param.size()
      if checkpoint_version == 1.0:
            # version 1.0 stores [num_heads * hidden_size * num_splits, :]
            lowerCamelCase__		:    List[Any]   =  (num_heads, hidden_size, num_splits) + input_shape[1:]
            lowerCamelCase__		:    int   =  param.view(*SCREAMING_SNAKE_CASE__						)
            lowerCamelCase__		:    Union[str, Any]   =  param.transpose(0   , 2						)
            lowerCamelCase__		:    Tuple   =  param.transpose(1   , 2						).contiguous()
      elif checkpoint_version >= 2.0:
            # other versions store [num_heads * num_splits * hidden_size, :]
            lowerCamelCase__		:    str   =  (num_heads, num_splits, hidden_size) + input_shape[1:]
            lowerCamelCase__		:    Dict   =  param.view(*SCREAMING_SNAKE_CASE__						)
            lowerCamelCase__		:    List[Any]   =  param.transpose(0   , 1						).contiguous()
      lowerCamelCase__		:    Optional[int]   =  param.view(*SCREAMING_SNAKE_CASE__						)
      return param




def 				lowercase_		(		_A      :   Union[str, Any]   , _A      :   Dict   , _A      :   Any						):

      """simple docstring"""




      lowerCamelCase__		:    Union[str, Any]   =  {}

      # old versions did not store training args
      lowerCamelCase__		:    Tuple   =  input_state_dict.get("args"   , SCREAMING_SNAKE_CASE__						)
      if ds_args is not None:
            # do not make the user write a config file when the exact dimensions/sizes are already in the checkpoint
            # from pprint import pprint
            # pprint(vars(ds_args))

            lowerCamelCase__		:    Dict   =  ds_args.tokenizer_type
            lowerCamelCase__		:    str   =  ds_args.padded_vocab_size
            lowerCamelCase__		:    List[str]   =  ds_args.max_position_embeddings
            lowerCamelCase__		:    List[Any]   =  ds_args.hidden_size
            lowerCamelCase__		:    Dict   =  ds_args.num_layers
            lowerCamelCase__		:    str   =  ds_args.num_attention_heads
            lowerCamelCase__		:    Any   =  ds_args.ffn_hidden_size if 'ffn_hidden_size' in ds_args else 4 * ds_args.hidden_size
            # pprint(config)

      # The number of heads.
      lowerCamelCase__		:    List[str]   =  config.num_attention_heads
      # The hidden_size per head.
      lowerCamelCase__		:    Dict   =  config.hidden_size // heads
      # Megatron-LM checkpoint version
      if "checkpoint_version" in input_state_dict.keys():
            lowerCamelCase__		:    Any   =  input_state_dict['checkpoint_version']
      else:
            lowerCamelCase__		:    Any   =  0.0

      # The model.
      lowerCamelCase__		:    List[Any]   =  input_state_dict['model']
      # The language model.
      lowerCamelCase__		:    Any   =  model['language_model']
      # The embeddings.
      lowerCamelCase__		:    Any   =  lm['embedding']

      # The word embeddings.
      lowerCamelCase__		:    Union[str, Any]   =  embeddings['word_embeddings']['weight']
      # Truncate the embedding table to vocab_size rows.
      lowerCamelCase__		:    List[str]   =  word_embeddings[: config.vocab_size, :]
      # Store the word embeddings.
      lowerCamelCase__		:    Any   =  word_embeddings

      # The position embeddings.
      lowerCamelCase__		:    int   =  embeddings['position_embeddings']['weight']
      assert pos_embeddings.size(0						) == config.max_position_embeddings and pos_embeddings.size(1						) == config.hidden_size
      # Store the position embeddings.
      lowerCamelCase__		:    str   =  pos_embeddings

      # The token-type embeddings.
      lowerCamelCase__		:    Optional[Any]   =  embeddings['tokentype_embeddings']['weight']
      # Store the position embeddings.
      lowerCamelCase__		:    List[Any]   =  tokentype_embeddings

      # The transformer.
      lowerCamelCase__		:    Optional[Any]   =  lm['transformer'] if 'transformer' in lm.keys() else lm['encoder']

      # The regex to extract layer names.
      lowerCamelCase__		:    Optional[Any]   =  re.compile(r"layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)"						)

      # The simple map of names for "automated" rules.
      lowerCamelCase__		:    Any   =  {
          'attention.dense': '.attention.output.dense.',
          'self_attention.dense': '.attention.output.dense.',
          'mlp.dense_h_to_4h': '.intermediate.dense.',
          'mlp.dense_4h_to_h': '.output.dense.',
      }

      # Keep track of the attention/query/value tensor.
      lowerCamelCase__		:    Tuple   =  None

      # Extract the layers.
      for key, val in transformer.items():
            # Match the name.
            lowerCamelCase__		:    Optional[int]   =  layer_re.match(SCREAMING_SNAKE_CASE__						)

            # Stop if that's not a layer
            if m is None:
                  break

            # The index of the layer.
            lowerCamelCase__		:    Union[str, Any]   =  int(m.group(1						)						)
            # The name of the operation.
            lowerCamelCase__		:    Optional[int]   =  m.group(2						)
            # Is it a weight or a bias?
            lowerCamelCase__		:    List[str]   =  m.group(3						)

            # The name of the layer.
            lowerCamelCase__		:    str   =  F"bert.encoder.layer.{layer_idx}"

            # For layernorm(s), simply store the layer norm.
            if op_name.endswith("layernorm"						):
                  lowerCamelCase__		:    Any   =  'attention.ln' if op_name.startswith("input"						) else 'ln'
                  lowerCamelCase__		:    List[str]   =  val

            # Transpose the QKV matrix.
            elif (
                op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
            ) and weight_or_bias == "weight":
                  # Make sure the QKV pointer is nil.
                  assert attention_qkv_weight is None, ""

                  lowerCamelCase__		:    Union[str, Any]   =  fix_query_key_value_ordering(SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__   , 3   , SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__						)
                  # Store the tensor as we need the bias as well to interleave QKV and biases.
                  lowerCamelCase__		:    List[str]   =  out_val

            # Transpose the bias.
            elif (
                op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
            ) and weight_or_bias == "bias":
                  # Make sure we read the weight tensor.
                  assert attention_qkv_weight is not None, ""

                  # Split the QKV matrix into Q, K and V. Megatron stores Q,K,V interleaved.
                  lowerCamelCase__		:    str   =  attention_qkv_weight[0 * config.hidden_size : 1 * config.hidden_size, :]
                  lowerCamelCase__		:    Optional[int]   =  attention_qkv_weight[1 * config.hidden_size : 2 * config.hidden_size, :]
                  lowerCamelCase__		:    List[Any]   =  attention_qkv_weight[2 * config.hidden_size : 3 * config.hidden_size, :]

                  lowerCamelCase__		:    Any   =  fix_query_key_value_ordering(SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__   , 3   , SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__						)
                  # Split the bias.
                  lowerCamelCase__		:    Tuple   =  out_val[0 * config.hidden_size : 1 * config.hidden_size]
                  lowerCamelCase__		:    str   =  out_val[1 * config.hidden_size : 2 * config.hidden_size]
                  lowerCamelCase__		:    Any   =  out_val[2 * config.hidden_size : 3 * config.hidden_size]

                  # Store.
                  lowerCamelCase__		:    Dict   =  q
                  lowerCamelCase__		:    Union[str, Any]   =  q_bias
                  lowerCamelCase__		:    str   =  k
                  lowerCamelCase__		:    str   =  k_bias
                  lowerCamelCase__		:    int   =  v
                  lowerCamelCase__		:    Tuple   =  v_bias

                  # Clear the stored tensor.
                  lowerCamelCase__		:    Any   =  None

            # Copy weights and biases as is.
            elif weight_or_bias in ["weight", "bias"]:
                  lowerCamelCase__		:    Tuple   =  megatron_to_transformers[op_name]
                  lowerCamelCase__		:    Dict   =  val

    # The final layernorm.
      lowerCamelCase__		:    List[Any]   =  transformer['final_layernorm.weight']
      lowerCamelCase__		:    Optional[int]   =  transformer['final_layernorm.bias']

      # The pooler.
      lowerCamelCase__		:    Any   =  lm['pooler']

      # Store the matrix and the bias.
      lowerCamelCase__		:    List[Any]   =  pooler['dense.weight']
      lowerCamelCase__		:    int   =  pooler['dense.bias']

      # The LM head from Megatron (for RACE).
      lowerCamelCase__		:    List[str]   =  model['lm_head']

      # The transform matrix.
      lowerCamelCase__		:    int   =  lm_head['dense.weight']
      lowerCamelCase__		:    List[Any]   =  lm_head['dense.bias']

      # The transform LN.
      lowerCamelCase__		:    Optional[int]   =  lm_head['layernorm.weight']
      lowerCamelCase__		:    List[str]   =  lm_head['layernorm.bias']

      # For the decoder, we replicate the weights.
      lowerCamelCase__		:    Tuple   =  word_embeddings
      lowerCamelCase__		:    Any   =  lm_head['bias']

      # The classifier from Megatron (for MLNI).
      lowerCamelCase__		:    Optional[int]   =  model['binary_head']

      # Store the classifier.
      lowerCamelCase__		:    int   =  binary_head['weight']
      lowerCamelCase__		:    int   =  binary_head['bias']

      # It should be done!
      return output_state_dict




def 				lowercase_		(		):

      """simple docstring"""




      lowerCamelCase__		:    Union[str, Any]   =  argparse.ArgumentParser()
      parser.add_argument("--print-checkpoint-structure"   , action="store_true"						)
      parser.add_argument("path_to_checkpoint"   , type=SCREAMING_SNAKE_CASE__   , help="Path to the ZIP file containing the checkpoint"						)
      parser.add_argument(
          "--config_file"   , default=""   , type=SCREAMING_SNAKE_CASE__   , help="An optional config json file describing the pre-trained model."   , )
      lowerCamelCase__		:    Optional[Any]   =  parser.parse_args()

      # Extract the basename.
      lowerCamelCase__		:    Dict   =  os.path.dirname(args.path_to_checkpoint						)

      # Load the model.
      # the .zip is very optional, let's keep it for backward compatibility
      print(F"Extracting PyTorch state dictionary from \"{args.path_to_checkpoint}\""						)
      if args.path_to_checkpoint.endswith(".zip"						):
            with zipfile.ZipFile(args.path_to_checkpoint   , "r"						) as checkpoint:
                  with checkpoint.open("release/mp_rank_00/model_optim_rng.pt"						) as pytorch_dict:
                        lowerCamelCase__		:    List[Any]   =  torch.load(SCREAMING_SNAKE_CASE__   , map_location="cpu"						)
      else:
            lowerCamelCase__		:    str   =  torch.load(args.path_to_checkpoint   , map_location="cpu"						)

      if args.config_file == "":
            # Default config of megatron-bert 345m
            lowerCamelCase__		:    Tuple   =  MegatronBertConfig()

            # different megatron-bert-*-345m models have different vocab sizes, so override the default
            # config (which is for megatron-bert-cased-345m) with the actual vocab dimension
            lowerCamelCase__		:    str   =  input_state_dict['model']['lm_head']['bias'].numel()
      else:
            lowerCamelCase__		:    List[str]   =  MegatronBertConfig.from_json_file(args.config_file						)

      # Convert.
      print("Converting"						)
      lowerCamelCase__		:    int   =  convert_megatron_checkpoint(SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__						)

      # Print the structure of converted state dict.
      if args.print_checkpoint_structure:
            recursive_print(SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__						)

      # Store the config to file.
      print("Saving config"						)
      config.save_pretrained(SCREAMING_SNAKE_CASE__						)

      # Store the state_dict to file.
      lowerCamelCase__		:    str   =  os.path.join(SCREAMING_SNAKE_CASE__   , "pytorch_model.bin"						)
      print(F"Saving checkpoint to \"{output_checkpoint_file}\""						)
      torch.save(SCREAMING_SNAKE_CASE__   , SCREAMING_SNAKE_CASE__						)


####################################################################################################

if __name__ == "__main__":
    main()

####################################################################################################






import itertools
import random
import unittest

import numpy as np

from transformers import EncodecFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
    import torch


A				:   Union[str, Any]		= random.Random()

def 				lowercase_		(		_A      :   Any   , _A      :   int=1.0   , _A      :   Optional[int]=None   , _A      :   Union[str, Any]=None						):

      """simple docstring"""




      if rng is None:
            lowerCamelCase__		:    Any   =  global_rng

      lowerCamelCase__		:    Optional[Any]   =  []
      for batch_idx in range(shape[0]						):
            values.append([]						)
            for _ in range(shape[1]						):
                  values[-1].append(rng.random() * scale						)

      return values







@require_torch
class       _lowercase			(							unittest.TestCase):


       """simple docstring"""




       def __init__(       self							:      Tuple ,				__lowerCamelCase							:      str ,				__lowerCamelCase							:      Union[str, Any]=7 ,				__lowerCamelCase							:      int=400 ,				__lowerCamelCase							:      Tuple=2000 ,				__lowerCamelCase							:      int=1 ,				__lowerCamelCase							:      int=0.0 ,				__lowerCamelCase							:      int=24000 ,				__lowerCamelCase							:      Union[str, Any]=True ,				):


             '''simple docstring'''



             lowerCamelCase__		:    str   =  parent
             lowerCamelCase__		:    Union[str, Any]   =  batch_size
             lowerCamelCase__		:    str   =  min_seq_length
             lowerCamelCase__		:    int   =  max_seq_length
             lowerCamelCase__		:    List[Any]   =  (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
             lowerCamelCase__		:    List[str]   =  feature_size
             lowerCamelCase__		:    List[Any]   =  padding_value
             lowerCamelCase__		:    Union[str, Any]   =  sampling_rate
             lowerCamelCase__		:    Tuple   =  return_attention_mask





       def 						lowerCAmelCase					(       self							:      List[Any]	):


             '''simple docstring'''



             return {
                 "feature_size": self.feature_size,
                 "padding_value": self.padding_value,
                 "sampling_rate": self.sampling_rate,
                 "return_attention_mask": self.return_attention_mask,
             }





       def 						lowerCAmelCase					(       self							:      List[str] ,				__lowerCamelCase							:      Tuple=False ,				__lowerCamelCase							:      Optional[int]=False	):


             '''simple docstring'''



             def _flatten(__lowerCamelCase							:      Dict	):
                   return list(itertools.chain(*__lowerCamelCase	)	)

             if equal_length:
                   lowerCamelCase__		:    List[Any]   =  floats_list((self.batch_size, self.max_seq_length)	)
             else:
                   # make sure that inputs increase in size
                   lowerCamelCase__		:    Dict   =  [
                       _flatten(floats_list((x, self.feature_size)	)	)
                       for x in range(self.min_seq_length ,				self.max_seq_length ,				self.seq_length_diff	)
                   ]

             if numpify:
                   lowerCamelCase__		:    Optional[int]   =  [np.asarray(__lowerCamelCase	) for x in audio_inputs]

             return audio_inputs







@require_torch
class       _lowercase			(							lowercase__							,  unittest.TestCase):


       """simple docstring"""




       A__ 						=      EncodecFeatureExtractor





       def 						lowerCAmelCase					(       self							:      Union[str, Any]	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[int]   =  EnCodecFeatureExtractionTester(self	)





       def 						lowerCAmelCase					(       self							:      int	):


             '''simple docstring'''



             lowerCamelCase__		:    Tuple   =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
             # create three inputs of length 800, 1000, and 1200
             lowerCamelCase__		:    Union[str, Any]   =  [floats_list((1, x)	)[0] for x in range(800 ,				1400 ,				200	)]
             lowerCamelCase__		:    Dict   =  [np.asarray(__lowerCamelCase	) for audio_input in audio_inputs]

             # Test not batched input
             lowerCamelCase__		:    Dict   =  feat_extract(audio_inputs[0] ,				return_tensors="np"	).input_values
             lowerCamelCase__		:    Optional[int]   =  feat_extract(np_audio_inputs[0] ,				return_tensors="np"	).input_values
             self.assertTrue(np.allclose(__lowerCamelCase ,				__lowerCamelCase ,				atol=1E-3	)	)

             # Test batched
             lowerCamelCase__		:    Dict   =  feat_extract(__lowerCamelCase ,				padding=__lowerCamelCase ,				return_tensors="np"	).input_values
             lowerCamelCase__		:    Any   =  feat_extract(__lowerCamelCase ,				padding=__lowerCamelCase ,				return_tensors="np"	).input_values
             for enc_seq_a, enc_seq_a in zip(__lowerCamelCase ,				__lowerCamelCase	):
                   self.assertTrue(np.allclose(__lowerCamelCase ,				__lowerCamelCase ,				atol=1E-3	)	)





       def 						lowerCAmelCase					(       self							:      Tuple	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[Any]   =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
             lowerCamelCase__		:    Tuple   =  np.random.rand(100	).astype(np.floataa	)
             lowerCamelCase__		:    str   =  np_audio_inputs.tolist()

             for inputs in [py_audio_inputs, np_audio_inputs]:
                   lowerCamelCase__		:    List[Any]   =  feature_extractor.pad([{"input_values": inputs}] ,				return_tensors="np"	)
                   self.assertTrue(np_processed.input_values.dtype == np.floataa	)
                   lowerCamelCase__		:    int   =  feature_extractor.pad([{"input_values": inputs}] ,				return_tensors="pt"	)
                   self.assertTrue(pt_processed.input_values.dtype == torch.floataa	)





       def 						lowerCAmelCase					(       self							:      str ,				__lowerCamelCase							:      Any	):


             '''simple docstring'''



             from datasets import load_dataset

             lowerCamelCase__		:    List[Any]   =  load_dataset("hf-internal-testing/librispeech_asr_dummy" ,				"clean" ,				split="validation"	)
             # automatic decoding with librispeech
             lowerCamelCase__		:    Union[str, Any]   =  ds.sort("id"	).select(range(__lowerCamelCase	)	)[:num_samples]["audio"]

             return [x["array"] for x in audio_samples]





       def 						lowerCAmelCase					(       self							:      List[Any]	):


             '''simple docstring'''



             lowerCamelCase__		:    List[Any]   =  torch.tensor(
                 [2.3_8_0_4E-0_3, 2.0_7_5_2E-0_3, 1.9_8_3_6E-0_3, 2.1_0_5_7E-0_3, 1.6_1_7_4E-0_3,
                  3.0_5_1_8E-0_4, 9.1_5_5_3E-0_5, 3.3_5_6_9E-0_4, 9.7_6_5_6E-0_4, 1.8_3_1_1E-0_3,
                  2.0_1_4_2E-0_3, 2.1_0_5_7E-0_3, 1.7_3_9_5E-0_3, 4.5_7_7_6E-0_4, -3.9_6_7_3E-0_4,
                  4.5_7_7_6E-0_4, 1.0_0_7_1E-0_3, 9.1_5_5_3E-0_5, 4.8_8_2_8E-0_4, 1.1_5_9_7E-0_3,
                  7.3_2_4_2E-0_4, 9.4_6_0_4E-0_4, 1.8_0_0_5E-0_3, 1.8_3_1_1E-0_3, 8.8_5_0_1E-0_4,
                  4.2_7_2_5E-0_4, 4.8_8_2_8E-0_4, 7.3_2_4_2E-0_4, 1.0_9_8_6E-0_3, 2.1_0_5_7E-0_3]	)
             # fmt: on
             lowerCamelCase__		:    str   =  self._load_datasamples(1	)
             lowerCamelCase__		:    Any   =  EncodecFeatureExtractor()
             lowerCamelCase__		:    Optional[int]   =  feature_extractor(__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(input_values.shape ,				(1, 1, 93680)	)
             self.assertTrue(torch.allclose(input_values[0, 0, :30] ,				__lowerCamelCase ,				atol=1E-6	)	)





       def 						lowerCAmelCase					(       self							:      Optional[int]	):


             '''simple docstring'''



             lowerCamelCase__		:    Any   =  torch.tensor(
                 [2.3_8_0_4E-0_3, 2.0_7_5_2E-0_3, 1.9_8_3_6E-0_3, 2.1_0_5_7E-0_3, 1.6_1_7_4E-0_3,
                  3.0_5_1_8E-0_4, 9.1_5_5_3E-0_5, 3.3_5_6_9E-0_4, 9.7_6_5_6E-0_4, 1.8_3_1_1E-0_3,
                  2.0_1_4_2E-0_3, 2.1_0_5_7E-0_3, 1.7_3_9_5E-0_3, 4.5_7_7_6E-0_4, -3.9_6_7_3E-0_4,
                  4.5_7_7_6E-0_4, 1.0_0_7_1E-0_3, 9.1_5_5_3E-0_5, 4.8_8_2_8E-0_4, 1.1_5_9_7E-0_3,
                  7.3_2_4_2E-0_4, 9.4_6_0_4E-0_4, 1.8_0_0_5E-0_3, 1.8_3_1_1E-0_3, 8.8_5_0_1E-0_4,
                  4.2_7_2_5E-0_4, 4.8_8_2_8E-0_4, 7.3_2_4_2E-0_4, 1.0_9_8_6E-0_3, 2.1_0_5_7E-0_3]	)
             # fmt: on
             lowerCamelCase__		:    Any   =  self._load_datasamples(1	)
             lowerCamelCase__		:    int   =  [np.tile(input_audio[0][None] ,				reps=(2, 1)	)]
             input_audio[0][1] *= 0.5
             lowerCamelCase__		:    List[str]   =  EncodecFeatureExtractor(feature_size=2	)
             lowerCamelCase__		:    Optional[Any]   =  feature_extractor(__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(input_values.shape ,				(1, 2, 93680)	)
             self.assertTrue(torch.allclose(input_values[0, 0, :30] ,				__lowerCamelCase ,				atol=1E-6	)	)
             self.assertTrue(torch.allclose(input_values[0, 1, :30] ,				EXPECTED_INPUT_VALUES * 0.5 ,				atol=1E-6	)	)





       def 						lowerCAmelCase					(       self							:      Dict	):


             '''simple docstring'''



             lowerCamelCase__		:    Tuple   =  self._load_datasamples(2	)
             # would be easier if the stride was like
             lowerCamelCase__		:    Optional[Any]   =  EncodecFeatureExtractor(feature_size=1 ,				chunk_length_s=1 ,				overlap=0.0_1	)

             # pad and trunc raise an error ?
             with self.assertRaisesRegex(
                 __lowerCamelCase ,				"^Both padding and truncation were set. Make sure you only set one.$" ,				):
                   lowerCamelCase__		:    str   =  feature_extractor(
                       __lowerCamelCase ,				padding="max_length" ,				truncation=__lowerCamelCase ,				return_tensors="pt"	).input_values

             # truncate to chunk
             lowerCamelCase__		:    Optional[int]   =  feature_extractor(__lowerCamelCase ,				truncation=__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(2, 1, 71520)	)  # 2 chunks

             # force truncate to max_length
             lowerCamelCase__		:    List[str]   =  feature_extractor(
                 __lowerCamelCase ,				truncation=__lowerCamelCase ,				max_length=48000 ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(2, 1, 48000)	)

             # pad to chunk
             lowerCamelCase__		:    int   =  feature_extractor(__lowerCamelCase ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(padded_outputs.shape ,				(2, 1, 95280)	)

             # pad to chunk
             lowerCamelCase__		:    Optional[int]   =  feature_extractor(__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(2, 1, 95280)	)

             # force pad to max length
             lowerCamelCase__		:    Tuple   =  feature_extractor(
                 __lowerCamelCase ,				padding="max_length" ,				max_length=100000 ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(2, 1, 100000)	)

             # force no pad
             with self.assertRaisesRegex(
                 __lowerCamelCase ,				"^Unable to create tensor, you should probably activate padding with 'padding=True' to have batched tensors with the same length.$" ,				):
                   lowerCamelCase__		:    Optional[Any]   =  feature_extractor(__lowerCamelCase ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values

             lowerCamelCase__		:    str   =  feature_extractor(input_audio[0] ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(1, 1, 93680)	)

             # no pad if no chunk_length_s
             lowerCamelCase__		:    Tuple   =  None
             with self.assertRaisesRegex(
                 __lowerCamelCase ,				"^Unable to create tensor, you should probably activate padding with 'padding=True' to have batched tensors with the same length.$" ,				):
                   lowerCamelCase__		:    List[str]   =  feature_extractor(__lowerCamelCase ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values

             lowerCamelCase__		:    Tuple   =  feature_extractor(input_audio[0] ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(1, 1, 93680)	)

             # no pad if no overlap
             lowerCamelCase__		:    List[Any]   =  2
             lowerCamelCase__		:    List[Any]   =  None
             with self.assertRaisesRegex(
                 __lowerCamelCase ,				"^Unable to create tensor, you should probably activate padding with 'padding=True' to have batched tensors with the same length.$" ,				):
                   lowerCamelCase__		:    Optional[Any]   =  feature_extractor(__lowerCamelCase ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values

             lowerCamelCase__		:    Optional[Any]   =  feature_extractor(input_audio[0] ,				padding=__lowerCamelCase ,				return_tensors="pt"	).input_values
             self.assertEquals(truncated_outputs.shape ,				(1, 1, 93680)	)


