





'''simple docstring'''







import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

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


snake_case_   :				Optional[int]		=     logging.get_logger(__name__)


snake_case_   :				int		=     {'vocab_file': 'vocab.json', 'merges_file': 'merges.txt'}

# See all BART models at https://huggingface.co/models?filter=bart
snake_case_   :				List[str]		=     {
    'vocab_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/vocab.json',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/vocab.json',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json',
    },
    'merges_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/merges.txt',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/merges.txt',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt',
    },
}

snake_case_   :				Optional[Any]		=     {
    'facebook/bart-base': 10_24,
    'facebook/bart-large': 10_24,
    'facebook/bart-large-mnli': 10_24,
    'facebook/bart-large-cnn': 10_24,
    'facebook/bart-large-xsum': 10_24,
    'yjernite/bart_eli5': 10_24,
}





@lru_cache()
def 				lowerCamelCase_					(     )       ->		Union[str, Any]:
						UpperCAmelCase_	:					Optional[Any]     =     (
						    list(range(ord('''!'''      ),	ord('''~'''      ) + 1      )      ) + list(range(ord('''¡'''      ),	ord('''¬'''      ) + 1      )      ) + list(range(ord('''®'''      ),	ord('''ÿ'''      ) + 1      )      )
						)
						UpperCAmelCase_	:					Optional[Any]     =     bs[:]
						UpperCAmelCase_	:					Union[str, Any]     =     0
						for b in range(2**8      ):
												if b not in bs:
																		bs.append(a__      )
																		cs.append(2**8 + n      )
																		n += 1
						UpperCAmelCase_	:					Optional[Any]     =     [chr(a__      ) for n in cs]
						return dict(zip(a__,	a__      )      )





def 				lowerCamelCase_					(     SCREAMING_SNAKE_CASE__					:	Union[str, Any]      )       ->		str:
						UpperCAmelCase_	:					Dict     =     set()
						UpperCAmelCase_	:					Any     =     word[0]
						for char in word[1:]:
												pairs.add((prev_char, char)      )
												UpperCAmelCase_	:					str     =     char
						return pairs


class     __a					(lowerCamelCase   ):
							__a    :      int  	=				VOCAB_FILES_NAMES
							__a    :      Any  	=				PRETRAINED_VOCAB_FILES_MAP
							__a    :      Tuple  	=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__a    :      Dict  	=				["input_ids", "attention_mask"]

							def __init__(							self							:   Any				,   __magic_name__							:   Optional[Any]				,   __magic_name__							:   Tuple				,   __magic_name__							:   Union[str, Any]="replace"				,   __magic_name__							:   Optional[int]="<s>"				,   __magic_name__							:   str="</s>"				,   __magic_name__							:   Any="</s>"				,   __magic_name__							:   Optional[Any]="<s>"				,   __magic_name__							:   List[Any]="<unk>"				,   __magic_name__							:   Union[str, Any]="<pad>"				,   __magic_name__							:   int="<mask>"				,   __magic_name__							:   int=False				,   **__magic_name__							:   Optional[int]				,   )    ->				Tuple:







													"""simple docstring"""
													UpperCAmelCase_	:					Union[str, Any]     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else bos_token
													UpperCAmelCase_	:					List[str]     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else eos_token
													UpperCAmelCase_	:					Any     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else sep_token
													UpperCAmelCase_	:					Tuple     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else cls_token
													UpperCAmelCase_	:					Tuple     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else unk_token
													UpperCAmelCase_	:					str     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else pad_token

													# Mask token behave like a normal word, i.e. include the space before it
													UpperCAmelCase_	:					Any     =     AddedToken(_snake_case				,   lstrip=_snake_case				,   rstrip=_snake_case      ) if isinstance(_snake_case				,   _snake_case      ) else mask_token

													super().__init__(
													    errors=_snake_case				,   bos_token=_snake_case				,   eos_token=_snake_case				,   unk_token=_snake_case				,   sep_token=_snake_case				,   cls_token=_snake_case				,   pad_token=_snake_case				,   mask_token=_snake_case				,   add_prefix_space=_snake_case				,   **_snake_case				,   )

													with open(_snake_case				,   encoding='''utf-8'''      ) as vocab_handle:
																			UpperCAmelCase_	:					Any     =     json.load(_snake_case      )
													UpperCAmelCase_	:					Optional[int]     =     {v: k for k, v in self.encoder.items()}
													UpperCAmelCase_	:					Any     =     errors  # how to handle errors in decoding
													UpperCAmelCase_	:					List[str]     =     bytes_to_unicode()
													UpperCAmelCase_	:					Union[str, Any]     =     {v: k for k, v in self.byte_encoder.items()}
													with open(_snake_case				,   encoding='''utf-8'''      ) as merges_handle:
																			UpperCAmelCase_	:					Optional[Any]     =     merges_handle.read().split('''\n'''      )[1:-1]
													UpperCAmelCase_	:					Optional[Any]     =     [tuple(merge.split()      ) for merge in bpe_merges]
													UpperCAmelCase_	:					int     =     dict(zip(_snake_case				,   range(len(_snake_case      )      )      )      )
													UpperCAmelCase_	:					Optional[int]     =     {}
													UpperCAmelCase_	:					List[Any]     =     add_prefix_space

													# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
													UpperCAmelCase_	:					Optional[int]     =     re.compile(R'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''      )

							@property
							def 			UpperCAmelCase__		(							self							:   Optional[Any]      )    ->				str:







													"""simple docstring"""
													return len(self.encoder      )

							def 			UpperCAmelCase__		(							self							:   int      )    ->				Tuple:







													"""simple docstring"""
													return dict(self.encoder				,   **self.added_tokens_encoder      )

							def 			UpperCAmelCase__		(							self							:   int				,   __magic_name__							:   Optional[int]      )    ->				int:







													"""simple docstring"""




													if token in self.cache:
																			return self.cache[token]
													UpperCAmelCase_	:					Any     =     tuple(_snake_case      )
													UpperCAmelCase_	:					Dict     =     get_pairs(_snake_case      )

													if not pairs:
																			return token

													while True:
																			UpperCAmelCase_	:					Tuple     =     min(_snake_case				,   key=lambda __magic_name__      : self.bpe_ranks.get(_snake_case				,   float('''inf'''      )      )      )
																			if bigram not in self.bpe_ranks:
																									break
																			UpperCAmelCase_  ,				UpperCAmelCase_	:					str     =     bigram
																			UpperCAmelCase_	:					Optional[Any]     =     []
																			UpperCAmelCase_	:					Union[str, Any]     =     0
																			while i < len(_snake_case      ):
																									try:
																															UpperCAmelCase_	:					List[Any]     =     word.index(_snake_case				,   _snake_case      )
																									except ValueError:
																															new_word.extend(word[i:]      )
																															break
																									else:
																															new_word.extend(word[i:j]      )
																															UpperCAmelCase_	:					Any     =     j

																									if word[i] == first and i < len(_snake_case      ) - 1 and word[i + 1] == second:
																															new_word.append(first + second      )
																															i += 2
																									else:
																															new_word.append(word[i]      )
																															i += 1
																			UpperCAmelCase_	:					Tuple     =     tuple(_snake_case      )
																			UpperCAmelCase_	:					Optional[int]     =     new_word
																			if len(_snake_case      ) == 1:
																									break
																			else:
																									UpperCAmelCase_	:					Union[str, Any]     =     get_pairs(_snake_case      )
													UpperCAmelCase_	:					Union[str, Any]     =     ''' '''.join(_snake_case      )
													UpperCAmelCase_	:					List[Any]     =     word
													return word

							def 			UpperCAmelCase__		(							self							:   Union[str, Any]				,   __magic_name__							:   Tuple      )    ->				Any:







													"""simple docstring"""
													UpperCAmelCase_	:					Union[str, Any]     =     []
													for token in re.findall(self.pat				,   _snake_case      ):
																			UpperCAmelCase_	:					List[str]     =     ''''''.join(
																			    self.byte_encoder[b] for b in token.encode('''utf-8'''      )      )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
																			bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_snake_case      ).split(''' '''      )      )
													return bpe_tokens

							def 			UpperCAmelCase__		(							self							:   Dict				,   __magic_name__							:   Optional[int]      )    ->				int:







													"""simple docstring"""
													return self.encoder.get(_snake_case				,   self.encoder.get(self.unk_token      )      )

							def 			UpperCAmelCase__		(							self							:   List[str]				,   __magic_name__							:   Union[str, Any]      )    ->				Optional[int]:







													"""simple docstring"""
													return self.decoder.get(_snake_case      )

							def 			UpperCAmelCase__		(							self							:   List[Any]				,   __magic_name__							:   Union[str, Any]      )    ->				Optional[Any]:







													"""simple docstring"""
													UpperCAmelCase_	:					Optional[Any]     =     ''''''.join(_snake_case      )
													UpperCAmelCase_	:					Union[str, Any]     =     bytearray([self.byte_decoder[c] for c in text]      ).decode('''utf-8'''				,   errors=self.errors      )
													return text

							def 			UpperCAmelCase__		(							self							:   int				,   __magic_name__							:   Union[str, Any]				,   __magic_name__							:   Optional[int] = None      )    ->				Tuple[str]:







													"""simple docstring"""




													if not os.path.isdir(_snake_case      ):
																			logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""      )
																			return
													UpperCAmelCase_	:					Optional[Any]     =     os.path.join(
													    _snake_case				,   (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']      )
													UpperCAmelCase_	:					Tuple     =     os.path.join(
													    _snake_case				,   (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']      )

													with open(_snake_case				,   '''w'''				,   encoding='''utf-8'''      ) as f:
																			f.write(json.dumps(self.encoder				,   indent=2				,   sort_keys=_snake_case				,   ensure_ascii=_snake_case      ) + '''\n'''      )

													UpperCAmelCase_	:					Tuple     =     0
													with open(_snake_case				,   '''w'''				,   encoding='''utf-8'''      ) as writer:
																			writer.write('''#version: 0.2\n'''      )
																			for bpe_tokens, token_index in sorted(self.bpe_ranks.items()				,   key=lambda __magic_name__      : kv[1]      ):
																									if index != token_index:
																															logger.warning(
																															    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																															    ''' Please check that the tokenizer is not corrupted!'''      )
																															UpperCAmelCase_	:					List[str]     =     token_index
																									writer.write(''' '''.join(_snake_case      ) + '''\n'''      )
																									index += 1

													return vocab_file, merge_file

							def 			UpperCAmelCase__		(							self							:   Optional[int]				,   __magic_name__							:   Optional[int]				,   __magic_name__							:   Optional[int] = None      )    ->				List[int]:







													"""simple docstring"""




													if token_ids_a is None:
																			return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
													UpperCAmelCase_	:					Union[str, Any]     =     [self.cls_token_id]
													UpperCAmelCase_	:					Optional[int]     =     [self.sep_token_id]
													return cls + token_ids_a + sep + sep + token_ids_a + sep

							def 			UpperCAmelCase__		(							self							:   List[Any]				,   __magic_name__							:   Tuple				,   __magic_name__							:   List[str] = None				,   __magic_name__							:   Union[str, Any] = False      )    ->				List[int]:







													"""simple docstring"""




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

													if token_ids_a is None:
																			return [1] + ([0] * len(_snake_case      )) + [1]
													return [1] + ([0] * len(_snake_case      )) + [1, 1] + ([0] * len(_snake_case      )) + [1]

							def 			UpperCAmelCase__		(							self							:   Dict				,   __magic_name__							:   List[str]				,   __magic_name__							:   Tuple = None      )    ->				List[int]:







													"""simple docstring"""
													UpperCAmelCase_	:					Optional[Any]     =     [self.sep_token_id]
													UpperCAmelCase_	:					Union[str, Any]     =     [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 + sep + token_ids_a + sep      ) * [0]



							def 			UpperCAmelCase__		(							self							:   Optional[int]				,   __magic_name__							:   Dict				,   __magic_name__							:   Optional[int]=False				,   **__magic_name__							:   Any      )    ->				Optional[int]:







													"""simple docstring"""
													UpperCAmelCase_	:					int     =     kwargs.pop('''add_prefix_space'''				,   self.add_prefix_space      )
													if (is_split_into_words or add_prefix_space) and (len(_snake_case      ) > 0 and not text[0].isspace()):
																			UpperCAmelCase_	:					Tuple     =     ''' ''' + text
													return (text, kwargs)







import tempfile
import unittest

import numpy as np
from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import BertConfig, is_flax_available
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_flax


if is_flax_available():
					import os

					from flax.core.frozen_dict import unfreeze
					from flax.traverse_util import flatten_dict

					from transformers import FlaxBertModel

					A     :	int    = '0.12'  # assumed parallelism: 8







@require_flax
@is_staging_test
class 			__A( unittest.TestCase			):




				@classmethod
				def 				SCREAMING_SNAKE_CASE_		(     cls				)    ->      Union[str, Any]:


								'''simple docstring'''




								__a										=    TOKEN
								HfFolder.save_token(_snake_case				)




				@classmethod
				def 				SCREAMING_SNAKE_CASE_		(     cls				)    ->      Union[str, Any]:


								'''simple docstring'''




								try:
												delete_repo(token=cls._token		,     repo_id='''test-model-flax'''				)
								except HTTPError:
												pass

								try:
												delete_repo(token=cls._token		,     repo_id='''valid_org/test-model-flax-org'''				)
								except HTTPError:
												pass




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    BertConfig(
								    vocab_size=99		,     hidden_size=32		,     num_hidden_layers=5		,     num_attention_heads=4		,     intermediate_size=37				)
								__a										=    FlaxBertModel(_snake_case				)
								model.push_to_hub('''test-model-flax'''		,     use_auth_token=self._token				)

								__a										=    FlaxBertModel.from_pretrained(F"""{USER}/test-model-flax"""				)

								__a										=    flatten_dict(unfreeze(model.params				)				)
								__a										=    flatten_dict(unfreeze(new_model.params				)				)

								for key in base_params.keys():
												__a										=    (base_params[key] - new_params[key]).sum().item()
												self.assertLessEqual(_snake_case		,     1E-3		,     msg=F"""{key} not identical"""				)

								# Reset repo
								delete_repo(token=self._token		,     repo_id='''test-model-flax'''				)

								# Push to hub via save_pretrained
								with tempfile.TemporaryDirectory() as tmp_dir:
												model.save_pretrained(_snake_case		,     repo_id='''test-model-flax'''		,     push_to_hub=_snake_case		,     use_auth_token=self._token				)

								__a										=    FlaxBertModel.from_pretrained(F"""{USER}/test-model-flax"""				)

								__a										=    flatten_dict(unfreeze(model.params				)				)
								__a										=    flatten_dict(unfreeze(new_model.params				)				)

								for key in base_params.keys():
												__a										=    (base_params[key] - new_params[key]).sum().item()
												self.assertLessEqual(_snake_case		,     1E-3		,     msg=F"""{key} not identical"""				)




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


								'''simple docstring'''




								__a										=    BertConfig(
								    vocab_size=99		,     hidden_size=32		,     num_hidden_layers=5		,     num_attention_heads=4		,     intermediate_size=37				)
								__a										=    FlaxBertModel(_snake_case				)
								model.push_to_hub('''valid_org/test-model-flax-org'''		,     use_auth_token=self._token				)

								__a										=    FlaxBertModel.from_pretrained('''valid_org/test-model-flax-org'''				)

								__a										=    flatten_dict(unfreeze(model.params				)				)
								__a										=    flatten_dict(unfreeze(new_model.params				)				)

								for key in base_params.keys():
												__a										=    (base_params[key] - new_params[key]).sum().item()
												self.assertLessEqual(_snake_case		,     1E-3		,     msg=F"""{key} not identical"""				)

								# Reset repo
								delete_repo(token=self._token		,     repo_id='''valid_org/test-model-flax-org'''				)

								# Push to hub via save_pretrained
								with tempfile.TemporaryDirectory() as tmp_dir:
												model.save_pretrained(
												    _snake_case		,     repo_id='''valid_org/test-model-flax-org'''		,     push_to_hub=_snake_case		,     use_auth_token=self._token				)

								__a										=    FlaxBertModel.from_pretrained('''valid_org/test-model-flax-org'''				)

								__a										=    flatten_dict(unfreeze(model.params				)				)
								__a										=    flatten_dict(unfreeze(new_model.params				)				)

								for key in base_params.keys():
												__a										=    (base_params[key] - new_params[key]).sum().item()
												self.assertLessEqual(_snake_case		,     1E-3		,     msg=F"""{key} not identical"""				)



def     __lowerCAmelCase			(			a__						,    a__  )       ->  str:
				__a										=    True
				__a										=    flatten_dict(modela.params  )
				__a										=    flatten_dict(modela.params  )
				for key in flat_params_a.keys():
								if np.sum(np.abs(flat_params_a[key] - flat_params_a[key]  )  ) > 1e-4:
												__a										=    False

				return models_are_equal







@require_flax
class 			__A( unittest.TestCase			):




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


								'''simple docstring'''




								__a										=    BertConfig.from_pretrained('''hf-internal-testing/tiny-bert-flax-only'''				)
								__a										=    FlaxBertModel(_snake_case				)

								__a										=    '''bert'''
								with tempfile.TemporaryDirectory() as tmp_dir:
												model.save_pretrained(os.path.join(_snake_case		,     _snake_case				)				)

												with self.assertRaises(_snake_case				):
																__a										=    FlaxBertModel.from_pretrained(_snake_case				)

												__a										=    FlaxBertModel.from_pretrained(_snake_case		,     subfolder=_snake_case				)

								self.assertTrue(check_models_equal(_snake_case		,     _snake_case				)				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    BertConfig.from_pretrained('''hf-internal-testing/tiny-bert-flax-only'''				)
								__a										=    FlaxBertModel(_snake_case				)

								__a										=    '''bert'''
								with tempfile.TemporaryDirectory() as tmp_dir:
												model.save_pretrained(os.path.join(_snake_case		,     _snake_case				)		,     max_shard_size='''10KB'''				)

												with self.assertRaises(_snake_case				):
																__a										=    FlaxBertModel.from_pretrained(_snake_case				)

												__a										=    FlaxBertModel.from_pretrained(_snake_case		,     subfolder=_snake_case				)

								self.assertTrue(check_models_equal(_snake_case		,     _snake_case				)				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      int:


								'''simple docstring'''




								__a										=    '''bert'''
								__a										=    '''hf-internal-testing/tiny-random-bert-subfolder'''

								with self.assertRaises(_snake_case				):
												__a										=    FlaxBertModel.from_pretrained(_snake_case				)

								__a										=    FlaxBertModel.from_pretrained(_snake_case		,     subfolder=_snake_case				)

								self.assertIsNotNone(_snake_case				)




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


								'''simple docstring'''




								__a										=    '''bert'''
								__a										=    '''hf-internal-testing/tiny-random-bert-sharded-subfolder'''
								with self.assertRaises(_snake_case				):
												__a										=    FlaxBertModel.from_pretrained(_snake_case				)

								__a										=    FlaxBertModel.from_pretrained(_snake_case		,     subfolder=_snake_case				)

								self.assertIsNotNone(_snake_case				)