


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			:						Union[str, Any]        =  random.Random()






def 					UpperCamelCase   (      __magic_name__	:				Tuple       ,       __magic_name__	:				List[Any]=1.0       ,       __magic_name__	:				Dict=None       ,       __magic_name__	:				List[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



@require_torch
@require_torchaudio
class 				A       (			unittest.TestCase				):
					'''simple docstring'''






					def __init__(self       :		Union[str, Any] ,   _UpperCAmelCase       :		Union[str, Any] ,   _UpperCAmelCase       :		List[Any]=7 ,   _UpperCAmelCase       :		int=400 ,   _UpperCAmelCase       :		Tuple=2000 ,   _UpperCAmelCase       :		str=10 ,   _UpperCAmelCase       :		Union[str, Any]=160 ,   _UpperCAmelCase       :		List[Any]=8 ,   _UpperCAmelCase       :		List[str]=0.0 ,   _UpperCAmelCase       :		int=4000 ,   _UpperCAmelCase       :		Optional[int]=False ,   _UpperCAmelCase       :		Optional[int]=True ,   )		->  Tuple:




							"""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__	     =						padding_value
							lowercase__	     =						sampling_rate
							lowercase__	     =						return_attention_mask
							lowercase__	     =						do_normalize
							lowercase__	     =						feature_size
							lowercase__	     =						chunk_length
							lowercase__	     =						hop_length

					def     lowerCamelCase__	(self       :		str						)		->  Any:




							"""simple docstring"""


							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       :		Dict ,   _UpperCAmelCase       :		str=False ,   _UpperCAmelCase       :		Optional[int]=False						)		->  Tuple:




							"""simple docstring"""


							def _flatten(_UpperCAmelCase       :		Any						):
									return list(itertools.chain(*_UpperCAmelCase						)						)

							if equal_length:
									lowercase__	     =						[floats_list((self.max_seq_length, self.feature_size)						) for _ in range(self.batch_size						)]
							else:
									# make sure that inputs increase in size
									lowercase__	     =						[
									    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(_UpperCAmelCase						) for x in speech_inputs]
							return speech_inputs






@require_torch
@require_torchaudio
class 				A       (			UpperCAmelCase__					,    unittest.TestCase				):
					'''simple docstring'''






					A__     =					WhisperFeatureExtractor if is_speech_available() else None

					def     lowerCamelCase__	(self       :		List[Any]						)		->  Any:




							"""simple docstring"""


							lowercase__	     =						WhisperFeatureExtractionTester(self						)

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




							"""simple docstring"""


							lowercase__	     =						self.feature_extraction_class(**self.feat_extract_dict						)

							with tempfile.TemporaryDirectory() as tmpdirname:
									lowercase__	     =						feat_extract_first.save_pretrained(_UpperCAmelCase						)[0]
									check_json_file_has_correct_format(_UpperCAmelCase						)
									lowercase__	     =						self.feature_extraction_class.from_pretrained(_UpperCAmelCase						)

							lowercase__	     =						feat_extract_first.to_dict()
							lowercase__	     =						feat_extract_second.to_dict()
							lowercase__	     =						feat_extract_first.mel_filters
							lowercase__	     =						feat_extract_second.mel_filters
							self.assertTrue(np.allclose(_UpperCAmelCase ,   _UpperCAmelCase						)						)
							self.assertEqual(_UpperCAmelCase ,   _UpperCAmelCase						)

					def     lowerCamelCase__	(self       :		List[Any]						)		->  List[Any]:




							"""simple docstring"""


							lowercase__	     =						self.feature_extraction_class(**self.feat_extract_dict						)

							with tempfile.TemporaryDirectory() as tmpdirname:
									lowercase__	     =						os.path.join(_UpperCAmelCase ,   """feat_extract.json"""						)
									feat_extract_first.to_json_file(_UpperCAmelCase						)
									lowercase__	     =						self.feature_extraction_class.from_json_file(_UpperCAmelCase						)

							lowercase__	     =						feat_extract_first.to_dict()
							lowercase__	     =						feat_extract_second.to_dict()
							lowercase__	     =						feat_extract_first.mel_filters
							lowercase__	     =						feat_extract_second.mel_filters
							self.assertTrue(np.allclose(_UpperCAmelCase ,   _UpperCAmelCase						)						)
							self.assertEqual(_UpperCAmelCase ,   _UpperCAmelCase						)

					def     lowerCamelCase__	(self       :		List[str]						)		->  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 ,   1400 ,   200						)]
							lowercase__	     =						[np.asarray(_UpperCAmelCase						) for speech_input in speech_inputs]

							# Test feature size
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   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
							lowercase__	     =						feature_extractor(speech_inputs[0] ,   return_tensors="""np"""						).input_features
							lowercase__	     =						feature_extractor(np_speech_inputs[0] ,   return_tensors="""np"""						).input_features
							self.assertTrue(np.allclose(_UpperCAmelCase ,   _UpperCAmelCase ,   atol=1E-3						)						)

							# Test batched
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""np"""						).input_features
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""np"""						).input_features
							for enc_seq_a, enc_seq_a in zip(_UpperCAmelCase ,   _UpperCAmelCase						):
									self.assertTrue(np.allclose(_UpperCAmelCase ,   _UpperCAmelCase ,   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(_UpperCAmelCase						)
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""np"""						).input_features
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""np"""						).input_features
							for enc_seq_a, enc_seq_a in zip(_UpperCAmelCase ,   _UpperCAmelCase						):
									self.assertTrue(np.allclose(_UpperCAmelCase ,   _UpperCAmelCase ,   atol=1E-3						)						)

							# Test truncation required
							lowercase__	     =						[floats_list((1, x)						)[0] for x in range(200 ,   (feature_extractor.n_samples + 500) ,   200						)]
							lowercase__	     =						[np.asarray(_UpperCAmelCase						) for speech_input in speech_inputs]

							lowercase__	     =						[x[: feature_extractor.n_samples] for x in speech_inputs]
							lowercase__	     =						[np.asarray(_UpperCAmelCase						) for speech_input in speech_inputs_truncated]

							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""np"""						).input_features
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""np"""						).input_features
							for enc_seq_a, enc_seq_a in zip(_UpperCAmelCase ,   _UpperCAmelCase						):
									self.assertTrue(np.allclose(_UpperCAmelCase ,   _UpperCAmelCase ,   atol=1E-3						)						)

					def     lowerCamelCase__	(self       :		List[str]						)		->  Any:




							"""simple docstring"""


							import torch

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

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

					def     lowerCamelCase__	(self       :		List[str] ,   _UpperCAmelCase       :		Union[str, Any]						)		->  List[Any]:




							"""simple docstring"""


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

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

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




							"""simple docstring"""


							lowercase__	     =						torch.tensor(
							    [
							        0.1_193, -0.0_946, -0.1_098, -0.0_196, 0.0_225, -0.0_690, -0.1_736, 0.0_951,
							        0.0_971, -0.0_817, -0.0_702, 0.0_162, 0.0_260, 0.0_017, -0.0_192, -0.1_678,
							        0.0_709, -0.1_867, -0.0_655, -0.0_274, -0.0_234, -0.1_884, -0.0_516, -0.0_554,
							        -0.0_274, -0.1_425, -0.1_423, 0.0_837, 0.0_377, -0.0_854
							    ]						)
							# fmt: on

							lowercase__	     =						self._load_datasamples(1						)
							lowercase__	     =						WhisperFeatureExtractor()
							lowercase__	     =						feature_extractor(_UpperCAmelCase ,   return_tensors="""pt"""						).input_features
							self.assertEqual(input_features.shape ,   (1, 80, 3000)						)
							self.assertTrue(torch.allclose(input_features[0, 0, :30] ,   _UpperCAmelCase ,   atol=1E-4						)						)


					def     lowerCamelCase__	(self       :		Optional[Any]						)		->  Optional[Any]:




							"""simple docstring"""


							lowercase__	     =						self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							lowercase__	     =						self._load_datasamples(1						)[0]
							lowercase__	     =						((audio - audio.min()) / (audio.max() - audio.min())) * 6_5535  # Rescale to [0, 65535] to show issue
							lowercase__	     =						feat_extract.zero_mean_unit_var_norm([audio] ,   attention_mask=_UpperCAmelCase						)[0]

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


'''simple docstring'''




# 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 							snake_case__				(     _A:   Dict="no" ,  _A:   str = default_json_config_file ,  _A:   bool = False					)       ->      List[str]:

		'''simple docstring'''


		lowerCAmelCase     =    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
		lowerCAmelCase     =    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}"					)
		lowerCAmelCase     =    {
		    """compute_environment""": """LOCAL_MACHINE""",
		    """mixed_precision""": mixed_precision,
		}
		if torch.cuda.is_available():
				lowerCAmelCase     =    torch.cuda.device_count()
				lowerCAmelCase     =    num_gpus
				lowerCAmelCase     =    False
				if num_gpus > 1:
						lowerCAmelCase     =    """MULTI_GPU"""
				else:
						lowerCAmelCase     =    """NO"""
		elif is_xpu_available() and use_xpu:
				lowerCAmelCase     =    torch.xpu.device_count()
				lowerCAmelCase     =    num_xpus
				lowerCAmelCase     =    False
				if num_xpus > 1:
						lowerCAmelCase     =    """MULTI_XPU"""
				else:
						lowerCAmelCase     =    """NO"""
		elif is_npu_available():
				lowerCAmelCase     =    torch.npu.device_count()
				lowerCAmelCase     =    num_npus
				lowerCAmelCase     =    False
				if num_npus > 1:
						lowerCAmelCase     =    """MULTI_NPU"""
				else:
						lowerCAmelCase     =    """NO"""
		else:
				lowerCAmelCase     =    0
				lowerCAmelCase     =    True
				lowerCAmelCase     =    1
				lowerCAmelCase     =    """NO"""
		lowerCAmelCase     =    ClusterConfig(**_A					)
		config.to_json_file(_A					)
		return path






def 							snake_case__				(     _A:   str ,  _A:   Any					)       ->      Optional[Any]:

		'''simple docstring'''


		lowerCAmelCase     =    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 							snake_case__				(     _A:   Optional[int]					)       ->      Optional[int]:

		'''simple docstring'''


		lowerCAmelCase     =    write_basic_config(args.mixed_precision ,  args.save_location					)
		if config_file:
				print(f"accelerate configuration saved at {config_file}"					)




