






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

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


a_				=     logging.get_logger(__name__)
a_				=     {'''vocab_file''': '''spiece.model'''}

a_				=     {
    '''vocab_file''': {
        '''albert-base-v1''': '''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''',
        '''albert-large-v1''': '''https://huggingface.co/albert-large-v1/resolve/main/spiece.model''',
        '''albert-xlarge-v1''': '''https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model''',
        '''albert-xxlarge-v1''': '''https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model''',
        '''albert-base-v2''': '''https://huggingface.co/albert-base-v2/resolve/main/spiece.model''',
        '''albert-large-v2''': '''https://huggingface.co/albert-large-v2/resolve/main/spiece.model''',
        '''albert-xlarge-v2''': '''https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model''',
        '''albert-xxlarge-v2''': '''https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model''',
    }
}

a_				=     {
    '''albert-base-v1''': 512,
    '''albert-large-v1''': 512,
    '''albert-xlarge-v1''': 512,
    '''albert-xxlarge-v1''': 512,
    '''albert-base-v2''': 512,
    '''albert-large-v2''': 512,
    '''albert-xlarge-v2''': 512,
    '''albert-xxlarge-v2''': 512,
}

a_				=     '''▁'''


class   lowercase__  (      A__						):
	a_       =VOCAB_FILES_NAMES
	a_       =PRETRAINED_VOCAB_FILES_MAP
	a_       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES

	def __init__(				self		,     __UpperCAmelCase		,     __UpperCAmelCase=True		,     __UpperCAmelCase=True		,     __UpperCAmelCase=False		,     __UpperCAmelCase="[CLS]"		,     __UpperCAmelCase="[SEP]"		,     __UpperCAmelCase="<unk>"		,     __UpperCAmelCase="[SEP]"		,     __UpperCAmelCase="<pad>"		,     __UpperCAmelCase="[CLS]"		,     __UpperCAmelCase="[MASK]"		,     __UpperCAmelCase = None		,     **__UpperCAmelCase		,     )->       Optional[Any]:







							'''simple docstring'''





							lowerCAmelCase__			   =    (
							    AddedToken(_lowerCamelCase		,     lstrip=_lowerCamelCase		,     rstrip=_lowerCamelCase		,     normalized=_lowerCamelCase							)
							    if isinstance(_lowerCamelCase		,     _lowerCamelCase							)
							    else mask_token
							)

							lowerCAmelCase__			   =    {} if sp_model_kwargs is None else sp_model_kwargs

							super().__init__(
							    do_lower_case=_lowerCamelCase		,     remove_space=_lowerCamelCase		,     keep_accents=_lowerCamelCase		,     bos_token=_lowerCamelCase		,     eos_token=_lowerCamelCase		,     unk_token=_lowerCamelCase		,     sep_token=_lowerCamelCase		,     pad_token=_lowerCamelCase		,     cls_token=_lowerCamelCase		,     mask_token=_lowerCamelCase		,     sp_model_kwargs=self.sp_model_kwargs		,     **_lowerCamelCase		,     )

							lowerCAmelCase__			   =    do_lower_case
							lowerCAmelCase__			   =    remove_space
							lowerCAmelCase__			   =    keep_accents
							lowerCAmelCase__			   =    vocab_file

							lowerCAmelCase__			   =    spm.SentencePieceProcessor(**self.sp_model_kwargs							)
							self.sp_model.Load(_lowerCamelCase							)

	@property
	def 		UpperCAmelCase	(				self							)->       Tuple:







							'''simple docstring'''





							return len(self.sp_model							)

	def 		UpperCAmelCase	(				self							)->       str:







							'''simple docstring'''





							lowerCAmelCase__			   =    {self.convert_ids_to_tokens(_lowerCamelCase							): i for i in range(self.vocab_size							)}
							vocab.update(self.added_tokens_encoder							)
							return vocab

	def __getstate__(				self							)->       Optional[int]:







							'''simple docstring'''





							lowerCAmelCase__			   =    self.__dict__.copy()
							lowerCAmelCase__			   =    None
							return state

	def __setstate__(				self		,     __UpperCAmelCase							)->       Any:







							'''simple docstring'''





							lowerCAmelCase__			   =    d

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

							lowerCAmelCase__			   =    spm.SentencePieceProcessor(**self.sp_model_kwargs							)
							self.sp_model.Load(self.vocab_file							)

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       int:







							'''simple docstring'''





							if self.remove_space:
													lowerCAmelCase__			   =    ' '.join(inputs.strip().split()							)
							else:
													lowerCAmelCase__			   =    inputs
							lowerCAmelCase__			   =    outputs.replace("``"		,     "\""							).replace("\'\'"		,     "\""							)

							if not self.keep_accents:
													lowerCAmelCase__			   =    unicodedata.normalize("NFKD"		,     _lowerCamelCase							)
													lowerCAmelCase__			   =    ''.join([c for c in outputs if not unicodedata.combining(_lowerCamelCase							)]							)
							if self.do_lower_case:
													lowerCAmelCase__			   =    outputs.lower()

							return outputs

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       Optional[Any]:







							'''simple docstring'''





							lowerCAmelCase__			   =    self.preprocess_text(_lowerCamelCase							)
							lowerCAmelCase__			   =    self.sp_model.encode(_lowerCamelCase		,     out_type=_lowerCamelCase							)
							lowerCAmelCase__			   =    []
							for piece in pieces:
													if len(_lowerCamelCase							) > 1 and piece[-1] == str(","							) and piece[-2].isdigit():
																			lowerCAmelCase__			   =    self.sp_model.EncodeAsPieces(piece[:-1].replace(_lowerCamelCase		,     ""							)							)
																			if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																									if len(cur_pieces[0]							) == 1:
																															lowerCAmelCase__			   =    cur_pieces[1:]
																									else:
																															lowerCAmelCase__			   =    cur_pieces[0][1:]
																			cur_pieces.append(piece[-1]							)
																			new_pieces.extend(_lowerCamelCase							)
													else:
																			new_pieces.append(_lowerCamelCase							)

							return new_pieces

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       Optional[Any]:







							'''simple docstring'''





							return self.sp_model.PieceToId(_lowerCamelCase							)

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       Any:







							'''simple docstring'''





							return self.sp_model.IdToPiece(_lowerCamelCase							)

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase							)->       Union[str, Any]:







							'''simple docstring'''





							lowerCAmelCase__			   =    []
							lowerCAmelCase__			   =    ''
							lowerCAmelCase__			   =    False
							for token in tokens:
													# make sure that special tokens are not decoded using sentencepiece model
													if token in self.all_special_tokens:
																			if not prev_is_special:
																									out_string += " "
																			out_string += self.sp_model.decode(_lowerCamelCase							) + token
																			lowerCAmelCase__			   =    True
																			lowerCAmelCase__			   =    []
													else:
																			current_sub_tokens.append(_lowerCamelCase							)
																			lowerCAmelCase__			   =    False
							out_string += self.sp_model.decode(_lowerCamelCase							)
							return out_string.strip()

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase		,     __UpperCAmelCase = None							)->       Dict:







							'''simple docstring'''





							lowerCAmelCase__			   =    [self.sep_token_id]
							lowerCAmelCase__			   =    [self.cls_token_id]
							if token_ids_a is None:
													return cls + token_ids_a + sep
							return cls + token_ids_a + sep + token_ids_a + sep

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase		,     __UpperCAmelCase = None		,     __UpperCAmelCase = False							)->       Optional[int]:







							'''simple docstring'''





							if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=_lowerCamelCase		,     token_ids_a=_lowerCamelCase		,     already_has_special_tokens=_lowerCamelCase							)

							if token_ids_a is not None:
													return [1] + ([0] * len(_lowerCamelCase							)) + [1] + ([0] * len(_lowerCamelCase							)) + [1]
							return [1] + ([0] * len(_lowerCamelCase							)) + [1]

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase		,     __UpperCAmelCase = None							)->       int:







							'''simple docstring'''





							lowerCAmelCase__			   =    [self.sep_token_id]
							lowerCAmelCase__			   =    [self.cls_token_id]

							if token_ids_a is None:
													return len(cls + token_ids_a + sep							) * [0]
							return len(cls + token_ids_a + sep							) * [0] + len(token_ids_a + sep							) * [1]


	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase		,     __UpperCAmelCase = None							)->       Tuple:







							'''simple docstring'''





							if not os.path.isdir(_lowerCamelCase							):
													logger.error(F"Vocabulary path ({save_directory}) should be a directory"							)
													return
							lowerCAmelCase__			   =    os.path.join(
							    _lowerCamelCase		,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]							)

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

							return (out_vocab_file,)








"""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__       ):



     def  lowerCamelCase_						(							self :List[str]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[Any]   	=tempfile.mkdtemp()
      UpperCamelCase_     :					str   	=5

      # Realm tok
      UpperCamelCase_     :					List[Any]   	=[
          '[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',
      ]
      UpperCamelCase_     :					List[str]   	=os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )
      UpperCamelCase_     :					int   	=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]   )   )

      UpperCamelCase_     :					str   	=os.path.join(self.tmpdirname      ,  'realm_block_records'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )



     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



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



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      shutil.rmtree(self.tmpdirname   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



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



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



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



     def  lowerCamelCase_						(							self :Optional[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Union[str, 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=_lowerCamelCase      ,  )
      return block_records



     def  lowerCamelCase_						(							self :Any   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[Any]   	=RealmRetriever(
          block_records=self.get_dummy_block_records()      ,  tokenizer=self.get_tokenizer()      ,  )
      return retriever



     def  lowerCamelCase_						(							self :int   ):







      '''simple docstring'''



      UpperCamelCase_     :					Any   	=self.get_config()
      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Tuple   	=retriever.tokenizer

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

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Any   	=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  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



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

      UpperCamelCase_     :					Optional[int]   	=np.array([0, 3, 5]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[int]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					int   	=tokenizer(
          ['the fourth', 'longer longer']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Union[str, Any]   	=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  lowerCamelCase_						(							self :List[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      retriever.save_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )

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

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