


'''simple docstring'''




import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_a   :   str												=   logging.get_logger(__name__)


_a   :   List[Any]												=   "▁"

_a   :   str												=   {"vocab_file": "spiece.model"}

_a   :   List[Any]												=   {
    "vocab_file": {
        "google/reformer-crime-and-punishment": (
            "https://huggingface.co/google/reformer-crime-and-punishment/resolve/main/spiece.model"
        )
    }
}

_a   :   Optional[Any]												=   {
    "google/reformer-crime-and-punishment": 524_288,
}




class 			_lowercase (	__lowercase							):
   _SCREAMING_SNAKE_CASE  :  Any          =	VOCAB_FILES_NAMES
   _SCREAMING_SNAKE_CASE  :  Optional[int]          =	PRETRAINED_VOCAB_FILES_MAP
   _SCREAMING_SNAKE_CASE  :  List[str]          =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   _SCREAMING_SNAKE_CASE  :  Optional[int]          =	["input_ids", "attention_mask"]
   def __init__(						self       :      Optional[int]       ,   SCREAMING_SNAKE_CASE_       :      List[Any]       ,   SCREAMING_SNAKE_CASE_       :      Tuple="</s>"       ,   SCREAMING_SNAKE_CASE_       :      Optional[Any]="<unk>"       ,   SCREAMING_SNAKE_CASE_       :      Optional[int]=[]       ,   SCREAMING_SNAKE_CASE_       :      Optional[Dict[str, Any]] = None       ,   **SCREAMING_SNAKE_CASE_       :      Dict       ,   ) ->							None:
        __snake_case           =						{} if sp_model_kwargs is None else sp_model_kwargs

        super().__init__(
            eos_token=SCREAMING_SNAKE_CASE_       ,   unk_token=SCREAMING_SNAKE_CASE_       ,   additional_special_tokens=SCREAMING_SNAKE_CASE_       ,   sp_model_kwargs=self.sp_model_kwargs       ,   **SCREAMING_SNAKE_CASE_       ,   )

        __snake_case           =						vocab_file
        __snake_case           =						spm.SentencePieceProcessor(**self.sp_model_kwargs      )
        self.sp_model.Load(SCREAMING_SNAKE_CASE_      )
   @property
   def a  (						self       :      List[str]      ) ->							Dict:
        return self.sp_model.get_piece_size()
   def a  (						self       :      Union[str, Any]      ) ->							Dict[str, int]:
        __snake_case           =						{self.convert_ids_to_tokens(SCREAMING_SNAKE_CASE_      ): i for i in range(self.vocab_size      )}
        vocab.update(self.added_tokens_encoder      )
        return vocab
   def __getstate__(						self       :      Optional[Any]      ) ->							Dict:
        __snake_case           =						self.__dict__.copy()
        __snake_case           =						None
        return state
   def __setstate__(						self       :      List[str]       ,   SCREAMING_SNAKE_CASE_       :      Dict      ) ->							Optional[int]:
        __snake_case           =						d

        # for backward compatibility
        if not hasattr(self       ,   'sp_model_kwargs'      ):
             __snake_case           =						{}

        __snake_case           =						spm.SentencePieceProcessor(**self.sp_model_kwargs      )
        self.sp_model.Load(self.vocab_file      )
   def a  (						self       :      int       ,   SCREAMING_SNAKE_CASE_       :      str      ) ->							List[str]:
        return self.sp_model.encode(SCREAMING_SNAKE_CASE_       ,   out_type=SCREAMING_SNAKE_CASE_      )
   def a  (						self       :      Optional[Any]       ,   SCREAMING_SNAKE_CASE_       :      Dict      ) ->							Dict:
        return self.sp_model.piece_to_id(SCREAMING_SNAKE_CASE_      )
   def a  (						self       :      Tuple       ,   SCREAMING_SNAKE_CASE_       :      str      ) ->							List[Any]:
        if index < self.sp_model.get_piece_size():
             __snake_case           =						self.sp_model.IdToPiece(SCREAMING_SNAKE_CASE_      )
        return token
   def a  (						self       :      Union[str, Any]       ,   SCREAMING_SNAKE_CASE_       :      Optional[Any]      ) ->							Dict:
        __snake_case           =						[]
        __snake_case           =						''
        for token in tokens:
             # make sure that special tokens are not decoded using sentencepiece model
             if token in self.all_special_tokens:
                  out_string += self.sp_model.decode(SCREAMING_SNAKE_CASE_      ) + token
                  __snake_case           =						[]
             else:
                  current_sub_tokens.append(SCREAMING_SNAKE_CASE_      )
        out_string += self.sp_model.decode(SCREAMING_SNAKE_CASE_      )
        return out_string.strip()





   def a  (						self       :      List[str]       ,   SCREAMING_SNAKE_CASE_       :      str       ,   SCREAMING_SNAKE_CASE_       :      Optional[str] = None      ) ->							Tuple[str]:
        if not os.path.isdir(SCREAMING_SNAKE_CASE_      ):
             logger.error(f'Vocabulary path ({save_directory}) should be a directory'      )
             return
        __snake_case           =						os.path.join(
            SCREAMING_SNAKE_CASE_       ,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']      )

        if os.path.abspath(self.vocab_file      ) != os.path.abspath(SCREAMING_SNAKE_CASE_      ) and os.path.isfile(self.vocab_file      ):
             copyfile(self.vocab_file       ,   SCREAMING_SNAKE_CASE_      )
        elif not os.path.isfile(self.vocab_file      ):
             with open(SCREAMING_SNAKE_CASE_       ,   'wb'      ) as fi:
                  __snake_case           =						self.sp_model.serialized_model_proto()
                  fi.write(SCREAMING_SNAKE_CASE_      )

        return (out_vocab_file,)









import unittest

from transformers import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
			from PIL import Image
else:


			class 						lowercase    :
									'''simple docstring'''


									@staticmethod
									def       lowerCamelCase_				(  *snake_case    :				Dict       ,	**snake_case    :				List[Any]	):



														'''simple docstring'''


														pass


@is_pipeline_test
@require_torch
@require_vision
class 						lowercase    (							unittest.TestCase):
						'''simple docstring'''


						UpperCAmelCase			:		Any		=		MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING







						def       lowerCamelCase_				(  self    :				List[Any]       ,	snake_case    :				List[str]       ,	snake_case    :				int       ,	snake_case    :				Optional[Any]	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				pipeline('visual-question-answering'       ,	model='hf-internal-testing/tiny-vilt-random-vqa'	)
											SCREAMING_SNAKE_CASE							:     Any						=				[
											    {
											        'image': Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'	),
											        'question': 'How many cats are there?',
											    },
											    {
											        'image': './tests/fixtures/tests_samples/COCO/000000039769.png',
											        'question': 'How many cats are there?',
											    },
											]
											return vqa_pipeline, examples







						def       lowerCamelCase_				(  self    :				int       ,	snake_case    :				int       ,	snake_case    :				Dict	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				vqa_pipeline(snake_case       ,	top_k=1	)
											self.assertEqual(
											    snake_case       ,	[
											        [{'score': ANY(snake_case	), 'answer': ANY(snake_case	)}],
											        [{'score': ANY(snake_case	), 'answer': ANY(snake_case	)}],
											    ]       ,	)







						@require_torch
						def       lowerCamelCase_				(  self    :				Dict	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Tuple						=				pipeline('visual-question-answering'       ,	model='hf-internal-testing/tiny-vilt-random-vqa'	)
											SCREAMING_SNAKE_CASE							:     List[Any]						=				'./tests/fixtures/tests_samples/COCO/000000039769.png'
											SCREAMING_SNAKE_CASE							:     Any						=				'How many cats are there?'

											SCREAMING_SNAKE_CASE							:     Optional[Any]						=				vqa_pipeline(image=snake_case       ,	question='How many cats are there?'       ,	top_k=2	)
											self.assertEqual(
											    snake_case       ,	[{'score': ANY(snake_case	), 'answer': ANY(snake_case	)}, {'score': ANY(snake_case	), 'answer': ANY(snake_case	)}]	)

											SCREAMING_SNAKE_CASE							:     int						=				vqa_pipeline({'image': image, 'question': question}       ,	top_k=2	)
											self.assertEqual(
											    snake_case       ,	[{'score': ANY(snake_case	), 'answer': ANY(snake_case	)}, {'score': ANY(snake_case	), 'answer': ANY(snake_case	)}]	)







						@slow
						@require_torch
						def       lowerCamelCase_				(  self    :				Dict	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     List[Any]						=				pipeline('visual-question-answering'       ,	model='dandelin/vilt-b32-finetuned-vqa'	)
											SCREAMING_SNAKE_CASE							:     int						=				'./tests/fixtures/tests_samples/COCO/000000039769.png'
											SCREAMING_SNAKE_CASE							:     Dict						=				'How many cats are there?'

											SCREAMING_SNAKE_CASE							:     int						=				vqa_pipeline(image=snake_case       ,	question=snake_case       ,	top_k=2	)
											self.assertEqual(
											    nested_simplify(snake_case       ,	decimals=4	)       ,	[{'score': 0.8799, 'answer': '2'}, {'score': 0.296, 'answer': '1'}]	)

											SCREAMING_SNAKE_CASE							:     List[str]						=				vqa_pipeline({'image': image, 'question': question}       ,	top_k=2	)
											self.assertEqual(
											    nested_simplify(snake_case       ,	decimals=4	)       ,	[{'score': 0.8799, 'answer': '2'}, {'score': 0.296, 'answer': '1'}]	)

											SCREAMING_SNAKE_CASE							:     Any						=				vqa_pipeline(
											    [{'image': image, 'question': question}, {'image': image, 'question': question}]       ,	top_k=2	)
											self.assertEqual(
											    nested_simplify(snake_case       ,	decimals=4	)       ,	[[{'score': 0.8799, 'answer': '2'}, {'score': 0.296, 'answer': '1'}]] * 2       ,	)







						@require_tf
						@unittest.skip('Visual question answering not implemented in TF'	)
						def       lowerCamelCase_				(  self    :				str	):



											'''simple docstring'''


											pass