




'''simple docstring'''



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


UpperCAmelCase_       :  List[str]      =		random.Random()







def 			_UpperCamelCase (_lowerCamelCase							:     List[str]   ,	_lowerCamelCase							:     Tuple=1.0   ,	_lowerCamelCase							:     Tuple=None   ,	_lowerCamelCase							:     Dict=None	)->   Optional[int]:




     '''simple docstring'''





     if rng is None:
          __snake_case    					=  global_rng

     __snake_case    					=  []
     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     lowerCAmelCase   ( unittest.TestCase):


       def __init__(   self    ,  __SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE=7    ,  __SCREAMING_SNAKE_CASE=400    ,  __SCREAMING_SNAKE_CASE=2000    ,  __SCREAMING_SNAKE_CASE=1    ,  __SCREAMING_SNAKE_CASE=0.0    ,  __SCREAMING_SNAKE_CASE=2_4000    ,  __SCREAMING_SNAKE_CASE=True    ,  )      ->  Dict:



            '''simple docstring'''




            __snake_case    					=  parent
            __snake_case    					=  batch_size
            __snake_case    					=  min_seq_length
            __snake_case    					=  max_seq_length
            __snake_case    					=  (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
            __snake_case    					=  feature_size
            __snake_case    					=  padding_value
            __snake_case    					=  sampling_rate
            __snake_case    					=  return_attention_mask


       def       lowerCAmelCase      (   self  )      ->  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    ,  __SCREAMING_SNAKE_CASE=False    ,  __SCREAMING_SNAKE_CASE=False  )      ->  Any:



            '''simple docstring'''




            def _flatten(__SCREAMING_SNAKE_CASE  ):
                 return list(itertools.chain(*__SCREAMING_SNAKE_CASE  )  )

            if equal_length:
                 __snake_case    					=  floats_list((self.batch_size, self.max_seq_length)  )
            else:
                 # make sure that inputs increase in size
                 __snake_case    					=  [
                     _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:
                 __snake_case    					=  [np.asarray(__SCREAMING_SNAKE_CASE  ) for x in audio_inputs]

            return audio_inputs





@require_torch
class     lowerCAmelCase   ( __lowerCAmelCase				,   unittest.TestCase):
       __lowercase  :      Optional[int]        =			EncodecFeatureExtractor


       def       lowerCAmelCase      (   self  )      ->  int:



            '''simple docstring'''




            __snake_case    					=  EnCodecFeatureExtractionTester(self  )


       def       lowerCAmelCase      (   self  )      ->  Optional[int]:



            '''simple docstring'''




            __snake_case    					=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
            # create three inputs of length 800, 1000, and 1200
            __snake_case    					=  [floats_list((1, x)  )[0] for x in range(800    ,  1400    ,  200  )]
            __snake_case    					=  [np.asarray(__SCREAMING_SNAKE_CASE  ) for audio_input in audio_inputs]

            # Test not batched input
            __snake_case    					=  feat_extract(audio_inputs[0]    ,  return_tensors='''np'''  ).input_values
            __snake_case    					=  feat_extract(np_audio_inputs[0]    ,  return_tensors='''np'''  ).input_values
            self.assertTrue(np.allclose(__SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE    ,  atol=1E-3  )  )

            # Test batched
            __snake_case    					=  feat_extract(__SCREAMING_SNAKE_CASE    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''np'''  ).input_values
            __snake_case    					=  feat_extract(__SCREAMING_SNAKE_CASE    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''np'''  ).input_values
            for enc_seq_a, enc_seq_a in zip(__SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE  ):
                 self.assertTrue(np.allclose(__SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE    ,  atol=1E-3  )  )


       def       lowerCAmelCase      (   self  )      ->  int:



            '''simple docstring'''




            __snake_case    					=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
            __snake_case    					=  np.random.rand(100  ).astype(np.floataa  )
            __snake_case    					=  np_audio_inputs.tolist()

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


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



            '''simple docstring'''




            from datasets import load_dataset

            __snake_case    					=  load_dataset('''hf-internal-testing/librispeech_asr_dummy'''    ,  '''clean'''    ,  split='''validation'''  )
            # automatic decoding with librispeech
            __snake_case    					=  ds.sort('''id'''  ).select(range(__SCREAMING_SNAKE_CASE  )  )[:num_samples]['''audio''']

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


       def       lowerCAmelCase      (   self  )      ->  Any:



            '''simple docstring'''




            __snake_case    					=  torch.tensor(
                [2.38_04E-03, 2.07_52E-03, 1.98_36E-03, 2.10_57E-03, 1.61_74E-03,
                 3.05_18E-04, 9.15_53E-05, 3.35_69E-04, 9.76_56E-04, 1.83_11E-03,
                 2.01_42E-03, 2.10_57E-03, 1.73_95E-03, 4.57_76E-04, -3.96_73E-04,
                 4.57_76E-04, 1.00_71E-03, 9.15_53E-05, 4.88_28E-04, 1.15_97E-03,
                 7.32_42E-04, 9.46_04E-04, 1.80_05E-03, 1.83_11E-03, 8.85_01E-04,
                 4.27_25E-04, 4.88_28E-04, 7.32_42E-04, 1.09_86E-03, 2.10_57E-03]  )
            # fmt: on
            __snake_case    					=  self._load_datasamples(1  )
            __snake_case    					=  EncodecFeatureExtractor()
            __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(input_values.shape    ,  (1, 1, 9_3680)  )
            self.assertTrue(torch.allclose(input_values[0, 0, :30]    ,  __SCREAMING_SNAKE_CASE    ,  atol=1E-6  )  )


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



            '''simple docstring'''




            __snake_case    					=  torch.tensor(
                [2.38_04E-03, 2.07_52E-03, 1.98_36E-03, 2.10_57E-03, 1.61_74E-03,
                 3.05_18E-04, 9.15_53E-05, 3.35_69E-04, 9.76_56E-04, 1.83_11E-03,
                 2.01_42E-03, 2.10_57E-03, 1.73_95E-03, 4.57_76E-04, -3.96_73E-04,
                 4.57_76E-04, 1.00_71E-03, 9.15_53E-05, 4.88_28E-04, 1.15_97E-03,
                 7.32_42E-04, 9.46_04E-04, 1.80_05E-03, 1.83_11E-03, 8.85_01E-04,
                 4.27_25E-04, 4.88_28E-04, 7.32_42E-04, 1.09_86E-03, 2.10_57E-03]  )
            # fmt: on
            __snake_case    					=  self._load_datasamples(1  )
            __snake_case    					=  [np.tile(input_audio[0][None]    ,  reps=(2, 1)  )]
            input_audio[0][1] *= 0.5
            __snake_case    					=  EncodecFeatureExtractor(feature_size=2  )
            __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(input_values.shape    ,  (1, 2, 9_3680)  )
            self.assertTrue(torch.allclose(input_values[0, 0, :30]    ,  __SCREAMING_SNAKE_CASE    ,  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'''




            __snake_case    					=  self._load_datasamples(2  )
            # would be easier if the stride was like
            __snake_case    					=  EncodecFeatureExtractor(feature_size=1    ,  chunk_length_s=1    ,  overlap=0.01  )

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

            # truncate to chunk
            __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  truncation=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (2, 1, 7_1520)  )  # 2 chunks

            # force truncate to max_length
            __snake_case    					=  feature_extractor(
                __SCREAMING_SNAKE_CASE    ,  truncation=__SCREAMING_SNAKE_CASE    ,  max_length=4_8000    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (2, 1, 4_8000)  )

            # pad to chunk
            __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(padded_outputs.shape    ,  (2, 1, 9_5280)  )

            # pad to chunk
            __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (2, 1, 9_5280)  )

            # force pad to max length
            __snake_case    					=  feature_extractor(
                __SCREAMING_SNAKE_CASE    ,  padding='''max_length'''    ,  max_length=10_0000    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (2, 1, 10_0000)  )

            # force no pad
            with self.assertRaisesRegex(
                __SCREAMING_SNAKE_CASE    ,  '''^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'''    ,  ):
                 __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values

            __snake_case    					=  feature_extractor(input_audio[0]    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (1, 1, 9_3680)  )

            # no pad if no chunk_length_s
            __snake_case    					=  None
            with self.assertRaisesRegex(
                __SCREAMING_SNAKE_CASE    ,  '''^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'''    ,  ):
                 __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values

            __snake_case    					=  feature_extractor(input_audio[0]    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (1, 1, 9_3680)  )

            # no pad if no overlap
            __snake_case    					=  2
            __snake_case    					=  None
            with self.assertRaisesRegex(
                __SCREAMING_SNAKE_CASE    ,  '''^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'''    ,  ):
                 __snake_case    					=  feature_extractor(__SCREAMING_SNAKE_CASE    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values

            __snake_case    					=  feature_extractor(input_audio[0]    ,  padding=__SCREAMING_SNAKE_CASE    ,  return_tensors='''pt'''  ).input_values
            self.assertEquals(truncated_outputs.shape    ,  (1, 1, 9_3680)  )












'''simple docstring'''



# Copyright 2022 The HuggingFace Team. 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.
import os
from typing import List







def 			_UpperCamelCase (_lowerCamelCase							:     List[str]   ,	_lowerCamelCase							:     str	)->   str:




     '''simple docstring'''





     if name != "training_function" and name != "main":
          raise ValueError(f'''Incorrect function name passed: {name}, choose either \'main\' or \'training_function\''''	)
     __snake_case					,    __snake_case    					=  [], False
     for line in lines:
          if not found_start and f'''def {name}''' in line:
               __snake_case    					=  True
               good_lines.append(_lowerCamelCase	)
               continue
          if found_start:
               if name == "training_function" and "def main" in line:
                    return good_lines
               if name == "main" and "if __name__" in line:
                    return good_lines
               good_lines.append(_lowerCamelCase	)







def 			_UpperCamelCase (_lowerCamelCase							:     List[str]	)->   int:




     '''simple docstring'''





     return [line for line in lines if not line.lstrip().startswith('''#'''	) and line != "\n"]







def 			_UpperCamelCase (_lowerCamelCase							:     str   ,	_lowerCamelCase							:     str   ,	_lowerCamelCase							:     bool   ,	_lowerCamelCase							:     str = None	)->   Optional[Any]:




     '''simple docstring'''





     with open(_lowerCamelCase   ,	'''r'''	) as f:
          __snake_case    					=  f.readlines()
     with open(os.path.abspath(os.path.join('''examples'''   ,	'''nlp_example.py'''	)	)   ,	'''r'''	) as f:
          __snake_case    					=  f.readlines()
     with open(_lowerCamelCase   ,	'''r'''	) as f:
          __snake_case    					=  f.readlines()
     if secondary_filename is not None:
          with open(_lowerCamelCase   ,	'''r'''	) as f:
               __snake_case    					=  f.readlines()

    # This is our base, we remove all the code from here in our `full_filename` and `feature_filename` to find the new content
     if parser_only:
          __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''main'''	)	)
          __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''main'''	)	)
          __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''main'''	)	)
          if secondary_filename is not None:
               __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''main'''	)	)
     else:
          __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''training_function'''	)	)
          __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''training_function'''	)	)
          __snake_case    					=  clean_lines(get_function_contents_by_name(_lowerCamelCase   ,	'''training_function'''	)	)
          if secondary_filename is not None:
               __snake_case    					=  clean_lines(
                   get_function_contents_by_name(_lowerCamelCase   ,	'''training_function'''	)	)

     __snake_case    					=  '''train_dataloader, eval_dataloader = get_dataloaders(accelerator, batch_size)\n'''

     # Specific code in our script that differs from the full version, aka what is new
     __snake_case    					=  []
     __snake_case    					=  []  # We keep track of the idxs just in case it's a repeated statement
     __snake_case    					=  iter(_lowerCamelCase	)
     for i in range(len(_lowerCamelCase	) - 1	):
          if i not in passed_idxs:
               __snake_case    					=  next(_lowerCamelCase	)
               if (line not in full_file_func) and (line.lstrip() != _dl_line):
                    if "TESTING_MOCKED_DATALOADERS" not in line:
                         new_feature_code.append(_lowerCamelCase	)
                         passed_idxs.append(_lowerCamelCase	)
                    else:
                         # Skip over the `config['num_epochs'] = 2` statement
                         __snake_case    					=  next(_lowerCamelCase	)

    # Extract out just the new parts from the full_file_training_func
     __snake_case    					=  []
     __snake_case    					=  []  # We keep track of the idxs just in case it's a repeated statement
     for i, line in enumerate(_lowerCamelCase	):
          if i not in passed_idxs:
               if (line not in full_file_func) and (line.lstrip() != _dl_line):
                    if "TESTING_MOCKED_DATALOADERS" not in line:
                         new_full_example_parts.append(_lowerCamelCase	)
                         passed_idxs.append(_lowerCamelCase	)

    # Finally, get the overall diff
     __snake_case    					=  [line for line in new_feature_code if line not in new_full_example_parts]
     if secondary_filename is not None:
          __snake_case    					=  [line for line in full_file_contents if line not in secondary_file_func]
          __snake_case    					=  [line for line in diff_from_example if line not in diff_from_two]

     return diff_from_example






