


import itertools
import random
import unittest

import numpy as np

from transformers import is_speech_available
from transformers.testing_utils import require_torch, require_torchaudio

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
						from transformers import SpeechaTextFeatureExtractor

snake_case							: Optional[Any]      =   random.Random()






def 							__lowercase     (				__lowerCAmelCase							:       Optional[Any]      ,					__lowerCAmelCase							:       Dict=1.0      ,					__lowerCAmelCase							:       Optional[Any]=None      ,					__lowerCAmelCase							:       List[Any]=None     ):
							if rng is None:
														a__    =  global_rng

							a__    =  []
							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   ):
					def __init__(   self     :Any    ,__snake_case     :Tuple    ,__snake_case     :Optional[int]=7    ,__snake_case     :List[Any]=4_00    ,__snake_case     :Optional[int]=20_00    ,__snake_case     :Tuple=24    ,__snake_case     :Union[str, Any]=24    ,__snake_case     :int=0.0    ,__snake_case     :str=1_60_00    ,__snake_case     :int=True    ,__snake_case     :Any=True    ,)      ->					Optional[Any]:
												a__    =  parent
												a__    =  batch_size
												a__    =  min_seq_length
												a__    =  max_seq_length
												a__    =  (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
												a__    =  feature_size
												a__    =  num_mel_bins
												a__    =  padding_value
												a__    =  sampling_rate
												a__    =  return_attention_mask
												a__    =  do_normalize
					def 				lowerCamelCase__(   self     :str   )      ->					Any:
												return {
												    "feature_size": self.feature_size,
												    "num_mel_bins": self.num_mel_bins,
												    "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     :List[Any]    ,__snake_case     :List[Any]=False    ,__snake_case     :Optional[int]=False   )      ->					Dict:
												def _flatten(__snake_case     :str   ):
																			return list(itertools.chain(*__snake_case   )   )

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







@require_torch
@require_torchaudio
class  snake_case_     (lowerCamelCase_			,		unittest.TestCase   ):
					UpperCAmelCase__      :					str					   =    SpeechaTextFeatureExtractor if is_speech_available() else None
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					Dict:
												a__    =  SpeechaTextFeatureExtractionTester(self   )
					def 				lowerCamelCase__(   self     :int    ,__snake_case     :List[str]   )      ->					Optional[int]:
												self.assertTrue(np.all(np.mean(__snake_case    ,axis=0   ) < 1E-3   )   )
												self.assertTrue(np.all(np.abs(np.var(__snake_case    ,axis=0   ) - 1   ) < 1E-3   )   )
					def 				lowerCamelCase__(   self     :List[Any]   )      ->					Tuple:
												# Tests that all call wrap to encode_plus and batch_encode_plus
												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												# create three inputs of length 800, 1000, and 1200
												a__    =  [floats_list((1, x)   )[0] for x in range(8_00    ,14_00    ,2_00   )]
												a__    =  [np.asarray(__snake_case   ) for speech_input in speech_inputs]

												# Test feature size
												a__    =  feature_extractor(__snake_case    ,padding=__snake_case    ,return_tensors='np'   ).input_features
												self.assertTrue(input_features.ndim == 3   )
												self.assertTrue(input_features.shape[-1] == feature_extractor.feature_size   )

												# Test not batched input
												a__    =  feature_extractor(speech_inputs[0]    ,return_tensors='np'   ).input_features
												a__    =  feature_extractor(np_speech_inputs[0]    ,return_tensors='np'   ).input_features
												self.assertTrue(np.allclose(__snake_case    ,__snake_case    ,atol=1E-3   )   )

												# Test batched
												a__    =  feature_extractor(__snake_case    ,return_tensors='np'   ).input_features
												a__    =  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.
												a__    =  [floats_list((1, x)   )[0] for x in (8_00, 8_00, 8_00)]
												a__    =  np.asarray(__snake_case   )
												a__    =  feature_extractor(__snake_case    ,return_tensors='np'   ).input_features
												a__    =  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     :List[str]   )      ->					str:
												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												a__    =  [floats_list((1, x)   )[0] for x in range(8_00    ,14_00    ,2_00   )]

												a__    =  ['longest', 'max_length', 'do_not_pad']
												a__    =  [None, 16, None]
												for max_length, padding in zip(__snake_case    ,__snake_case   ):
																			a__    =  feature_extractor(
																			    __snake_case    ,padding=__snake_case    ,max_length=__snake_case    ,return_attention_mask=__snake_case   )
																			a__    =  inputs.input_features
																			a__    =  inputs.attention_mask
																			a__    =  [np.sum(__snake_case   ) for x in attention_mask]

																			self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]   )
																			self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]   )
																			self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]   )
					def 				lowerCamelCase__(   self     :Dict   )      ->					Dict:
												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												a__    =  [floats_list((1, x)   )[0] for x in range(8_00    ,14_00    ,2_00   )]

												a__    =  ['longest', 'max_length', 'do_not_pad']
												a__    =  [None, 16, None]
												for max_length, padding in zip(__snake_case    ,__snake_case   ):
																			a__    =  feature_extractor(
																			    __snake_case    ,max_length=__snake_case    ,padding=__snake_case    ,return_tensors='np'    ,return_attention_mask=__snake_case   )
																			a__    =  inputs.input_features
																			a__    =  inputs.attention_mask
																			a__    =  [np.sum(__snake_case   ) for x in attention_mask]

																			self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]   )
																			self.assertTrue(input_features[0][fbank_feat_lengths[0] :].sum() < 1E-6   )
																			self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]   )
																			self.assertTrue(input_features[0][fbank_feat_lengths[1] :].sum() < 1E-6   )
																			self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]   )
					def 				lowerCamelCase__(   self     :Optional[int]   )      ->					int:
												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												a__    =  [floats_list((1, x)   )[0] for x in range(8_00    ,14_00    ,2_00   )]
												a__    =  feature_extractor(
												    __snake_case    ,padding='max_length'    ,max_length=4    ,truncation=__snake_case    ,return_tensors='np'    ,return_attention_mask=__snake_case    ,)
												a__    =  inputs.input_features
												a__    =  inputs.attention_mask
												a__    =  np.sum(attention_mask == 1    ,axis=1   )

												self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]   )
												self._check_zero_mean_unit_variance(input_features[1]   )
												self._check_zero_mean_unit_variance(input_features[2]   )
					def 				lowerCamelCase__(   self     :Tuple   )      ->					List[str]:
												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												a__    =  [floats_list((1, x)   )[0] for x in range(8_00    ,14_00    ,2_00   )]
												a__    =  feature_extractor(
												    __snake_case    ,padding='longest'    ,max_length=4    ,truncation=__snake_case    ,return_tensors='np'    ,return_attention_mask=__snake_case    ,)
												a__    =  inputs.input_features
												a__    =  inputs.attention_mask
												a__    =  np.sum(attention_mask == 1    ,axis=1   )

												self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]   )
												self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]   )
												self._check_zero_mean_unit_variance(input_features[2]   )

												# make sure that if max_length < longest -> then pad to max_length
												self.assertEqual(input_features.shape    ,(3, 4, 24)   )

												a__    =  [floats_list((1, x)   )[0] for x in range(8_00    ,14_00    ,2_00   )]
												a__    =  feature_extractor(
												    __snake_case    ,padding='longest'    ,max_length=16    ,truncation=__snake_case    ,return_tensors='np'    ,return_attention_mask=__snake_case    ,)
												a__    =  inputs.input_features
												a__    =  inputs.attention_mask
												a__    =  np.sum(attention_mask == 1    ,axis=1   )

												self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]   )
												self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]   )
												self._check_zero_mean_unit_variance(input_features[2]   )

												# make sure that if max_length < longest -> then pad to max_length
												self.assertEqual(input_features.shape    ,(3, 6, 24)   )
					def 				lowerCamelCase__(   self     :str   )      ->					int:
												import torch

												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												a__    =  np.random.rand(1_00    ,32   ).astype(np.floataa   )
												a__    =  np_speech_inputs.tolist()

												for inputs in [py_speech_inputs, np_speech_inputs]:
																			a__    =  feature_extractor.pad([{'input_features': inputs}]    ,return_tensors='np'   )
																			self.assertTrue(np_processed.input_features.dtype == np.floataa   )
																			a__    =  feature_extractor.pad([{'input_features': inputs}]    ,return_tensors='pt'   )
																			self.assertTrue(pt_processed.input_features.dtype == torch.floataa   )
					def 				lowerCamelCase__(   self     :Dict    ,__snake_case     :List[str]   )      ->					Optional[Any]:
												from datasets import load_dataset

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

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



					def 				lowerCamelCase__(   self     :str   )      ->					List[str]:
												# fmt: off
												a__    =  np.array([
												    -1.57_45, -1.77_13, -1.70_20, -1.60_69, -1.22_50, -1.11_05, -0.90_72, -0.82_41,
												    -1.23_10, -0.80_98, -0.33_20, -0.41_01, -0.79_85, -0.49_96, -0.82_13, -0.91_28,
												    -1.04_20, -1.12_86, -1.04_40, -0.79_99, -0.84_05, -1.22_75, -1.54_43, -1.46_25,
												]   )
												# fmt: on

												a__    =  self._load_datasamples(1   )
												a__    =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
												a__    =  feature_extractor(__snake_case    ,return_tensors='pt'   ).input_features
												self.assertEquals(input_features.shape    ,(1, 5_84, 24)   )
												self.assertTrue(np.allclose(input_features[0, 0, :30]    ,__snake_case    ,atol=1E-4   )   )








import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
						import torch
						from torch import nn

						from transformers import (
						    Adafactor,
						    AdamW,
						    get_constant_schedule,
						    get_constant_schedule_with_warmup,
						    get_cosine_schedule_with_warmup,
						    get_cosine_with_hard_restarts_schedule_with_warmup,
						    get_inverse_sqrt_schedule,
						    get_linear_schedule_with_warmup,
						    get_polynomial_decay_schedule_with_warmup,
						)






def 							__lowercase     (				__lowerCAmelCase							:       str      ,					__lowerCAmelCase							:       Any=1_0     ):
							a__    =  []
							for _ in range(__lowerCAmelCase     ):
														lrs.append(scheduler.get_lr()[0]     )
														scheduler.step()
							return lrs






def 							__lowercase     (				__lowerCAmelCase							:       Optional[int]      ,					__lowerCAmelCase							:       List[str]=1_0     ):
							a__    =  []
							for step in range(__lowerCAmelCase     ):
														lrs.append(scheduler.get_lr()[0]     )
														scheduler.step()
														if step == num_steps // 2:
																					with tempfile.TemporaryDirectory() as tmpdirname:
																												a__    =  os.path.join(__lowerCAmelCase      ,					'schedule.bin'     )
																												torch.save(scheduler.state_dict()      ,					__lowerCAmelCase     )

																												a__    =  torch.load(__lowerCAmelCase     )
																												scheduler.load_state_dict(__lowerCAmelCase     )
							return lrs







@require_torch
class  snake_case_     (unittest.TestCase   ):
					def 				lowerCamelCase__(   self     :Optional[int]    ,__snake_case     :List[Any]    ,__snake_case     :int    ,__snake_case     :Union[str, Any]   )      ->					int:
												self.assertEqual(len(__snake_case   )    ,len(__snake_case   )   )
												for a, b in zip(__snake_case    ,__snake_case   ):
																			self.assertAlmostEqual(__snake_case    ,__snake_case    ,delta=__snake_case   )
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					str:
												a__    =  torch.tensor([0.1, -0.2, -0.1]    ,requires_grad=__snake_case   )
												a__    =  torch.tensor([0.4, 0.2, -0.5]   )
												a__    =  nn.MSELoss()
												# No warmup, constant schedule, no gradient clipping
												a__    =  AdamW(params=[w]    ,lr=2E-1    ,weight_decay=0.0   )
												for _ in range(1_00   ):
																			a__    =  criterion(__snake_case    ,__snake_case   )
																			loss.backward()
																			optimizer.step()
																			w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
																			w.grad.zero_()
												self.assertListAlmostEqual(w.tolist()    ,[0.4, 0.2, -0.5]    ,tol=1E-2   )



					def 				lowerCamelCase__(   self     :Tuple   )      ->					int:
												a__    =  torch.tensor([0.1, -0.2, -0.1]    ,requires_grad=__snake_case   )
												a__    =  torch.tensor([0.4, 0.2, -0.5]   )
												a__    =  nn.MSELoss()
												# No warmup, constant schedule, no gradient clipping
												a__    =  Adafactor(
												    params=[w]    ,lr=1E-2    ,eps=(1E-30, 1E-3)    ,clip_threshold=1.0    ,decay_rate=-0.8    ,betaa=__snake_case    ,weight_decay=0.0    ,relative_step=__snake_case    ,scale_parameter=__snake_case    ,warmup_init=__snake_case    ,)
												for _ in range(10_00   ):
																			a__    =  criterion(__snake_case    ,__snake_case   )
																			loss.backward()
																			optimizer.step()
																			w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
																			w.grad.zero_()
												self.assertListAlmostEqual(w.tolist()    ,[0.4, 0.2, -0.5]    ,tol=1E-2   )







@require_torch
class  snake_case_     (unittest.TestCase   ):
					UpperCAmelCase__      :					str					   =    nn.Linear(5_0			,		5_0   ) if is_torch_available() else None
					UpperCAmelCase__      :					Dict					   =    AdamW(m.parameters()			,		lr=1_0.0   ) if is_torch_available() else None
					UpperCAmelCase__      :					Optional[Any]					   =    1_0
					def 				lowerCamelCase__(   self     :Optional[Any]    ,__snake_case     :Optional[int]    ,__snake_case     :Tuple    ,__snake_case     :int    ,__snake_case     :Any=None   )      ->					Optional[Any]:
												self.assertEqual(len(__snake_case   )    ,len(__snake_case   )   )
												for a, b in zip(__snake_case    ,__snake_case   ):
																			self.assertAlmostEqual(__snake_case    ,__snake_case    ,delta=__snake_case    ,msg=__snake_case   )



					def 				lowerCamelCase__(   self     :Tuple   )      ->					List[Any]:
												a__    =  {'num_warmup_steps': 2, 'num_training_steps': 10}
												# schedulers doct format
												# function: (sched_args_dict, expected_learning_rates)
												a__    =  {
												    get_constant_schedule: ({}, [10.0] * self.num_steps),
												    get_constant_schedule_with_warmup: (
												        {'num_warmup_steps': 4},
												        [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
												    ),
												    get_linear_schedule_with_warmup: (
												        {**common_kwargs},
												        [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
												    ),
												    get_cosine_schedule_with_warmup: (
												        {**common_kwargs},
												        [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
												    ),
												    get_cosine_with_hard_restarts_schedule_with_warmup: (
												        {**common_kwargs, 'num_cycles': 2},
												        [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
												    ),
												    get_polynomial_decay_schedule_with_warmup: (
												        {**common_kwargs, 'power': 2.0, 'lr_end': 1E-7},
												        [0.0, 5.0, 10.0, 7.6_56, 5.6_25, 3.9_06, 2.5, 1.4_06, 0.6_25, 0.1_56],
												    ),
												    get_inverse_sqrt_schedule: (
												        {'num_warmup_steps': 2},
												        [0.0, 5.0, 10.0, 8.1_65, 7.0_71, 6.3_25, 5.7_74, 5.3_45, 5.0, 4.7_14],
												    ),
												}

												for scheduler_func, data in scheds.items():
																			a__   ,      a__    =  data

																			a__    =  scheduler_func(self.optimizer    ,**__snake_case   )
																			self.assertEqual(len([scheduler.get_lr()[0]]   )    ,1   )
																			a__    =  unwrap_schedule(__snake_case    ,self.num_steps   )
																			self.assertListAlmostEqual(
																			    __snake_case    ,__snake_case    ,tol=1E-2    ,msg=F'failed for {scheduler_func} in normal scheduler'    ,)

																			a__    =  scheduler_func(self.optimizer    ,**__snake_case   )
																			if scheduler_func.__name__ != "get_constant_schedule":
																										LambdaScheduleWrapper.wrap_scheduler(__snake_case   )  # wrap to test picklability of the schedule
																			a__    =  unwrap_and_save_reload_schedule(__snake_case    ,self.num_steps   )
																			self.assertListEqual(__snake_case    ,__snake_case    ,msg=F'failed for {scheduler_func} in save and reload'   )







class  snake_case_     :
					def __init__(   self     :Tuple    ,__snake_case     :str   )      ->					Any:
												a__    =  fn
					def __call__(   self     :List[str]    ,*__snake_case     :Optional[Any]    ,**__snake_case     :Optional[int]   )      ->					Union[str, Any]:
												return self.fn(*__snake_case    ,**__snake_case   )



					@classmethod
					def 				lowerCamelCase__(   self     :Tuple    ,__snake_case     :Union[str, Any]   )      ->					Dict:
												a__    =  list(map(self    ,scheduler.lr_lambdas   )   )




