


"""simple docstring"""




import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


__magic_name__	:   str          =							logging.get_logger(__name__)

__magic_name__	:   Optional[Any]          =							{
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__magic_name__	:   str          =							{
    'vocab_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/vocab.json'},
    'merges_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/merges.txt'},
}

__magic_name__	:   Optional[int]          =							{
    'openai-gpt': 5_1_2,
}





def 			a_   (  lowercase__   :Any    ):
			__lowerCamelCase        =			text.strip()
			if not text:
						return []
			__lowerCamelCase        =			text.split()
			return tokens




class  __snake_case      (lowerCamelCase     ):




							def __init__(    self:						List[str]    ,       A_:						List[str]=True    ,       A_:						List[Any]=None    ,       A_:						Any=True    ,       A_:						Any=None    ,       A_:						str=True    ,       ):
										if never_split is None:
													__lowerCamelCase        =			[]
										__lowerCamelCase        =			do_lower_case
										__lowerCamelCase        =			set(A_      )
										__lowerCamelCase        =			tokenize_chinese_chars
										__lowerCamelCase        =			strip_accents
										__lowerCamelCase        =			do_split_on_punc




							def      __a      (    self:						Union[str, Any]    ,       A_:						Union[str, Any]    ,       A_:						int=None      ):
										__lowerCamelCase        =			self.never_split.union(set(A_      )      ) if never_split else self.never_split
										__lowerCamelCase        =			self._clean_text(A_      )

										# This was added on November 1st, 2018 for the multilingual and Chinese
										# models. This is also applied to the English models now, but it doesn't
										# matter since the English models were not trained on any Chinese data
										# and generally don't have any Chinese data in them (there are Chinese
										# characters in the vocabulary because Wikipedia does have some Chinese
										# words in the English Wikipedia.).
										if self.tokenize_chinese_chars:
													__lowerCamelCase        =			self._tokenize_chinese_chars(A_      )
										# prevents treating the same character with different unicode codepoints as different characters
										__lowerCamelCase        =			unicodedata.normalize("""NFC"""    ,       A_      )
										__lowerCamelCase        =			whitespace_tokenize(A_      )
										__lowerCamelCase        =			[]
										for token in orig_tokens:
													if token not in never_split:
																if self.do_lower_case:
																			__lowerCamelCase        =			token.lower()
																			if self.strip_accents is not False:
																						__lowerCamelCase        =			self._run_strip_accents(A_      )
																elif self.strip_accents:
																			__lowerCamelCase        =			self._run_strip_accents(A_      )
													split_tokens.extend(self._run_split_on_punc(A_    ,       A_      )      )

										__lowerCamelCase        =			whitespace_tokenize(""" """.join(A_      )      )
										return output_tokens




							def      __a      (    self:						Optional[int]    ,       A_:						Union[str, Any]      ):
										__lowerCamelCase        =			unicodedata.normalize("""NFD"""    ,       A_      )
										__lowerCamelCase        =			[]
										for char in text:
													__lowerCamelCase        =			unicodedata.category(A_      )
													if cat == "Mn":
																continue
													output.append(A_      )
										return "".join(A_      )




							def      __a      (    self:						Dict    ,       A_:						int    ,       A_:						Union[str, Any]=None      ):
										if not self.do_split_on_punc or (never_split is not None and text in never_split):
													return [text]
										__lowerCamelCase        =			list(A_      )
										__lowerCamelCase        =			0
										__lowerCamelCase        =			True
										__lowerCamelCase        =			[]
										while i < len(A_      ):
													__lowerCamelCase        =			chars[i]
													if _is_punctuation(A_      ):
																output.append([char]      )
																__lowerCamelCase        =			True
													else:
																if start_new_word:
																			output.append([]      )
																__lowerCamelCase        =			False
																output[-1].append(A_      )
													i += 1

										return ["".join(A_      ) for x in output]




							def      __a      (    self:						List[str]    ,       A_:						Optional[int]      ):
										__lowerCamelCase        =			[]
										for char in text:
													__lowerCamelCase        =			ord(A_      )
													if self._is_chinese_char(A_      ):
																output.append(""" """      )
																output.append(A_      )
																output.append(""" """      )
													else:
																output.append(A_      )
										return "".join(A_      )




							def      __a      (    self:						List[Any]    ,       A_:						Optional[int]      ):
										# This defines a "chinese character" as anything in the CJK Unicode block:
										#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
										#
										# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
										# despite its name. The modern Korean Hangul alphabet is a different block,
										# as is Japanese Hiragana and Katakana. Those alphabets are used to write
										# space-separated words, so they are not treated specially and handled
										# like the all of the other languages.
										if (
										    (cp >= 0X4E00 and cp <= 0X9FFF)
										    or (cp >= 0X3400 and cp <= 0X4DBF)  #
										    or (cp >= 0X2_0000 and cp <= 0X2_A6DF)  #
										    or (cp >= 0X2_A700 and cp <= 0X2_B73F)  #
										    or (cp >= 0X2_B740 and cp <= 0X2_B81F)  #
										    or (cp >= 0X2_B820 and cp <= 0X2_CEAF)  #
										    or (cp >= 0XF900 and cp <= 0XFAFF)
										    or (cp >= 0X2_F800 and cp <= 0X2_FA1F)  #
										):  #
													return True

										return False




							def      __a      (    self:						Optional[Any]    ,       A_:						Tuple      ):
										__lowerCamelCase        =			[]
										for char in text:
													__lowerCamelCase        =			ord(A_      )
													if cp == 0 or cp == 0XFFFD or _is_control(A_      ):
																continue
													if _is_whitespace(A_      ):
																output.append(""" """      )
													else:
																output.append(A_      )
										return "".join(A_      )





def 			a_   (  lowercase__   :List[str]    ):
			__lowerCamelCase        =			set()
			__lowerCamelCase        =			word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)    )
						__lowerCamelCase        =			char
			return pairs





def 			a_   (  lowercase__   :Optional[Any]    ):
			__lowerCamelCase        =			text.replace("""—""", """-"""    )
			__lowerCamelCase        =			text.replace("""–""", """-"""    )
			__lowerCamelCase        =			text.replace("""―""", """-"""    )
			__lowerCamelCase        =			text.replace("""…""", """..."""    )
			__lowerCamelCase        =			text.replace("""´""", """'"""    )
			__lowerCamelCase        =			re.sub(r"""(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)""", r""" \1 """, lowercase__    )
			__lowerCamelCase        =			re.sub(r"""\s*\n\s*""", """ \n """, lowercase__    )
			__lowerCamelCase        =			re.sub(r"""[^\S\n]+""", """ """, lowercase__    )
			return text.strip()




class  __snake_case      (lowerCamelCase     ):
							__a				   =					VOCAB_FILES_NAMES
							__a				   =					PRETRAINED_VOCAB_FILES_MAP
							__a				   =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__a				   =					['''input_ids''', '''attention_mask''']




							def __init__(    self:						Union[str, Any]    ,       A_:						int    ,       A_:						Dict    ,       A_:						Dict="<unk>"    ,       **A_:						Union[str, Any]      ):
										super().__init__(unk_token=A_    ,       **A_      )

										try:
													import ftfy
													from spacy.lang.en import English

													__lowerCamelCase        =			English()
													__lowerCamelCase        =			_nlp.tokenizer
													__lowerCamelCase        =			ftfy.fix_text
										except ImportError:
													logger.warning("""ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy."""      )
													__lowerCamelCase        =			BasicTokenizer(do_lower_case=A_      )
													__lowerCamelCase        =			None

										with open(A_    ,       encoding="""utf-8"""      ) as vocab_handle:
													__lowerCamelCase        =			json.load(A_      )
										__lowerCamelCase        =			{v: k for k, v in self.encoder.items()}
										with open(A_    ,       encoding="""utf-8"""      ) as merges_handle:
													__lowerCamelCase        =			merges_handle.read().split("""\n"""      )[1:-1]
										__lowerCamelCase        =			[tuple(merge.split()      ) for merge in merges]
										__lowerCamelCase        =			dict(zip(A_    ,       range(len(A_      )      )      )      )
										__lowerCamelCase        =			{}




							@property
							def      __a      (    self:						List[str]      ):
										return True




							@property
							def      __a      (    self:						Optional[Any]      ):
										return len(self.encoder      )




							def      __a      (    self:						Optional[int]      ):
										return dict(self.encoder    ,       **self.added_tokens_encoder      )




							def      __a      (    self:						Union[str, Any]    ,       A_:						Tuple      ):
										__lowerCamelCase        =			tuple(token[:-1]      ) + (token[-1] + """</w>""",)
										if token in self.cache:
													return self.cache[token]
										__lowerCamelCase        =			get_pairs(A_      )

										if not pairs:
													return token + "</w>"

										while True:
													__lowerCamelCase        =			min(A_    ,       key=lambda A_      : self.bpe_ranks.get(A_    ,       float("""inf"""      )      )      )
													if bigram not in self.bpe_ranks:
																break
													__lowerCamelCase    ,__lowerCamelCase        =			bigram
													__lowerCamelCase        =			[]
													__lowerCamelCase        =			0
													while i < len(A_      ):
																try:
																			__lowerCamelCase        =			word.index(A_    ,       A_      )
																except ValueError:
																			new_word.extend(word[i:]      )
																			break
																else:
																			new_word.extend(word[i:j]      )
																			__lowerCamelCase        =			j

																if word[i] == first and i < len(A_      ) - 1 and word[i + 1] == second:
																			new_word.append(first + second      )
																			i += 2
																else:
																			new_word.append(word[i]      )
																			i += 1
													__lowerCamelCase        =			tuple(A_      )
													__lowerCamelCase        =			new_word
													if len(A_      ) == 1:
																break
													else:
																__lowerCamelCase        =			get_pairs(A_      )
										__lowerCamelCase        =			""" """.join(A_      )
										if word == "\n  </w>":
													__lowerCamelCase        =			"""\n</w>"""
										__lowerCamelCase        =			word
										return word




							def      __a      (    self:						Any    ,       A_:						Any      ):
										__lowerCamelCase        =			[]
										if self.fix_text is None:
													# Using BERT's BasicTokenizer
													__lowerCamelCase        =			self.nlp.tokenize(A_      )
													for token in text:
																split_tokens.extend(list(self.bpe(A_      ).split(""" """      )      )      )
										else:
													# Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
													__lowerCamelCase        =			self.nlp(text_standardize(self.fix_text(A_      )      )      )
													for token in text:
																split_tokens.extend(list(self.bpe(token.text.lower()      ).split(""" """      )      )      )
										return split_tokens




							def      __a      (    self:						List[Any]    ,       A_:						List[str]      ):
										return self.encoder.get(A_    ,       self.encoder.get(self.unk_token      )      )




							def      __a      (    self:						Dict    ,       A_:						Union[str, Any]      ):
										return self.decoder.get(A_    ,       self.unk_token      )




							def      __a      (    self:						List[Any]    ,       A_:						Any      ):
										__lowerCamelCase        =			"""""".join(A_      ).replace("""</w>"""    ,       """ """      ).strip()
										return out_string




							def      __a      (    self:						Union[str, Any]    ,       A_:						str    ,       A_:						Optional[str] = None      ):
										if not os.path.isdir(A_      ):
													logger.error(f'Vocabulary path ({save_directory}) should be a directory'      )
													return
										__lowerCamelCase        =			os.path.join(
										    A_    ,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]      )
										__lowerCamelCase        =			os.path.join(
										    A_    ,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]      )

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

										__lowerCamelCase        =			0
										with open(A_    ,       """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 A_      : 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!"""      )
																			__lowerCamelCase        =			token_index
																writer.write(""" """.join(A_      ) + """\n"""      )
																index += 1

										return vocab_file, merge_file








"""simple docstring"""




import argparse
import logging
import sys
from unittest.mock import patch

import run_glue_with_pabee

from transformers.testing_utils import TestCasePlus


logging.basicConfig(level=logging.DEBUG)

__magic_name__	:   Optional[int]          =							logging.getLogger()





def 			a_   (  ):
			__lowerCamelCase        =			argparse.ArgumentParser()
			parser.add_argument("""-f"""    )
			__lowerCamelCase        =			parser.parse_args()
			return args.f




class  __snake_case      (lowerCamelCase     ):




							def      __a      (    self:						Optional[int]      ):
										__lowerCamelCase        =			logging.StreamHandler(sys.stdout      )
										logger.addHandler(A_      )

										__lowerCamelCase        =			self.get_auto_remove_tmp_dir()
										__lowerCamelCase        =			f'\n            run_glue_with_pabee.py\n            --model_type albert\n            --model_name_or_path albert-base-v2\n            --data_dir ./tests/fixtures/tests_samples/MRPC/\n            --output_dir {tmp_dir}\n            --overwrite_output_dir\n            --task_name mrpc\n            --do_train\n            --do_eval\n            --per_gpu_train_batch_size=2\n            --per_gpu_eval_batch_size=1\n            --learning_rate=2e-5\n            --max_steps=50\n            --warmup_steps=2\n            --seed=42\n            --max_seq_length=128\n            '.split()

										with patch.object(A_    ,       """argv"""    ,       A_      ):
													__lowerCamelCase        =			run_glue_with_pabee.main()
													for value in result.values():
																self.assertGreaterEqual(A_    ,       0.75      )




