


'''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__				     =		random.Random()






def __UpperCamelCase   (			__lowerCamelCase		: Union[str, Any]      ,  __lowerCamelCase		: str=1.0      ,  __lowerCamelCase		: Optional[Any]=None      ,  __lowerCamelCase		: Optional[int]=None				)      ->     str:







     '''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 						__SCREAMING_SNAKE_CASE					(      unittest.TestCase ):






     def __init__(			self     ,							__UpperCamelCase     ,							__UpperCamelCase=7     ,							__UpperCamelCase=400     ,							__UpperCamelCase=2_000     ,							__UpperCamelCase=10     ,							__UpperCamelCase=160     ,							__UpperCamelCase=8     ,							__UpperCamelCase=0.0     ,							__UpperCamelCase=4_000     ,							__UpperCamelCase=False     ,							__UpperCamelCase=True     ,							)       ->    List[str]:
          _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  a_					(			self		)       ->    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  a_					(			self     ,							__UpperCamelCase=False     ,							__UpperCamelCase=False		)       ->    Union[str, Any]:
          def _flatten(__UpperCamelCase		):
               return list(itertools.chain(*__UpperCamelCase		)		)

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







@require_torch
@require_torchaudio
class 						__SCREAMING_SNAKE_CASE					(      lowerCamelCase__ ,			unittest.TestCase ):
     UpperCAmelCase 					=				WhisperFeatureExtractor if is_speech_available() else None






     def  a_					(			self		)       ->    Union[str, Any]:
          _a						=		WhisperFeatureExtractionTester(self		)






     def  a_					(			self		)       ->    Union[str, Any]:
          _a						=		self.feature_extraction_class(**self.feat_extract_dict		)

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

          _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(__UpperCamelCase     ,							__UpperCamelCase		)		)
          self.assertEqual(__UpperCamelCase     ,							__UpperCamelCase		)






     def  a_					(			self		)       ->    List[str]:
          _a						=		self.feature_extraction_class(**self.feat_extract_dict		)

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

          _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(__UpperCamelCase     ,							__UpperCamelCase		)		)
          self.assertEqual(__UpperCamelCase     ,							__UpperCamelCase		)






     def  a_					(			self		)       ->    List[Any]:
          # 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(800     ,							1_400     ,							200		)]
          _a						=		[np.asarray(__UpperCamelCase		) for speech_input in speech_inputs]

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

          # Test batched
          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="np"		).input_features
          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="np"		).input_features
          for enc_seq_a, enc_seq_a in zip(__UpperCamelCase     ,							__UpperCamelCase		):
               self.assertTrue(np.allclose(__UpperCamelCase     ,							__UpperCamelCase     ,							atol=1e-3		)		)

          # Test 2-D numpy arrays are batched.
          _a						=		[floats_list((1, x)		)[0] for x in (800, 800, 800)]
          _a						=		np.asarray(__UpperCamelCase		)
          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="np"		).input_features
          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="np"		).input_features
          for enc_seq_a, enc_seq_a in zip(__UpperCamelCase     ,							__UpperCamelCase		):
               self.assertTrue(np.allclose(__UpperCamelCase     ,							__UpperCamelCase     ,							atol=1e-3		)		)

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

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

          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="np"		).input_features
          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="np"		).input_features
          for enc_seq_a, enc_seq_a in zip(__UpperCamelCase     ,							__UpperCamelCase		):
               self.assertTrue(np.allclose(__UpperCamelCase     ,							__UpperCamelCase     ,							atol=1e-3		)		)






     def  a_					(			self		)       ->    Union[str, Any]:
          import torch

          _a						=		self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
          _a						=		np.random.rand(100     ,							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  a_					(			self     ,							__UpperCamelCase		)       ->    Union[str, Any]:
          _a						=		load_dataset("hf-internal-testing/librispeech_asr_dummy"     ,							"clean"     ,							split="validation"		)
          # automatic decoding with librispeech
          _a						=		ds.sort("id"		).select(range(__UpperCamelCase		)		)[:num_samples]["audio"]

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






     def  a_					(			self		)       ->    str:
          # fmt: off
          _a						=		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

          _a						=		self._load_datasamples(1		)
          _a						=		WhisperFeatureExtractor()
          _a						=		feature_extractor(__UpperCamelCase     ,							return_tensors="pt"		).input_features
          self.assertEqual(input_features.shape     ,							(1, 80, 3_000)		)
          self.assertTrue(torch.allclose(input_features[0, 0, :30]     ,							__UpperCamelCase     ,							atol=1e-4		)		)






     def  a_					(			self		)       ->    Optional[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())) * 65_535  # Rescale to [0, 65535] to show issue
          _a						=		feat_extract.zero_mean_unit_var_norm([audio]     ,							attention_mask=__UpperCamelCase		)[0]

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




'''simple docstring'''

from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowercase__				     =		{
    "configuration_mctct": ["MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP", "MCTCTConfig"],
    "feature_extraction_mctct": ["MCTCTFeatureExtractor"],
    "processing_mctct": ["MCTCTProcessor"],
}


try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowercase__				     =		[
       "MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST",
       "MCTCTForCTC",
       "MCTCTModel",
       "MCTCTPreTrainedModel",
   ]


if TYPE_CHECKING:
   from .configuration_mctct import MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP, MCTCTConfig
   from .feature_extraction_mctct import MCTCTFeatureExtractor
   from .processing_mctct import MCTCTProcessor

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_mctct import MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST, MCTCTForCTC, MCTCTModel, MCTCTPreTrainedModel

else:
   import sys

   lowercase__				     =		_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)

