


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

lowerCamelCase											=       random.Random()




def 	a_      (    SCREAMING_SNAKE_CASE__					:      Optional[Any]		,			SCREAMING_SNAKE_CASE__					:      Dict=1.0		,			SCREAMING_SNAKE_CASE__					:      Optional[int]=None		,			SCREAMING_SNAKE_CASE__					:      str=None     ):
       '''simple docstring'''





       if rng is None:
              _lowerCamelCase       :		List[Any]		=global_rng

       _lowerCamelCase       :		Union[str, 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 A  (   unittest.TestCase     ):






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




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






      def        lowerCamelCase       (					self						:     List[str] )					->						List[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        lowerCamelCase       (					self						:     int			,   lowercase_						:     List[Any]=False			,   lowercase_						:     Tuple=False )					->						str:
             """simple docstring"""




             def _flatten(lowercase_						:     Union[str, Any] ):
                    return list(itertools.chain(*lowercase_ ) )

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




@require_torch
@require_torchaudio
class A  (   UpperCamelCase_    ,							unittest.TestCase     ):
      UpperCamelCase__							:  Any      =WhisperFeatureExtractor if is_speech_available() else None






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




             _lowerCamelCase       :		Optional[int]		=WhisperFeatureExtractionTester(self )






      def        lowerCamelCase       (					self						:     Tuple )					->						Tuple:
             """simple docstring"""




             _lowerCamelCase       :		Tuple		=self.feature_extraction_class(**self.feat_extract_dict )

             with tempfile.TemporaryDirectory() as tmpdirname:
                    _lowerCamelCase       :		Any		=feat_extract_first.save_pretrained(lowercase_ )[0]
                    check_json_file_has_correct_format(lowercase_ )
                    _lowerCamelCase       :		List[str]		=self.feature_extraction_class.from_pretrained(lowercase_ )

             _lowerCamelCase       :		Any		=feat_extract_first.to_dict()
             _lowerCamelCase       :		Optional[int]		=feat_extract_second.to_dict()
             _lowerCamelCase       :		int		=feat_extract_first.mel_filters
             _lowerCamelCase       :		Tuple		=feat_extract_second.mel_filters
             self.assertTrue(np.allclose(lowercase_			,   lowercase_ ) )
             self.assertEqual(lowercase_			,   lowercase_ )






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




             _lowerCamelCase       :		Union[str, Any]		=self.feature_extraction_class(**self.feat_extract_dict )

             with tempfile.TemporaryDirectory() as tmpdirname:
                    _lowerCamelCase       :		Tuple		=os.path.join(lowercase_			,   'feat_extract.json' )
                    feat_extract_first.to_json_file(lowercase_ )
                    _lowerCamelCase       :		Any		=self.feature_extraction_class.from_json_file(lowercase_ )

             _lowerCamelCase       :		List[Any]		=feat_extract_first.to_dict()
             _lowerCamelCase       :		Optional[Any]		=feat_extract_second.to_dict()
             _lowerCamelCase       :		Optional[int]		=feat_extract_first.mel_filters
             _lowerCamelCase       :		Tuple		=feat_extract_second.mel_filters
             self.assertTrue(np.allclose(lowercase_			,   lowercase_ ) )
             self.assertEqual(lowercase_			,   lowercase_ )






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




             _lowerCamelCase       :		List[str]		=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
             # create three inputs of length 800, 1000, and 1200
             _lowerCamelCase       :		Tuple		=[floats_list((1, x) )[0] for x in range(800			,   1400			,   200 )]
             _lowerCamelCase       :		Dict		=[np.asarray(lowercase_ ) for speech_input in speech_inputs]

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

             # Test batched
             _lowerCamelCase       :		Any		=feature_extractor(lowercase_			,   return_tensors='np' ).input_features
             _lowerCamelCase       :		Dict		=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.
             _lowerCamelCase       :		Union[str, Any]		=[floats_list((1, x) )[0] for x in (800, 800, 800)]
             _lowerCamelCase       :		List[Any]		=np.asarray(lowercase_ )
             _lowerCamelCase       :		Optional[int]		=feature_extractor(lowercase_			,   return_tensors='np' ).input_features
             _lowerCamelCase       :		Tuple		=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
             _lowerCamelCase       :		List[Any]		=[floats_list((1, x) )[0] for x in range(200			,   (feature_extractor.n_samples + 500)			,   200 )]
             _lowerCamelCase       :		Dict		=[np.asarray(lowercase_ ) for speech_input in speech_inputs]

             _lowerCamelCase       :		Union[str, Any]		=[x[: feature_extractor.n_samples] for x in speech_inputs]
             _lowerCamelCase       :		List[str]		=[np.asarray(lowercase_ ) for speech_input in speech_inputs_truncated]

             _lowerCamelCase       :		Any		=feature_extractor(lowercase_			,   return_tensors='np' ).input_features
             _lowerCamelCase       :		List[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        lowerCamelCase       (					self						:     Optional[Any] )					->						Optional[Any]:
             """simple docstring"""




             import torch

             _lowerCamelCase       :		List[Any]		=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
             _lowerCamelCase       :		Dict		=np.random.rand(100			,   32 ).astype(np.floataa )
             _lowerCamelCase       :		Union[str, Any]		=np_speech_inputs.tolist()

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






      def        lowerCamelCase       (					self						:     int			,   lowercase_						:     Tuple )					->						int:
             """simple docstring"""




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

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






      def        lowerCamelCase       (					self						:     int )					->						Tuple:
             """simple docstring"""




             _lowerCamelCase       :		List[str]		=torch.tensor(
                 [
                     0.1193, -0.0946, -0.1098, -0.0196, 0.0225, -0.0690, -0.1736, 0.0951,
                     0.0971, -0.0817, -0.0702, 0.0162, 0.0260, 0.0017, -0.0192, -0.1678,
                     0.0709, -0.1867, -0.0655, -0.0274, -0.0234, -0.1884, -0.0516, -0.0554,
                     -0.0274, -0.1425, -0.1423, 0.0837, 0.0377, -0.0854
                 ] )
             # fmt: on

             _lowerCamelCase       :		Union[str, Any]		=self._load_datasamples(1 )
             _lowerCamelCase       :		Dict		=WhisperFeatureExtractor()
             _lowerCamelCase       :		List[Any]		=feature_extractor(lowercase_			,   return_tensors='pt' ).input_features
             self.assertEqual(input_features.shape			,   (1, 80, 3000) )
             self.assertTrue(torch.allclose(input_features[0, 0, :30]			,   lowercase_			,   atol=1E-4 ) )






      def        lowerCamelCase       (					self						:     Any )					->						Any:
             """simple docstring"""




             _lowerCamelCase       :		Optional[int]		=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
             _lowerCamelCase       :		Optional[Any]		=self._load_datasamples(1 )[0]
             _lowerCamelCase       :		Any		=((audio - audio.min()) / (audio.max() - audio.min())) * 6_5535  # Rescale to [0, 65535] to show issue
             _lowerCamelCase       :		List[str]		=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 os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
 import ctypes
 import msvcrt  # noqa




 class A  (   ctypes.Structure     ):
       # _fields is a specific attr expected by ctypes
       UpperCamelCase__							:  List[Any]      =[('size', ctypes.c_int), ('visible', ctypes.c_byte)]




def 	a_      (    ):
       '''simple docstring'''





       if os.name == "nt":
              _lowerCamelCase       :		Optional[Any]		=CursorInfo()
              _lowerCamelCase       :		Dict		=ctypes.windll.kernelaa.GetStdHandle(-11     )
              ctypes.windll.kernelaa.GetConsoleCursorInfo(SCREAMING_SNAKE_CASE__		,			ctypes.byref(SCREAMING_SNAKE_CASE__     )     )
              _lowerCamelCase       :		Any		=False
              ctypes.windll.kernelaa.SetConsoleCursorInfo(SCREAMING_SNAKE_CASE__		,			ctypes.byref(SCREAMING_SNAKE_CASE__     )     )
       elif os.name == "posix":
              sys.stdout.write('\033[?25l'     )
              sys.stdout.flush()




def 	a_      (    ):
       '''simple docstring'''





       if os.name == "nt":
              _lowerCamelCase       :		Any		=CursorInfo()
              _lowerCamelCase       :		Optional[Any]		=ctypes.windll.kernelaa.GetStdHandle(-11     )
              ctypes.windll.kernelaa.GetConsoleCursorInfo(SCREAMING_SNAKE_CASE__		,			ctypes.byref(SCREAMING_SNAKE_CASE__     )     )
              _lowerCamelCase       :		Union[str, Any]		=True
              ctypes.windll.kernelaa.SetConsoleCursorInfo(SCREAMING_SNAKE_CASE__		,			ctypes.byref(SCREAMING_SNAKE_CASE__     )     )
       elif os.name == "posix":
              sys.stdout.write('\033[?25h'     )
              sys.stdout.flush()




@contextmanager
def 	a_      (    ):
       '''simple docstring'''





       try:
              hide_cursor()
              yield
       finally:
              show_cursor()


