


"""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 						_A  (  _a  ):







    """simple docstring"""




    def 	__snake_case							(					self				:    Union[str, Any]):
      a						:		Any      							=			tempfile.mkdtemp()
      a						:		str      							=			5

      # Realm tok
      a						:		Dict      							=			[
          "[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						:		Tuple      							=			os.path.join(self.tmpdirname				, "realm_tokenizer")
      os.makedirs(__UpperCAmelCase				, exist_ok=__UpperCAmelCase)
      a						:		Optional[Any]      							=			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						:		Dict      							=			os.path.join(self.tmpdirname				, "realm_block_records")
      os.makedirs(__UpperCAmelCase				, exist_ok=__UpperCAmelCase)






    def 	__snake_case							(					self				:    int):
      return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname				, "realm_tokenizer"))






    def 	__snake_case							(					self				:    Dict):
      shutil.rmtree(self.tmpdirname)






    def 	__snake_case							(					self				:    List[Any]):
      a						:		Tuple      							=			RealmConfig(num_block_records=self.num_block_records)
      return config






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






    def 	__snake_case							(					self				:    Any):
      a						:		List[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 	__snake_case							(					self				:    List[str]):
      a						:		Union[str, Any]      							=			RealmRetriever(
          block_records=self.get_dummy_block_records()				, tokenizer=self.get_tokenizer()				, )
      return retriever






    def 	__snake_case							(					self				:    Optional[int]):
      a						:		Union[str, Any]      							=			self.get_config()
      a						:		List[str]      							=			self.get_dummy_retriever()
      a						:		Optional[Any]      							=			retriever.tokenizer

      a						:		int      							=			np.array([0, 3]				, dtype="long")
      a						:		List[str]      							=			tokenizer(["Test question"]).input_ids
      a						:		Union[str, Any]      							=			tokenizer(
          ["the fourth"]				, add_special_tokens=__UpperCAmelCase				, return_token_type_ids=__UpperCAmelCase				, return_attention_mask=__UpperCAmelCase				, ).input_ids
      a						:		List[str]      							=			config.reader_seq_len

      a							,		a							,		a							,		a						:		Optional[Any]      							=			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 	__snake_case							(					self				:    str):
      a						:		Union[str, Any]      							=			self.get_config()
      a						:		List[str]      							=			self.get_dummy_retriever()
      a						:		str      							=			retriever.tokenizer

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

      a							,		a							,		a							,		a						:		Tuple      							=			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 	__snake_case							(					self				:    Optional[Any]):
      a						:		str      							=			self.get_dummy_retriever()
      retriever.save_pretrained(os.path.join(self.tmpdirname				, "realm_block_records"))

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

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







import unittest

from datasets import load_dataset

from transformers.pipelines import pipeline
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_torch, slow



@is_pipeline_test
@require_torch
class   snake_case_       (		unittest.TestCase       ):






   @require_torch
   def       __UpperCamelCase     (      self				:     Optional[int]  )					->						List[Any]:
       lowercase__ :   Union[str, Any]    					=				pipeline(
           task="zero-shot-audio-classification"	,						model="hf-internal-testing/tiny-clap-htsat-unfused"  )
       lowercase__ :   List[str]    					=				load_dataset("ashraq/esc50"  )
       lowercase__ :   List[Any]    					=				dataset["train"]["audio"][-1]["array"]
       lowercase__ :   Dict    					=				audio_classifier(lowercase_	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]  )
       self.assertEqual(
           nested_simplify(lowercase_  )	,						[{"score": 0.5_01, "label": "Sound of a dog"}, {"score": 0.4_99, "label": "Sound of vaccum cleaner"}]	,						)






   @unittest.skip("No models are available in TF"  )
   def       __UpperCamelCase     (      self				:     str  )					->						Optional[int]:
       pass






   @slow
   @require_torch
   def       __UpperCamelCase     (      self				:     List[str]  )					->						int:
       lowercase__ :   Tuple    					=				pipeline(
           task="zero-shot-audio-classification"	,						model="laion/clap-htsat-unfused"	,						)
       # This is an audio of a dog
       lowercase__ :   Union[str, Any]    					=				load_dataset("ashraq/esc50"  )
       lowercase__ :   Tuple    					=				dataset["train"]["audio"][-1]["array"]
       lowercase__ :   List[Any]    					=				audio_classifier(lowercase_	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]  )

       self.assertEqual(
           nested_simplify(lowercase_  )	,						[
               {"score": 0.9_99, "label": "Sound of a dog"},
               {"score": 0.0_01, "label": "Sound of vaccum cleaner"},
           ]	,						)

       lowercase__ :   int    					=				audio_classifier([audio] * 5	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]  )
       self.assertEqual(
           nested_simplify(lowercase_  )	,						[
               [
                   {"score": 0.9_99, "label": "Sound of a dog"},
                   {"score": 0.0_01, "label": "Sound of vaccum cleaner"},
               ],
           ]
           * 5	,						)
       lowercase__ :   Tuple    					=				audio_classifier(
           [audio] * 5	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]	,						batch_size=5  )
       self.assertEqual(
           nested_simplify(lowercase_  )	,						[
               [
                   {"score": 0.9_99, "label": "Sound of a dog"},
                   {"score": 0.0_01, "label": "Sound of vaccum cleaner"},
               ],
           ]
           * 5	,						)







   @unittest.skip("No models are available in TF"  )
   def       __UpperCamelCase     (      self				:     Union[str, Any]  )					->						Dict:
       pass



