
'''simple docstring'''


import warnings
from typing import Any, Dict, 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 ...utils import PaddingStrategy, TensorType, logging


__lowercase							:   List[Any]					=     logging.get_logger(__name__)

class 	__lowercase				(  _lowercase    ):
			lowerCamelCase			:		Tuple	      =			["input_values", "attention_mask"]

			def __init__(self			,   A = 1			,   A = 1_6_0_0_0			,   A = 0.0			,   A = False			,   A = 8_0			,   A = 1_6			,   A = 6_4			,   A = "hann_window"			,   A = 1.0			,   A = 8_0			,   A = 7_6_0_0			,   A = 1E-10			,   A = 2			,   A = True			,   **A			,   ):
								super().__init__(feature_size=A			,   sampling_rate=A			,   padding_value=A			,   **A      )
								lowerCamelCase_					:      Optional[int]     =		do_normalize
								lowerCamelCase_					:      Dict     =		return_attention_mask

								lowerCamelCase_					:      List[Any]     =		num_mel_bins
								lowerCamelCase_					:      List[Any]     =		hop_length
								lowerCamelCase_					:      Any     =		win_length
								lowerCamelCase_					:      List[str]     =		win_function
								lowerCamelCase_					:      Tuple     =		frame_signal_scale
								lowerCamelCase_					:      Dict     =		fmin
								lowerCamelCase_					:      List[Any]     =		fmax
								lowerCamelCase_					:      Any     =		mel_floor
								lowerCamelCase_					:      Tuple     =		reduction_factor

								lowerCamelCase_					:      Tuple     =		win_length * sampling_rate // 1_0_0_0
								lowerCamelCase_					:      Union[str, Any]     =		hop_length * sampling_rate // 1_0_0_0
								lowerCamelCase_					:      Dict     =		optimal_fft_length(self.sample_size      )
								lowerCamelCase_					:      Optional[Any]     =		(self.n_fft // 2) + 1

								lowerCamelCase_					:      Optional[int]     =		window_function(window_length=self.sample_size			,   name=self.win_function			,   periodic=A      )

								lowerCamelCase_					:      List[str]     =		mel_filter_bank(
								    num_frequency_bins=self.n_freqs			,   num_mel_filters=self.num_mel_bins			,   min_frequency=self.fmin			,   max_frequency=self.fmax			,   sampling_rate=self.sampling_rate			,   norm='''slaney'''			,   mel_scale='''slaney'''			,   )

								if frame_signal_scale != 1.0:
													warnings.warn(
													    '''The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers'''			,   A			,   )
								if reduction_factor != 2.0:
													warnings.warn(
													    '''The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'''			,   A			,   )

			@staticmethod
			# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
			def        UpperCAmelCase__ (A			,   A			,   A = 0.0      ):
								if attention_mask is not None:
													lowerCamelCase_					:      str     =		np.array(A			,   np.intaa      )
													lowerCamelCase_					:      Optional[Any]     =		[]

													for vector, length in zip(A			,   attention_mask.sum(-1      )      ):
																		lowerCamelCase_					:      Optional[Any]     =		(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7      )
																		if length < normed_slice.shape[0]:
																							lowerCamelCase_					:      List[str]     =		padding_value

																		normed_input_values.append(A      )
								else:
													lowerCamelCase_					:      int     =		[(x - x.mean()) / np.sqrt(x.var() + 1E-7      ) for x in input_values]

								return normed_input_values

			def        UpperCAmelCase__ (self			,   A			,   ):
								lowerCamelCase_					:      List[str]     =		spectrogram(
								    A			,   window=self.window			,   frame_length=self.sample_size			,   hop_length=self.sample_stride			,   fft_length=self.n_fft			,   mel_filters=self.mel_filters			,   mel_floor=self.mel_floor			,   log_mel='''log10'''			,   )
								return log_mel_spec.T

			def __call__(self			,   A = None			,   A = None			,   A = False			,   A = None			,   A = False			,   A = None			,   A = None			,   A = None			,   A = None			,   **A			,   ):
								if audio is None and audio_target is None:
													raise ValueError('''You must provide either `audio` or `audio_target` values.'''      )

								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 audio 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.'''      )

								if audio is not None:
													lowerCamelCase_					:      str     =		self._process_audio(
													    A			,   A			,   A			,   A			,   A			,   A			,   A			,   A			,   **A			,   )
								else:
													lowerCamelCase_					:      List[Any]     =		None

								if audio_target is not None:
													lowerCamelCase_					:      Dict     =		self._process_audio(
													    A			,   A			,   A			,   A			,   A			,   A			,   A			,   A			,   **A			,   )

													if inputs is None:
																		return inputs_target
													else:
																		lowerCamelCase_					:      Optional[int]     =		inputs_target['''input_values''']
																		lowerCamelCase_					:      Union[str, Any]     =		inputs_target.get('''attention_mask'''      )
																		if decoder_attention_mask is not None:
																							lowerCamelCase_					:      List[Any]     =		decoder_attention_mask

								return inputs

			def        UpperCAmelCase__ (self			,   A			,   A = False			,   A = False			,   A = None			,   A = False			,   A = None			,   A = None			,   A = None			,   **A			,   ):
								lowerCamelCase_					:      List[Any]     =		isinstance(A			,   np.ndarray      ) and len(speech.shape      ) > 1
								if is_batched_numpy and len(speech.shape      ) > 2:
													raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""      )
								lowerCamelCase_					:      List[Any]     =		is_batched_numpy or (
								    isinstance(A			,   (list, tuple)      ) and (isinstance(speech[0]			,   (np.ndarray, tuple, list)      ))
								)

								if is_batched:
													lowerCamelCase_					:      Optional[int]     =		[np.asarray(A			,   dtype=np.floataa      ) for speech in speech]
								elif not is_batched and not isinstance(A			,   np.ndarray      ):
													lowerCamelCase_					:      Tuple     =		np.asarray(A			,   dtype=np.floataa      )
								elif isinstance(A			,   np.ndarray      ) and speech.dtype is np.dtype(np.floataa      ):
													lowerCamelCase_					:      Union[str, Any]     =		speech.astype(np.floataa      )

								# always return batch
								if not is_batched:
													lowerCamelCase_					:      List[Any]     =		[speech]

								# needed to make pad() work on spectrogram inputs
								lowerCamelCase_					:      List[str]     =		self.feature_size

								# convert into correct format for padding
								if is_target:
													lowerCamelCase_					:      List[Any]     =		[self._extract_mel_features(A      ) for waveform in speech]
													lowerCamelCase_					:      Dict     =		BatchFeature({'''input_values''': features}      )
													lowerCamelCase_					:      Any     =		self.num_mel_bins
								else:
													lowerCamelCase_					:      Tuple     =		BatchFeature({'''input_values''': speech}      )

								lowerCamelCase_					:      Tuple     =		self.pad(
								    A			,   padding=A			,   max_length=A			,   truncation=A			,   pad_to_multiple_of=A			,   return_attention_mask=A			,   **A			,   )

								lowerCamelCase_					:      Union[str, Any]     =		feature_size_hack

								# convert input values to correct format
								lowerCamelCase_					:      int     =		padded_inputs['''input_values''']
								if not isinstance(input_values[0]			,   np.ndarray      ):
													lowerCamelCase_					:      Any     =		[np.asarray(A			,   dtype=np.floataa      ) for array in input_values]
								elif (
								    not isinstance(A			,   np.ndarray      )
								    and isinstance(input_values[0]			,   np.ndarray      )
								    and input_values[0].dtype is np.dtype(np.floataa      )
								):
													lowerCamelCase_					:      Tuple     =		[array.astype(np.floataa      ) for array in input_values]
								elif isinstance(A			,   np.ndarray      ) and input_values.dtype is np.dtype(np.floataa      ):
													lowerCamelCase_					:      List[Any]     =		input_values.astype(np.floataa      )

								# convert attention_mask to correct format
								lowerCamelCase_					:      List[Any]     =		padded_inputs.get('''attention_mask'''      )
								if attention_mask is not None:
													lowerCamelCase_					:      Any     =		[np.asarray(A			,   dtype=np.intaa      ) for array in attention_mask]

								# zero-mean and unit-variance normalization
								if not is_target and self.do_normalize:
													lowerCamelCase_					:      Optional[int]     =		(
													    attention_mask
													    if self._get_padding_strategies(A			,   max_length=A      ) is not PaddingStrategy.DO_NOT_PAD
													    else None
													)
													lowerCamelCase_					:      str     =		self.zero_mean_unit_var_norm(
													    padded_inputs['''input_values''']			,   attention_mask=A			,   padding_value=self.padding_value      )

								if return_tensors is not None:
													lowerCamelCase_					:      Union[str, Any]     =		padded_inputs.convert_to_tensors(A      )

								return padded_inputs






			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      str     =		super().to_dict()

								# Don't serialize these as they are derived from the other properties.
								lowerCamelCase_					:      Any     =		['''window''', '''mel_filters''', '''sample_size''', '''sample_stride''', '''n_fft''', '''n_freqs''']
								for name in names:
													if name in output:
																		del output[name]

								return output




'''simple docstring'''


import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer

class 	__lowercase				(  _lowercase    ):

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Any     =		tempfile.mkdtemp()
								lowerCamelCase_					:      Optional[int]     =		5

								# Realm tok
								lowerCamelCase_					:      Any     =		[
								    '''[UNK]''',
								    '''[CLS]''',
								    '''[SEP]''',
								    '''[PAD]''',
								    '''[MASK]''',
								    '''test''',
								    '''question''',
								    '''this''',
								    '''is''',
								    '''the''',
								    '''first''',
								    '''second''',
								    '''third''',
								    '''fourth''',
								    '''fifth''',
								    '''record''',
								    '''want''',
								    '''##want''',
								    '''##ed''',
								    '''wa''',
								    '''un''',
								    '''runn''',
								    '''##ing''',
								    ''',''',
								    '''low''',
								    '''lowest''',
								]
								lowerCamelCase_					:      Optional[int]     =		os.path.join(self.tmpdirname			,   '''realm_tokenizer'''      )
								os.makedirs(A			,   exist_ok=A      )
								lowerCamelCase_					:      str     =		os.path.join(A			,   VOCAB_FILES_NAMES['''vocab_file''']      )
								with open(self.vocab_file			,   '''w'''			,   encoding='''utf-8'''      ) as vocab_writer:
													vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]      )      )

								lowerCamelCase_					:      List[Any]     =		os.path.join(self.tmpdirname			,   '''realm_block_records'''      )
								os.makedirs(A			,   exist_ok=A      )

			def        UpperCAmelCase__ (self      ):
								return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname			,   '''realm_tokenizer'''      )      )

			def        UpperCAmelCase__ (self      ):
								shutil.rmtree(self.tmpdirname      )

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      int     =		RealmConfig(num_block_records=self.num_block_records      )
								return config

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Optional[int]     =		Dataset.from_dict(
								    {
								        '''id''': ['''0''', '''1'''],
								        '''question''': ['''foo''', '''bar'''],
								        '''answers''': [['''Foo''', '''Bar'''], ['''Bar''']],
								    }      )
								return dataset

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Any     =		np.array(
								    [
								        B'''This is the first record''',
								        B'''This is the second record''',
								        B'''This is the third record''',
								        B'''This is the fourth record''',
								        B'''This is the fifth record''',
								        B'''This is a longer longer longer record''',
								    ]			,   dtype=A			,   )
								return block_records

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Dict     =		RealmRetriever(
								    block_records=self.get_dummy_block_records()			,   tokenizer=self.get_tokenizer()			,   )
								return retriever

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Optional[Any]     =		self.get_config()
								lowerCamelCase_					:      List[Any]     =		self.get_dummy_retriever()
								lowerCamelCase_					:      Dict     =		retriever.tokenizer

								lowerCamelCase_					:      List[Any]     =		np.array([0, 3]			,   dtype='''long'''      )
								lowerCamelCase_					:      int     =		tokenizer(['''Test question''']      ).input_ids
								lowerCamelCase_					:      Any     =		tokenizer(
								    ['''the fourth''']			,   add_special_tokens=A			,   return_token_type_ids=A			,   return_attention_mask=A			,   ).input_ids
								lowerCamelCase_					:      Any     =		config.reader_seq_len

								lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_					:      Tuple     =		retriever(
								    A			,   A			,   answer_ids=A			,   max_length=A			,   return_tensors='''np'''      )

								self.assertEqual(len(A      )			,   2      )
								self.assertEqual(len(A      )			,   2      )
								self.assertEqual(len(A      )			,   2      )
								self.assertEqual(concat_inputs.input_ids.shape			,   (2, 1_0)      )
								self.assertEqual(concat_inputs.attention_mask.shape			,   (2, 1_0)      )
								self.assertEqual(concat_inputs.token_type_ids.shape			,   (2, 1_0)      )
								self.assertEqual(concat_inputs.special_tokens_mask.shape			,   (2, 1_0)      )
								self.assertEqual(
								    tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]      )			,   ['''[CLS]''', '''test''', '''question''', '''[SEP]''', '''this''', '''is''', '''the''', '''first''', '''record''', '''[SEP]''']			,   )
								self.assertEqual(
								    tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]      )			,   ['''[CLS]''', '''test''', '''question''', '''[SEP]''', '''this''', '''is''', '''the''', '''fourth''', '''record''', '''[SEP]''']			,   )

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      str     =		self.get_config()
								lowerCamelCase_					:      Optional[Any]     =		self.get_dummy_retriever()
								lowerCamelCase_					:      Optional[int]     =		retriever.tokenizer

								lowerCamelCase_					:      int     =		np.array([0, 3, 5]			,   dtype='''long'''      )
								lowerCamelCase_					:      List[Any]     =		tokenizer(['''Test question''']      ).input_ids
								lowerCamelCase_					:      int     =		tokenizer(
								    ['''the fourth''', '''longer longer''']			,   add_special_tokens=A			,   return_token_type_ids=A			,   return_attention_mask=A			,   ).input_ids
								lowerCamelCase_					:      Tuple     =		config.reader_seq_len

								lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_					:      int     =		retriever(
								    A			,   A			,   answer_ids=A			,   max_length=A			,   return_tensors='''np'''      )

								self.assertEqual([False, True, True]			,   A      )
								self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]			,   A      )
								self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]			,   A      )






			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Any     =		self.get_dummy_retriever()
								retriever.save_pretrained(os.path.join(self.tmpdirname			,   '''realm_block_records'''      )      )

								# Test local path
								lowerCamelCase_					:      str     =		retriever.from_pretrained(os.path.join(self.tmpdirname			,   '''realm_block_records'''      )      )
								self.assertEqual(retriever.block_records[0]			,   B'''This is the first record'''      )

								# Test mocked remote path
								with patch('''transformers.models.realm.retrieval_realm.hf_hub_download'''      ) as mock_hf_hub_download:
													lowerCamelCase_					:      Dict     =		os.path.join(
													    os.path.join(self.tmpdirname			,   '''realm_block_records'''      )			,   _REALM_BLOCK_RECORDS_FILENAME      )
													lowerCamelCase_					:      str     =		RealmRetriever.from_pretrained('''google/realm-cc-news-pretrained-openqa'''      )

								self.assertEqual(retriever.block_records[0]			,   B'''This is the first record'''      )


