






import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

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


lowerCamelCase	:							Optional[Any]				=logging.get_logger(__name__)





class 			__a  (    __lowerCamelCase       ):
 _lowerCAmelCase       :       str        = ["""input_features""", """is_longer"""]







 def __init__(		self					:   Union[str, Any]							,   SCREAMING_SNAKE_CASE					:   str=64							,   SCREAMING_SNAKE_CASE					:   Any=4_80_00							,   SCREAMING_SNAKE_CASE					:   List[Any]=4_80							,   SCREAMING_SNAKE_CASE					:   Optional[int]=10							,   SCREAMING_SNAKE_CASE					:   Optional[int]=10_24							,   SCREAMING_SNAKE_CASE					:   int=0.0							,   SCREAMING_SNAKE_CASE					:   List[Any]=False							,   SCREAMING_SNAKE_CASE					:   float = 0							,   SCREAMING_SNAKE_CASE					:   float = 1_40_00							,   SCREAMING_SNAKE_CASE					:   int = None							,   SCREAMING_SNAKE_CASE					:   str = "fusion"							,   SCREAMING_SNAKE_CASE					:   str = "repeatpad"							,   **SCREAMING_SNAKE_CASE					:   List[Any]							,   ):


      '''simple docstring'''






      super().__init__(
          feature_size=__lowercase							,   sampling_rate=__lowercase							,   padding_value=__lowercase							,   return_attention_mask=__lowercase							,   **__lowercase							,   )
      UpperCamelCase__			:   Optional[Any]       				=  top_db
      UpperCamelCase__			:   Tuple       				=  truncation
      UpperCamelCase__			:   Tuple       				=  padding
      UpperCamelCase__			:   List[Any]       				=  fft_window_size
      UpperCamelCase__			:   List[Any]       				=  (fft_window_size >> 1) + 1
      UpperCamelCase__			:   str       				=  hop_length
      UpperCamelCase__			:   Dict       				=  max_length_s
      UpperCamelCase__			:   List[str]       				=  max_length_s * sampling_rate
      UpperCamelCase__			:   List[Any]       				=  sampling_rate
      UpperCamelCase__			:   Any       				=  frequency_min
      UpperCamelCase__			:   Dict       				=  frequency_max
      UpperCamelCase__			:   Tuple       				=  mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins							,   num_mel_filters=__lowercase							,   min_frequency=__lowercase							,   max_frequency=__lowercase							,   sampling_rate=__lowercase							,   norm=__lowercase							,   mel_scale="htk"							,   )
      UpperCamelCase__			:   Dict       				=  mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins							,   num_mel_filters=__lowercase							,   min_frequency=__lowercase							,   max_frequency=__lowercase							,   sampling_rate=__lowercase							,   norm="slaney"							,   mel_scale="slaney"							,   )







 def 					__lowercase					(		self					:   int						):


      '''simple docstring'''






      UpperCamelCase__			:   Dict       				=  copy.deepcopy(self.__dict__						)
      UpperCamelCase__			:   Optional[int]       				=  self.__class__.__name__
      if "mel_filters" in output:
           del output["mel_filters"]
      if "mel_filters_slaney" in output:
           del output["mel_filters_slaney"]
      return output







 def 					__lowercase					(		self					:   Union[str, Any]							,   SCREAMING_SNAKE_CASE					:   np.array							,   SCREAMING_SNAKE_CASE					:   Optional[np.array] = None						):


      '''simple docstring'''






      UpperCamelCase__			:   List[Any]       				=  spectrogram(
          __lowercase							,   window_function(self.fft_window_size							,   "hann"						)							,   frame_length=self.fft_window_size							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=__lowercase							,   log_mel="dB"							,   )
      return log_mel_spectrogram.T







 def 					__lowercase					(		self					:   Optional[Any]							,   SCREAMING_SNAKE_CASE					:   List[str]							,   SCREAMING_SNAKE_CASE					:   Tuple							,   SCREAMING_SNAKE_CASE					:   List[str]						):


      '''simple docstring'''






      UpperCamelCase__			:   Dict       				=  np.array_split(list(range(0							,   total_frames - chunk_frames + 1						)						)							,   3						)
      if len(ranges[1]						) == 0:
           # if the audio is too short, we just use the first chunk
           UpperCamelCase__			:   Tuple       				=  [0]
      if len(ranges[2]						) == 0:
           # if the audio is too short, we just use the first chunk
           UpperCamelCase__			:   str       				=  [0]
      # randomly choose index for each part
      UpperCamelCase__			:   Dict       				=  np.random.choice(ranges[0]						)
      UpperCamelCase__			:   Any       				=  np.random.choice(ranges[1]						)
      UpperCamelCase__			:   Dict       				=  np.random.choice(ranges[2]						)

      UpperCamelCase__			:   Optional[int]       				=  mel[idx_front : idx_front + chunk_frames, :]
      UpperCamelCase__			:   Optional[Any]       				=  mel[idx_middle : idx_middle + chunk_frames, :]
      UpperCamelCase__			:   List[str]       				=  mel[idx_back : idx_back + chunk_frames, :]

      UpperCamelCase__			:   Optional[Any]       				=  torch.tensor(mel[None, None, :]						)
      UpperCamelCase__			:   Any       				=  torch.nn.functional.interpolate(
          __lowercase							,   size=[chunk_frames, 64]							,   mode="bilinear"							,   align_corners=__lowercase						)
      UpperCamelCase__			:   List[Any]       				=  mel_shrink[0][0].numpy()
      UpperCamelCase__			:   Union[str, Any]       				=  np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]							,   axis=0						)
      return mel_fusion







 def 					__lowercase					(		self					:   Any							,   SCREAMING_SNAKE_CASE					:   np.array							,   SCREAMING_SNAKE_CASE					:   str							,   SCREAMING_SNAKE_CASE					:   int							,   SCREAMING_SNAKE_CASE					:   List[str]						):


      '''simple docstring'''






      if waveform.shape[0] > max_length:
           if truncation == "rand_trunc":
                UpperCamelCase__			:   Tuple       				=  True
                # random crop to max_length (for compatibility) -> this should be handled by self.pad
                UpperCamelCase__			:   List[str]       				=  len(__lowercase						) - max_length
                UpperCamelCase__			:   Any       				=  np.random.randint(0							,   overflow + 1						)
                UpperCamelCase__			:   Tuple       				=  waveform[idx : idx + max_length]
                UpperCamelCase__			:   Tuple       				=  self._np_extract_fbank_features(__lowercase							,   self.mel_filters_slaney						)[None, :]
           elif truncation == "fusion":
                UpperCamelCase__			:   str       				=  self._np_extract_fbank_features(__lowercase							,   self.mel_filters						)
                UpperCamelCase__			:   Union[str, Any]       				=  max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                UpperCamelCase__			:   Union[str, Any]       				=  mel.shape[0]
                if chunk_frames == total_frames:
                     # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                     # In this case, we just use the whole audio.
                     UpperCamelCase__			:   Tuple       				=  np.stack([mel, mel, mel, mel]							,   axis=0						)
                     UpperCamelCase__			:   List[Any]       				=  False
                else:
                     UpperCamelCase__			:   List[Any]       				=  self._random_mel_fusion(__lowercase							,   __lowercase							,   __lowercase						)
                     UpperCamelCase__			:   Dict       				=  True
           else:
                raise NotImplementedError(F'data_truncating {truncation} not implemented'						)

      else:
           UpperCamelCase__			:   Optional[int]       				=  False
           # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
           if waveform.shape[0] < max_length:
                if padding == "repeat":
                     UpperCamelCase__			:   List[str]       				=  int(max_length / len(__lowercase						)						)
                     UpperCamelCase__			:   List[str]       				=  np.stack(np.tile(__lowercase							,   n_repeat + 1						)						)[:max_length]
                if padding == "repeatpad":
                     UpperCamelCase__			:   Union[str, Any]       				=  int(max_length / len(__lowercase						)						)
                     UpperCamelCase__			:   List[str]       				=  np.stack(np.tile(__lowercase							,   __lowercase						)						)
                UpperCamelCase__			:   int       				=  np.pad(__lowercase							,   (0, max_length - waveform.shape[0])							,   mode="constant"							,   constant_values=0						)

           if truncation == "fusion":
                UpperCamelCase__			:   Tuple       				=  self._np_extract_fbank_features(__lowercase							,   self.mel_filters						)
                UpperCamelCase__			:   Optional[int]       				=  np.stack([input_mel, input_mel, input_mel, input_mel]							,   axis=0						)
           else:
                UpperCamelCase__			:   List[Any]       				=  self._np_extract_fbank_features(__lowercase							,   self.mel_filters_slaney						)[None, :]

      return input_mel, longer







 def __call__(		self					:   Dict							,   SCREAMING_SNAKE_CASE					:   Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,   SCREAMING_SNAKE_CASE					:   str = None							,   SCREAMING_SNAKE_CASE					:   Optional[str] = None							,   SCREAMING_SNAKE_CASE					:   Optional[int] = None							,   SCREAMING_SNAKE_CASE					:   Optional[int] = None							,   SCREAMING_SNAKE_CASE					:   Optional[Union[str, TensorType]] = None							,   **SCREAMING_SNAKE_CASE					:   Optional[int]							,   ):


      '''simple docstring'''






      UpperCamelCase__			:   Optional[int]       				=  truncation if truncation is not None else self.truncation
      UpperCamelCase__			:   Optional[int]       				=  padding if padding else self.padding

      if sampling_rate is not None:
           if sampling_rate != self.sampling_rate:
                raise ValueError(
                    F'The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'
                    F' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'
                    F' was sampled with {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."						)

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

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

      # always return batch
      if not is_batched:
           UpperCamelCase__			:   Dict       				=  [np.asarray(__lowercase						)]

      # convert to mel spectrogram, truncate and pad if needed.
      UpperCamelCase__			:   Any       				=  [
          self._get_input_mel(__lowercase							,   max_length if max_length else self.nb_max_samples							,   __lowercase							,   __lowercase						)
          for waveform in raw_speech
      ]

      UpperCamelCase__			:   Any       				=  []
      UpperCamelCase__			:   Tuple       				=  []
      for mel, longer in padded_inputs:
           input_mel.append(__lowercase						)
           is_longer.append(__lowercase						)

      if truncation == "fusion" and sum(__lowercase						) == 0:
           # if no audio is longer than 10s, then randomly select one audio to be longer
           UpperCamelCase__			:   Optional[int]       				=  np.random.randint(0							,   len(__lowercase						)						)
           UpperCamelCase__			:   List[str]       				=  True

      if isinstance(input_mel[0]							,   __lowercase						):
           UpperCamelCase__			:   Optional[int]       				=  [np.asarray(__lowercase							,   dtype=np.floataa						) for feature in input_mel]

      # is_longer is a list of bool
      UpperCamelCase__			:   Dict       				=  [[longer] for longer in is_longer]

      UpperCamelCase__			:   Dict       				=  {'''input_features''': input_mel, '''is_longer''': is_longer}
      UpperCamelCase__			:   str       				=  BatchFeature(__lowercase						)

      if return_tensors is not None:
           UpperCamelCase__			:   List[str]       				=  input_features.convert_to_tensors(__lowercase						)

      return input_features




import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import PaddingStrategy, logging
from .tokenization_realm import RealmTokenizer


A__								=						logging.get_logger(__name__)

A__								=						{'''vocab_file''': '''vocab.txt''', '''tokenizer_file''': '''tokenizer.json'''}

A__								=						{
    '''vocab_file''': {
        '''google/realm-cc-news-pretrained-embedder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/vocab.txt'''
        ),
        '''google/realm-cc-news-pretrained-encoder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/vocab.txt'''
        ),
        '''google/realm-cc-news-pretrained-scorer''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/vocab.txt'''
        ),
        '''google/realm-cc-news-pretrained-openqa''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/vocab.txt'''
        ),
        '''google/realm-orqa-nq-openqa''': '''https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/vocab.txt''',
        '''google/realm-orqa-nq-reader''': '''https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/vocab.txt''',
        '''google/realm-orqa-wq-openqa''': '''https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/vocab.txt''',
        '''google/realm-orqa-wq-reader''': '''https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/vocab.txt''',
    },
    '''tokenizer_file''': {
        '''google/realm-cc-news-pretrained-embedder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/tokenizer.jsont'''
        ),
        '''google/realm-cc-news-pretrained-encoder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/tokenizer.json'''
        ),
        '''google/realm-cc-news-pretrained-scorer''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/tokenizer.json'''
        ),
        '''google/realm-cc-news-pretrained-openqa''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-nq-openqa''': (
            '''https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-nq-reader''': (
            '''https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-wq-openqa''': (
            '''https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-wq-reader''': (
            '''https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/tokenizer.json'''
        ),
    },
}

A__								=						{
    '''google/realm-cc-news-pretrained-embedder''': 512,
    '''google/realm-cc-news-pretrained-encoder''': 512,
    '''google/realm-cc-news-pretrained-scorer''': 512,
    '''google/realm-cc-news-pretrained-openqa''': 512,
    '''google/realm-orqa-nq-openqa''': 512,
    '''google/realm-orqa-nq-reader''': 512,
    '''google/realm-orqa-wq-openqa''': 512,
    '''google/realm-orqa-wq-reader''': 512,
}

A__								=						{
    '''google/realm-cc-news-pretrained-embedder''': {'''do_lower_case''': True},
    '''google/realm-cc-news-pretrained-encoder''': {'''do_lower_case''': True},
    '''google/realm-cc-news-pretrained-scorer''': {'''do_lower_case''': True},
    '''google/realm-cc-news-pretrained-openqa''': {'''do_lower_case''': True},
    '''google/realm-orqa-nq-openqa''': {'''do_lower_case''': True},
    '''google/realm-orqa-nq-reader''': {'''do_lower_case''': True},
    '''google/realm-orqa-wq-openqa''': {'''do_lower_case''': True},
    '''google/realm-orqa-wq-reader''': {'''do_lower_case''': True},
}


class       a							(			__lowerCamelCase				):
		__lowerCAmelCase	:    List[Any]											=					VOCAB_FILES_NAMES
		__lowerCAmelCase	:    Optional[Any]											=					PRETRAINED_VOCAB_FILES_MAP
		__lowerCAmelCase	:    Optional[Any]											=					PRETRAINED_INIT_CONFIGURATION
		__lowerCAmelCase	:    Optional[Any]											=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		__lowerCAmelCase	:    Optional[Any]											=					RealmTokenizer



		def __init__(     self     :Optional[Any]   ,__lowercase     :Dict=None   ,__lowercase     :Optional[int]=None   ,__lowercase     :Optional[Any]=True   ,__lowercase     :Optional[int]="[UNK]"   ,__lowercase     :List[str]="[SEP]"   ,__lowercase     :List[str]="[PAD]"   ,__lowercase     :int="[CLS]"   ,__lowercase     :str="[MASK]"   ,__lowercase     :Dict=True   ,__lowercase     :List[str]=None   ,**__lowercase     :Any   ,):
						super().__init__(
						    __lowercase   ,tokenizer_file=__lowercase   ,do_lower_case=__lowercase   ,unk_token=__lowercase   ,sep_token=__lowercase   ,pad_token=__lowercase   ,cls_token=__lowercase   ,mask_token=__lowercase   ,tokenize_chinese_chars=__lowercase   ,strip_accents=__lowercase   ,**__lowercase   ,)

						snake_case__	:      List[str]     		=	json.loads(self.backend_tokenizer.normalizer.__getstate__()  )
						if (
						    normalizer_state.get('''lowercase'''   ,__lowercase  ) != do_lower_case
						    or normalizer_state.get('''strip_accents'''   ,__lowercase  ) != strip_accents
						    or normalizer_state.get('''handle_chinese_chars'''   ,__lowercase  ) != tokenize_chinese_chars
						):
										snake_case__	:      Optional[int]     		=	getattr(__lowercase   ,normalizer_state.pop('''type'''  )  )
										snake_case__	:      List[Any]     		=	do_lower_case
										snake_case__	:      Optional[Any]     		=	strip_accents
										snake_case__	:      List[str]     		=	tokenize_chinese_chars
										snake_case__	:      Dict     		=	normalizer_class(**__lowercase  )

						snake_case__	:      Tuple     		=	do_lower_case



		def 	__lowerCamelCase			(     self     :Union[str, Any]   ,__lowercase     :Union[str, Any]   ,**__lowercase     :Any  ):
						snake_case__	:      Dict     		=	PaddingStrategy.MAX_LENGTH

						snake_case__	:      List[str]     		=	text
						snake_case__	:      int     		=	kwargs.pop('''text_pair'''   ,__lowercase  )
						snake_case__	:      Optional[int]     		=	kwargs.pop('''return_tensors'''   ,__lowercase  )

						snake_case__	:      str     		=	{
						    '''input_ids''': [],
						    '''attention_mask''': [],
						    '''token_type_ids''': [],
						}

						for idx, candidate_text in enumerate(__lowercase  ):
										if batch_text_pair is not None:
														snake_case__	:      Optional[int]     		=	batch_text_pair[idx]
										else:
														snake_case__	:      Tuple     		=	None

										snake_case__	:      List[str]     		=	super().__call__(__lowercase   ,__lowercase   ,return_tensors=__lowercase   ,**__lowercase  )

										snake_case__	:      Optional[Any]     		=	encoded_candidates.get('''input_ids'''  )
										snake_case__	:      Optional[Any]     		=	encoded_candidates.get('''attention_mask'''  )
										snake_case__	:      List[Any]     		=	encoded_candidates.get('''token_type_ids'''  )

										if encoded_input_ids is not None:
														output_data["input_ids"].append(__lowercase  )
										if encoded_attention_mask is not None:
														output_data["attention_mask"].append(__lowercase  )
										if encoded_token_type_ids is not None:
														output_data["token_type_ids"].append(__lowercase  )

						snake_case__	:      Any     		=	{key: item for key, item in output_data.items() if len(__lowercase  ) != 0}

						return BatchEncoding(__lowercase   ,tensor_type=__lowercase  )



		def 	__lowerCamelCase			(     self     :List[Any]   ,__lowercase     :Tuple   ,__lowercase     :Tuple=None  ):
						snake_case__	:      List[str]     		=	[self.cls_token_id] + token_ids_a + [self.sep_token_id]

						if token_ids_a:
										output += token_ids_a + [self.sep_token_id]

						return output



		def 	__lowerCamelCase			(     self     :List[str]   ,__lowercase     :List[int]   ,__lowercase     :Optional[List[int]] = None  ):
						snake_case__	:      Tuple     		=	[self.sep_token_id]
						snake_case__	:      Union[str, Any]     		=	[self.cls_token_id]
						if token_ids_a is None:
										return len(cls + token_ids_a + sep  ) * [0]
						return len(cls + token_ids_a + sep  ) * [0] + len(token_ids_a + sep  ) * [1]






		def 	__lowerCamelCase			(     self     :Optional[Any]   ,__lowercase     :str   ,__lowercase     :Optional[str] = None  ):
						snake_case__	:      Tuple     		=	self._tokenizer.model.save(__lowercase   ,name=__lowercase  )
						return tuple(__lowercase  )

