






'''simple docstring'''







import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


_A     :      Dict   					=get_tests_dir('''fixtures''')
_A     :      List[Any]   					=get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
_A     :      List[Any]   					=get_tests_dir('''fixtures/dummy-config.json''')

class _lowercase     (    unittest.TestCase		):




       def 		lowerCamelCase_    (							self:    Tuple					):
              lowerCamelCase__ :     List[Any]	  =	0




       def 		lowerCamelCase_    (							self:    Any					):
              lowerCamelCase__ :     Dict	  =	AutoFeatureExtractor.from_pretrained("""facebook/wav2vec2-base-960h"""					)
              self.assertIsInstance(__lowerCAmelCase      ,							__lowerCAmelCase					)




       def 		lowerCamelCase_    (							self:    Optional[Any]					):
              lowerCamelCase__ :     Union[str, Any]	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase					)
              self.assertIsInstance(__lowerCAmelCase      ,							__lowerCAmelCase					)




       def 		lowerCamelCase_    (							self:    List[str]					):
              with tempfile.TemporaryDirectory() as tmpdirname:
                     lowerCamelCase__ :     str	  =	WavaVecaConfig()

                     # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
                     lowerCamelCase__ :     Optional[int]	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase					).to_dict()

                     config_dict.pop("""feature_extractor_type"""					)
                     lowerCamelCase__ :     Tuple	  =	WavaVecaFeatureExtractor(**__lowerCAmelCase					)

                     # save in new folder
                     model_config.save_pretrained(__lowerCAmelCase					)
                     config.save_pretrained(__lowerCAmelCase					)

                     lowerCamelCase__ :     Union[str, Any]	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase					)

                     # make sure private variable is not incorrectly saved
                     lowerCamelCase__ :     str	  =	json.loads(config.to_json_string()					)
                     self.assertTrue("""_processor_class""" not in dict_as_saved					)

              self.assertIsInstance(__lowerCAmelCase      ,							__lowerCAmelCase					)




       def 		lowerCamelCase_    (							self:    int					):
              lowerCamelCase__ :     int	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase					)
              self.assertIsInstance(__lowerCAmelCase      ,							__lowerCAmelCase					)




       def 		lowerCamelCase_    (							self:    Dict					):
              with self.assertRaisesRegex(
                  __lowerCAmelCase      ,							"""bert-base is not a local folder and is not a valid model identifier"""					):
                     lowerCamelCase__ :     str	  =	AutoFeatureExtractor.from_pretrained("""bert-base"""					)




       def 		lowerCamelCase_    (							self:    str					):
              with self.assertRaisesRegex(
                  __lowerCAmelCase      ,							R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""					):
                     lowerCamelCase__ :     Union[str, Any]	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase      ,							revision="""aaaaaa"""					)




       def 		lowerCamelCase_    (							self:    Optional[Any]					):
              with self.assertRaisesRegex(
                  __lowerCAmelCase      ,							"""hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""      ,							):
                     lowerCamelCase__ :     Tuple	  =	AutoFeatureExtractor.from_pretrained("""hf-internal-testing/config-no-model"""					)




       def 		lowerCamelCase_    (							self:    Union[str, Any]					):
              # If remote code is not set, we will time out when asking whether to load the model.
              with self.assertRaises(__lowerCAmelCase					):
                     lowerCamelCase__ :     Dict	  =	AutoFeatureExtractor.from_pretrained(
                         """hf-internal-testing/test_dynamic_feature_extractor"""					)
              # If remote code is disabled, we can't load this config.
              with self.assertRaises(__lowerCAmelCase					):
                     lowerCamelCase__ :     str	  =	AutoFeatureExtractor.from_pretrained(
                         """hf-internal-testing/test_dynamic_feature_extractor"""      ,							trust_remote_code=__lowerCAmelCase					)

              lowerCamelCase__ :     List[Any]	  =	AutoFeatureExtractor.from_pretrained(
                  """hf-internal-testing/test_dynamic_feature_extractor"""      ,							trust_remote_code=__lowerCAmelCase					)
              self.assertEqual(feature_extractor.__class__.__name__      ,							"""NewFeatureExtractor"""					)

              # Test feature extractor can be reloaded.
              with tempfile.TemporaryDirectory() as tmp_dir:
                     feature_extractor.save_pretrained(__lowerCAmelCase					)
                     lowerCamelCase__ :     Dict	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase      ,							trust_remote_code=__lowerCAmelCase					)
              self.assertEqual(reloaded_feature_extractor.__class__.__name__      ,							"""NewFeatureExtractor"""					)




       def 		lowerCamelCase_    (							self:    Union[str, Any]					):
              try:
                     AutoConfig.register("""custom"""      ,							__lowerCAmelCase					)
                     AutoFeatureExtractor.register(__lowerCAmelCase      ,							__lowerCAmelCase					)
                     # Trying to register something existing in the Transformers library will raise an error
                     with self.assertRaises(__lowerCAmelCase					):
                            AutoFeatureExtractor.register(__lowerCAmelCase      ,							__lowerCAmelCase					)

                     # Now that the config is registered, it can be used as any other config with the auto-API
                     lowerCamelCase__ :     Optional[Any]	  =	CustomFeatureExtractor.from_pretrained(__lowerCAmelCase					)
                     with tempfile.TemporaryDirectory() as tmp_dir:
                            feature_extractor.save_pretrained(__lowerCAmelCase					)
                            lowerCamelCase__ :     Dict	  =	AutoFeatureExtractor.from_pretrained(__lowerCAmelCase					)
                            self.assertIsInstance(__lowerCAmelCase      ,							__lowerCAmelCase					)

              finally:
                     if "custom" in CONFIG_MAPPING._extra_content:
                            del CONFIG_MAPPING._extra_content["custom"]
                     if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                            del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]




       def 		lowerCamelCase_    (							self:    str					):

              class _lowercase     (    lowerCAmelCase_		):
                     a    				=	True

              try:
                     AutoConfig.register("""custom"""      ,							__lowerCAmelCase					)
                     AutoFeatureExtractor.register(__lowerCAmelCase      ,							__lowerCAmelCase					)
                     # If remote code is not set, the default is to use local
                     lowerCamelCase__ :     Tuple	  =	AutoFeatureExtractor.from_pretrained(
                         """hf-internal-testing/test_dynamic_feature_extractor"""					)
                     self.assertEqual(feature_extractor.__class__.__name__      ,							"""NewFeatureExtractor"""					)
                     self.assertTrue(feature_extractor.is_local					)

                     # If remote code is disabled, we load the local one.
                     lowerCamelCase__ :     Dict	  =	AutoFeatureExtractor.from_pretrained(
                         """hf-internal-testing/test_dynamic_feature_extractor"""      ,							trust_remote_code=__lowerCAmelCase					)
                     self.assertEqual(feature_extractor.__class__.__name__      ,							"""NewFeatureExtractor"""					)
                     self.assertTrue(feature_extractor.is_local					)

                     # If remote is enabled, we load from the Hub
                     lowerCamelCase__ :     int	  =	AutoFeatureExtractor.from_pretrained(
                         """hf-internal-testing/test_dynamic_feature_extractor"""      ,							trust_remote_code=__lowerCAmelCase					)
                     self.assertEqual(feature_extractor.__class__.__name__      ,							"""NewFeatureExtractor"""					)
                     self.assertTrue(not hasattr(__lowerCAmelCase      ,							"""is_local"""					)					)

              finally:
                     if "custom" in CONFIG_MAPPING._extra_content:
                            del CONFIG_MAPPING._extra_content["custom"]
                     if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                            del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]









'''simple docstring'''







from typing import List, Optional, Union

import numpy as np

from ....audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ....feature_extraction_sequence_utils import SequenceFeatureExtractor
from ....feature_extraction_utils import BatchFeature
from ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


_A     :      List[str]   					=logging.get_logger(__name__)

class _lowercase     (    _lowercase		):
							a    				=	["""input_features""", """attention_mask"""]




							def __init__(							self:    Optional[int]      ,							UpperCamelCase__:    str=80      ,							UpperCamelCase__:    Union[str, Any]=16_000      ,							UpperCamelCase__:    str=0.0      ,							UpperCamelCase__:    Tuple=10      ,							UpperCamelCase__:    Dict=25      ,							UpperCamelCase__:    List[Any]="hamming_window"      ,							UpperCamelCase__:    Tuple=32_768.0      ,							UpperCamelCase__:    List[Any]=0.97      ,							UpperCamelCase__:    Optional[int]=1.0      ,							UpperCamelCase__:    str=True      ,							UpperCamelCase__:    Optional[Any]=True      ,							UpperCamelCase__:    Optional[int]=False      ,							**UpperCamelCase__:    Optional[int]      ,							):
														super().__init__(feature_size=UpperCamelCase__      ,							sampling_rate=UpperCamelCase__      ,							padding_value=UpperCamelCase__      ,							**UpperCamelCase__					)

														lowerCamelCase__ :     Dict	  =	feature_size
														lowerCamelCase__ :     Tuple	  =	sampling_rate
														lowerCamelCase__ :     Union[str, Any]	  =	padding_value
														lowerCamelCase__ :     Union[str, Any]	  =	hop_length
														lowerCamelCase__ :     Any	  =	win_length
														lowerCamelCase__ :     Tuple	  =	frame_signal_scale
														lowerCamelCase__ :     Dict	  =	preemphasis_coeff
														lowerCamelCase__ :     Optional[Any]	  =	mel_floor
														lowerCamelCase__ :     Any	  =	normalize_means
														lowerCamelCase__ :     Optional[int]	  =	normalize_vars
														lowerCamelCase__ :     Dict	  =	win_function
														lowerCamelCase__ :     Dict	  =	return_attention_mask

														lowerCamelCase__ :     Optional[Any]	  =	win_length * sampling_rate // 1_000
														lowerCamelCase__ :     Tuple	  =	hop_length * sampling_rate // 1_000

														lowerCamelCase__ :     List[Any]	  =	optimal_fft_length(self.sample_size					)
														lowerCamelCase__ :     List[str]	  =	(self.n_fft // 2) + 1




							def 		lowerCamelCase_    (							self:    Union[str, Any]      ,							UpperCamelCase__:    np.array					):
														if self.win_function == "hamming_window":
																					lowerCamelCase__ :     Dict	  =	window_function(window_length=self.sample_size      ,							name=self.win_function      ,							periodic=UpperCamelCase__					)
														else:
																					lowerCamelCase__ :     Union[str, Any]	  =	window_function(window_length=self.sample_size      ,							name=self.win_function					)

														lowerCamelCase__ :     List[Any]	  =	mel_filter_bank(
														    num_frequency_bins=self.n_freqs      ,							num_mel_filters=self.feature_size      ,							min_frequency=0.0      ,							max_frequency=self.sampling_rate / 2.0      ,							sampling_rate=self.sampling_rate      ,							)

														lowerCamelCase__ :     Tuple	  =	spectrogram(
														    one_waveform * self.frame_signal_scale      ,							window=UpperCamelCase__      ,							frame_length=self.sample_size      ,							hop_length=self.sample_stride      ,							fft_length=self.n_fft      ,							center=UpperCamelCase__      ,							preemphasis=self.preemphasis_coeff      ,							mel_filters=UpperCamelCase__      ,							mel_floor=self.mel_floor      ,							log_mel="""log"""      ,							)
														return msfc_features.T




							def 		lowerCamelCase_    (							self:    Optional[int]      ,							UpperCamelCase__:    List[str]      ,							UpperCamelCase__:    List[Any]      ,							UpperCamelCase__:    Union[str, Any]					):
														# make sure we normalize float32 arrays
														if self.normalize_means:
																					lowerCamelCase__ :     List[str]	  =	x[:input_length].mean(axis=0					)
																					lowerCamelCase__ :     List[str]	  =	np.subtract(UpperCamelCase__      ,							UpperCamelCase__					)
														if self.normalize_vars:
																					lowerCamelCase__ :     Union[str, Any]	  =	x[:input_length].std(axis=0					)
																					lowerCamelCase__ :     str	  =	np.divide(UpperCamelCase__      ,							UpperCamelCase__					)

														if input_length < x.shape[0]:
																					lowerCamelCase__ :     List[Any]	  =	padding_value

														# make sure array is in float32
														lowerCamelCase__ :     str	  =	x.astype(np.floataa					)

														return x




							def 		lowerCamelCase_    (							self:    str      ,							UpperCamelCase__:    List[np.ndarray]      ,							UpperCamelCase__:    Optional[np.ndarray] = None					):
														lowerCamelCase__ :     Optional[int]	  =	attention_mask.sum(-1					) if attention_mask is not None else [x.shape[0] for x in input_features]
														return [self._normalize_one(UpperCamelCase__      ,							UpperCamelCase__      ,							self.padding_value					) for x, n in zip(UpperCamelCase__      ,							UpperCamelCase__					)]




							def __call__(							self:    List[str]      ,							UpperCamelCase__:    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,							UpperCamelCase__:    Union[bool, str, PaddingStrategy] = False      ,							UpperCamelCase__:    Optional[int] = None      ,							UpperCamelCase__:    bool = False      ,							UpperCamelCase__:    Optional[int] = None      ,							UpperCamelCase__:    Optional[bool] = None      ,							UpperCamelCase__:    Optional[Union[str, TensorType]] = None      ,							UpperCamelCase__:    Optional[int] = None      ,							**UpperCamelCase__:    Tuple      ,							):

														if sampling_rate is not None:
																					if sampling_rate != self.sampling_rate:
																												raise ValueError(
																												    F'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
																												    F''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
																												    F''' {self.sampling_rate} and not {sampling_rate}.'''					)
														else:
																					logger.warning(
																					    """It is strongly recommended to pass the ``sampling_rate`` argument to this function. """
																					    """Failing to do so can result in silent errors that might be hard to debug."""					)

														lowerCamelCase__ :     List[str]	  =	isinstance(UpperCamelCase__      ,							np.ndarray					) and len(raw_speech.shape					) > 1
														if is_batched_numpy and len(raw_speech.shape					) > 2:
																					raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''					)
														lowerCamelCase__ :     List[str]	  =	is_batched_numpy or (
														    isinstance(UpperCamelCase__      ,							(list, tuple)					) and (isinstance(raw_speech[0]      ,							(np.ndarray, tuple, list)					))
														)

														if is_batched:
																					lowerCamelCase__ :     str	  =	[np.asarray(UpperCamelCase__      ,							dtype=np.floataa					) for speech in raw_speech]
														elif not is_batched and not isinstance(UpperCamelCase__      ,							np.ndarray					):
																					lowerCamelCase__ :     List[Any]	  =	np.asarray(UpperCamelCase__      ,							dtype=np.floataa					)
														elif isinstance(UpperCamelCase__      ,							np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
																					lowerCamelCase__ :     Optional[Any]	  =	raw_speech.astype(np.floataa					)

														# always return batch
														if not is_batched:
																					lowerCamelCase__ :     Optional[Any]	  =	[raw_speech]

														# extract fbank features
														lowerCamelCase__ :     str	  =	[self._extract_mfsc_features(UpperCamelCase__					) for one_waveform in raw_speech]

														# convert into correct format for padding
														lowerCamelCase__ :     str	  =	BatchFeature({"""input_features""": features}					)

														lowerCamelCase__ :     str	  =	self.pad(
														    UpperCamelCase__      ,							padding=UpperCamelCase__      ,							max_length=UpperCamelCase__      ,							truncation=UpperCamelCase__      ,							pad_to_multiple_of=UpperCamelCase__      ,							return_attention_mask=UpperCamelCase__      ,							**UpperCamelCase__      ,							)
														# make sure list is in array format
														lowerCamelCase__ :     str	  =	padded_inputs.get("""input_features"""					)
														if isinstance(input_features[0]      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     Dict	  =	[np.asarray(UpperCamelCase__      ,							dtype=np.floataa					) for feature in input_features]

														lowerCamelCase__ :     List[str]	  =	padded_inputs.get("""attention_mask"""					)
														if attention_mask is not None:
																					lowerCamelCase__ :     Any	  =	[np.asarray(UpperCamelCase__      ,							dtype=np.intaa					) for array in attention_mask]

														if self.normalize_means or self.normalize_vars:
																					lowerCamelCase__ :     str	  =	(
																					    np.array(UpperCamelCase__      ,							dtype=np.intaa					)
																					    if self._get_padding_strategies(UpperCamelCase__      ,							max_length=UpperCamelCase__					) is not PaddingStrategy.DO_NOT_PAD
																					    and padding
																					    else None
																					)
																					lowerCamelCase__ :     Union[str, Any]	  =	self.normalize(
																					    padded_inputs["""input_features"""]      ,							attention_mask=UpperCamelCase__					)

														if return_tensors is not None:
																					lowerCamelCase__ :     Dict	  =	padded_inputs.convert_to_tensors(UpperCamelCase__					)

														return padded_inputs

