



'''simple docstring'''





import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
       from transformers import WhisperFeatureExtractor

if is_torch_available():
       import torch

__a  			=						random.Random()





def 		__snake_case(							_lowerCAmelCase	,  _lowerCAmelCase=1.0	,  _lowerCAmelCase=None	,  _lowerCAmelCase=None						)						->					int:
      if rng is None:
            snake_case__    :       List[Any]       		= global_rng

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

      return values







@require_torch
@require_torchaudio
class        UpperCAmelCase_			(			unittest.TestCase      ):
    """simple docstring"""







    def __init__(       self   :	List[Any]					,	snake_case_   :	str					,	snake_case_   :	Dict=7					,	snake_case_   :	Any=400					,	snake_case_   :	Any=2_000					,	snake_case_   :	Optional[int]=10					,	snake_case_   :	str=160					,	snake_case_   :	Dict=8					,	snake_case_   :	Union[str, Any]=0.0					,	snake_case_   :	Optional[Any]=4_000					,	snake_case_   :	Union[str, Any]=False					,	snake_case_   :	Optional[Any]=True					,	):
          snake_case__    :       str       		= parent
          snake_case__    :       Optional[Any]       		= batch_size
          snake_case__    :       str       		= min_seq_length
          snake_case__    :       List[Any]       		= max_seq_length
          snake_case__    :       str       		= (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
          snake_case__    :       List[Any]       		= padding_value
          snake_case__    :       List[str]       		= sampling_rate
          snake_case__    :       List[str]       		= return_attention_mask
          snake_case__    :       Union[str, Any]       		= do_normalize
          snake_case__    :       int       		= feature_size
          snake_case__    :       Tuple       		= chunk_length
          snake_case__    :       Optional[int]       		= hop_length




    def lowerCamelCase			(       self   :	Optional[Any]		):
          return {
              "feature_size": self.feature_size,
              "hop_length": self.hop_length,
              "chunk_length": self.chunk_length,
              "padding_value": self.padding_value,
              "sampling_rate": self.sampling_rate,
              "return_attention_mask": self.return_attention_mask,
              "do_normalize": self.do_normalize,
          }




    def lowerCamelCase			(       self   :	Union[str, Any]					,	snake_case_   :	Optional[Any]=False					,	snake_case_   :	Any=False		):
          def _flatten(snake_case_   :	Tuple		):
                return list(itertools.chain(*snake_case_		)		)

          if equal_length:
                snake_case__    :       Optional[int]       		= [floats_list((self.max_seq_length, self.feature_size)		) for _ in range(self.batch_size		)]
          else:
                # make sure that inputs increase in size
                snake_case__    :       str       		= [
                    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__    :       Dict       		= [np.asarray(snake_case_		) for x in speech_inputs]
          return speech_inputs







@require_torch
@require_torchaudio
class        UpperCAmelCase_			(			_a						,						unittest.TestCase      ):
    """simple docstring"""







    lowercase   =						WhisperFeatureExtractor if is_speech_available() else None




    def lowerCamelCase			(       self   :	Dict		):
          snake_case__    :       List[Any]       		= WhisperFeatureExtractionTester(self		)




    def lowerCamelCase			(       self   :	Optional[Any]		):
          snake_case__    :       int       		= self.feature_extraction_class(**self.feat_extract_dict		)

          with tempfile.TemporaryDirectory() as tmpdirname:
                snake_case__    :       List[str]       		= feat_extract_first.save_pretrained(snake_case_		)[0]
                check_json_file_has_correct_format(snake_case_		)
                snake_case__    :       str       		= self.feature_extraction_class.from_pretrained(snake_case_		)

          snake_case__    :       Any       		= feat_extract_first.to_dict()
          snake_case__    :       Union[str, Any]       		= feat_extract_second.to_dict()
          snake_case__    :       Union[str, Any]       		= feat_extract_first.mel_filters
          snake_case__    :       List[str]       		= feat_extract_second.mel_filters
          self.assertTrue(np.allclose(snake_case_					,	snake_case_		)		)
          self.assertEqual(snake_case_					,	snake_case_		)




    def lowerCamelCase			(       self   :	List[str]		):
          snake_case__    :       Optional[Any]       		= self.feature_extraction_class(**self.feat_extract_dict		)

          with tempfile.TemporaryDirectory() as tmpdirname:
                snake_case__    :       List[str]       		= os.path.join(snake_case_					,	"""feat_extract.json"""		)
                feat_extract_first.to_json_file(snake_case_		)
                snake_case__    :       Dict       		= self.feature_extraction_class.from_json_file(snake_case_		)

          snake_case__    :       str       		= feat_extract_first.to_dict()
          snake_case__    :       List[str]       		= feat_extract_second.to_dict()
          snake_case__    :       Any       		= feat_extract_first.mel_filters
          snake_case__    :       List[str]       		= feat_extract_second.mel_filters
          self.assertTrue(np.allclose(snake_case_					,	snake_case_		)		)
          self.assertEqual(snake_case_					,	snake_case_		)




    def lowerCamelCase			(       self   :	List[str]		):
          # Tests that all call wrap to encode_plus and batch_encode_plus
          snake_case__    :       Union[str, Any]       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
          # create three inputs of length 800, 1000, and 1200
          snake_case__    :       int       		= [floats_list((1, x)		)[0] for x in range(800					,	1_400					,	200		)]
          snake_case__    :       Tuple       		= [np.asarray(snake_case_		) for speech_input in speech_inputs]

          # Test feature size
          snake_case__    :       int       		= feature_extractor(snake_case_					,	padding="""max_length"""					,	return_tensors="""np"""		).input_features
          self.assertTrue(input_features.ndim == 3		)
          self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames		)
          self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size		)

          # Test not batched input
          snake_case__    :       int       		= feature_extractor(speech_inputs[0]					,	return_tensors="""np"""		).input_features
          snake_case__    :       Optional[Any]       		= feature_extractor(np_speech_inputs[0]					,	return_tensors="""np"""		).input_features
          self.assertTrue(np.allclose(snake_case_					,	snake_case_					,	atol=1E-3		)		)

          # Test batched
          snake_case__    :       Tuple       		= feature_extractor(snake_case_					,	return_tensors="""np"""		).input_features
          snake_case__    :       int       		= feature_extractor(snake_case_					,	return_tensors="""np"""		).input_features
          for enc_seq_a, enc_seq_a in zip(snake_case_					,	snake_case_		):
                self.assertTrue(np.allclose(snake_case_					,	snake_case_					,	atol=1E-3		)		)

          # Test 2-D numpy arrays are batched.
          snake_case__    :       str       		= [floats_list((1, x)		)[0] for x in (800, 800, 800)]
          snake_case__    :       Dict       		= np.asarray(snake_case_		)
          snake_case__    :       Union[str, Any]       		= feature_extractor(snake_case_					,	return_tensors="""np"""		).input_features
          snake_case__    :       List[str]       		= feature_extractor(snake_case_					,	return_tensors="""np"""		).input_features
          for enc_seq_a, enc_seq_a in zip(snake_case_					,	snake_case_		):
                self.assertTrue(np.allclose(snake_case_					,	snake_case_					,	atol=1E-3		)		)

          # Test truncation required
          snake_case__    :       Dict       		= [floats_list((1, x)		)[0] for x in range(200					,	(feature_extractor.n_samples + 500)					,	200		)]
          snake_case__    :       List[str]       		= [np.asarray(snake_case_		) for speech_input in speech_inputs]

          snake_case__    :       int       		= [x[: feature_extractor.n_samples] for x in speech_inputs]
          snake_case__    :       int       		= [np.asarray(snake_case_		) for speech_input in speech_inputs_truncated]

          snake_case__    :       int       		= feature_extractor(snake_case_					,	return_tensors="""np"""		).input_features
          snake_case__    :       List[str]       		= feature_extractor(snake_case_					,	return_tensors="""np"""		).input_features
          for enc_seq_a, enc_seq_a in zip(snake_case_					,	snake_case_		):
                self.assertTrue(np.allclose(snake_case_					,	snake_case_					,	atol=1E-3		)		)




    def lowerCamelCase			(       self   :	Union[str, Any]		):
          import torch

          snake_case__    :       int       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
          snake_case__    :       Dict       		= np.random.rand(100					,	32		).astype(np.floataa		)
          snake_case__    :       List[str]       		= np_speech_inputs.tolist()

          for inputs in [py_speech_inputs, np_speech_inputs]:
                snake_case__    :       Tuple       		= feature_extractor.pad([{"""input_features""": inputs}]					,	return_tensors="""np"""		)
                self.assertTrue(np_processed.input_features.dtype == np.floataa		)
                snake_case__    :       List[Any]       		= feature_extractor.pad([{"""input_features""": inputs}]					,	return_tensors="""pt"""		)
                self.assertTrue(pt_processed.input_features.dtype == torch.floataa		)




    def lowerCamelCase			(       self   :	Dict					,	snake_case_   :	Any		):
          snake_case__    :       Optional[Any]       		= load_dataset("""hf-internal-testing/librispeech_asr_dummy"""					,	"""clean"""					,	split="""validation"""		)
          # automatic decoding with librispeech
          snake_case__    :       int       		= ds.sort("""id"""		).select(range(snake_case_		)		)[:num_samples]["""audio"""]

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




    def lowerCamelCase			(       self   :	int		):
          # fmt: off
          snake_case__    :       str       		= torch.tensor(
              [
                  0.1193, -0.0946, -0.1098, -0.0196, 0.0225, -0.0690, -0.1736, 0.0951,
                  0.0971, -0.0817, -0.0702, 0.0162, 0.0260, 0.0017, -0.0192, -0.1678,
                  0.0709, -0.1867, -0.0655, -0.0274, -0.0234, -0.1884, -0.0516, -0.0554,
                  -0.0274, -0.1425, -0.1423, 0.0837, 0.0377, -0.0854
              ]		)
          # fmt: on

          snake_case__    :       Any       		= self._load_datasamples(1		)
          snake_case__    :       List[Any]       		= WhisperFeatureExtractor()
          snake_case__    :       Dict       		= feature_extractor(snake_case_					,	return_tensors="""pt"""		).input_features
          self.assertEqual(input_features.shape					,	(1, 80, 3_000)		)
          self.assertTrue(torch.allclose(input_features[0, 0, :30]					,	snake_case_					,	atol=1E-4		)		)




    def lowerCamelCase			(       self   :	int		):
          snake_case__    :       Tuple       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
          snake_case__    :       Optional[int]       		= self._load_datasamples(1		)[0]
          snake_case__    :       int       		= ((audio - audio.min()) / (audio.max() - audio.min())) * 65_535  # Rescale to [0, 65535] to show issue
          snake_case__    :       int       		= feat_extract.zero_mean_unit_var_norm([audio]					,	attention_mask=snake_case_		)[0]

          self.assertTrue(np.all(np.mean(snake_case_		) < 1E-3		)		)
          self.assertTrue(np.all(np.abs(np.var(snake_case_		) - 1		) < 1E-3		)		)





'''simple docstring'''





import baseaa
import io
import json
import os
from copy import deepcopy

from ..optimizer import AcceleratedOptimizer
from ..scheduler import AcceleratedScheduler







class        UpperCAmelCase_			:
    """simple docstring"""







    def __init__(       self   :	Tuple					,	snake_case_   :	List[str]		):
          if isinstance(snake_case_					,	snake_case_		):
                # Don't modify user's data should they want to reuse it (e.g. in tests), because once we
                # modified it, it will not be accepted here again, since `auto` values would have been overridden
                snake_case__    :       Optional[Any]       		= deepcopy(snake_case_		)
          elif os.path.exists(snake_case_		):
                with io.open(snake_case_					,	"""r"""					,	encoding="""utf-8"""		) as f:
                      snake_case__    :       Union[str, Any]       		= json.load(snake_case_		)
          else:
                try:
                      snake_case__    :       List[Any]       		= baseaa.urlsafe_baadecode(snake_case_		).decode("""utf-8"""		)
                      snake_case__    :       Optional[int]       		= json.loads(snake_case_		)
                except (UnicodeDecodeError, AttributeError, ValueError):
                      raise ValueError(
                          f"Expected a string path to an existing deepspeed config, or a dictionary, or a base64 encoded string. Received: {config_file_or_dict}"		)

          snake_case__    :       str       		= config

          self.set_stage_and_offload()




    def lowerCamelCase			(       self   :	int		):
          # zero stage - this is done as early as possible, before model is created, to allow
          # ``is_deepspeed_zero3_enabled`` query and getting to the early deepspeed config object
          # during ``zero.Init()`` which needs to know the dtype, and some other hparams.
          snake_case__    :       Union[str, Any]       		= self.get_value("""zero_optimization.stage"""					,	-1		)

          # offload
          snake_case__    :       Optional[Any]       		= False
          if self.is_zeroa() or self.is_zeroa():
                snake_case__    :       int       		= set(["""cpu""", """nvme"""]		)
                snake_case__    :       Any       		= set(
                    [
                        self.get_value("""zero_optimization.offload_optimizer.device"""		),
                        self.get_value("""zero_optimization.offload_param.device"""		),
                    ]		)
                if len(offload_devices & offload_devices_valid		) > 0:
                      snake_case__    :       List[Any]       		= True




    def lowerCamelCase			(       self   :	Any					,	snake_case_   :	Optional[Any]		):
          snake_case__    :       Optional[Any]       		= self.config

          # find the config node of interest if it exists
          snake_case__    :       List[Any]       		= ds_key_long.split("""."""		)
          snake_case__    :       Tuple       		= nodes.pop()
          for node in nodes:
                snake_case__    :       Optional[Any]       		= config.get(snake_case_		)
                if config is None:
                      return None, ds_key

          return config, ds_key




    def lowerCamelCase			(       self   :	List[Any]					,	snake_case_   :	int					,	snake_case_   :	List[str]=None		):
          snake_case__		,							snake_case__    :       Dict       		= self.find_config_node(snake_case_		)
          if config is None:
                return default
          return config.get(snake_case_					,	snake_case_		)




    def lowerCamelCase			(       self   :	Dict					,	snake_case_   :	Optional[int]					,	snake_case_   :	int=False		):
          snake_case__    :       Dict       		= self.config

          # find the config node of interest if it exists
          snake_case__    :       str       		= ds_key_long.split("""."""		)
          for node in nodes:
                snake_case__    :       Union[str, Any]       		= config
                snake_case__    :       List[Any]       		= config.get(snake_case_		)
                if config is None:
                      if must_exist:
                            raise ValueError(f"Can't find {ds_key_long} entry in the config: {self.config}"		)
                      else:
                            return

        # if found remove it
          if parent_config is not None:
                parent_config.pop(snake_case_		)




    def lowerCamelCase			(       self   :	str					,	snake_case_   :	Optional[Any]		):
          snake_case__    :       Dict       		= self.get_value(snake_case_		)
          return False if value is None else bool(snake_case_		)




    def lowerCamelCase			(       self   :	List[str]					,	snake_case_   :	Union[str, Any]		):
          snake_case__    :       Optional[int]       		= self.get_value(snake_case_		)
          return False if value is None else not bool(snake_case_		)




    def lowerCamelCase			(       self   :	Optional[Any]		):
          return self._stage == 2




    def lowerCamelCase			(       self   :	Dict		):
          return self._stage == 3




    def lowerCamelCase			(       self   :	List[Any]		):
          return self._offload







class        UpperCAmelCase_			:
    """simple docstring"""







    def __init__(       self   :	Any					,	snake_case_   :	List[Any]		):
          snake_case__    :       Tuple       		= engine




    def lowerCamelCase			(       self   :	Dict					,	snake_case_   :	Optional[Any]					,	**snake_case_   :	int		):
          # runs backpropagation and handles mixed precision
          self.engine.backward(snake_case_					,	**snake_case_		)

          # Deepspeed's `engine.step` performs the following operations:
          # - gradient accumulation check
          # - gradient clipping
          # - optimizer step
          # - zero grad
          # - checking overflow
          # - lr_scheduler step (only if engine.lr_scheduler is not None)
          self.engine.step()
          # and this plugin overrides the above calls with no-ops when Accelerate runs under
          # Deepspeed, but allows normal functionality for non-Deepspeed cases thus enabling a simple
          # training loop that works transparently under many training regimes.







class        UpperCAmelCase_			(			_a      ):
    """simple docstring"""







    def __init__(       self   :	Tuple					,	snake_case_   :	Any		):
          super().__init__(snake_case_					,	device_placement=snake_case_					,	scaler=snake_case_		)
          snake_case__    :       Optional[int]       		= hasattr(self.optimizer					,	"""overflow"""		)




    def lowerCamelCase			(       self   :	int					,	snake_case_   :	Optional[Any]=None		):
          pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed




    def lowerCamelCase			(       self   :	List[Any]		):
          pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed




    @property
    def lowerCamelCase			(       self   :	List[str]		):
          if self.__has_overflow__:
                return self.optimizer.overflow
          return False







class        UpperCAmelCase_			(			_a      ):
    """simple docstring"""







    def __init__(       self   :	Tuple					,	snake_case_   :	Optional[Any]					,	snake_case_   :	Tuple		):
          super().__init__(snake_case_					,	snake_case_		)




    def lowerCamelCase			(       self   :	Optional[int]		):
          pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed







class        UpperCAmelCase_			:
    """simple docstring"""







    def __init__(       self   :	Optional[int]					,	snake_case_   :	Dict					,	snake_case_   :	int=0.001					,	snake_case_   :	Optional[Any]=0					,	**snake_case_   :	List[Any]		):
          snake_case__    :       int       		= params
          snake_case__    :       Optional[Any]       		= lr
          snake_case__    :       Any       		= weight_decay
          snake_case__    :       Optional[Any]       		= kwargs







class        UpperCAmelCase_			:
    """simple docstring"""







    def __init__(       self   :	Dict					,	snake_case_   :	str					,	snake_case_   :	Optional[int]=None					,	snake_case_   :	Any=0					,	**snake_case_   :	List[str]		):
          snake_case__    :       List[Any]       		= optimizer
          snake_case__    :       List[Any]       		= total_num_steps
          snake_case__    :       List[Any]       		= warmup_num_steps
          snake_case__    :       Optional[Any]       		= kwargs
