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







 """simple docstring"""






 def      UpperCAmelCase							(					self							)  ->					Any:
        _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      UpperCAmelCase							(					self							)  ->					RealmTokenizer:
        return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname		,				"""realm_tokenizer"""							)							)



 def      UpperCAmelCase							(					self							)  ->					List[Any]:
        shutil.rmtree(self.tmpdirname							)



 def      UpperCAmelCase							(					self							)  ->					List[Any]:
        _A        =  RealmConfig(num_block_records=self.num_block_records							)
        return config



 def      UpperCAmelCase							(					self							)  ->					List[str]:
        _A        =  Dataset.from_dict(
            {
                """id""": ["""0""", """1"""],
                """question""": ["""foo""", """bar"""],
                """answers""": [["""Foo""", """Bar"""], ["""Bar"""]],
            }							)
        return dataset



 def      UpperCAmelCase							(					self							)  ->					str:
        _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      UpperCAmelCase							(					self							)  ->					Dict:
        _A        =  RealmRetriever(
            block_records=self.get_dummy_block_records()		,				tokenizer=self.get_tokenizer()		,				)
        return retriever



 def      UpperCAmelCase							(					self							)  ->					Optional[int]:
        _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      UpperCAmelCase							(					self							)  ->					Dict:
        _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      UpperCAmelCase							(					self							)  ->					Optional[Any]:
        _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"""							)





import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_SCREAMING_SNAKE_CASE										=						logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE										=						{
    'BridgeTower/bridgetower-base': 'https://huggingface.co/BridgeTower/bridgetower-base/blob/main/config.json',
    'BridgeTower/bridgetower-base-itm-mlm': (
        'https://huggingface.co/BridgeTower/bridgetower-base-itm-mlm/blob/main/config.json'
    ),
}



class 				a							(			__lowerCAmelCase       ):







 """simple docstring"""






 lowerCamelCase	:Optional[Any]    			=		'''bridgetower_vision_model'''



 def __init__(					self		,				lowerCAmelCase_=7_68		,				lowerCAmelCase_=12		,				lowerCAmelCase_=3		,				lowerCAmelCase_=16		,				lowerCAmelCase_=2_88		,				lowerCAmelCase_=1		,				lowerCAmelCase_=1E-05		,				lowerCAmelCase_=False		,				lowerCAmelCase_=True		,				lowerCAmelCase_=False		,				**lowerCAmelCase_		,				)  ->					Any:
        super().__init__(**lowerCAmelCase_							)
        _A        =  hidden_size
        _A        =  num_hidden_layers
        _A        =  num_channels
        _A        =  patch_size
        _A        =  image_size
        _A        =  initializer_factor
        _A        =  layer_norm_eps
        _A        =  stop_gradient
        _A        =  share_layernorm
        _A        =  remove_last_layer



 @classmethod
 def      UpperCAmelCase							(					cls		,				lowerCAmelCase_		,				**lowerCAmelCase_							)  ->					"PretrainedConfig":
        _A   ,  _A        =  cls.get_config_dict(lowerCAmelCase_		,				**lowerCAmelCase_							)

        if config_dict.get("""model_type"""							) == "bridgetower":
               _A        =  config_dict["""text_config"""]

        if "model_type" in config_dict and hasattr(cls		,				"""model_type"""							) and config_dict["model_type"] != cls.model_type:
               logger.warning(
                   F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
                   F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''							)

        return cls.from_dict(lowerCAmelCase_		,				**lowerCAmelCase_							)



class 				a							(			__lowerCAmelCase       ):







 """simple docstring"""






 lowerCamelCase	:Union[str, Any]    			=		'''bridgetower_text_model'''



 def __init__(					self		,				lowerCAmelCase_=5_02_65		,				lowerCAmelCase_=7_68		,				lowerCAmelCase_=12		,				lowerCAmelCase_=12		,				lowerCAmelCase_=1		,				lowerCAmelCase_=30_72		,				lowerCAmelCase_="gelu"		,				lowerCAmelCase_=0.1		,				lowerCAmelCase_=0.1		,				lowerCAmelCase_=5_14		,				lowerCAmelCase_=1		,				lowerCAmelCase_=1E-05		,				lowerCAmelCase_=1		,				lowerCAmelCase_=0		,				lowerCAmelCase_=2		,				lowerCAmelCase_="absolute"		,				lowerCAmelCase_=True		,				**lowerCAmelCase_		,				)  ->					Optional[Any]:
        super().__init__(**lowerCAmelCase_							)

        _A        =  vocab_size
        _A        =  hidden_size
        _A        =  num_hidden_layers
        _A        =  num_attention_heads
        _A        =  hidden_act
        _A        =  initializer_factor
        _A        =  intermediate_size
        _A        =  hidden_dropout_prob
        _A        =  attention_probs_dropout_prob
        _A        =  max_position_embeddings
        _A        =  type_vocab_size
        _A        =  layer_norm_eps
        _A        =  position_embedding_type
        _A        =  use_cache
        _A        =  pad_token_id
        _A        =  bos_token_id
        _A        =  eos_token_id



 @classmethod
 def      UpperCAmelCase							(					cls		,				lowerCAmelCase_		,				**lowerCAmelCase_							)  ->					"PretrainedConfig":
        _A   ,  _A        =  cls.get_config_dict(lowerCAmelCase_		,				**lowerCAmelCase_							)

        if config_dict.get("""model_type"""							) == "bridgetower":
               _A        =  config_dict["""text_config"""]

        if "model_type" in config_dict and hasattr(cls		,				"""model_type"""							) and config_dict["model_type"] != cls.model_type:
               logger.warning(
                   F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
                   F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''							)

        return cls.from_dict(lowerCAmelCase_		,				**lowerCAmelCase_							)



class 				a							(			__lowerCAmelCase       ):







 """simple docstring"""






 lowerCamelCase	:Optional[int]    			=		'''bridgetower'''



 def __init__(					self		,				lowerCAmelCase_=True		,				lowerCAmelCase_="gelu"		,				lowerCAmelCase_=7_68		,				lowerCAmelCase_=1		,				lowerCAmelCase_=1E-05		,				lowerCAmelCase_=False		,				lowerCAmelCase_="add"		,				lowerCAmelCase_=12		,				lowerCAmelCase_=6		,				lowerCAmelCase_=False		,				lowerCAmelCase_=False		,				lowerCAmelCase_=None		,				lowerCAmelCase_=None		,				**lowerCAmelCase_		,				)  ->					int:
        # TODO: remove this once the Hub files are updated.
        _A        =  kwargs.pop("""text_config_dict"""		,				lowerCAmelCase_							)
        _A        =  kwargs.pop("""vision_config_dict"""		,				lowerCAmelCase_							)

        super().__init__(**lowerCAmelCase_							)
        _A        =  share_cross_modal_transformer_layers
        _A        =  hidden_act
        _A        =  hidden_size
        _A        =  initializer_factor
        _A        =  layer_norm_eps
        _A        =  share_link_tower_layers
        _A        =  link_tower_type
        _A        =  num_attention_heads
        _A        =  num_hidden_layers
        _A        =  tie_word_embeddings
        _A        =  init_layernorm_from_vision_encoder

        if text_config is None:
               _A        =  {}
               logger.info("""`text_config` is `None`. Initializing the `BridgeTowerTextConfig` with default values."""							)

        if vision_config is None:
               _A        =  {}
               logger.info("""`vision_config` is `None`. Initializing the `BridgeTowerVisionConfig` with default values."""							)

        _A        =  BridgeTowerTextConfig(**lowerCAmelCase_							)
        _A        =  BridgeTowerVisionConfig(**lowerCAmelCase_							)



 @classmethod
 def      UpperCAmelCase							(					cls		,				lowerCAmelCase_		,				lowerCAmelCase_		,				**lowerCAmelCase_							)  ->					Any:
        return cls(text_config=text_config.to_dict()		,				vision_config=vision_config.to_dict()		,				**lowerCAmelCase_							)



 def      UpperCAmelCase							(					self							)  ->					Union[str, Any]:
        _A        =  copy.deepcopy(self.__dict__							)
        _A        =  self.text_config.to_dict()
        _A        =  self.vision_config.to_dict()
        _A        =  self.__class__.model_type
        return output




