



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		(       UpperCAmelCase__	):

   def 		A							(      self			:       Dict							)       ->				Any:


    """simple docstring"""






    a						=		tempfile.mkdtemp()
    a						=		5

    # Realm tok
    a						=		[
        "[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						=		os.path.join(self.tmpdirname							,   "realm_tokenizer"							)
    os.makedirs(__lowerCAmelCase							,   exist_ok=__lowerCAmelCase							)
    a						=		os.path.join(__lowerCAmelCase							,   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						=		os.path.join(self.tmpdirname							,   "realm_block_records"							)
    os.makedirs(__lowerCAmelCase							,   exist_ok=__lowerCAmelCase							)

   def 		A							(      self			:       List[str]							)       ->				RealmTokenizer:


    """simple docstring"""






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

   def 		A							(      self			:       Dict							)       ->				int:


    """simple docstring"""






    shutil.rmtree(self.tmpdirname							)

   def 		A							(      self			:       List[str]							)       ->				List[str]:


    """simple docstring"""






    a						=		RealmConfig(num_block_records=self.num_block_records							)
    return config

   def 		A							(      self			:       str							)       ->				Optional[Any]:


    """simple docstring"""






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

   def 		A							(      self			:       Tuple							)       ->				Union[str, Any]:


    """simple docstring"""






    a						=		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=__lowerCAmelCase							,   )
    return block_records

   def 		A							(      self			:       Any							)       ->				Dict:


    """simple docstring"""






    a						=		RealmRetriever(
        block_records=self.get_dummy_block_records()							,   tokenizer=self.get_tokenizer()							,   )
    return retriever

   def 		A							(      self			:       List[Any]							)       ->				Optional[Any]:


    """simple docstring"""






    a						=		self.get_config()
    a						=		self.get_dummy_retriever()
    a						=		retriever.tokenizer

    a						=		np.array([0, 3]							,   dtype="long"							)
    a						=		tokenizer(["Test question"]							).input_ids
    a						=		tokenizer(
        ["the fourth"]							,   add_special_tokens=__lowerCAmelCase							,   return_token_type_ids=__lowerCAmelCase							,   return_attention_mask=__lowerCAmelCase							,   ).input_ids
    a						=		config.reader_seq_len

    a		,  a		,  a		,  a						=		retriever(
        __lowerCAmelCase							,   __lowerCAmelCase							,   answer_ids=__lowerCAmelCase							,   max_length=__lowerCAmelCase							,   return_tensors="np"							)

    self.assertEqual(len(__lowerCAmelCase							)							,   2							)
    self.assertEqual(len(__lowerCAmelCase							)							,   2							)
    self.assertEqual(len(__lowerCAmelCase							)							,   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 		A							(      self			:       int							)       ->				int:


    """simple docstring"""






    a						=		self.get_config()
    a						=		self.get_dummy_retriever()
    a						=		retriever.tokenizer

    a						=		np.array([0, 3, 5]							,   dtype="long"							)
    a						=		tokenizer(["Test question"]							).input_ids
    a						=		tokenizer(
        ["the fourth", "longer longer"]							,   add_special_tokens=__lowerCAmelCase							,   return_token_type_ids=__lowerCAmelCase							,   return_attention_mask=__lowerCAmelCase							,   ).input_ids
    a						=		config.reader_seq_len

    a		,  a		,  a		,  a						=		retriever(
        __lowerCAmelCase							,   __lowerCAmelCase							,   answer_ids=__lowerCAmelCase							,   max_length=__lowerCAmelCase							,   return_tensors="np"							)

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




   def 		A							(      self			:       Union[str, Any]							)       ->				Tuple:


    """simple docstring"""






    a						=		self.get_dummy_retriever()
    retriever.save_pretrained(os.path.join(self.tmpdirname							,   "realm_block_records"							)							)

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

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









from math import ceil
from typing import List, Optional, Union

import numpy as np

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


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



class 		_lowercase		(       UpperCAmelCase__	):
   _UpperCAmelCase     	=			['''audio_values''', '''audio_mask''']

   def __init__(      self			:       List[Any]							,   __lowerCAmelCase			:       Dict=2048							,   __lowerCAmelCase			:       List[Any]=1							,   __lowerCAmelCase			:       Dict=[16, 16]							,   __lowerCAmelCase			:       str=128							,   __lowerCAmelCase			:       Optional[int]=4_4100							,   __lowerCAmelCase			:       int=86							,   __lowerCAmelCase			:       Optional[Any]=2048							,   __lowerCAmelCase			:       str=0.0							,   **__lowerCAmelCase			:       Optional[int]							,   )       ->				Union[str, Any]:


    """simple docstring"""






    super().__init__(
        feature_size=__lowerCAmelCase							,   sampling_rate=__lowerCAmelCase							,   padding_value=__lowerCAmelCase							,   **__lowerCAmelCase							,   )

    a						=		spectrogram_length
    a						=		num_channels
    a						=		patch_size
    a						=		feature_size // self.patch_size[1]
    a						=		n_fft
    a						=		sampling_rate // hop_length_to_sampling_rate
    a						=		sampling_rate
    a						=		padding_value
    a						=		mel_filter_bank(
        num_frequency_bins=1 + n_fft // 2							,   num_mel_filters=__lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=2_2_0_5_0.0							,   sampling_rate=__lowerCAmelCase							,   norm="slaney"							,   mel_scale="slaney"							,   ).T

   def 		A							(      self			:       List[str]							,   __lowerCAmelCase			:       np.array							)       ->				np.ndarray:


    """simple docstring"""






    a						=		spectrogram(
        __lowerCAmelCase							,   window_function(self.n_fft							,   "hann"							)							,   frame_length=self.n_fft							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=self.mel_filters.T							,   log_mel="dB"							,   db_range=8_0.0							,   )
    a						=		log_spec[:, :-1]
    a						=		log_spec - 2_0.0
    a						=		np.clip(log_spec / 4_0.0							,   -2.0							,   0.0							) + 1.0
    return log_spec




   def __call__(      self			:       Union[str, Any]							,   __lowerCAmelCase			:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,   __lowerCAmelCase			:       Optional[Union[str, TensorType]] = None							,   __lowerCAmelCase			:       Optional[bool] = True							,   __lowerCAmelCase			:       Optional[int] = None							,   __lowerCAmelCase			:       bool = False							,   __lowerCAmelCase			:       bool = False							,   **__lowerCAmelCase			:       Optional[int]							,   )       ->				BatchFeature:


    """simple docstring"""






    if sampling_rate is not None:
     if sampling_rate != self.sampling_rate:
      raise ValueError(
          "This feature extractor is set to support sampling rate"
          f""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
          f""" 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."							)

    a						=		isinstance(__lowerCAmelCase							,   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}"""							)
    a						=		is_batched_numpy or (
        isinstance(__lowerCAmelCase							,   (list, tuple)							) and (isinstance(raw_speech[0]							,   (np.ndarray, tuple, list)							))
    )
    if is_batched:
     a						=		[np.asarray([speech]							,   dtype=np.floataa							).T for speech in raw_speech]
    elif not is_batched and not isinstance(__lowerCAmelCase							,   np.ndarray							):
     a						=		np.asarray(__lowerCAmelCase							,   dtype=np.floataa							)
    elif isinstance(__lowerCAmelCase							,   np.ndarray							) and raw_speech.dtype is np.dtype(np.floataa							):
     a						=		raw_speech.astype(np.floataa							)
    # always return batch
    if not is_batched:
     a						=		[np.asarray([raw_speech]							).T]

    # Convert audio signals to log mel spectrograms, truncate by time axis
    a						=		[
        self._np_extract_fbank_features(waveform.squeeze()							).T[: self.spectrogram_length] for waveform in raw_speech
    ]
    if isinstance(audio_features[0]							,   __lowerCAmelCase							):
     a						=		[np.asarray(__lowerCAmelCase							,   dtype=np.floataa							) for feature in audio_features]

    # Create audio attention mask
    a						=		max(
        [ceil(feature.shape[0] / self.patch_size[0]							) * self.freq_len for feature in audio_features]							)  # The maximum number of audio patches in a batch
    if return_attention_mask:
     a						=		[
         (ceil(feature.shape[0] / self.patch_size[0]							) * self.freq_len) * [1]
         + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]							) * self.freq_len) * [0]
         for feature in audio_features
     ]
     a						=		np.array(__lowerCAmelCase							).astype(np.floataa							)

    # convert into correct format for padding
    a						=		max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
    a						=		np.ones([len(__lowerCAmelCase							), 1, max_time_len, self.feature_size]							).astype(np.floataa							)
    a						=		padded_audio_features * self.padding_value
    for i in range(len(__lowerCAmelCase							)							):
     a						=		audio_features[i]
     a						=		feature

    # return as BatchFeature
    if return_attention_mask:
     a						=		{"audio_values": padded_audio_features, "audio_mask": audio_mask}
    else:
     a						=		{"audio_values": padded_audio_features}

    a						=		BatchFeature(data=__lowerCAmelCase							,   tensor_type=__lowerCAmelCase							)
    return encoded_inputs

