



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

_lowercase		:				str      		=random.Random()

def      lowerCAmelCase_					(  _lowercase    :       List[str]						,       _lowercase    :       List[str]=1.0						,       _lowercase    :       Optional[Any]=None						,       _lowercase    :       List[Any]=None)							->	Optional[int]:







			"""simple docstring"""
			if rng is None:
						a__       :						List[str]							=  global_rng

			a__       :						List[Any]							=  []
			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        snake_case__    (unittest.TestCase		):







						"""simple docstring"""




						def __init__(       self ,       __lowercase ,       __lowercase=7 ,       __lowercase=4_0_0 ,       __lowercase=2_0_0_0 ,       __lowercase=1_0 ,       __lowercase=1_6_0 ,       __lowercase=8 ,       __lowercase=0.0 ,       __lowercase=4_0_0_0 ,       __lowercase=False ,       __lowercase=True ,       ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						Dict							=  parent
									a__       :						List[Any]							=  batch_size
									a__       :						List[Any]							=  min_seq_length
									a__       :						int							=  max_seq_length
									a__       :						str							=  (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
									a__       :						Union[str, Any]							=  padding_value
									a__       :						Dict							=  sampling_rate
									a__       :						Any							=  return_attention_mask
									a__       :						Tuple							=  do_normalize
									a__       :						List[str]							=  feature_size
									a__       :						int							=  chunk_length
									a__       :						Optional[Any]							=  hop_length

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				str:
									"""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 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase=False ,       __lowercase=False  ) ->				Optional[int]:
									"""simple docstring"""




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

									if equal_length:
												a__       :						str							=  [floats_list((self.max_seq_length, self.feature_size)  ) for _ in range(self.batch_size  )]
									else:
												# make sure that inputs increase in size
												a__       :						Union[str, Any]							=  [
												    floats_list((x, self.feature_size)  )
												    for x in range(self.min_seq_length ,       self.max_seq_length ,       self.seq_length_diff  )
												]
									if numpify:
												a__       :						Tuple							=  [np.asarray(__lowercase  ) for x in speech_inputs]
									return speech_inputs




@require_torch
@require_torchaudio
class        snake_case__    (A__ ,     unittest.TestCase		):







						"""simple docstring"""




						__lowerCAmelCase					:Optional[Any]      =					WhisperFeatureExtractor if is_speech_available() else None

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						List[Any]							=  WhisperFeatureExtractionTester(self  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						Optional[int]							=  self.feature_extraction_class(**self.feat_extract_dict  )

									with tempfile.TemporaryDirectory() as tmpdirname:
												a__       :						int							=  feat_extract_first.save_pretrained(__lowercase  )[0]
												check_json_file_has_correct_format(__lowercase  )
												a__       :						Optional[int]							=  self.feature_extraction_class.from_pretrained(__lowercase  )

									a__       :						Optional[Any]							=  feat_extract_first.to_dict()
									a__       :						Union[str, Any]							=  feat_extract_second.to_dict()
									a__       :						int							=  feat_extract_first.mel_filters
									a__       :						Union[str, Any]							=  feat_extract_second.mel_filters
									self.assertTrue(np.allclose(__lowercase ,       __lowercase  )  )
									self.assertEqual(__lowercase ,       __lowercase  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				List[Any]:
									"""simple docstring"""




									a__       :						Any							=  self.feature_extraction_class(**self.feat_extract_dict  )

									with tempfile.TemporaryDirectory() as tmpdirname:
												a__       :						Union[str, Any]							=  os.path.join(__lowercase ,       """feat_extract.json"""  )
												feat_extract_first.to_json_file(__lowercase  )
												a__       :						Any							=  self.feature_extraction_class.from_json_file(__lowercase  )

									a__       :						str							=  feat_extract_first.to_dict()
									a__       :						Tuple							=  feat_extract_second.to_dict()
									a__       :						Any							=  feat_extract_first.mel_filters
									a__       :						Tuple							=  feat_extract_second.mel_filters
									self.assertTrue(np.allclose(__lowercase ,       __lowercase  )  )
									self.assertEqual(__lowercase ,       __lowercase  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Optional[Any]:
									"""simple docstring"""




									a__       :						int							=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
									# create three inputs of length 800, 1000, and 1200
									a__       :						Optional[Any]							=  [floats_list((1, x)  )[0] for x in range(8_0_0 ,       1_4_0_0 ,       2_0_0  )]
									a__       :						Optional[Any]							=  [np.asarray(__lowercase  ) for speech_input in speech_inputs]

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

									# Test batched
									a__       :						List[Any]							=  feature_extractor(__lowercase ,       return_tensors="""np"""  ).input_features
									a__       :						str							=  feature_extractor(__lowercase ,       return_tensors="""np"""  ).input_features
									for enc_seq_a, enc_seq_a in zip(__lowercase ,       __lowercase  ):
												self.assertTrue(np.allclose(__lowercase ,       __lowercase ,       atol=1E-3  )  )

									# Test 2-D numpy arrays are batched.
									a__       :						Any							=  [floats_list((1, x)  )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
									a__       :						Tuple							=  np.asarray(__lowercase  )
									a__       :						Optional[int]							=  feature_extractor(__lowercase ,       return_tensors="""np"""  ).input_features
									a__       :						str							=  feature_extractor(__lowercase ,       return_tensors="""np"""  ).input_features
									for enc_seq_a, enc_seq_a in zip(__lowercase ,       __lowercase  ):
												self.assertTrue(np.allclose(__lowercase ,       __lowercase ,       atol=1E-3  )  )

									# Test truncation required
									a__       :						int							=  [floats_list((1, x)  )[0] for x in range(2_0_0 ,       (feature_extractor.n_samples + 5_0_0) ,       2_0_0  )]
									a__       :						Tuple							=  [np.asarray(__lowercase  ) for speech_input in speech_inputs]

									a__       :						str							=  [x[: feature_extractor.n_samples] for x in speech_inputs]
									a__       :						Any							=  [np.asarray(__lowercase  ) for speech_input in speech_inputs_truncated]

									a__       :						str							=  feature_extractor(__lowercase ,       return_tensors="""np"""  ).input_features
									a__       :						Union[str, Any]							=  feature_extractor(__lowercase ,       return_tensors="""np"""  ).input_features
									for enc_seq_a, enc_seq_a in zip(__lowercase ,       __lowercase  ):
												self.assertTrue(np.allclose(__lowercase ,       __lowercase ,       atol=1E-3  )  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									import torch

									a__       :						Optional[int]							=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
									a__       :						Optional[int]							=  np.random.rand(1_0_0 ,       3_2  ).astype(np.floataa  )
									a__       :						Optional[int]							=  np_speech_inputs.tolist()

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

						def 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase  ) ->				str:
									"""simple docstring"""




									a__       :						Union[str, Any]							=  load_dataset("""hf-internal-testing/librispeech_asr_dummy""" ,       """clean""" ,       split="""validation"""  )
									# automatic decoding with librispeech
									a__       :						Tuple							=  ds.sort("""id"""  ).select(range(__lowercase  )  )[:num_samples]["""audio"""]

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

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						Tuple							=  torch.tensor(
									    [
									        0.1_1_9_3, -0.0_9_4_6, -0.1_0_9_8, -0.0_1_9_6, 0.0_2_2_5, -0.0_6_9_0, -0.1_7_3_6, 0.0_9_5_1,
									        0.0_9_7_1, -0.0_8_1_7, -0.0_7_0_2, 0.0_1_6_2, 0.0_2_6_0, 0.0_0_1_7, -0.0_1_9_2, -0.1_6_7_8,
									        0.0_7_0_9, -0.1_8_6_7, -0.0_6_5_5, -0.0_2_7_4, -0.0_2_3_4, -0.1_8_8_4, -0.0_5_1_6, -0.0_5_5_4,
									        -0.0_2_7_4, -0.1_4_2_5, -0.1_4_2_3, 0.0_8_3_7, 0.0_3_7_7, -0.0_8_5_4
									    ]  )
									# fmt: on

									a__       :						Tuple							=  self._load_datasamples(1  )
									a__       :						Union[str, Any]							=  WhisperFeatureExtractor()
									a__       :						List[Any]							=  feature_extractor(__lowercase ,       return_tensors="""pt"""  ).input_features
									self.assertEqual(input_features.shape ,       (1, 8_0, 3_0_0_0)  )
									self.assertTrue(torch.allclose(input_features[0, 0, :3_0] ,       __lowercase ,       atol=1E-4  )  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						Union[str, Any]							=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
									a__       :						Optional[int]							=  self._load_datasamples(1  )[0]
									a__       :						List[str]							=  ((audio - audio.min()) / (audio.max() - audio.min())) * 6_5_5_3_5  # Rescale to [0, 65535] to show issue
									a__       :						Optional[int]							=  feat_extract.zero_mean_unit_var_norm([audio] ,       attention_mask=__lowercase  )[0]

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









import logging
from dataclasses import dataclass, field
from typing import Optional

from seqaseq_trainer import arg_to_scheduler

from transformers import TrainingArguments


_lowercase		:				str      		=logging.getLogger(__name__)

@dataclass
class        snake_case__    (A__		):







						"""simple docstring"""




						__lowerCAmelCase					:Optional[float]      =					field(
						    default=0.0 ,     metadata={"help": "The label smoothing epsilon to apply (if not zero)."}		)
						__lowerCAmelCase					:bool      =					field(default=A__ ,     metadata={"help": "Whether to SortishSamler or not."}		)
						__lowerCAmelCase					:bool      =					field(
						    default=A__ ,     metadata={"help": "Whether to use generate to calculate generative metrics (ROUGE, BLEU)."}		)
						__lowerCAmelCase					:bool      =					field(default=A__ ,     metadata={"help": "whether to use adafactor"}		)
						__lowerCAmelCase					:Optional[float]      =					field(
						    default=A__ ,     metadata={"help": "Encoder layer dropout probability. Goes into model.config."}		)
						__lowerCAmelCase					:Optional[float]      =					field(
						    default=A__ ,     metadata={"help": "Decoder layer dropout probability. Goes into model.config."}		)
						__lowerCAmelCase					:Optional[float]      =					field(default=A__ ,     metadata={"help": "Dropout probability. Goes into model.config."}		)
						__lowerCAmelCase					:Optional[float]      =					field(
						    default=A__ ,     metadata={"help": "Attention dropout probability. Goes into model.config."}		)
						__lowerCAmelCase					:Optional[str]      =					field(
						    default="linear" ,     metadata={"help": f"Which lr scheduler to use. Selected in {sorted(arg_to_scheduler.keys()		)}"} ,     )




