





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        SCREAMING_SNAKE_CASE__					(		A__ ):



 def 							UpperCamelCase     (				self					):
        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(__snake_case,exist_ok=__snake_case					)
        A__		   =		os.path.join(__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__		   =		os.path.join(self.tmpdirname,'''realm_block_records'''					)
        os.makedirs(__snake_case,exist_ok=__snake_case					)



 def 							UpperCamelCase     (				self					):
        return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname,'''realm_tokenizer'''					)					)



 def 							UpperCamelCase     (				self					):
        shutil.rmtree(self.tmpdirname					)



 def 							UpperCamelCase     (				self					):
        A__		   =		RealmConfig(num_block_records=self.num_block_records					)
        return config



 def 							UpperCamelCase     (				self					):
        A__		   =		Dataset.from_dict(
            {
                '''id''': ['''0''', '''1'''],
                '''question''': ['''foo''', '''bar'''],
                '''answers''': [['''Foo''', '''Bar'''], ['''Bar''']],
            }					)
        return dataset



 def 							UpperCamelCase     (				self					):
        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=__snake_case,)
        return block_records



 def 							UpperCamelCase     (				self					):
        A__		   =		RealmRetriever(
            block_records=self.get_dummy_block_records(),tokenizer=self.get_tokenizer(),)
        return retriever



 def 							UpperCamelCase     (				self					):
        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=__snake_case,return_token_type_ids=__snake_case,return_attention_mask=__snake_case,).input_ids
        A__		   =		config.reader_seq_len

        A__   ,      A__   ,      A__   ,      A__		   =		retriever(
            __snake_case,__snake_case,answer_ids=__snake_case,max_length=__snake_case,return_tensors='''np'''					)

        self.assertEqual(len(__snake_case					),2					)
        self.assertEqual(len(__snake_case					),2					)
        self.assertEqual(len(__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 							UpperCamelCase     (				self					):
        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=__snake_case,return_token_type_ids=__snake_case,return_attention_mask=__snake_case,).input_ids
        A__		   =		config.reader_seq_len

        A__   ,      A__   ,      A__   ,      A__		   =		retriever(
            __snake_case,__snake_case,answer_ids=__snake_case,max_length=__snake_case,return_tensors='''np'''					)

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




 def 							UpperCamelCase     (				self					):
        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'''					)












'''simple docstring'''




import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType
class 			_lowerCAmelCase		(			A__	):






     """simple docstring"""






     snake_case_						      =   ["image_processor", "tokenizer"]
     snake_case_						      =   "LayoutLMv2ImageProcessor"
     snake_case_						      =   ("LayoutXLMTokenizer", "LayoutXLMTokenizerFast")



     def __init__(   self							:			List[Any]   ,       __snake_case							:			Union[str, Any]=None   ,       __snake_case							:			Optional[int]=None   ,       **__snake_case							:			Union[str, Any]						)->	str:
        if "feature_extractor" in kwargs:
           warnings.warn(
               """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
               """ instead."""   ,       __snake_case   ,       )
           snake_case 						=   kwargs.pop("""feature_extractor"""						)

        snake_case 						=   image_processor if image_processor is not None else feature_extractor
        if image_processor is None:
           raise ValueError("""You need to specify an `image_processor`."""						)
        if tokenizer is None:
           raise ValueError("""You need to specify a `tokenizer`."""						)

        super().__init__(__snake_case   ,       __snake_case						)



     def __call__(   self							:			Optional[Any]   ,       __snake_case							:			Dict   ,       __snake_case							:			Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None   ,       __snake_case							:			Optional[Union[PreTokenizedInput, List[PreTokenizedInput]]] = None   ,       __snake_case							:			Union[List[List[int]], List[List[List[int]]]] = None   ,       __snake_case							:			Optional[Union[List[int], List[List[int]]]] = None   ,       __snake_case							:			bool = True   ,       __snake_case							:			Union[bool, str, PaddingStrategy] = False   ,       __snake_case							:			Union[bool, str, TruncationStrategy] = None   ,       __snake_case							:			Optional[int] = None   ,       __snake_case							:			int = 0   ,       __snake_case							:			Optional[int] = None   ,       __snake_case							:			Optional[bool] = None   ,       __snake_case							:			Optional[bool] = None   ,       __snake_case							:			bool = False   ,       __snake_case							:			bool = False   ,       __snake_case							:			bool = False   ,       __snake_case							:			bool = False   ,       __snake_case							:			bool = True   ,       __snake_case							:			Optional[Union[str, TensorType]] = None   ,       **__snake_case							:			List[str]   ,       )->	BatchEncoding:
        # verify input
        if self.image_processor.apply_ocr and (boxes is not None):
           raise ValueError(
               """You cannot provide bounding boxes """
               """if you initialized the image processor with apply_ocr set to True."""						)

        if self.image_processor.apply_ocr and (word_labels is not None):
           raise ValueError(
               """You cannot provide word labels if you initialized the image processor with apply_ocr set to True."""						)

        if return_overflowing_tokens is True and return_offsets_mapping is False:
           raise ValueError("""You cannot return overflowing tokens without returning the offsets mapping."""						)

        # first, apply the image processor
        snake_case 						=   self.image_processor(images=__snake_case   ,       return_tensors=__snake_case						)

        # second, apply the tokenizer
        if text is not None and self.image_processor.apply_ocr and text_pair is None:
           if isinstance(__snake_case   ,       __snake_case						):
              snake_case 						=   [text]  # add batch dimension (as the image processor always adds a batch dimension)
           snake_case 						=   features["""words"""]

        snake_case 						=   self.tokenizer(
            text=text if text is not None else features["""words"""]   ,       text_pair=text_pair if text_pair is not None else None   ,       boxes=boxes if boxes is not None else features["""boxes"""]   ,       word_labels=__snake_case   ,       add_special_tokens=__snake_case   ,       padding=__snake_case   ,       truncation=__snake_case   ,       max_length=__snake_case   ,       stride=__snake_case   ,       pad_to_multiple_of=__snake_case   ,       return_token_type_ids=__snake_case   ,       return_attention_mask=__snake_case   ,       return_overflowing_tokens=__snake_case   ,       return_special_tokens_mask=__snake_case   ,       return_offsets_mapping=__snake_case   ,       return_length=__snake_case   ,       verbose=__snake_case   ,       return_tensors=__snake_case   ,       **__snake_case   ,       )

        # add pixel values
        snake_case 						=   features.pop("""pixel_values"""						)
        if return_overflowing_tokens is True:
           snake_case 						=   self.get_overflowing_images(__snake_case   ,       encoded_inputs["""overflow_to_sample_mapping"""]						)
        snake_case 						=   images

        return encoded_inputs



     def      lowerCAmelCase			(   self							:			Any   ,       __snake_case							:			List[Any]   ,       __snake_case							:			str						)->	List[str]:
        # in case there's an overflow, ensure each `input_ids` sample is mapped to its corresponding image
        snake_case 						=   []
        for sample_idx in overflow_to_sample_mapping:
           images_with_overflow.append(images[sample_idx]						)

        if len(__snake_case						) != len(__snake_case						):
           raise ValueError(
               """Expected length of images to be the same as the length of `overflow_to_sample_mapping`, but got"""
               f''' {len(__snake_case						)} and {len(__snake_case						)}'''						)

        return images_with_overflow



     def      lowerCAmelCase			(   self							:			int   ,       *__snake_case							:			Optional[int]   ,       **__snake_case							:			Tuple						)->	str:
        return self.tokenizer.batch_decode(*__snake_case   ,       **__snake_case						)



     def      lowerCAmelCase			(   self							:			Dict   ,       *__snake_case							:			Tuple   ,       **__snake_case							:			Optional[int]						)->	List[Any]:
        return self.tokenizer.decode(*__snake_case   ,       **__snake_case						)



     @property
     def      lowerCAmelCase			(   self							:			str						)->	int:
        return ["input_ids", "bbox", "attention_mask", "image"]



     @property
     def      lowerCAmelCase			(   self							:			Union[str, Any]						)->	List[str]:
        warnings.warn(
            """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""   ,       __snake_case   ,       )
        return self.image_processor_class



     @property
     def      lowerCAmelCase			(   self							:			Optional[Any]						)->	Tuple:
        warnings.warn(
            """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""   ,       __snake_case   ,       )
        return self.image_processor

