

import os
import unittest

from transformers import MobileBertTokenizer, MobileBertTokenizerFast
from transformers.models.bert.tokenization_bert import (
    VOCAB_FILES_NAMES,
    BasicTokenizer,
    WordpieceTokenizer,
    _is_control,
    _is_punctuation,
    _is_whitespace,
)
from transformers.testing_utils import require_tokenizers, slow

from ...test_tokenization_common import TokenizerTesterMixin, filter_non_english



@require_tokenizers
class        _a		(__magic_name__						,  unittest.TestCase							):


					'''simple docstring'''






					UpperCAmelCase__:  Optional[Any]    				=			MobileBertTokenizer
					UpperCAmelCase__:  Optional[Any]    				=			MobileBertTokenizerFast
					UpperCAmelCase__:  Tuple    				=			True
					UpperCAmelCase__:  Union[str, Any]    				=			True
					UpperCAmelCase__:  List[Any]    				=			filter_non_english
					UpperCAmelCase__:  List[Any]    				=			'''google/mobilebert-uncased'''






					def     __A     (					self     ):
						super().setUp()

						A__				:       Tuple	  =						[
						    """[UNK]""",
						    """[CLS]""",
						    """[SEP]""",
						    """[PAD]""",
						    """[MASK]""",
						    """want""",
						    """##want""",
						    """##ed""",
						    """wa""",
						    """un""",
						    """runn""",
						    """##ing""",
						    """,""",
						    """low""",
						    """lowest""",
						]
						A__				:       int	  =						os.path.join(self.tmpdirname       ,			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__				:       Tuple	  =						[
						    (tokenizer_def[0], self.pre_trained_model_path, tokenizer_def[2])  # else the 'google/' prefix is stripped
						    for tokenizer_def in self.tokenizers_list
						]






					def     __A     (					self       ,			A__     ):
						A__				:       Dict	  =						"""UNwant\u00E9d,running"""
						A__				:       str	  =						"""unwanted, running"""
						return input_text, output_text






					def     __A     (					self     ):
						A__				:       List[str]	  =						self.tokenizer_class(self.vocab_file     )

						A__				:       Any	  =						tokenizer.tokenize("""UNwant\u00E9d,running"""     )
						self.assertListEqual(A__       ,			["""un""", """##want""", """##ed""", """,""", """runn""", """##ing"""]     )
						self.assertListEqual(tokenizer.convert_tokens_to_ids(A__     )       ,			[9, 6, 7, 12, 10, 11]     )






					def     __A     (					self     ):
						if not self.test_rust_tokenizer:
							return

						A__				:       Optional[Any]	  =						self.get_tokenizer()
						A__				:       Dict	  =						self.get_rust_tokenizer()

						A__				:       Tuple	  =						"""UNwant\u00E9d,running"""

						A__				:       int	  =						tokenizer.tokenize(A__     )
						A__				:       int	  =						rust_tokenizer.tokenize(A__     )
						self.assertListEqual(A__       ,			A__     )

						A__				:       Optional[Any]	  =						tokenizer.encode(A__       ,			add_special_tokens=A__     )
						A__				:       Tuple	  =						rust_tokenizer.encode(A__       ,			add_special_tokens=A__     )
						self.assertListEqual(A__       ,			A__     )

						A__				:       List[str]	  =						self.get_rust_tokenizer()
						A__				:       Tuple	  =						tokenizer.encode(A__     )
						A__				:       str	  =						rust_tokenizer.encode(A__     )
						self.assertListEqual(A__       ,			A__     )

						# With lower casing
						A__				:       int	  =						self.get_tokenizer(do_lower_case=A__     )
						A__				:       Dict	  =						self.get_rust_tokenizer(do_lower_case=A__     )

						A__				:       Tuple	  =						"""UNwant\u00E9d,running"""

						A__				:       Union[str, Any]	  =						tokenizer.tokenize(A__     )
						A__				:       str	  =						rust_tokenizer.tokenize(A__     )
						self.assertListEqual(A__       ,			A__     )

						A__				:       Union[str, Any]	  =						tokenizer.encode(A__       ,			add_special_tokens=A__     )
						A__				:       Any	  =						rust_tokenizer.encode(A__       ,			add_special_tokens=A__     )
						self.assertListEqual(A__       ,			A__     )

						A__				:       List[Any]	  =						self.get_rust_tokenizer()
						A__				:       Optional[Any]	  =						tokenizer.encode(A__     )
						A__				:       Union[str, Any]	  =						rust_tokenizer.encode(A__     )
						self.assertListEqual(A__       ,			A__     )






					def     __A     (					self     ):
						A__				:       Tuple	  =						BasicTokenizer()

						self.assertListEqual(tokenizer.tokenize("""ah\u535A\u63A8zz"""     )       ,			["""ah""", """\u535A""", """\u63A8""", """zz"""]     )






					def     __A     (					self     ):
						A__				:       Dict	  =						BasicTokenizer(do_lower_case=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHeLLo!how  \n Are yoU?  """     )       ,			["""hello""", """!""", """how""", """are""", """you""", """?"""]     )
						self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""     )       ,			["""hello"""]     )






					def     __A     (					self     ):
						A__				:       Union[str, Any]	  =						BasicTokenizer(do_lower_case=A__       ,			strip_accents=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """     )       ,			["""hällo""", """!""", """how""", """are""", """you""", """?"""]     )
						self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""     )       ,			["""h\u00E9llo"""]     )






					def     __A     (					self     ):
						A__				:       List[Any]	  =						BasicTokenizer(do_lower_case=A__       ,			strip_accents=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """     )       ,			["""hallo""", """!""", """how""", """are""", """you""", """?"""]     )
						self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""     )       ,			["""hello"""]     )






					def     __A     (					self     ):
						A__				:       Tuple	  =						BasicTokenizer(do_lower_case=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """     )       ,			["""hallo""", """!""", """how""", """are""", """you""", """?"""]     )
						self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""     )       ,			["""hello"""]     )






					def     __A     (					self     ):
						A__				:       Tuple	  =						BasicTokenizer(do_lower_case=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHeLLo!how  \n Are yoU?  """     )       ,			["""HeLLo""", """!""", """how""", """Are""", """yoU""", """?"""]     )






					def     __A     (					self     ):
						A__				:       Any	  =						BasicTokenizer(do_lower_case=A__       ,			strip_accents=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """     )       ,			["""HäLLo""", """!""", """how""", """Are""", """yoU""", """?"""]     )






					def     __A     (					self     ):
						A__				:       Optional[int]	  =						BasicTokenizer(do_lower_case=A__       ,			strip_accents=A__     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """     )       ,			["""HaLLo""", """!""", """how""", """Are""", """yoU""", """?"""]     )






					def     __A     (					self     ):
						A__				:       Optional[int]	  =						BasicTokenizer(do_lower_case=A__       ,			never_split=["""[UNK]"""]     )

						self.assertListEqual(
						    tokenizer.tokenize(""" \tHeLLo!how  \n Are yoU? [UNK]"""     )       ,			["""HeLLo""", """!""", """how""", """Are""", """yoU""", """?""", """[UNK]"""]     )






					def     __A     (					self     ):
						A__				:       int	  =						["""[UNK]""", """[CLS]""", """[SEP]""", """want""", """##want""", """##ed""", """wa""", """un""", """runn""", """##ing"""]

						A__				:       Optional[Any]	  =						{}
						for i, token in enumerate(A__     ):
							A__				:       Tuple	  =						i
						A__				:       Tuple	  =						WordpieceTokenizer(vocab=A__       ,			unk_token="""[UNK]"""     )

						self.assertListEqual(tokenizer.tokenize(""""""     )       ,			[]     )

						self.assertListEqual(tokenizer.tokenize("""unwanted running"""     )       ,			["""un""", """##want""", """##ed""", """runn""", """##ing"""]     )

						self.assertListEqual(tokenizer.tokenize("""unwantedX running"""     )       ,			["""[UNK]""", """runn""", """##ing"""]     )






					def     __A     (					self     ):
						self.assertTrue(_is_whitespace(""" """     )     )
						self.assertTrue(_is_whitespace("""\t"""     )     )
						self.assertTrue(_is_whitespace("""\r"""     )     )
						self.assertTrue(_is_whitespace("""\n"""     )     )
						self.assertTrue(_is_whitespace("""\u00A0"""     )     )

						self.assertFalse(_is_whitespace("""A"""     )     )
						self.assertFalse(_is_whitespace("""-"""     )     )






					def     __A     (					self     ):
						self.assertTrue(_is_control("""\u0005"""     )     )

						self.assertFalse(_is_control("""A"""     )     )
						self.assertFalse(_is_control(""" """     )     )
						self.assertFalse(_is_control("""\t"""     )     )
						self.assertFalse(_is_control("""\r"""     )     )






					def     __A     (					self     ):
						self.assertTrue(_is_punctuation("""-"""     )     )
						self.assertTrue(_is_punctuation("""$"""     )     )
						self.assertTrue(_is_punctuation("""`"""     )     )
						self.assertTrue(_is_punctuation("""."""     )     )

						self.assertFalse(_is_punctuation("""A"""     )     )
						self.assertFalse(_is_punctuation(""" """     )     )






					def     __A     (					self     ):
						A__				:       Dict	  =						self.get_tokenizer()
						A__				:       Optional[int]	  =						self.get_rust_tokenizer()

						# Example taken from the issue https://github.com/huggingface/tokenizers/issues/340
						self.assertListEqual([tokenizer.tokenize(A__     ) for t in ["""Test""", """\xad""", """test"""]]       ,			[["""[UNK]"""], [], ["""[UNK]"""]]     )

						self.assertListEqual(
						    [rust_tokenizer.tokenize(A__     ) for t in ["""Test""", """\xad""", """test"""]]       ,			[["""[UNK]"""], [], ["""[UNK]"""]]     )






					@slow
					def     __A     (					self     ):
						A__				:       Optional[Any]	  =						self.tokenizer_class.from_pretrained("""google/mobilebert-uncased"""     )

						A__				:       Tuple	  =						tokenizer.encode("""sequence builders"""       ,			add_special_tokens=A__     )
						A__				:       List[Any]	  =						tokenizer.encode("""multi-sequence build"""       ,			add_special_tokens=A__     )

						A__				:       Any	  =						tokenizer.build_inputs_with_special_tokens(A__     )
						A__				:       List[str]	  =						tokenizer.build_inputs_with_special_tokens(A__       ,			A__     )

						assert encoded_sentence == [101] + text + [102]
						assert encoded_pair == [101] + text + [102] + text_a + [102]






					def     __A     (					self     ):
						for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
							with self.subTest(F"""{tokenizer.__class__.__name__} ({pretrained_name})"""     ):
								A__				:       Dict	  =						self.rust_tokenizer_class.from_pretrained(A__       ,			**A__     )

								A__				:       Optional[Any]	  =						F"""A, naïve {tokenizer_r.mask_token} AllenNLP sentence."""
								A__				:       int	  =						tokenizer_r.encode_plus(
								    A__       ,			return_attention_mask=A__       ,			return_token_type_ids=A__       ,			return_offsets_mapping=A__       ,			add_special_tokens=A__       ,			)

								A__				:       Union[str, Any]	  =						tokenizer_r.do_lower_case if hasattr(A__       ,			"""do_lower_case"""     ) else False
								A__				:       Any	  =						(
								    [
								        ((0, 0), tokenizer_r.cls_token),
								        ((0, 1), """A"""),
								        ((1, 2), ""","""),
								        ((3, 5), """na"""),
								        ((5, 6), """##ï"""),
								        ((6, 8), """##ve"""),
								        ((9, 15), tokenizer_r.mask_token),
								        ((16, 21), """Allen"""),
								        ((21, 23), """##NL"""),
								        ((23, 24), """##P"""),
								        ((25, 33), """sentence"""),
								        ((33, 34), """."""),
								        ((0, 0), tokenizer_r.sep_token),
								    ]
								    if not do_lower_case
								    else [
								        ((0, 0), tokenizer_r.cls_token),
								        ((0, 1), """a"""),
								        ((1, 2), ""","""),
								        ((3, 8), """naive"""),
								        ((9, 15), tokenizer_r.mask_token),
								        ((16, 21), """allen"""),
								        ((21, 23), """##nl"""),
								        ((23, 24), """##p"""),
								        ((25, 33), """sentence"""),
								        ((33, 34), """."""),
								        ((0, 0), tokenizer_r.sep_token),
								    ]
								)

								self.assertEqual(
								    [e[1] for e in expected_results]       ,			tokenizer_r.convert_ids_to_tokens(tokens["""input_ids"""]     )     )
								self.assertEqual([e[0] for e in expected_results]       ,			tokens["""offset_mapping"""]     )






					def     __A     (					self     ):
						A__				:       Tuple	  =						["""的""", """人""", """有"""]
						A__				:       Optional[Any]	  =						"""""".join(A__     )
						for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
							with self.subTest(F"""{tokenizer.__class__.__name__} ({pretrained_name})"""     ):
								A__				:       Union[str, Any]	  =						True
								A__				:       Optional[Any]	  =						self.tokenizer_class.from_pretrained(A__       ,			**A__     )
								A__				:       str	  =						self.rust_tokenizer_class.from_pretrained(A__       ,			**A__     )

								A__				:       int	  =						tokenizer_p.encode(A__       ,			add_special_tokens=A__     )
								A__				:       List[Any]	  =						tokenizer_r.encode(A__       ,			add_special_tokens=A__     )

								A__				:       str	  =						tokenizer_r.convert_ids_to_tokens(A__     )
								A__				:       Dict	  =						tokenizer_p.convert_ids_to_tokens(A__     )

								# it is expected that each Chinese character is not preceded by "##"
								self.assertListEqual(A__       ,			A__     )
								self.assertListEqual(A__       ,			A__     )

								A__				:       Optional[int]	  =						False
								A__				:       Union[str, Any]	  =						self.rust_tokenizer_class.from_pretrained(A__       ,			**A__     )
								A__				:       Union[str, Any]	  =						self.tokenizer_class.from_pretrained(A__       ,			**A__     )

								A__				:       Dict	  =						tokenizer_r.encode(A__       ,			add_special_tokens=A__     )
								A__				:       Dict	  =						tokenizer_p.encode(A__       ,			add_special_tokens=A__     )

								A__				:       Optional[int]	  =						tokenizer_r.convert_ids_to_tokens(A__     )
								A__				:       Any	  =						tokenizer_p.convert_ids_to_tokens(A__     )

								# it is expected that only the first Chinese character is not preceded by "##".
								A__				:       Any	  =						[
								    F"""##{token}""" if idx != 0 else token for idx, token in enumerate(A__     )
								]
								self.assertListEqual(A__       ,			A__     )
								self.assertListEqual(A__       ,			A__     )






import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


A_			:					Any    			=     logging.get_logger(__name__)



class        _a		:


					'''simple docstring'''






					UpperCAmelCase__:  str    				=			None



@experimental
def 		UpperCamelCase   (lowercase_:					int       , lowercase_:					Any       , lowercase_:					str       , lowercase_:					str       , lowercase_:					List[str]       , lowercase_:					List[Any]       , lowercase_:					Union[str, Any]   )	-> Union[str, Any]:
	if ParallelBackendConfig.backend_name is None:
		return _map_with_multiprocessing_pool(
		    lowercase_       , lowercase_       , lowercase_       , lowercase_       , lowercase_       , lowercase_       , lowercase_   )

	return _map_with_joblib(lowercase_       , lowercase_       , lowercase_       , lowercase_       , lowercase_       , lowercase_       , lowercase_   )

def 		UpperCamelCase   (lowercase_:					Tuple       , lowercase_:					List[Any]       , lowercase_:					Tuple       , lowercase_:					Tuple       , lowercase_:					List[str]       , lowercase_:					Dict       , lowercase_:					Tuple   )	-> Tuple:
	A__				:       Union[str, Any]	  =						num_proc if num_proc <= len(lowercase_   ) else len(lowercase_   )
	A__				:       str	  =						[]  # We organize the splits ourselve (contiguous splits)
	for index in range(lowercase_   ):
		A__				:       Tuple	  =						len(lowercase_   ) // num_proc
		A__				:       Tuple	  =						len(lowercase_   ) % num_proc
		A__				:       Optional[Any]	  =						div * index + min(lowercase_       , lowercase_   )
		A__				:       Union[str, Any]	  =						start + div + (1 if index < mod else 0)
		split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)   )

	if len(lowercase_   ) != sum(len(i[1]   ) for i in split_kwds   ):
		raise ValueError(
		    f"""Error dividing inputs iterable among processes. """
		    f"""Total number of objects {len(lowercase_   )}, """
		    f"""length: {sum(len(i[1]   ) for i in split_kwds   )}"""   )

	logger.info(
	    f"""Spawning {num_proc} processes for {len(lowercase_   )} objects in slices of {[len(i[1]   ) for i in split_kwds]}"""   )
	A__  ,  A__				:       Optional[int]	  =						None, None
	if not disable_tqdm:
		A__  ,  A__				:       List[str]	  =						(RLock(),), tqdm.set_lock
	with Pool(lowercase_       , initargs=lowercase_       , initializer=lowercase_   ) as pool:
		A__				:       Optional[Any]	  =						pool.map(lowercase_       , lowercase_   )
	logger.info(f"""Finished {num_proc} processes"""   )
	A__				:       Tuple	  =						[obj for proc_res in mapped for obj in proc_res]
	logger.info(f"""Unpacked {len(lowercase_   )} objects"""   )

	return mapped

def 		UpperCamelCase   (lowercase_:					Union[str, Any]       , lowercase_:					Dict       , lowercase_:					str       , lowercase_:					Union[str, Any]       , lowercase_:					List[Any]       , lowercase_:					Dict       , lowercase_:					Optional[Any]   )	-> Union[str, Any]:
	# progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
	# and it requires monkey-patching joblib internal classes which is subject to change
	import joblib

	with joblib.parallel_backend(ParallelBackendConfig.backend_name       , n_jobs=lowercase_   ):
		return joblib.Parallel()(
		    joblib.delayed(lowercase_   )((function, obj, types, None, True, None)   ) for obj in iterable   )


@experimental
@contextlib.contextmanager
def 		UpperCamelCase   (lowercase_:					str   )	-> str:
	A__				:       int	  =						backend_name

	if backend_name == "spark":
		from joblibspark import register_spark

		register_spark()

		# TODO: call create_cache_and_write_probe if "download" in steps
		# TODO: raise NotImplementedError when Dataset.map etc is called

	try:
		yield
	finally:
		A__				:       Optional[Any]	  =						None



