
import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, ByTaTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
		A		: Any								=     '''pt'''
elif is_tf_available():
		A		: int								=     '''tf'''
else:
		A		: Dict								=     '''jax'''



class    a_ (			_a					,			unittest.TestCase    ):
			a	:			int					   =							ByTaTokenizer
			a	:			str					   =							False






			def    UpperCamelCase_				(      self  ):
									super().setUp()
									_lowercase					       =					ByTaTokenizer()
									tokenizer.save_pretrained(self.tmpdirname  )






			@cached_property
			def    UpperCamelCase_				(      self  ):
									return ByTaTokenizer.from_pretrained("""google/byt5-small"""  )






			def    UpperCamelCase_				(      self				,			**__UpperCamelCase  ):
									return self.tokenizer_class.from_pretrained(self.tmpdirname				,			**__UpperCamelCase  )






			def    UpperCamelCase_				(      self				,			__UpperCamelCase				,			__UpperCamelCase=False				,			__UpperCamelCase=20				,			__UpperCamelCase=5  ):
									# XXX The default common tokenizer tests assume that every ID is decodable on its own.
									# This assumption is invalid for ByT5 because single bytes might not be
									# valid utf-8 (byte 128 for instance).
									# Here we're overriding the smallest possible method to provide
									# a clean sequence without making the same assumption.

									_lowercase					       =					[]
									for i in range(len(__UpperCamelCase  )  ):
															try:
																					_lowercase					       =					tokenizer.decode([i]				,			clean_up_tokenization_spaces=__UpperCamelCase  )
															except UnicodeDecodeError:
																					pass
															toks.append((i, tok)  )

									_lowercase					       =					list(filter(lambda __UpperCamelCase  : re.match(r"""^[ a-zA-Z]+$"""				,			t[1]  )				,			__UpperCamelCase  )  )
									_lowercase					       =					list(filter(lambda __UpperCamelCase  : [t[0]] == tokenizer.encode(t[1]				,			add_special_tokens=__UpperCamelCase  )				,			__UpperCamelCase  )  )
									if max_length is not None and len(__UpperCamelCase  ) > max_length:
															_lowercase					       =					toks[:max_length]
									if min_length is not None and len(__UpperCamelCase  ) < min_length and len(__UpperCamelCase  ) > 0:
															while len(__UpperCamelCase  ) < min_length:
																					_lowercase					       =					toks + toks
        # toks_str = [t[1] for t in toks]
									_lowercase					       =					[t[0] for t in toks]

									# Ensure consistency
									_lowercase					       =					tokenizer.decode(__UpperCamelCase				,			clean_up_tokenization_spaces=__UpperCamelCase  )
									if " " not in output_txt and len(__UpperCamelCase  ) > 1:
															_lowercase					       =					(
															    tokenizer.decode([toks_ids[0]]				,			clean_up_tokenization_spaces=__UpperCamelCase  )
															    + """ """
															    + tokenizer.decode(toks_ids[1:]				,			clean_up_tokenization_spaces=__UpperCamelCase  )
															)
									if with_prefix_space:
															_lowercase					       =					""" """ + output_txt
									_lowercase					       =					tokenizer.encode(__UpperCamelCase				,			add_special_tokens=__UpperCamelCase  )
									return output_txt, output_ids






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.ta_base_tokenizer
									_lowercase					       =					tokenizer(["""hi</s>""", """I went to the gym</s>""", """</s>"""]  )
									_lowercase					       =					tokenizer(["""hi""", """I went to the gym""", """"""]  )
									self.assertListEqual(batch_with_eos_added["""input_ids"""]				,			batch_without_eos_added["""input_ids"""]  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.ta_base_tokenizer
									_lowercase					       =					"""Unicode €."""
									_lowercase					       =					tokenizer(__UpperCamelCase  )
									_lowercase					       =					[88, 113, 108, 102, 114, 103, 104, 35, 229, 133, 175, 49, 1]
									self.assertEqual(encoded["""input_ids"""]				,			__UpperCamelCase  )

									# decoding
									_lowercase					       =					tokenizer.decode(__UpperCamelCase  )
									self.assertEqual(__UpperCamelCase				,			"""Unicode €.</s>"""  )

									_lowercase					       =					tokenizer("""e è é ê ë"""  )
									_lowercase					       =					[104, 35, 198, 171, 35, 198, 172, 35, 198, 173, 35, 198, 174, 1]
									self.assertEqual(encoded["""input_ids"""]				,			__UpperCamelCase  )
									# decoding
									_lowercase					       =					tokenizer.decode(__UpperCamelCase  )
									self.assertEqual(__UpperCamelCase				,			"""e è é ê ë</s>"""  )

									# encode/decode, but with `encode` instead of `__call__`
									self.assertEqual(tokenizer.decode(tokenizer.encode("""e è é ê ë"""  )  )				,			"""e è é ê ë</s>"""  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.ta_base_tokenizer
									_lowercase					       =					["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
									# fmt: off
									_lowercase					       =					[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 1, 0]
									# fmt: on
									_lowercase					       =					tokenizer(__UpperCamelCase				,			padding=__UpperCamelCase				,			return_tensors=__UpperCamelCase  )
									self.assertIsInstance(__UpperCamelCase				,			__UpperCamelCase  )

									if FRAMEWORK != "jax":
															_lowercase					       =					list(batch.input_ids.numpy()[0]  )
									else:
															_lowercase					       =					list(batch.input_ids.tolist()[0]  )

									self.assertListEqual(__UpperCamelCase				,			__UpperCamelCase  )

									self.assertEqual((2, 37)				,			batch.input_ids.shape  )
									self.assertEqual((2, 37)				,			batch.attention_mask.shape  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.ta_base_tokenizer
									_lowercase					       =					["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
									_lowercase					       =					tokenizer(__UpperCamelCase				,			padding=__UpperCamelCase				,			return_tensors=__UpperCamelCase  )
									# check if input_ids are returned and no decoder_input_ids
									self.assertIn("""input_ids"""				,			__UpperCamelCase  )
									self.assertIn("""attention_mask"""				,			__UpperCamelCase  )
									self.assertNotIn("""decoder_input_ids"""				,			__UpperCamelCase  )
									self.assertNotIn("""decoder_attention_mask"""				,			__UpperCamelCase  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.ta_base_tokenizer
									_lowercase					       =					[
									    """Summary of the text.""",
									    """Another summary.""",
									]
									_lowercase					       =					tokenizer(
									    text_target=__UpperCamelCase				,			max_length=32				,			padding="""max_length"""				,			truncation=__UpperCamelCase				,			return_tensors=__UpperCamelCase  )
									self.assertEqual(32				,			targets["""input_ids"""].shape[1]  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.ta_base_tokenizer
									_lowercase					       =					["""A long paragraph for summarization. </s>"""]
									_lowercase					       =					["""Summary of the text. </s>"""]
									# fmt: off
									_lowercase					       =					[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 35, 1]
									_lowercase					       =					[86, 120, 112, 112, 100, 117, 124, 35, 114, 105, 35, 119, 107, 104, 35, 119, 104, 123, 119, 49, 35, 1]
									# fmt: on

									_lowercase					       =					tokenizer(__UpperCamelCase				,			text_target=__UpperCamelCase  )

									self.assertEqual(__UpperCamelCase				,			batch["""input_ids"""][0]  )
									self.assertEqual(__UpperCamelCase				,			batch["""labels"""][0]  )






			def    UpperCamelCase_				(      self  ):
									# safety check on max_len default value so we are sure the test works
									_lowercase					       =					self.get_tokenizers()
									for tokenizer in tokenizers:
															with self.subTest(f"""{tokenizer.__class__.__name__}"""  ):
																					self.assertNotEqual(tokenizer.model_max_length				,			42  )

        # Now let's start the test
									_lowercase					       =					self.get_tokenizers()
									for tokenizer in tokenizers:
															with self.subTest(f"""{tokenizer.__class__.__name__}"""  ):
																					# Isolate this from the other tests because we save additional tokens/etc
																					_lowercase					       =					tempfile.mkdtemp()

																					_lowercase					       =					""" He is very happy, UNwant\u00E9d,running"""
																					_lowercase					       =					tokenizer.encode(__UpperCamelCase				,			add_special_tokens=__UpperCamelCase  )
																					tokenizer.save_pretrained(__UpperCamelCase  )

																					_lowercase					       =					tokenizer.__class__.from_pretrained(__UpperCamelCase  )
																					_lowercase					       =					after_tokenizer.encode(__UpperCamelCase				,			add_special_tokens=__UpperCamelCase  )
																					self.assertListEqual(__UpperCamelCase				,			__UpperCamelCase  )

																					shutil.rmtree(__UpperCamelCase  )

									_lowercase					       =					self.get_tokenizers(model_max_length=42  )
									for tokenizer in tokenizers:
															with self.subTest(f"""{tokenizer.__class__.__name__}"""  ):
																					# Isolate this from the other tests because we save additional tokens/etc
																					_lowercase					       =					tempfile.mkdtemp()

																					_lowercase					       =					""" He is very happy, UNwant\u00E9d,running"""
																					tokenizer.add_tokens(["""bim""", """bambam"""]  )
																					_lowercase					       =					tokenizer.additional_special_tokens
																					additional_special_tokens.append("""new_additional_special_token"""  )
																					tokenizer.add_special_tokens({"""additional_special_tokens""": additional_special_tokens}  )
																					_lowercase					       =					tokenizer.encode(__UpperCamelCase				,			add_special_tokens=__UpperCamelCase  )
																					tokenizer.save_pretrained(__UpperCamelCase  )

																					_lowercase					       =					tokenizer.__class__.from_pretrained(__UpperCamelCase  )
																					_lowercase					       =					after_tokenizer.encode(__UpperCamelCase				,			add_special_tokens=__UpperCamelCase  )
																					self.assertListEqual(__UpperCamelCase				,			__UpperCamelCase  )
																					self.assertIn("""new_additional_special_token"""				,			after_tokenizer.additional_special_tokens  )
																					self.assertEqual(after_tokenizer.model_max_length				,			42  )

																					_lowercase					       =					tokenizer.__class__.from_pretrained(__UpperCamelCase				,			model_max_length=43  )
																					self.assertEqual(tokenizer.model_max_length				,			43  )

																					shutil.rmtree(__UpperCamelCase  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					[]
									if self.test_slow_tokenizer:
															tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())  )

									if self.test_rust_tokenizer:
															tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())  )

									for tokenizer_class, tokenizer_utils in tokenizer_list:
															with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer_utils.save_pretrained(__UpperCamelCase  )

																					with open(os.path.join(__UpperCamelCase				,			"""special_tokens_map.json"""  )				,			encoding="""utf-8"""  ) as json_file:
																											_lowercase					       =					json.load(__UpperCamelCase  )

																					with open(os.path.join(__UpperCamelCase				,			"""tokenizer_config.json"""  )				,			encoding="""utf-8"""  ) as json_file:
																											_lowercase					       =					json.load(__UpperCamelCase  )

																					_lowercase					       =					[f"""<extra_id_{i}>""" for i in range(125  )]

																					_lowercase					       =					added_tokens_extra_ids + [
																					    """an_additional_special_token"""
																					]
																					_lowercase					       =					added_tokens_extra_ids + [
																					    """an_additional_special_token"""
																					]

																					with open(os.path.join(__UpperCamelCase				,			"""special_tokens_map.json"""  )				,			"""w"""				,			encoding="""utf-8"""  ) as outfile:
																											json.dump(__UpperCamelCase				,			__UpperCamelCase  )
																					with open(os.path.join(__UpperCamelCase				,			"""tokenizer_config.json"""  )				,			"""w"""				,			encoding="""utf-8"""  ) as outfile:
																											json.dump(__UpperCamelCase				,			__UpperCamelCase  )

																					# the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
																					# into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
																					# "special_tokens_map.json" files
																					_lowercase					       =					tokenizer_class.from_pretrained(
																					    __UpperCamelCase				,			)
																					self.assertIn(
																					    """an_additional_special_token"""				,			tokenizer_without_change_in_init.additional_special_tokens  )
																					# self.assertIn("an_additional_special_token",tokenizer_without_change_in_init.get_vocab()) # ByT5Tokenization no vocab
																					self.assertEqual(
																					    ["""an_additional_special_token"""]				,			tokenizer_without_change_in_init.convert_ids_to_tokens(
																					        tokenizer_without_change_in_init.convert_tokens_to_ids(["""an_additional_special_token"""]  )  )				,			)

																					# Now we test that we can change the value of additional_special_tokens in the from_pretrained
																					_lowercase					       =					added_tokens_extra_ids + [AddedToken("""a_new_additional_special_token"""				,			lstrip=__UpperCamelCase  )]
																					_lowercase					       =					tokenizer_class.from_pretrained(
																					    __UpperCamelCase				,			additional_special_tokens=__UpperCamelCase				,			)

																					self.assertIn("""a_new_additional_special_token"""				,			tokenizer.additional_special_tokens  )
																					self.assertEqual(
																					    ["""a_new_additional_special_token"""]				,			tokenizer.convert_ids_to_tokens(
																					        tokenizer.convert_tokens_to_ids(["""a_new_additional_special_token"""]  )  )				,			)






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					[]
									if self.test_slow_tokenizer:
															tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())  )

									if self.test_rust_tokenizer:
															tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())  )

									for tokenizer_class, tokenizer_utils in tokenizer_list:
															with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer_utils.save_pretrained(__UpperCamelCase  )

																					_lowercase					       =					tokenizer_class.from_pretrained(__UpperCamelCase  )

																					self.assertTrue(tokenizer.decode([255]  ) == """"""  )






			def    UpperCamelCase_				(      self  ):
									pass






			def    UpperCamelCase_				(      self  ):
									pass






			def    UpperCamelCase_				(      self  ):
									pass






			def    UpperCamelCase_				(      self  ):
									pass






			def    UpperCamelCase_				(      self  ):
									# The default common tokenizer tests uses invalid tokens for ByT5 that can only accept one-character strings
									# and special added tokens as tokens
									_lowercase					       =					self.get_tokenizers(fast=__UpperCamelCase				,			do_lower_case=__UpperCamelCase  )
									for tokenizer in tokenizers:
															with self.subTest(f"""{tokenizer.__class__.__name__}"""  ):
																					_lowercase					       =					["""t""", """h""", """i""", """s""", """ """, """i""", """s""", """ """, """a""", """ """, """t""", """e""", """x""", """t""", """</s>"""]
																					_lowercase					       =					tokenizer.convert_tokens_to_string(__UpperCamelCase  )

																					self.assertIsInstance(__UpperCamelCase				,			__UpperCamelCase  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.get_tokenizers()
									for tokenizer in tokenizers:
															with self.subTest(f"""{tokenizer.__class__.__name__}"""  ):
																					_lowercase					       =					[
																					    """bos_token""",
																					    """eos_token""",
																					    """unk_token""",
																					    """sep_token""",
																					    """pad_token""",
																					    """cls_token""",
																					    """mask_token""",
																					]

																					_lowercase					       =					0
																					_lowercase					       =					tokenizer.convert_ids_to_tokens(
																					    __UpperCamelCase				,			skip_special_tokens=__UpperCamelCase  )

																					for attr in attributes_list:
																											setattr(__UpperCamelCase				,			attr + """_id"""				,			__UpperCamelCase  )
																											self.assertEqual(getattr(__UpperCamelCase				,			__UpperCamelCase  )				,			__UpperCamelCase  )
																											self.assertEqual(getattr(__UpperCamelCase				,			attr + """_id"""  )				,			__UpperCamelCase  )

																											setattr(__UpperCamelCase				,			attr + """_id"""				,			__UpperCamelCase  )
																											self.assertEqual(getattr(__UpperCamelCase				,			__UpperCamelCase  )				,			__UpperCamelCase  )
																											self.assertEqual(getattr(__UpperCamelCase				,			attr + """_id"""  )				,			__UpperCamelCase  )

																					setattr(__UpperCamelCase				,			"""additional_special_tokens_ids"""				,			[]  )
																					self.assertListEqual(getattr(__UpperCamelCase				,			"""additional_special_tokens"""  )				,			[]  )
																					self.assertListEqual(getattr(__UpperCamelCase				,			"""additional_special_tokens_ids"""  )				,			[]  )

																					setattr(__UpperCamelCase				,			"""additional_special_tokens_ids"""				,			[token_id_to_test_setters]  )
																					self.assertListEqual(getattr(__UpperCamelCase				,			"""additional_special_tokens"""  )				,			[token_to_test_setters]  )
																					self.assertListEqual(getattr(__UpperCamelCase				,			"""additional_special_tokens_ids"""  )				,			[token_id_to_test_setters]  )

import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
		from sagemaker import Session, TrainingJobAnalytics
		from sagemaker.huggingface import HuggingFace



@pytest.mark.skipif(
    literal_eval(os.getenv('''TEST_SAGEMAKER'''					,			'''False'''    )    ) is not True					,			reason='''Skipping test because should only be run when releasing minor transformers version'''					,			)
@pytest.mark.usefixtures('''sm_env'''    )
@parameterized_class(
    [
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue_model_parallelism.py''',
            '''model_name_or_path''': '''roberta-large''',
            '''instance_type''': '''ml.p3dn.24xlarge''',
            '''results''': {'''train_runtime''': 1_6_0_0, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue.py''',
            '''model_name_or_path''': '''roberta-large''',
            '''instance_type''': '''ml.p3dn.24xlarge''',
            '''results''': {'''train_runtime''': 1_6_0_0, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
    ]    )
class    a_ (			unittest.TestCase    ):






			def    UpperCamelCase_				(      self  ):
									if self.framework == "pytorch":
															subprocess.run(
															    f"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()				,			encoding="""utf-8"""				,			check=__UpperCamelCase				,			)
									assert hasattr(self				,			"""env"""  )






			def    UpperCamelCase_				(      self				,			__UpperCamelCase  ):
									# configuration for running training on smdistributed Model Parallel
									_lowercase					       =					{
									    """enabled""": True,
									    """processes_per_host""": 8,
									}
									_lowercase					       =					{
									    """enabled""": True,
									    """parameters""": {
									        """microbatches""": 4,
									        """placement_strategy""": """spread""",
									        """pipeline""": """interleaved""",
									        """optimize""": """speed""",
									        """partitions""": 4,
									        """ddp""": True,
									    },
									}

									_lowercase					       =					{"""smdistributed""": {"""modelparallel""": smp_options}, """mpi""": mpi_options}

									_lowercase					       =					"""trainer""" if self.script == """run_glue.py""" else """smtrainer"""
									# creates estimator
									return HuggingFace(
									    entry_point=self.script				,			source_dir=self.env.test_path				,			role=self.env.role				,			image_uri=self.env.image_uri				,			base_job_name=f"""{self.env.base_job_name}-{instance_count}-smp-{name_extension}"""				,			instance_count=__UpperCamelCase				,			instance_type=self.instance_type				,			debugger_hook_config=__UpperCamelCase				,			hyperparameters={
									        **self.env.hyperparameters,
									        """model_name_or_path""": self.model_name_or_path,
									        """max_steps""": 500,
									    }				,			metric_definitions=self.env.metric_definitions				,			distribution=__UpperCamelCase				,			py_version="""py36"""				,			)






			def    UpperCamelCase_				(      self				,			__UpperCamelCase  ):
									TrainingJobAnalytics(__UpperCamelCase  ).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv"""  )






			@parameterized.expand([(1,)]  )
			def    UpperCamelCase_				(      self				,			__UpperCamelCase  ):
									# create estimator
									_lowercase					       =					self.create_estimator(__UpperCamelCase  )

									# run training
									estimator.fit()

									# result dataframe
									_lowercase					       =					TrainingJobAnalytics(estimator.latest_training_job.name  ).dataframe()

									# extract kpis
									_lowercase					       =					list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""]  )
									_lowercase					       =					list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""]  )
									# get train time from SageMaker job, this includes starting, preprocessing, stopping
									_lowercase					       =					(
									    Session().describe_training_job(estimator.latest_training_job.name  ).get("""TrainingTimeInSeconds"""				,			999_999  )
									)

									# assert kpis
									assert train_runtime <= self.results["train_runtime"]
									assert all(t >= self.results["""eval_accuracy"""] for t in eval_accuracy  )
									assert all(t <= self.results["""eval_loss"""] for t in eval_loss  )

									# dump tests result into json file to share in PR
									with open(f"""{estimator.latest_training_job.name}.json"""				,			"""w"""  ) as outfile:
															json.dump({"""train_time""": train_runtime, """eval_accuracy""": eval_accuracy, """eval_loss""": eval_loss}				,			__UpperCamelCase  )