



"""simple docstring"""

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__    :  Optional[int]							    =	random.Random()




def __lowercase			(							_a		,			_a=1.0		,			_a=None		,			_a=None    ):
    if rng is None:
        snake_case_ :     Optional[int]				=	global_rng

    snake_case_ :     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
@require_torchaudio
class        _UpperCAmelCase       (			unittest.TestCase):

 def __init__(     self		:       str  ,     lowercase_		:       List[str]  ,     lowercase_		:       Tuple=7  ,     lowercase_		:       Union[str, Any]=400  ,     lowercase_		:       str=2000  ,     lowercase_		:       str=10  ,     lowercase_		:       Tuple=160  ,     lowercase_		:       List[str]=8  ,     lowercase_		:       Optional[int]=0.0  ,     lowercase_		:       Union[str, Any]=4000  ,     lowercase_		:       Optional[int]=False  ,     lowercase_		:       List[str]=True  ,     ):
     snake_case_ :     List[Any]				=	parent
     snake_case_ :     Optional[Any]				=	batch_size
     snake_case_ :     int				=	min_seq_length
     snake_case_ :     Tuple				=	max_seq_length
     snake_case_ :     List[str]				=	(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
     snake_case_ :     List[Any]				=	padding_value
     snake_case_ :     int				=	sampling_rate
     snake_case_ :     List[str]				=	return_attention_mask
     snake_case_ :     Union[str, Any]				=	do_normalize
     snake_case_ :     Tuple				=	feature_size
     snake_case_ :     str				=	chunk_length
     snake_case_ :     Any				=	hop_length

 def 						_snake_case					(     self		:       Dict				):
     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 						_snake_case					(     self		:       Any  ,     lowercase_		:       List[str]=False  ,     lowercase_		:       Any=False				):
     def _flatten(lowercase_		:       Optional[int]				):
         return list(itertools.chain(*lowercase_				)				)

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





@require_torch
@require_torchaudio
class        _UpperCAmelCase       (			UpperCAmelCase__    ,     unittest.TestCase):
 _lowerCAmelCase   :			Union[str, Any]							=		WhisperFeatureExtractor if is_speech_available() else None

 def 						_snake_case					(     self		:       List[Any]				):
     snake_case_ :     Optional[int]				=	WhisperFeatureExtractionTester(self				)

 def 						_snake_case					(     self		:       Optional[Any]				):
     snake_case_ :     int				=	self.feature_extraction_class(**self.feat_extract_dict				)

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

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

 def 						_snake_case					(     self		:       Dict				):
     snake_case_ :     Any				=	self.feature_extraction_class(**self.feat_extract_dict				)

     with tempfile.TemporaryDirectory() as tmpdirname:
         snake_case_ :     List[Any]				=	os.path.join(lowercase_  ,     '''feat_extract.json'''				)
         feat_extract_first.to_json_file(lowercase_				)
         snake_case_ :     Dict				=	self.feature_extraction_class.from_json_file(lowercase_				)

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

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

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

     # Test batched
     snake_case_ :     Optional[int]				=	feature_extractor(lowercase_  ,     return_tensors='''np'''				).input_features
     snake_case_ :     Optional[int]				=	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.
     snake_case_ :     int				=	[floats_list((1, x)				)[0] for x in (800, 800, 800)]
     snake_case_ :     List[str]				=	np.asarray(lowercase_				)
     snake_case_ :     Optional[Any]				=	feature_extractor(lowercase_  ,     return_tensors='''np'''				).input_features
     snake_case_ :     Optional[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				)				)

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

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

     snake_case_ :     Dict				=	feature_extractor(lowercase_  ,     return_tensors='''np'''				).input_features
     snake_case_ :     Optional[int]				=	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 						_snake_case					(     self		:       int				):
     import torch

     snake_case_ :     List[Any]				=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()				)
     snake_case_ :     int				=	np.random.rand(100  ,     32				).astype(np.floataa				)
     snake_case_ :     Optional[int]				=	np_speech_inputs.tolist()

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

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

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

 def 						_snake_case					(     self		:       str				):
     snake_case_ :     Any				=	torch.tensor(
         [
             0.11_93, -0.09_46, -0.10_98, -0.01_96, 0.02_25, -0.06_90, -0.17_36, 0.09_51,
             0.09_71, -0.08_17, -0.07_02, 0.01_62, 0.02_60, 0.00_17, -0.01_92, -0.16_78,
             0.07_09, -0.18_67, -0.06_55, -0.02_74, -0.02_34, -0.18_84, -0.05_16, -0.05_54,
             -0.02_74, -0.14_25, -0.14_23, 0.08_37, 0.03_77, -0.08_54
         ]				)
     # fmt: on

     snake_case_ :     Any				=	self._load_datasamples(1				)
     snake_case_ :     int				=	WhisperFeatureExtractor()
     snake_case_ :     str				=	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 						_snake_case					(     self		:       Any				):
     snake_case_ :     Dict				=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()				)
     snake_case_ :     str				=	self._load_datasamples(1				)[0]
     snake_case_ :     List[Any]				=	((audio - audio.min()) / (audio.max() - audio.min())) * 65535  # Rescale to [0, 65535] to show issue
     snake_case_ :     Union[str, Any]				=	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				)				)









'''simple docstring'''







class A__       :






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




             '''simple docstring'''
             _a    : Tuple     					=0
             _a    : Any     					=0
             _a    : int     					={}






      def __UpperCAmelCase       ( self			:Any     ,						SCREAMING_SNAKE_CASE			:List[str]	)       ->					Optional[int]:




             '''simple docstring'''
             if vertex not in self.adjacency:
                    _a    : Dict     					={}
                    self.num_vertices += 1






      def __UpperCAmelCase       ( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:str     ,						SCREAMING_SNAKE_CASE			:str     ,						SCREAMING_SNAKE_CASE			:Any	)       ->					List[str]:




             '''simple docstring'''
             self.add_vertex(SCREAMING_SNAKE_CASE	)
             self.add_vertex(SCREAMING_SNAKE_CASE	)

             if head == tail:
                    return

             _a    : Any     					=weight
             _a    : Tuple     					=weight






      def __UpperCAmelCase       ( self			:Dict	)       ->					Optional[int]:




             '''simple docstring'''
             _a    : Union[str, Any]     					=self.get_edges()
             for edge in edges:
                    _a					,    _a					,    _a    : List[str]     					=edge
                    edges.remove((tail, head, weight)	)
             for i in range(len(SCREAMING_SNAKE_CASE	)	):
                    _a    : str     					=list(edges[i]	)

             edges.sort(key=lambda SCREAMING_SNAKE_CASE	: e[2]	)
             for i in range(len(SCREAMING_SNAKE_CASE	) - 1	):
                    if edges[i][2] >= edges[i + 1][2]:
                           _a    : Union[str, Any]     					=edges[i][2] + 1
             for edge in edges:
                    _a					,    _a					,    _a    : Tuple     					=edge
                    _a    : Tuple     					=weight
                    _a    : List[Any]     					=weight






      def __str__( self			:int	)       ->					str:




             '''simple docstring'''
             _a    : int     					=""""""
             for tail in self.adjacency:
                    for head in self.adjacency[tail]:
                           _a    : str     					=self.adjacency[head][tail]
                           string += f"{head} -> {tail} == {weight}\n"
             return string.rstrip("""\n"""	)






      def __UpperCAmelCase       ( self			:Optional[int]	)       ->					Optional[Any]:




             '''simple docstring'''
             _a    : Union[str, Any]     					=[]
             for tail in self.adjacency:
                    for head in self.adjacency[tail]:
                           output.append((tail, head, self.adjacency[head][tail])	)
             return output






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




             '''simple docstring'''
             return self.adjacency.keys()






      @staticmethod
      def __UpperCAmelCase       ( SCREAMING_SNAKE_CASE			:Dict=None     ,						SCREAMING_SNAKE_CASE			:List[Any]=None	)       ->					Optional[int]:




             '''simple docstring'''
             _a    : str     					=Graph()
             if vertices is None:
                    _a    : Union[str, Any]     					=[]
             if edges is None:
                    _a    : List[Any]     					=[]
             for vertex in vertices:
                    g.add_vertex(SCREAMING_SNAKE_CASE	)
             for edge in edges:
                    g.add_edge(*SCREAMING_SNAKE_CASE	)
             return g







      class A__       :






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




                   '''simple docstring'''
                   _a    : Optional[int]     					={}
                   _a    : List[str]     					={}






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




                   '''simple docstring'''
                   return len(self.parent	)






            def __UpperCAmelCase       ( self			:Tuple     ,						SCREAMING_SNAKE_CASE			:Tuple	)       ->					Dict:




                   '''simple docstring'''
                   if item in self.parent:
                          return self.find(SCREAMING_SNAKE_CASE	)

                   _a    : Optional[Any]     					=item
                   _a    : List[str]     					=0
                   return item






            def __UpperCAmelCase       ( self			:int     ,						SCREAMING_SNAKE_CASE			:Dict	)       ->					List[str]:




                   '''simple docstring'''
                   if item not in self.parent:
                          return self.make_set(SCREAMING_SNAKE_CASE	)
                   if item != self.parent[item]:
                          _a    : str     					=self.find(self.parent[item]	)
                   return self.parent[item]






            def __UpperCAmelCase       ( self			:Optional[int]     ,						SCREAMING_SNAKE_CASE			:Tuple     ,						SCREAMING_SNAKE_CASE			:List[Any]	)       ->					Optional[Any]:




                   '''simple docstring'''
                   _a    : Optional[int]     					=self.find(SCREAMING_SNAKE_CASE	)
                   _a    : Dict     					=self.find(SCREAMING_SNAKE_CASE	)

                   if roota == roota:
                          return roota

                   if self.rank[roota] > self.rank[roota]:
                          _a    : Any     					=roota
                          return roota

                   if self.rank[roota] < self.rank[roota]:
                          _a    : List[str]     					=roota
                          return roota

                   if self.rank[roota] == self.rank[roota]:
                          self.rank[roota] += 1
                          _a    : List[Any]     					=roota
                          return roota
                   return None






      @staticmethod
      def __UpperCAmelCase       ( SCREAMING_SNAKE_CASE			:Dict	)       ->					Union[str, Any]:




             '''simple docstring'''
             _a    : Any     					=graph.num_vertices

             _a    : Union[str, Any]     					=Graph.UnionFind()
             _a    : Optional[int]     					=[]
             while num_components > 1:
                    _a    : str     					={}
                    for vertex in graph.get_vertices():
                           _a    : List[str]     					=-1

                    _a    : Any     					=graph.get_edges()
                    for edge in edges:
                           _a					,    _a					,    _a    : Tuple     					=edge
                           edges.remove((tail, head, weight)	)
                    for edge in edges:
                           _a					,    _a					,    _a    : Any     					=edge
                           _a    : Any     					=union_find.find(SCREAMING_SNAKE_CASE	)
                           _a    : List[Any]     					=union_find.find(SCREAMING_SNAKE_CASE	)
                           if seta != seta:
                                  if cheap_edge[seta] == -1 or cheap_edge[seta][2] > weight:
                                         _a    : Optional[int]     					=[head, tail, weight]

                                  if cheap_edge[seta] == -1 or cheap_edge[seta][2] > weight:
                                         _a    : List[Any]     					=[head, tail, weight]
                    for vertex in cheap_edge:
                           if cheap_edge[vertex] != -1:
                                  _a					,    _a					,    _a    : Optional[Any]     					=cheap_edge[vertex]
                                  if union_find.find(SCREAMING_SNAKE_CASE	) != union_find.find(SCREAMING_SNAKE_CASE	):
                                         union_find.union(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE	)
                                         mst_edges.append(cheap_edge[vertex]	)
                                         _a    : str     					=num_components - 1
             _a    : str     					=Graph.build(edges=SCREAMING_SNAKE_CASE	)
             return mst


