



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





							'''simple docstring'''

							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Union[str, Any]      						= tempfile.mkdtemp()
											a__	:							Optional[Any]      						= 5

											# Realm tok
											a__	:							List[str]      						= [
											    '[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__	:							Optional[Any]      						= os.path.join(self.tmpdirname ,					"realm_tokenizer"				)
											os.makedirs(SCREAMING_SNAKE_CASE_ ,					exist_ok=SCREAMING_SNAKE_CASE_				)
											a__	:							Any      						= os.path.join(SCREAMING_SNAKE_CASE_ ,					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__	:							str      						= os.path.join(self.tmpdirname ,					"realm_block_records"				)
											os.makedirs(SCREAMING_SNAKE_CASE_ ,					exist_ok=SCREAMING_SNAKE_CASE_				)



							def 		_A	(							self				):







											"""simple docstring"""



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



							def 		_A	(							self				):







											"""simple docstring"""



											shutil.rmtree(self.tmpdirname				)



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							List[str]      						= RealmConfig(num_block_records=self.num_block_records				)
											return config



							def 		_A	(							self				):







											"""simple docstring"""



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



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Optional[int]      						= 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=SCREAMING_SNAKE_CASE_ ,					)
											return block_records



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Any      						= RealmRetriever(
											    block_records=self.get_dummy_block_records() ,					tokenizer=self.get_tokenizer() ,					)
											return retriever



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Union[str, Any]      						= self.get_config()
											a__	:							int      						= self.get_dummy_retriever()
											a__	:							Any      						= retriever.tokenizer

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

											a__	:							Tuple      						= retriever(
											    SCREAMING_SNAKE_CASE_ ,					SCREAMING_SNAKE_CASE_ ,					answer_ids=SCREAMING_SNAKE_CASE_ ,					max_length=SCREAMING_SNAKE_CASE_ ,					return_tensors="np"				)

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







											"""simple docstring"""



											a__	:							Optional[Any]      						= self.get_config()
											a__	:							Any      						= self.get_dummy_retriever()
											a__	:							Optional[int]      						= retriever.tokenizer

											a__	:							Any      						= np.array([0, 3, 5] ,					dtype="long"				)
											a__	:							Optional[Any]      						= tokenizer(["Test question"]				).input_ids
											a__	:							List[str]      						= tokenizer(
											    ["the fourth", "longer longer"] ,					add_special_tokens=SCREAMING_SNAKE_CASE_ ,					return_token_type_ids=SCREAMING_SNAKE_CASE_ ,					return_attention_mask=SCREAMING_SNAKE_CASE_ ,					).input_ids
											a__	:							Union[str, Any]      						= config.reader_seq_len

											a__	:							str      						= retriever(
											    SCREAMING_SNAKE_CASE_ ,					SCREAMING_SNAKE_CASE_ ,					answer_ids=SCREAMING_SNAKE_CASE_ ,					max_length=SCREAMING_SNAKE_CASE_ ,					return_tensors="np"				)

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




							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							int      						= self.get_dummy_retriever()
											retriever.save_pretrained(os.path.join(self.tmpdirname ,					"realm_block_records"				)				)

											# Test local path
											a__	:							List[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__	:							Union[str, Any]      						= os.path.join(
															    os.path.join(self.tmpdirname ,					"realm_block_records"				) ,					_REALM_BLOCK_RECORDS_FILENAME				)
															a__	:							str      						= RealmRetriever.from_pretrained("google/realm-cc-news-pretrained-openqa"				)

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






'''simple docstring'''



# coding=utf-8
# Copyright 2023 The HuggingFace Inc. team.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# this script dumps information about the environment

import os
import platform
import sys


A__						:   List[str]				=    """3"""

print("""Python version:""", sys.version)

print("""OS platform:""", platform.platform())
print("""OS architecture:""", platform.machine())

try:
      import torch

      print("""Torch version:""", torch.__version__)
      print("""Cuda available:""", torch.cuda.is_available())
      print("""Cuda version:""", torch.version.cuda)
      print("""CuDNN version:""", torch.backends.cudnn.version())
      print("""Number of GPUs available:""", torch.cuda.device_count())
except ImportError:
      print("""Torch version:""", None)

try:
      import transformers

      print("""transformers version:""", transformers.__version__)
except ImportError:
      print("""transformers version:""", None)






