

import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


lowercase_									=   random.Random()

if is_torch_available():
  import torch
def      a      (			A__	:   Tuple      ,   A__	:   Dict=1.0      ,   A__	:   List[Any]=None      ,   A__	:   Optional[Any]=None     )    ->      Optional[Any]:





      """simple docstring"""



      if rng is None:
            _lowercase	       =global_rng

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

      return values





class  __lowerCAmelCase     (					unittest.TestCase							):




  def __init__(				self							,   lowerCAmelCase							,   lowerCAmelCase=7							,   lowerCAmelCase=400							,   lowerCAmelCase=2_000							,   lowerCAmelCase=1							,   lowerCAmelCase=0.0							,   lowerCAmelCase=16_000							,   lowerCAmelCase=True							,   lowerCAmelCase=True							,   )  -> List[str]:
        '''simple docstring'''






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




  def   A__     (				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,
            "do_normalize": self.do_normalize,
        }




  def   A__     (				self							,   lowerCAmelCase=False							,   lowerCAmelCase=False					)  -> Dict:
        '''simple docstring'''






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

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

        return speech_inputs





@require_torch
@require_torchaudio
class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE			,			unittest.TestCase							):
  _a									=  ASTFeatureExtractor




  def   A__     (				self					)  -> str:
        '''simple docstring'''






        _lowercase	       =ASTFeatureExtractionTester(self					)




  def   A__     (				self					)  -> List[str]:
        '''simple docstring'''






        _lowercase	       =self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
        # create three inputs of length 800, 1000, and 1200
        _lowercase	       =[floats_list((1, x)					)[0] for x in range(800							,   1_400							,   200					)]
        _lowercase	       =[np.asarray(lowerCAmelCase					) for speech_input in speech_inputs]

        # Test not batched input
        _lowercase	       =feat_extract(speech_inputs[0]							,   return_tensors='np'					).input_values
        _lowercase	       =feat_extract(np_speech_inputs[0]							,   return_tensors='np'					).input_values
        self.assertTrue(np.allclose(lowerCAmelCase							,   lowerCAmelCase							,   atol=1e-3					)					)

        # Test batched
        _lowercase	       =feat_extract(lowerCAmelCase							,   padding=lowerCAmelCase							,   return_tensors='np'					).input_values
        _lowercase	       =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					)					)

        # Test 2-D numpy arrays are batched.
        _lowercase	       =[floats_list((1, x)					)[0] for x in (800, 800, 800)]
        _lowercase	       =np.asarray(lowerCAmelCase					)
        _lowercase	       =feat_extract(lowerCAmelCase							,   return_tensors='np'					).input_values
        _lowercase	       =feat_extract(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					)					)




  @require_torch
  def   A__     (				self					)  -> Union[str, Any]:
        '''simple docstring'''






        import torch

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

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




  def   A__     (				self							,   lowerCAmelCase					)  -> str:
        '''simple docstring'''






        from datasets import load_dataset

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

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




  @require_torch
  def   A__     (				self					)  -> Any:
        '''simple docstring'''






        _lowercase	       =torch.tensor(
            [-0.9894, -1.2776, -0.9066, -1.2776, -0.9349, -1.2609, -1.0386, -1.2776,
             -1.1561, -1.2776, -1.2052, -1.2723, -1.2190, -1.2132, -1.2776, -1.1133,
             -1.1953, -1.1343, -1.1584, -1.2203, -1.1770, -1.2474, -1.2381, -1.1936,
             -0.9270, -0.8317, -0.8049, -0.7706, -0.7565, -0.7869]					)
        # fmt: on

        _lowercase	       =self._load_datasamples(1					)
        _lowercase	       =ASTFeatureExtractor()
        _lowercase	       =feature_extractor(lowerCAmelCase							,   return_tensors='pt'					).input_values
        self.assertEquals(input_values.shape							,   (1, 1_024, 128)					)
        self.assertTrue(torch.allclose(input_values[0, 0, :30]							,   lowerCAmelCase							,   atol=1e-4					)					)





# Copyright 2021 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.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


lowercase_									=   'Create a default config file for Accelerate with only a few flags set.'
def      a      (			A__	:   Optional[Any]="no"      ,   A__	:   str = default_json_config_file      ,   A__	:   bool = False     )    ->      Optional[int]:





      """simple docstring"""



      _lowercase	       =Path(A__     )
      path.parent.mkdir(parents=A__      ,   exist_ok=A__     )
      if path.exists():
            print(
                F'''Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`.'''     )
            return False
      _lowercase	       =mixed_precision.lower()
      if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
            raise ValueError(
                F'''`mixed_precision` should be one of \'no\', \'fp16\', \'bf16\', or \'fp8\'. Received {mixed_precision}'''     )
      _lowercase	       ={
          'compute_environment': 'LOCAL_MACHINE',
          'mixed_precision': mixed_precision,
      }
      if torch.cuda.is_available():
            _lowercase	       =torch.cuda.device_count()
            _lowercase	       =num_gpus
            _lowercase	       =False
            if num_gpus > 1:
                  _lowercase	       ='MULTI_GPU'
            else:
                  _lowercase	       ='NO'
      elif is_xpu_available() and use_xpu:
            _lowercase	       =torch.xpu.device_count()
            _lowercase	       =num_xpus
            _lowercase	       =False
            if num_xpus > 1:
                  _lowercase	       ='MULTI_XPU'
            else:
                  _lowercase	       ='NO'
      elif is_npu_available():
            _lowercase	       =torch.npu.device_count()
            _lowercase	       =num_npus
            _lowercase	       =False
            if num_npus > 1:
                  _lowercase	       ='MULTI_NPU'
            else:
                  _lowercase	       ='NO'
      else:
            _lowercase	       =0
            _lowercase	       =True
            _lowercase	       =1
            _lowercase	       ='NO'
      _lowercase	       =ClusterConfig(**A__     )
      config.to_json_file(A__     )
      return path
def      a      (			A__	:   Dict      ,   A__	:   Optional[Any]     )    ->      List[Any]:





      """simple docstring"""



      _lowercase	       =parser.add_parser('default'      ,   parents=A__      ,   help=A__      ,   formatter_class=A__     )
      parser.add_argument(
          '--config_file'      ,   default=A__      ,   help=(
              'The path to use to store the config file. Will default to a file named default_config.yaml in the cache '
              'location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '
              'such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '
              'with \'huggingface\'.'
          )      ,   dest='save_location'      ,   )

      parser.add_argument(
          '--mixed_precision'      ,   choices=['no', 'fp16', 'bf16']      ,   type=A__      ,   help='Whether or not to use mixed precision training. '
          'Choose between FP16 and BF16 (bfloat16) training. '
          'BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later.'      ,   default='no'      ,   )
      parser.set_defaults(func=A__     )
      return parser







def      a      (			A__	:   List[str]     )    ->      Any:





      """simple docstring"""



      _lowercase	       =write_basic_config(args.mixed_precision      ,   args.save_location     )
      if config_file:
            print(F'''accelerate configuration saved at {config_file}'''     )


