import itertools
import random
import unittest

import numpy as np

from transformers import BatchFeature, SpeechTaFeatureExtractor
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__           =							random.Random()





def  _lowerCamelCase					(   a_				: Optional[Any]						,			a_				: Optional[Any]=1.0						,			a_				: List[Any]=None						,			a_				: Union[str, Any]=None):

			if rng is None:
						lowerCamelCase						:Any						  =     global_rng

			lowerCamelCase						:List[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
class   _lowerCAmelCase		(     unittest.TestCase				):







							def __init__(				self     :    List[Any]	,					__snake_case     :    Dict	,					__snake_case     :    Optional[Any]=7	,					__snake_case     :    Any=400	,					__snake_case     :    Tuple=2000	,					__snake_case     :    Optional[Any]=1	,					__snake_case     :    Tuple=0.0	,					__snake_case     :    int=16000	,					__snake_case     :    List[str]=True	,					__snake_case     :    str=80	,					__snake_case     :    Union[str, Any]=16	,					__snake_case     :    str=64	,					__snake_case     :    Optional[Any]="hann_window"	,					__snake_case     :    Any=80	,					__snake_case     :    int=7600	,					__snake_case     :    Optional[int]=1e-1_0	,					__snake_case     :    int=True	,					):
										lowerCamelCase						:int						  =     parent
										lowerCamelCase						:Dict						  =     batch_size
										lowerCamelCase						:List[Any]						  =     min_seq_length
										lowerCamelCase						:List[Any]						  =     max_seq_length
										lowerCamelCase						:Optional[Any]						  =     (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
										lowerCamelCase						:Optional[Any]						  =     feature_size
										lowerCamelCase						:Optional[Any]						  =     padding_value
										lowerCamelCase						:Union[str, Any]						  =     sampling_rate
										lowerCamelCase						:Optional[Any]						  =     do_normalize
										lowerCamelCase						:Tuple						  =     num_mel_bins
										lowerCamelCase						:List[str]						  =     hop_length
										lowerCamelCase						:Any						  =     win_length
										lowerCamelCase						:Union[str, Any]						  =     win_function
										lowerCamelCase						:List[str]						  =     fmin
										lowerCamelCase						:Union[str, Any]						  =     fmax
										lowerCamelCase						:List[Any]						  =     mel_floor
										lowerCamelCase						:Optional[int]						  =     return_attention_mask







							def 				snake_case  (				self     :    List[str]		):
										return {
										    "feature_size": self.feature_size,
										    "padding_value": self.padding_value,
										    "sampling_rate": self.sampling_rate,
										    "do_normalize": self.do_normalize,
										    "num_mel_bins": self.num_mel_bins,
										    "hop_length": self.hop_length,
										    "win_length": self.win_length,
										    "win_function": self.win_function,
										    "fmin": self.fmin,
										    "fmax": self.fmax,
										    "mel_floor": self.mel_floor,
										    "return_attention_mask": self.return_attention_mask,
										}







							def 				snake_case  (				self     :    List[Any]	,					__snake_case     :    int=False	,					__snake_case     :    Tuple=False		):






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

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

										return speech_inputs







							def 				snake_case  (				self     :    int	,					__snake_case     :    Tuple=False	,					__snake_case     :    Union[str, Any]=False		):






										if equal_length:
													lowerCamelCase						:Union[str, Any]						  =     [floats_list((self.max_seq_length, self.num_mel_bins)		) for _ in range(self.batch_size		)]
										else:
													# make sure that inputs increase in size
													lowerCamelCase						:Optional[Any]						  =     [
													    floats_list((x, self.num_mel_bins)		)
													    for x in range(self.min_seq_length	,					self.max_seq_length	,					self.seq_length_diff		)
													]

										if numpify:
													lowerCamelCase						:Tuple						  =     [np.asarray(__lowerCamelCase		) for x in speech_inputs]

										return speech_inputs


@require_torch
class   _lowerCAmelCase		(     lowerCamelCase__      ,   unittest.TestCase				):
							_UpperCAmelCase						    =     SpeechTaFeatureExtractor







							def 				snake_case  (				self     :    str		):
										lowerCamelCase						:Tuple						  =     SpeechTaFeatureExtractionTester(self		)







							def 				snake_case  (				self     :    Union[str, Any]	,					__snake_case     :    Union[str, Any]		):
										self.assertTrue(np.all(np.mean(__lowerCamelCase	,					axis=0		) < 1e-3		)		)
										self.assertTrue(np.all(np.abs(np.var(__lowerCamelCase	,					axis=0		) - 1		) < 1e-3		)		)







							def 				snake_case  (				self     :    Any		):
										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						:Optional[int]						  =     [np.asarray(__lowerCamelCase		) for speech_input in speech_inputs]

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

										# Test batched
										lowerCamelCase						:int						  =     feat_extract(__lowerCamelCase	,					return_tensors='''np'''		).input_values
										lowerCamelCase						:str						  =     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		)		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:str						  =     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
										lowerCamelCase						:Optional[Any]						  =     [floats_list((1, x)		)[0] for x in range(800	,					1400	,					200		)]

										lowerCamelCase						:List[str]						  =     ['''longest''', '''max_length''', '''do_not_pad''']
										lowerCamelCase						:Optional[Any]						  =     [None, 1600, None]
										for max_length, padding in zip(__lowerCamelCase	,					__lowerCamelCase		):
													lowerCamelCase						:Any						  =     feat_extract(__lowerCamelCase	,					padding=__lowerCamelCase	,					max_length=__lowerCamelCase	,					return_tensors='''np'''		)
													lowerCamelCase						:Optional[int]						  =     processed.input_values

													self._check_zero_mean_unit_variance(input_values[0][:800]		)
													self.assertTrue(input_values[0][800:].sum() < 1e-6		)
													self._check_zero_mean_unit_variance(input_values[1][:1000]		)
													self.assertTrue(input_values[0][1000:].sum() < 1e-6		)
													self._check_zero_mean_unit_variance(input_values[2][:1200]		)







							def 				snake_case  (				self     :    List[Any]		):
										lowerCamelCase						:List[str]						  =     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
										lowerCamelCase						:List[str]						  =     range(800	,					1400	,					200		)
										lowerCamelCase						:Optional[Any]						  =     [floats_list((1, x)		)[0] for x in lengths]

										lowerCamelCase						:Union[str, Any]						  =     ['''longest''', '''max_length''', '''do_not_pad''']
										lowerCamelCase						:int						  =     [None, 1600, None]

										for max_length, padding in zip(__lowerCamelCase	,					__lowerCamelCase		):
													lowerCamelCase						:str						  =     feat_extract(__lowerCamelCase	,					max_length=__lowerCamelCase	,					padding=__lowerCamelCase		)
													lowerCamelCase						:Tuple						  =     processed.input_values

													self._check_zero_mean_unit_variance(input_values[0][:800]		)
													self._check_zero_mean_unit_variance(input_values[1][:1000]		)
													self._check_zero_mean_unit_variance(input_values[2][:1200]		)







							def 				snake_case  (				self     :    List[str]		):
										lowerCamelCase						:Dict						  =     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
										lowerCamelCase						:Optional[Any]						  =     [floats_list((1, x)		)[0] for x in range(800	,					1400	,					200		)]
										lowerCamelCase						:Optional[int]						  =     feat_extract(
										    __lowerCamelCase	,					truncation=__lowerCamelCase	,					max_length=1000	,					padding='''max_length'''	,					return_tensors='''np'''		)
										lowerCamelCase						:List[Any]						  =     processed.input_values

										self._check_zero_mean_unit_variance(input_values[0, :800]		)
										self._check_zero_mean_unit_variance(input_values[1]		)
										self._check_zero_mean_unit_variance(input_values[2]		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:List[Any]						  =     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
										lowerCamelCase						:List[str]						  =     [floats_list((1, x)		)[0] for x in range(800	,					1400	,					200		)]
										lowerCamelCase						:Union[str, Any]						  =     feat_extract(
										    __lowerCamelCase	,					truncation=__lowerCamelCase	,					max_length=1000	,					padding='''longest'''	,					return_tensors='''np'''		)
										lowerCamelCase						:Optional[int]						  =     processed.input_values

										self._check_zero_mean_unit_variance(input_values[0, :800]		)
										self._check_zero_mean_unit_variance(input_values[1, :1000]		)
										self._check_zero_mean_unit_variance(input_values[2]		)

										# make sure that if max_length < longest -> then pad to max_length
										self.assertTrue(input_values.shape == (3, 1000)		)

										lowerCamelCase						:Tuple						  =     [floats_list((1, x)		)[0] for x in range(800	,					1400	,					200		)]
										lowerCamelCase						:Optional[int]						  =     feat_extract(
										    __lowerCamelCase	,					truncation=__lowerCamelCase	,					max_length=2000	,					padding='''longest'''	,					return_tensors='''np'''		)
										lowerCamelCase						:Optional[Any]						  =     processed.input_values

										self._check_zero_mean_unit_variance(input_values[0, :800]		)
										self._check_zero_mean_unit_variance(input_values[1, :1000]		)
										self._check_zero_mean_unit_variance(input_values[2]		)

										# make sure that if max_length > longest -> then pad to longest
										self.assertTrue(input_values.shape == (3, 1200)		)







							def 				snake_case  (				self     :    Optional[int]		):
										lowerCamelCase						:List[Any]						  =     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
										lowerCamelCase						:Union[str, Any]						  =     np.random.rand(100		).astype(np.floataa		)
										lowerCamelCase						:Any						  =     np_speech_inputs.tolist()

										for inputs in [py_speech_inputs, np_speech_inputs]:
													lowerCamelCase						:str						  =     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 				snake_case  (				self     :    Optional[Any]		):
										lowerCamelCase						:str						  =     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
										# create three inputs of length 800, 1000, and 1200
										lowerCamelCase						:Any						  =     [floats_list((1, x)		)[0] for x in range(800	,					1400	,					200		)]
										lowerCamelCase						:Optional[Any]						  =     [np.asarray(__lowerCamelCase		) for speech_input in speech_inputs]

										# Test feature size
										lowerCamelCase						:List[str]						  =     feature_extractor(audio_target=__lowerCamelCase	,					padding=__lowerCamelCase	,					return_tensors='''np'''		).input_values
										self.assertTrue(input_values.ndim == 3		)
										self.assertTrue(input_values.shape[-1] == feature_extractor.num_mel_bins		)

										# Test not batched input
										lowerCamelCase						:List[str]						  =     feature_extractor(speech_inputs[0]	,					return_tensors='''np'''		).input_values
										lowerCamelCase						:Optional[int]						  =     feature_extractor(np_speech_inputs[0]	,					return_tensors='''np'''		).input_values
										self.assertTrue(np.allclose(__lowerCamelCase	,					__lowerCamelCase	,					atol=1e-3		)		)

										# Test batched
										lowerCamelCase						:Dict						  =     feature_extractor(__lowerCamelCase	,					return_tensors='''np'''		).input_values
										lowerCamelCase						:List[Any]						  =     feature_extractor(__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.
										lowerCamelCase						:Tuple						  =     [floats_list((1, x)		)[0] for x in (800, 800, 800)]
										lowerCamelCase						:Optional[int]						  =     np.asarray(__lowerCamelCase		)
										lowerCamelCase						:Dict						  =     feature_extractor(__lowerCamelCase	,					return_tensors='''np'''		).input_values
										lowerCamelCase						:Dict						  =     feature_extractor(__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 				snake_case  (				self     :    int		):
										lowerCamelCase						:Tuple						  =     self.feat_extract_tester.prepare_inputs_for_target()
										lowerCamelCase						:Union[str, Any]						  =     self.feature_extraction_class(**self.feat_extract_dict		)
										lowerCamelCase						:Optional[int]						  =     feat_extract.model_input_names[0]

										lowerCamelCase						:Dict						  =     BatchFeature({input_name: speech_inputs}		)

										self.assertTrue(all(len(__lowerCamelCase		) == len(__lowerCamelCase		) for x, y in zip(__lowerCamelCase	,					processed_features[input_name]		)		)		)

										lowerCamelCase						:str						  =     self.feat_extract_tester.prepare_inputs_for_target(equal_length=__lowerCamelCase		)
										lowerCamelCase						:List[Any]						  =     BatchFeature({input_name: speech_inputs}	,					tensor_type='''np'''		)

										lowerCamelCase						:int						  =     processed_features[input_name]

										if len(batch_features_input.shape		) < 3:
													lowerCamelCase						:List[str]						  =     batch_features_input[:, :, None]

										self.assertTrue(
										    batch_features_input.shape
										    == (self.feat_extract_tester.batch_size, len(speech_inputs[0]		), self.feat_extract_tester.num_mel_bins)		)







							@require_torch
							def 				snake_case  (				self     :    List[Any]		):
										lowerCamelCase						:List[Any]						  =     self.feat_extract_tester.prepare_inputs_for_target(equal_length=__lowerCamelCase		)
										lowerCamelCase						:List[Any]						  =     self.feature_extraction_class(**self.feat_extract_dict		)
										lowerCamelCase						:int						  =     feat_extract.model_input_names[0]

										lowerCamelCase						:Union[str, Any]						  =     BatchFeature({input_name: speech_inputs}	,					tensor_type='''pt'''		)

										lowerCamelCase						:Optional[int]						  =     processed_features[input_name]

										if len(batch_features_input.shape		) < 3:
													lowerCamelCase						:Optional[int]						  =     batch_features_input[:, :, None]

										self.assertTrue(
										    batch_features_input.shape
										    == (self.feat_extract_tester.batch_size, len(speech_inputs[0]		), self.feat_extract_tester.num_mel_bins)		)







							@require_torch
							def 				snake_case  (				self     :    str		):
										lowerCamelCase						:Optional[Any]						  =     self.feature_extraction_class(**self.feat_extract_dict		)
										lowerCamelCase						:List[str]						  =     self.feat_extract_tester.prepare_inputs_for_target()
										lowerCamelCase						:Optional[int]						  =     feat_extract.model_input_names[0]

										lowerCamelCase						:Union[str, Any]						  =     BatchFeature({input_name: speech_inputs}		)

										lowerCamelCase						:Optional[int]						  =     feat_extract.num_mel_bins  # hack!

										lowerCamelCase						:List[str]						  =     feat_extract.pad(__lowerCamelCase	,					padding='''longest'''	,					return_tensors='''np'''		)[input_name]
										lowerCamelCase						:Any						  =     feat_extract.pad(__lowerCamelCase	,					padding='''longest'''	,					return_tensors='''pt'''		)[input_name]

										self.assertTrue(abs(input_np.astype(np.floataa		).sum() - input_pt.numpy().astype(np.floataa		).sum()		) < 1e-2		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:List[str]						  =     self.feat_extract_dict
										lowerCamelCase						:Optional[Any]						  =     True
										lowerCamelCase						:Optional[Any]						  =     self.feature_extraction_class(**__lowerCamelCase		)
										lowerCamelCase						:int						  =     self.feat_extract_tester.prepare_inputs_for_target()
										lowerCamelCase						:Optional[int]						  =     [len(__lowerCamelCase		) for x in speech_inputs]
										lowerCamelCase						:Union[str, Any]						  =     feat_extract.model_input_names[0]

										lowerCamelCase						:Tuple						  =     BatchFeature({input_name: speech_inputs}		)

										lowerCamelCase						:str						  =     feat_extract.num_mel_bins  # hack!

										lowerCamelCase						:List[Any]						  =     feat_extract.pad(__lowerCamelCase	,					padding='''longest'''	,					return_tensors='''np'''		)
										self.assertIn('''attention_mask'''	,					__lowerCamelCase		)
										self.assertListEqual(list(processed.attention_mask.shape		)	,					list(processed[input_name].shape[:2]		)		)
										self.assertListEqual(processed.attention_mask.sum(-1		).tolist()	,					__lowerCamelCase		)







							def 				snake_case  (				self     :    Any		):
										lowerCamelCase						:List[Any]						  =     self.feat_extract_dict
										lowerCamelCase						:Any						  =     True
										lowerCamelCase						:List[str]						  =     self.feature_extraction_class(**__lowerCamelCase		)
										lowerCamelCase						:Any						  =     self.feat_extract_tester.prepare_inputs_for_target()
										lowerCamelCase						:Tuple						  =     [len(__lowerCamelCase		) for x in speech_inputs]
										lowerCamelCase						:str						  =     feat_extract.model_input_names[0]

										lowerCamelCase						:int						  =     BatchFeature({input_name: speech_inputs}		)
										lowerCamelCase						:Optional[int]						  =     min(__lowerCamelCase		)

										lowerCamelCase						:Dict						  =     feat_extract.num_mel_bins  # hack!

										lowerCamelCase						:Union[str, Any]						  =     feat_extract.pad(
										    __lowerCamelCase	,					padding='''max_length'''	,					max_length=__lowerCamelCase	,					truncation=__lowerCamelCase	,					return_tensors='''np'''		)
										self.assertIn('''attention_mask'''	,					__lowerCamelCase		)
										self.assertListEqual(
										    list(processed_pad.attention_mask.shape		)	,					[processed_pad[input_name].shape[0], max_length]		)
										self.assertListEqual(
										    processed_pad.attention_mask[:, :max_length].sum(-1		).tolist()	,					[max_length for x in speech_inputs]		)







							def 				snake_case  (				self     :    Any	,					__snake_case     :    List[Any]		):
										from datasets import load_dataset

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

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







							def 				snake_case  (				self     :    Dict		):
										lowerCamelCase						:Tuple						  =     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						:Dict						  =     self._load_datasamples(1		)
										lowerCamelCase						:Any						  =     SpeechTaFeatureExtractor()
										lowerCamelCase						:Dict						  =     feature_extractor(__lowerCamelCase	,					return_tensors='''pt'''		).input_values
										self.assertEquals(input_values.shape	,					(1, 93680)		)
										self.assertTrue(torch.allclose(input_values[0, :30]	,					__lowerCamelCase	,					atol=1e-6		)		)







							def 				snake_case  (				self     :    Optional[int]		):
										lowerCamelCase						:Union[str, Any]						  =     torch.tensor(
										    [-2.6_8_7_0, -3.0_1_0_4, -3.1_3_5_6, -3.5_3_5_2, -3.0_0_4_4, -3.0_3_5_3, -3.4_7_1_9, -3.6_7_7_7,
										     -3.1_5_2_0, -2.9_4_3_5, -2.6_5_5_3, -2.8_7_9_5, -2.9_9_4_4, -2.5_9_2_1, -3.0_2_7_9, -3.0_3_8_6,
										     -3.0_8_6_4, -3.1_2_9_1, -3.2_3_5_3, -2.7_4_4_4, -2.6_8_3_1, -2.7_2_8_7, -3.1_7_6_1, -3.1_5_7_1,
										     -3.2_7_2_6, -3.0_5_8_2, -3.1_0_0_7, -3.4_5_3_3, -3.4_6_9_5, -3.0_9_9_8]		)
										# fmt: on

										lowerCamelCase						:Tuple						  =     self._load_datasamples(1		)
										lowerCamelCase						:Optional[Any]						  =     SpeechTaFeatureExtractor()
										lowerCamelCase						:Any						  =     feature_extractor(audio_target=__lowerCamelCase	,					return_tensors='''pt'''		).input_values
										self.assertEquals(input_values.shape	,					(1, 366, 80)		)
										self.assertTrue(torch.allclose(input_values[0, 0, :30]	,					__lowerCamelCase	,					atol=1e-4		)		)




import ast
import os
import re
import shutil
import tempfile
import unittest
from unittest import mock

import torch

from accelerate.test_utils.examples import compare_against_test
from accelerate.test_utils.testing import TempDirTestCase, require_trackers, run_command, slow
from accelerate.utils import write_basic_config


# DataLoaders built from `test_samples/MRPC` for quick testing
# Should mock `{script_name}.get_dataloaders` via:
# @mock.patch("{script_name}.get_dataloaders", mocked_dataloaders)

A__           =							[
    """cross_validation.py""",
    """gradient_accumulation.py""",
    """local_sgd.py""",
    """multi_process_metrics.py""",
    """memory.py""",
    """automatic_gradient_accumulation.py""",
    """fsdp_with_peak_mem_tracking.py""",
    """deepspeed_with_config_support.py""",
    """megatron_lm_gpt_pretraining.py""",
]
class   _lowerCAmelCase		(     unittest.TestCase				):







							def 				snake_case  (				self     :    Any	,					__snake_case     :    str	,					__snake_case     :    bool	,					__snake_case     :    str = None	,					__snake_case     :    list = None		):
										lowerCamelCase						:Tuple						  =     None
										lowerCamelCase						:Tuple						  =     os.path.abspath(os.path.join('''examples'''	,					'''by_feature'''		)		)
										lowerCamelCase						:Optional[int]						  =     os.path.abspath('''examples'''		)
										for item in os.listdir(__snake_case		):
													if item not in EXCLUDE_EXAMPLES:
																lowerCamelCase						:Optional[int]						  =     os.path.join(__snake_case	,					__snake_case		)
																if os.path.isfile(__snake_case		) and ".py" in item_path:
																			with self.subTest(
																			    tested_script=__snake_case	,					feature_script=__snake_case	,					tested_section='''main()''' if parser_only else '''training_function()'''	,					):
																						lowerCamelCase						:Union[str, Any]						  =     compare_against_test(
																						    os.path.join(__snake_case	,					__snake_case		)	,					__snake_case	,					__snake_case	,					__snake_case		)
																						lowerCamelCase						:int						  =     '''\n'''.join(__snake_case		)
																						if special_strings is not None:
																									for string in special_strings:
																												lowerCamelCase						:int						  =     diff.replace(__snake_case	,					''''''		)
																						self.assertEqual(__snake_case	,					''''''		)







							def 				snake_case  (				self     :    Dict		):
										self.one_complete_example('''complete_nlp_example.py'''	,					__snake_case		)
										self.one_complete_example('''complete_nlp_example.py'''	,					__snake_case		)







							def 				snake_case  (				self     :    Optional[Any]		):
										lowerCamelCase						:Union[str, Any]						  =     os.path.abspath(os.path.join('''examples'''	,					'''cv_example.py'''		)		)
										lowerCamelCase						:Optional[int]						  =     [
										    ''' ''' * 16 + '''{\n\n''',
										    ''' ''' * 20 + '''"accuracy": eval_metric["accuracy"],\n\n''',
										    ''' ''' * 20 + '''"f1": eval_metric["f1"],\n\n''',
										    ''' ''' * 20 + '''"train_loss": total_loss.item() / len(train_dataloader),\n\n''',
										    ''' ''' * 20 + '''"epoch": epoch,\n\n''',
										    ''' ''' * 16 + '''},\n\n''',
										    ''' ''' * 16 + '''step=epoch,\n''',
										    ''' ''' * 12,
										    ''' ''' * 8 + '''for step, batch in enumerate(active_dataloader):\n''',
										]
										self.one_complete_example('''complete_cv_example.py'''	,					__snake_case	,					__snake_case	,					__snake_case		)
										self.one_complete_example('''complete_cv_example.py'''	,					__snake_case	,					__snake_case	,					__snake_case		)


@mock.patch.dict(os.environ      ,   {'TESTING_MOCKED_DATALOADERS': '1'}				)
class   _lowerCAmelCase		(     __SCREAMING_SNAKE_CASE				):
							_UpperCAmelCase						    =     False







							@classmethod
							def 				snake_case  (				cls     :    Optional[Any]		):
										super().setUpClass()
										lowerCamelCase						:Any						  =     tempfile.mkdtemp()
										lowerCamelCase						:Optional[int]						  =     os.path.join(cls._tmpdir	,					'''default_config.yml'''		)

										write_basic_config(save_location=cls.configPath		)
										lowerCamelCase						:List[str]						  =     ['''accelerate''', '''launch''', '''--config_file''', cls.configPath]







							@classmethod
							def 				snake_case  (				cls     :    Dict		):
										super().tearDownClass()
										shutil.rmtree(cls._tmpdir		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:Any						  =     F"\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps epoch\n        --output_dir {self.tmpdir}\n        ".split()
										run_command(self._launch_args + testargs		)
										self.assertTrue(os.path.exists(os.path.join(self.tmpdir	,					'''epoch_0'''		)		)		)







							def 				snake_case  (				self     :    List[Any]		):
										lowerCamelCase						:Tuple						  =     F"\n        examples/by_feature/checkpointing.py\n        --checkpointing_steps 1\n        --output_dir {self.tmpdir}\n        ".split()
										lowerCamelCase						:List[Any]						  =     run_command(self._launch_args + testargs		)
										self.assertTrue(os.path.exists(os.path.join(self.tmpdir	,					'''step_2'''		)		)		)







							def 				snake_case  (				self     :    List[str]		):
										lowerCamelCase						:Dict						  =     F"\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir	,					'epoch_0'		)}\n        ".split()
										lowerCamelCase						:Dict						  =     run_command(self._launch_args + testargs	,					return_stdout=__snake_case		)
										self.assertNotIn('''epoch 0:'''	,					__snake_case		)
										self.assertIn('''epoch 1:'''	,					__snake_case		)







							def 				snake_case  (				self     :    str		):
										lowerCamelCase						:List[Any]						  =     F"\n        examples/by_feature/checkpointing.py\n        --resume_from_checkpoint {os.path.join(self.tmpdir	,					'step_2'		)}\n        ".split()
										lowerCamelCase						:Optional[int]						  =     run_command(self._launch_args + testargs	,					return_stdout=__snake_case		)
										if torch.cuda.is_available():
													lowerCamelCase						:Union[str, Any]						  =     torch.cuda.device_count()
										else:
													lowerCamelCase						:Dict						  =     1
										if num_processes > 1:
													self.assertNotIn('''epoch 0:'''	,					__snake_case		)
													self.assertIn('''epoch 1:'''	,					__snake_case		)
										else:
													self.assertIn('''epoch 0:'''	,					__snake_case		)
													self.assertIn('''epoch 1:'''	,					__snake_case		)







							@slow
							def 				snake_case  (				self     :    Any		):
										lowerCamelCase						:Tuple						  =     '''
        examples/by_feature/cross_validation.py
        --num_folds 2
        '''.split()
										with mock.patch.dict(os.environ	,					{'''TESTING_MOCKED_DATALOADERS''': '''0'''}		):
													lowerCamelCase						:Dict						  =     run_command(self._launch_args + testargs	,					return_stdout=__snake_case		)
													lowerCamelCase						:Tuple						  =     re.findall('''({.+})'''	,					__snake_case		)
													lowerCamelCase						:Optional[Any]						  =     [r for r in results if '''accuracy''' in r][-1]
													lowerCamelCase						:List[str]						  =     ast.literal_eval(__snake_case		)
													self.assertGreaterEqual(results['''accuracy''']	,					0.7_5		)







							def 				snake_case  (				self     :    int		):
										lowerCamelCase						:Dict						  =     ['''examples/by_feature/multi_process_metrics.py''']
										run_command(self._launch_args + testargs		)







							@require_trackers
							@mock.patch.dict(os.environ	,					{'''WANDB_MODE''': '''offline'''}		)
							def 				snake_case  (				self     :    Any		):
										with tempfile.TemporaryDirectory() as tmpdir:
													lowerCamelCase						:Tuple						  =     F"\n            examples/by_feature/tracking.py\n            --with_tracking\n            --project_dir {tmpdir}\n            ".split()
													run_command(self._launch_args + testargs		)
													self.assertTrue(os.path.exists(os.path.join(__snake_case	,					'''tracking'''		)		)		)







							def 				snake_case  (				self     :    Tuple		):
										lowerCamelCase						:Tuple						  =     ['''examples/by_feature/gradient_accumulation.py''']
										run_command(self._launch_args + testargs		)







							def 				snake_case  (				self     :    Optional[Any]		):
										lowerCamelCase						:int						  =     ['''examples/by_feature/local_sgd.py''']
										run_command(self._launch_args + testargs		)



