


'''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

a_   :      List[str]				    =      random.Random()






def 		_A			(lowerCAmelCase__					:Optional[int]			,  lowerCAmelCase__					:List[Any]=1.0			,  lowerCAmelCase__					:List[str]=None			,  lowerCAmelCase__					:Dict=None						)				->					Union[str, Any]:




     '''simple docstring'''



     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 	a       (       unittest.TestCase    ):
 def __init__(	self ,			__magic_name__ ,			__magic_name__=7 ,			__magic_name__=4_00 ,			__magic_name__=20_00 ,			__magic_name__=10 ,			__magic_name__=1_60 ,			__magic_name__=8 ,			__magic_name__=0.0 ,			__magic_name__=40_00 ,			__magic_name__=False ,			__magic_name__=True ,			)   ->							Tuple:
      _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      =							padding_value
      _a      =							sampling_rate
      _a      =							return_attention_mask
      _a      =							do_normalize
      _a      =							feature_size
      _a      =							chunk_length
      _a      =							hop_length
 def 							__UpperCAmelCase						(	self  )   ->							Optional[int]:
      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 							__UpperCAmelCase						(	self ,			__magic_name__=False ,			__magic_name__=False  )   ->							Dict:
      def _flatten(__magic_name__  ):
           return list(itertools.chain(*__magic_name__  )  )

      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(__magic_name__  ) for x in speech_inputs]
      return speech_inputs





@require_torch
@require_torchaudio
class 	a       (       _SCREAMING_SNAKE_CASE					,			unittest.TestCase    ):
 _lowerCAmelCase	   =		WhisperFeatureExtractor if is_speech_available() else None
 def 							__UpperCAmelCase						(	self  )   ->							str:
      _a      =							WhisperFeatureExtractionTester(self  )
 def 							__UpperCAmelCase						(	self  )   ->							Dict:
      _a      =							self.feature_extraction_class(**self.feat_extract_dict  )

      with tempfile.TemporaryDirectory() as tmpdirname:
           _a      =							feat_extract_first.save_pretrained(__magic_name__  )[0]
           check_json_file_has_correct_format(__magic_name__  )
           _a      =							self.feature_extraction_class.from_pretrained(__magic_name__  )

      _a      =							feat_extract_first.to_dict()
      _a      =							feat_extract_second.to_dict()
      _a      =							feat_extract_first.mel_filters
      _a      =							feat_extract_second.mel_filters
      self.assertTrue(np.allclose(__magic_name__ ,			__magic_name__  )  )
      self.assertEqual(__magic_name__ ,			__magic_name__  )
 def 							__UpperCAmelCase						(	self  )   ->							Tuple:
      _a      =							self.feature_extraction_class(**self.feat_extract_dict  )

      with tempfile.TemporaryDirectory() as tmpdirname:
           _a      =							os.path.join(__magic_name__ ,			'feat_extract.json'  )
           feat_extract_first.to_json_file(__magic_name__  )
           _a      =							self.feature_extraction_class.from_json_file(__magic_name__  )

      _a      =							feat_extract_first.to_dict()
      _a      =							feat_extract_second.to_dict()
      _a      =							feat_extract_first.mel_filters
      _a      =							feat_extract_second.mel_filters
      self.assertTrue(np.allclose(__magic_name__ ,			__magic_name__  )  )
      self.assertEqual(__magic_name__ ,			__magic_name__  )
 def 							__UpperCAmelCase						(	self  )   ->							Optional[int]:
      # 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(__magic_name__  ) for speech_input in speech_inputs]

      # Test feature size
      _a      =							feature_extractor(__magic_name__ ,			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      =							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(__magic_name__ ,			__magic_name__ ,			atol=1e-3  )  )

      # Test batched
      _a      =							feature_extractor(__magic_name__ ,			return_tensors='np'  ).input_features
      _a      =							feature_extractor(__magic_name__ ,			return_tensors='np'  ).input_features
      for enc_seq_a, enc_seq_a in zip(__magic_name__ ,			__magic_name__  ):
           self.assertTrue(np.allclose(__magic_name__ ,			__magic_name__ ,			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(__magic_name__  )
      _a      =							feature_extractor(__magic_name__ ,			return_tensors='np'  ).input_features
      _a      =							feature_extractor(__magic_name__ ,			return_tensors='np'  ).input_features
      for enc_seq_a, enc_seq_a in zip(__magic_name__ ,			__magic_name__  ):
           self.assertTrue(np.allclose(__magic_name__ ,			__magic_name__ ,			atol=1e-3  )  )

      # Test truncation required
      _a      =							[floats_list((1, x)  )[0] for x in range(2_00 ,			(feature_extractor.n_samples + 5_00) ,			2_00  )]
      _a      =							[np.asarray(__magic_name__  ) for speech_input in speech_inputs]

      _a      =							[x[: feature_extractor.n_samples] for x in speech_inputs]
      _a      =							[np.asarray(__magic_name__  ) for speech_input in speech_inputs_truncated]

      _a      =							feature_extractor(__magic_name__ ,			return_tensors='np'  ).input_features
      _a      =							feature_extractor(__magic_name__ ,			return_tensors='np'  ).input_features
      for enc_seq_a, enc_seq_a in zip(__magic_name__ ,			__magic_name__  ):
           self.assertTrue(np.allclose(__magic_name__ ,			__magic_name__ ,			atol=1e-3  )  )
 def 							__UpperCAmelCase						(	self  )   ->							Union[str, Any]:
      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 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							Any:
      _a      =							load_dataset('hf-internal-testing/librispeech_asr_dummy' ,			'clean' ,			split='validation'  )
      # automatic decoding with librispeech
      _a      =							ds.sort('id'  ).select(range(__magic_name__  )  )[:num_samples]['audio']

      return [x["array"] for x in speech_samples]
 def 							__UpperCAmelCase						(	self  )   ->							List[str]:
      # fmt: off
      _a      =							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      =							self._load_datasamples(1  )
      _a      =							WhisperFeatureExtractor()
      _a      =							feature_extractor(__magic_name__ ,			return_tensors='pt'  ).input_features
      self.assertEqual(input_features.shape ,			(1, 80, 30_00)  )
      self.assertTrue(torch.allclose(input_features[0, 0, :30] ,			__magic_name__ ,			atol=1e-4  )  )




 def 							__UpperCAmelCase						(	self  )   ->							Union[str, Any]:
      _a      =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
      _a      =							self._load_datasamples(1  )[0]
      _a      =							((audio - audio.min()) / (audio.max() - audio.min())) * 6_55_35  # Rescale to [0, 65535] to show issue
      _a      =							feat_extract.zero_mean_unit_var_norm([audio] ,			attention_mask=__magic_name__  )[0]

      self.assertTrue(np.all(np.mean(__magic_name__  ) < 1e-3  )  )
      self.assertTrue(np.all(np.abs(np.var(__magic_name__  ) - 1  ) < 1e-3  )  )





'''simple docstring'''






import pytest
import requests

from datasets.utils.file_utils import http_head

from .utils import OfflineSimulationMode, RequestWouldHangIndefinitelyError, offline






@pytest.mark.integration
def 		_A			()				->					Optional[Any]:




     '''simple docstring'''



     with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT						):
          with pytest.raises(lowerCAmelCase__						):
               requests.request('GET'			,  'https://huggingface.co'						)
          with pytest.raises(requests.exceptions.ConnectTimeout						):
               requests.request('GET'			,  'https://huggingface.co'			,  timeout=1.0						)






@pytest.mark.integration
def 		_A			()				->					Any:




     '''simple docstring'''



     with offline(OfflineSimulationMode.CONNECTION_FAILS						):
          with pytest.raises(requests.exceptions.ConnectionError						):
               requests.request('GET'			,  'https://huggingface.co'						)






def 		_A			()				->					Dict:




     '''simple docstring'''



     with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1						):
          with pytest.raises(lowerCAmelCase__						):
               http_head('https://huggingface.co'						)

