





'''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__    :      Any						=    logging.get_logger(__name__)


class  SCREAMING_SNAKE_CASE    (a__							):
     lowerCAmelCase					=       ['''input_values''', '''attention_mask''']


     def __init__(						self						,	_UpperCAmelCase = 1						,	_UpperCAmelCase = 1_6000						,	_UpperCAmelCase = 0.0						,	_UpperCAmelCase = False						,	_UpperCAmelCase = 80						,	_UpperCAmelCase = 16						,	_UpperCAmelCase = 64						,	_UpperCAmelCase = "hann_window"						,	_UpperCAmelCase = 1.0						,	_UpperCAmelCase = 80						,	_UpperCAmelCase = 7600						,	_UpperCAmelCase = 1e-1_0						,	_UpperCAmelCase = 2						,	_UpperCAmelCase = True						,	**_UpperCAmelCase						,	):

          '''simple docstring'''






          super().__init__(feature_size=_UpperCAmelCase						,	sampling_rate=_UpperCAmelCase						,	padding_value=_UpperCAmelCase						,	**_UpperCAmelCase)
          __A						: Optional[Any]      =       do_normalize
          __A						: Any      =       return_attention_mask

          __A						: int      =       num_mel_bins
          __A						: str      =       hop_length
          __A						: Any      =       win_length
          __A						: int      =       win_function
          __A						: int      =       frame_signal_scale
          __A						: int      =       fmin
          __A						: int      =       fmax
          __A						: Any      =       mel_floor
          __A						: Tuple      =       reduction_factor

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

          __A						: Dict      =       window_function(window_length=self.sample_size						,	name=self.win_function						,	periodic=_UpperCAmelCase)

          __A						: Any      =       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'						,	_UpperCAmelCase						,	)
          if reduction_factor != 2.0:
               warnings.warn(
                   'The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'						,	_UpperCAmelCase						,	)


     @staticmethod
     # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
     def 							SCREAMING_SNAKE_CASE  (						_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase = 0.0):

          '''simple docstring'''






          if attention_mask is not None:
               __A						: Any      =       np.array(_UpperCAmelCase						,	np.intaa)
               __A						: Any      =       []

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

                    normed_input_values.append(_UpperCAmelCase)
          else:
               __A						: str      =       [(x - x.mean()) / np.sqrt(x.var() + 1e-7) for x in input_values]

          return normed_input_values


     def 							SCREAMING_SNAKE_CASE  (						self						,	_UpperCAmelCase						,	):

          '''simple docstring'''






          __A						: str      =       spectrogram(
              _UpperCAmelCase						,	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						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	_UpperCAmelCase = False						,	_UpperCAmelCase = None						,	_UpperCAmelCase = False						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	**_UpperCAmelCase						,	):

          '''simple docstring'''






          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:
               __A						: Optional[Any]      =       self._process_audio(
                   _UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	**_UpperCAmelCase						,	)
          else:
               __A						: Optional[int]      =       None

          if audio_target is not None:
               __A						: List[Any]      =       self._process_audio(
                   _UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	**_UpperCAmelCase						,	)

               if inputs is None:
                    return inputs_target
               else:
                    __A						: Optional[int]      =       inputs_target['input_values']
                    __A						: Optional[Any]      =       inputs_target.get('attention_mask')
                    if decoder_attention_mask is not None:
                         __A						: Optional[int]      =       decoder_attention_mask

          return inputs


     def 							SCREAMING_SNAKE_CASE  (						self						,	_UpperCAmelCase						,	_UpperCAmelCase = False						,	_UpperCAmelCase = False						,	_UpperCAmelCase = None						,	_UpperCAmelCase = False						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	**_UpperCAmelCase						,	):

          '''simple docstring'''






          __A						: List[str]      =       isinstance(_UpperCAmelCase						,	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}')
          __A						: Optional[int]      =       is_batched_numpy or (
              isinstance(_UpperCAmelCase						,	(list, tuple)) and (isinstance(speech[0]						,	(np.ndarray, tuple, list)))
          )

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

          # always return batch
          if not is_batched:
               __A						: List[str]      =       [speech]

          # needed to make pad() work on spectrogram inputs
          __A						: int      =       self.feature_size

          # convert into correct format for padding
          if is_target:
               __A						: str      =       [self._extract_mel_features(_UpperCAmelCase) for waveform in speech]
               __A						: List[Any]      =       BatchFeature({'input_values': features})
               __A						: List[Any]      =       self.num_mel_bins
          else:
               __A						: Optional[Any]      =       BatchFeature({'input_values': speech})

          __A						: Union[str, Any]      =       self.pad(
              _UpperCAmelCase						,	padding=_UpperCAmelCase						,	max_length=_UpperCAmelCase						,	truncation=_UpperCAmelCase						,	pad_to_multiple_of=_UpperCAmelCase						,	return_attention_mask=_UpperCAmelCase						,	**_UpperCAmelCase						,	)

          __A						: Tuple      =       feature_size_hack

          # convert input values to correct format
          __A						: Optional[Any]      =       padded_inputs['input_values']
          if not isinstance(input_values[0]						,	np.ndarray):
               __A						: Dict      =       [np.asarray(_UpperCAmelCase						,	dtype=np.floataa) for array in input_values]
          elif (
              not isinstance(_UpperCAmelCase						,	np.ndarray)
              and isinstance(input_values[0]						,	np.ndarray)
              and input_values[0].dtype is np.dtype(np.floataa)
          ):
               __A						: Any      =       [array.astype(np.floataa) for array in input_values]
          elif isinstance(_UpperCAmelCase						,	np.ndarray) and input_values.dtype is np.dtype(np.floataa):
               __A						: Any      =       input_values.astype(np.floataa)

          # convert attention_mask to correct format
          __A						: Dict      =       padded_inputs.get('attention_mask')
          if attention_mask is not None:
               __A						: Union[str, Any]      =       [np.asarray(_UpperCAmelCase						,	dtype=np.intaa) for array in attention_mask]

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

          if return_tensors is not None:
               __A						: List[Any]      =       padded_inputs.convert_to_tensors(_UpperCAmelCase)

          return padded_inputs


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: List[str]      =       super().to_dict()

          # Don't serialize these as they are derived from the other properties.
          __A						: 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  SCREAMING_SNAKE_CASE    (a__							):


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Any      =       tempfile.mkdtemp()
          __A						: str      =       5

          # Realm tok
          __A						: Union[str, 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',
          ]
          __A						: Union[str, Any]      =       os.path.join(self.tmpdirname						,	'realm_tokenizer')
          os.makedirs(_UpperCAmelCase						,	exist_ok=_UpperCAmelCase)
          __A						: Tuple      =       os.path.join(_UpperCAmelCase						,	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]))

          __A						: int      =       os.path.join(self.tmpdirname						,	'realm_block_records')
          os.makedirs(_UpperCAmelCase						,	exist_ok=_UpperCAmelCase)


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






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


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          shutil.rmtree(self.tmpdirname)


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Union[str, Any]      =       RealmConfig(num_block_records=self.num_block_records)
          return config


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Dict      =       Dataset.from_dict(
              {
                  'id': ['0', '1'],
                  'question': ['foo', 'bar'],
                  'answers': [['Foo', 'Bar'], ['Bar']],
              })
          return dataset


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: 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=_UpperCAmelCase						,	)
          return block_records


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Optional[Any]      =       RealmRetriever(
              block_records=self.get_dummy_block_records()						,	tokenizer=self.get_tokenizer()						,	)
          return retriever


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: List[Any]      =       self.get_config()
          __A						: str      =       self.get_dummy_retriever()
          __A						: List[str]      =       retriever.tokenizer

          __A						: Dict      =       np.array([0, 3]						,	dtype='long')
          __A						: Dict      =       tokenizer(['Test question']).input_ids
          __A						: Optional[Any]      =       tokenizer(
              ['the fourth']						,	add_special_tokens=_UpperCAmelCase						,	return_token_type_ids=_UpperCAmelCase						,	return_attention_mask=_UpperCAmelCase						,	).input_ids
          __A						: str      =       config.reader_seq_len

          __A ,__A ,__A ,__A						: List[str]      =       retriever(
              _UpperCAmelCase						,	_UpperCAmelCase						,	answer_ids=_UpperCAmelCase						,	max_length=_UpperCAmelCase						,	return_tensors='np')

          self.assertEqual(len(_UpperCAmelCase)						,	2)
          self.assertEqual(len(_UpperCAmelCase)						,	2)
          self.assertEqual(len(_UpperCAmelCase)						,	2)
          self.assertEqual(concat_inputs.input_ids.shape						,	(2, 10))
          self.assertEqual(concat_inputs.attention_mask.shape						,	(2, 10))
          self.assertEqual(concat_inputs.token_type_ids.shape						,	(2, 10))
          self.assertEqual(concat_inputs.special_tokens_mask.shape						,	(2, 10))
          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 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: int      =       self.get_config()
          __A						: Any      =       self.get_dummy_retriever()
          __A						: str      =       retriever.tokenizer

          __A						: Dict      =       np.array([0, 3, 5]						,	dtype='long')
          __A						: Tuple      =       tokenizer(['Test question']).input_ids
          __A						: Union[str, Any]      =       tokenizer(
              ['the fourth', 'longer longer']						,	add_special_tokens=_UpperCAmelCase						,	return_token_type_ids=_UpperCAmelCase						,	return_attention_mask=_UpperCAmelCase						,	).input_ids
          __A						: Dict      =       config.reader_seq_len

          __A ,__A ,__A ,__A						: str      =       retriever(
              _UpperCAmelCase						,	_UpperCAmelCase						,	answer_ids=_UpperCAmelCase						,	max_length=_UpperCAmelCase						,	return_tensors='np')

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


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Optional[Any]      =       self.get_dummy_retriever()
          retriever.save_pretrained(os.path.join(self.tmpdirname						,	'realm_block_records'))

          # Test local path
          __A						: 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:
               __A						: int      =       os.path.join(
                   os.path.join(self.tmpdirname						,	'realm_block_records')						,	_REALM_BLOCK_RECORDS_FILENAME)
               __A						: Tuple      =       RealmRetriever.from_pretrained('google/realm-cc-news-pretrained-openqa')

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