


"""simple docstring"""






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

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


_lowerCamelCase     =  logging.get_logger(__name__)


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

# See all LED models at https://huggingface.co/models?filter=LED
_lowerCamelCase     =  {
    'vocab_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json',
    },
    'merges_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt',
    },
    'tokenizer_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json',
    },
}

_lowerCamelCase     =  {
    'allenai/led-base-16384': 16384,
}


@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def __lowercase     (      ):
				SCREAMING_SNAKE_CASE__     							=				(
				    list(range(ord("!"  )						,					ord("~"  ) + 1  )  ) + list(range(ord("¡"  )						,					ord("¬"  ) + 1  )  ) + list(range(ord("®"  )						,					ord("ÿ"  ) + 1  )  )
				)
				SCREAMING_SNAKE_CASE__     							=				bs[:]
				SCREAMING_SNAKE_CASE__     							=				0
				for b in range(2**8  ):
								if b not in bs:
												bs.append(lowerCamelCase_  )
												cs.append(2**8 + n  )
												n += 1
				SCREAMING_SNAKE_CASE__     							=				[chr(lowerCamelCase_  ) for n in cs]
				return dict(zip(lowerCamelCase_						,					lowerCamelCase_  )  )


def __lowercase     (      lowerCamelCase_  :      Optional[Any]  ):
				SCREAMING_SNAKE_CASE__     							=				set()
				SCREAMING_SNAKE_CASE__     							=				word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)  )
								SCREAMING_SNAKE_CASE__     							=				char
				return pairs


class 			lowerCamelCase_		(			lowercase   ):





				"""simple docstring"""







				_lowerCAmelCase      :      Optional[int]      							=					VOCAB_FILES_NAMES
				_lowerCAmelCase      :      Optional[Any]      							=					PRETRAINED_VOCAB_FILES_MAP
				_lowerCAmelCase      :      Union[str, Any]      							=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				_lowerCAmelCase      :      List[Any]      							=					["input_ids", "attention_mask"]



				def __init__(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__       ,		UpperCAmelCase__="replace"       ,		UpperCAmelCase__="<s>"       ,		UpperCAmelCase__="</s>"       ,		UpperCAmelCase__="</s>"       ,		UpperCAmelCase__="<s>"       ,		UpperCAmelCase__="<unk>"       ,		UpperCAmelCase__="<pad>"       ,		UpperCAmelCase__="<mask>"       ,		UpperCAmelCase__=False       ,		**UpperCAmelCase__       ,		):
								SCREAMING_SNAKE_CASE__     							=				AddedToken(UpperCAmelCase__       ,		lstrip=UpperCAmelCase__       ,		rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__       ,		UpperCAmelCase__     ) else bos_token
								SCREAMING_SNAKE_CASE__     							=				AddedToken(UpperCAmelCase__       ,		lstrip=UpperCAmelCase__       ,		rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__       ,		UpperCAmelCase__     ) else eos_token
								SCREAMING_SNAKE_CASE__     							=				AddedToken(UpperCAmelCase__       ,		lstrip=UpperCAmelCase__       ,		rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__       ,		UpperCAmelCase__     ) else sep_token
								SCREAMING_SNAKE_CASE__     							=				AddedToken(UpperCAmelCase__       ,		lstrip=UpperCAmelCase__       ,		rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__       ,		UpperCAmelCase__     ) else cls_token
								SCREAMING_SNAKE_CASE__     							=				AddedToken(UpperCAmelCase__       ,		lstrip=UpperCAmelCase__       ,		rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__       ,		UpperCAmelCase__     ) else unk_token
								SCREAMING_SNAKE_CASE__     							=				AddedToken(UpperCAmelCase__       ,		lstrip=UpperCAmelCase__       ,		rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__       ,		UpperCAmelCase__     ) else pad_token

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

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

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

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



				@property
				# Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
				def   lowerCAmelCase__		(						self     ):
								return len(self.encoder     )



				def   lowerCAmelCase__		(						self     ):
								return dict(self.encoder       ,		**self.added_tokens_encoder     )



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								if token in self.cache:
												return self.cache[token]
								SCREAMING_SNAKE_CASE__     							=				tuple(UpperCAmelCase__     )
								SCREAMING_SNAKE_CASE__     							=				get_pairs(UpperCAmelCase__     )

								if not pairs:
												return token

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

																if word[i] == first and i < len(UpperCAmelCase__     ) - 1 and word[i + 1] == second:
																				new_word.append(first + second     )
																				i += 2
																else:
																				new_word.append(word[i]     )
																				i += 1
												SCREAMING_SNAKE_CASE__     							=				tuple(UpperCAmelCase__     )
												SCREAMING_SNAKE_CASE__     							=				new_word
												if len(UpperCAmelCase__     ) == 1:
																break
												else:
																SCREAMING_SNAKE_CASE__     							=				get_pairs(UpperCAmelCase__     )
								SCREAMING_SNAKE_CASE__     							=				" ".join(UpperCAmelCase__     )
								SCREAMING_SNAKE_CASE__     							=				word
								return word



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								SCREAMING_SNAKE_CASE__     							=				[]
								for token in re.findall(self.pat       ,		UpperCAmelCase__     ):
												SCREAMING_SNAKE_CASE__     							=				"".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(UpperCAmelCase__     ).split(" "     )     )
								return bpe_tokens



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								return self.encoder.get(UpperCAmelCase__       ,		self.encoder.get(self.unk_token     )     )



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								return self.decoder.get(UpperCAmelCase__     )



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								SCREAMING_SNAKE_CASE__     							=				"".join(UpperCAmelCase__     )
								SCREAMING_SNAKE_CASE__     							=				bytearray([self.byte_decoder[c] for c in text]     ).decode("utf-8"       ,		errors=self.errors     )
								return text



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__ = None     ):
								if not os.path.isdir(UpperCAmelCase__     ):
												logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''     )
												return
								SCREAMING_SNAKE_CASE__     							=				os.path.join(
								    UpperCAmelCase__       ,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]     )
								SCREAMING_SNAKE_CASE__     							=				os.path.join(
								    UpperCAmelCase__       ,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]     )

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

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

								return vocab_file, merge_file



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__ = None     ):
								if token_ids_a is None:
												return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
								SCREAMING_SNAKE_CASE__     							=				[self.cls_token_id]
								SCREAMING_SNAKE_CASE__     							=				[self.sep_token_id]
								return cls + token_ids_a + sep + sep + token_ids_a + sep



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__ = None       ,		UpperCAmelCase__ = False     ):
								if already_has_special_tokens:
												return super().get_special_tokens_mask(
												    token_ids_a=UpperCAmelCase__       ,		token_ids_a=UpperCAmelCase__       ,		already_has_special_tokens=UpperCAmelCase__     )

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



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__ = None     ):
								SCREAMING_SNAKE_CASE__     							=				[self.sep_token_id]
								SCREAMING_SNAKE_CASE__     							=				[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   lowerCAmelCase__		(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__=False       ,		**UpperCAmelCase__     ):
								SCREAMING_SNAKE_CASE__     							=				kwargs.pop("add_prefix_space"       ,		self.add_prefix_space     )
								if (is_split_into_words or add_prefix_space) and (len(UpperCAmelCase__     ) > 0 and not text[0].isspace()):
												SCREAMING_SNAKE_CASE__     							=				" " + text
								return (text, kwargs)






				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__       ,		UpperCAmelCase__ = None       ,		UpperCAmelCase__ = PaddingStrategy.DO_NOT_PAD       ,		UpperCAmelCase__ = None       ,		UpperCAmelCase__ = None       ,		):
								SCREAMING_SNAKE_CASE__     							=				super()._pad(
								    encoded_inputs=UpperCAmelCase__       ,		max_length=UpperCAmelCase__       ,		padding_strategy=UpperCAmelCase__       ,		pad_to_multiple_of=UpperCAmelCase__       ,		return_attention_mask=UpperCAmelCase__       ,		)

								# Load from model defaults
								if return_attention_mask is None:
												SCREAMING_SNAKE_CASE__     							=				"attention_mask" in self.model_input_names

								if return_attention_mask and "global_attention_mask" in encoded_inputs:
												SCREAMING_SNAKE_CASE__     							=				encoded_inputs[self.model_input_names[0]]
												# `global_attention_mask` need to have the same length as other (sequential) inputs.
												SCREAMING_SNAKE_CASE__     							=				len(encoded_inputs["global_attention_mask"]     ) != len(UpperCAmelCase__     )

												if needs_to_be_padded:
																SCREAMING_SNAKE_CASE__     							=				len(UpperCAmelCase__     ) - len(encoded_inputs["global_attention_mask"]     )

																if self.padding_side == "right":
																				# Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
																				SCREAMING_SNAKE_CASE__     							=				(
																				    encoded_inputs["global_attention_mask"] + [-1] * difference
																				)
																elif self.padding_side == "left":
																				SCREAMING_SNAKE_CASE__     							=				[-1] * difference + encoded_inputs[
																				    "global_attention_mask"
																				]
																else:
																				raise ValueError("Invalid padding strategy:" + str(self.padding_side     )     )

								return encoded_inputs




"""simple docstring"""






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

import pytest
from parameterized import 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.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 650, "eval_accuracy": 0.6, "eval_loss": 0.9},
        },
        {
            "framework": "tensorflow",
            "script": "run_tf.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 600, "eval_accuracy": 0.3, "eval_loss": 0.9},
        },
    ]   )
class 			lowerCamelCase_		(			unittest.TestCase   ):





				"""simple docstring"""







				def   lowerCAmelCase__		(						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   lowerCAmelCase__		(						self       ,		UpperCAmelCase__=1     ):
								# 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}-single'''       ,		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}       ,		metric_definitions=self.env.metric_definitions       ,		py_version="py36"       ,		)



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								TrainingJobAnalytics(UpperCAmelCase__     ).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''     )






				def   lowerCAmelCase__		(						self     ):
								# create estimator
								SCREAMING_SNAKE_CASE__     							=				self.create_estimator()

								# run training
								estimator.fit()

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

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

								# 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__     )
