






"""simple docstring"""
import json
from typing import Dict, List, Optional, Tuple, Union

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding, EncodedInput
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import PaddingStrategy, logging
from .tokenization_led import LEDTokenizer


SCREAMING_SNAKE_CASE__      						=  logging.get_logger(__name__)


SCREAMING_SNAKE_CASE__      						=  {"vocab_file": "vocab.json", "merges_file": "merges.txt", "tokenizer_file": "tokenizer.json"}

SCREAMING_SNAKE_CASE__      						=  {
    "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",
    },
}

SCREAMING_SNAKE_CASE__      						=  {
    "allenai/led-base-16384": 16_384,
}



class   lowerCAmelCase_    (			__a					):



				"""simple docstring"""
				_lowerCAmelCase     :    int			  =						VOCAB_FILES_NAMES
				_lowerCAmelCase     :    int			  =						PRETRAINED_VOCAB_FILES_MAP
				_lowerCAmelCase     :    Union[str, Any]			  =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				_lowerCAmelCase     :    Tuple			  =						LEDTokenizer
				_lowerCAmelCase     :    List[Any]			  =						["""input_ids""", """attention_mask"""]

				def __init__(				self		,    lowerCAmelCase=None		,    lowerCAmelCase=None		,    lowerCAmelCase=None		,    lowerCAmelCase="replace"		,    lowerCAmelCase="<s>"		,    lowerCAmelCase="</s>"		,    lowerCAmelCase="</s>"		,    lowerCAmelCase="<s>"		,    lowerCAmelCase="<unk>"		,    lowerCAmelCase="<pad>"		,    lowerCAmelCase="<mask>"		,    lowerCAmelCase=False		,    lowerCAmelCase=True		,    **lowerCAmelCase		,    ):






										"""simple docstring"""

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

										snake_case								=			json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()    )
										if pre_tok_state.get('add_prefix_space'		,    A__    ) != add_prefix_space:
																snake_case								=			getattr(A__		,    pre_tok_state.pop('type'    )    )
																snake_case								=			add_prefix_space
																snake_case								=			pre_tok_class(**A__    )

										snake_case								=			add_prefix_space

										# the pre_tokenizer is already updated in the GPT2TokenizerFast `__init__`
										snake_case								=			'post_processor'
										snake_case								=			getattr(self.backend_tokenizer		,    A__		,    A__    )
										if tokenizer_component_instance:
																snake_case								=			json.loads(tokenizer_component_instance.__getstate__()    )

																# The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
																if "sep" in state:
																						snake_case								=			tuple(state['sep']    )
																if "cls" in state:
																						snake_case								=			tuple(state['cls']    )

																snake_case								=			False

																if state.get('add_prefix_space'		,    A__    ) != add_prefix_space:
																						snake_case								=			add_prefix_space
																						snake_case								=			True

																if state.get('trim_offsets'		,    A__    ) != trim_offsets:
																						snake_case								=			trim_offsets
																						snake_case								=			True

																if changes_to_apply:
																						snake_case								=			getattr(A__		,    state.pop('type'    )    )
																						snake_case								=			component_class(**A__    )
																						setattr(self.backend_tokenizer		,    A__		,    A__    )

				@property
				# Copied from transformers.models.bart.tokenization_bart_fast.BartTokenizerFast.mask_token with BART->LED
				def 		snake_case (				self    ):






										"""simple docstring"""

										if self._mask_token is None:
																if self.verbose:
																						logger.error('Using mask_token, but it is not set yet.'    )
																return None
										return str(self._mask_token    )

				@mask_token.setter
				def 		snake_case (				self		,    lowerCAmelCase    ):






										"""simple docstring"""

										snake_case								=			AddedToken(A__		,    lstrip=A__		,    rstrip=A__    ) if isinstance(A__		,    A__    ) else value
										snake_case								=			value

				def 		snake_case (				self		,    *lowerCAmelCase		,    **lowerCAmelCase    ):






										"""simple docstring"""

										snake_case								=			kwargs.get('is_split_into_words'		,    A__    )

										if is_split_into_words and not self.add_prefix_space:
																raise ValueError(
																    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
																    'to use it with pretokenized inputs.'    )

										return super()._batch_encode_plus(*A__		,    **A__    )

				def 		snake_case (				self		,    *lowerCAmelCase		,    **lowerCAmelCase    ):






										"""simple docstring"""

										snake_case								=			kwargs.get('is_split_into_words'		,    A__    )

										if is_split_into_words and not self.add_prefix_space:
																raise ValueError(
																    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
																    'to use it with pretokenized inputs.'    )

										return super()._encode_plus(*A__		,    **A__    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = None    ):






										"""simple docstring"""

										snake_case								=			self._tokenizer.model.save(A__		,    name=A__    )
										return tuple(A__    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase=None    ):






										"""simple docstring"""

										snake_case								=			[self.bos_token_id] + token_ids_a + [self.eos_token_id]
										if token_ids_a is None:
																return output

										return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = None    ):






										"""simple docstring"""

										snake_case								=			[self.sep_token_id]
										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 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = None		,    lowerCAmelCase = PaddingStrategy.DO_NOT_PAD		,    lowerCAmelCase = None		,    lowerCAmelCase = None		,    ):






										"""simple docstring"""

										snake_case								=			super()._pad(
										    encoded_inputs=A__		,    max_length=A__		,    padding_strategy=A__		,    pad_to_multiple_of=A__		,    return_attention_mask=A__		,    )

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

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

																if needs_to_be_padded:
																						snake_case								=			len(A__    ) - 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`
																												snake_case								=			(
																												    encoded_inputs['global_attention_mask'] + [-1] * difference
																												)
																						elif self.padding_side == "left":
																												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 subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch



class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

										snake_case								=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

										snake_case								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

										# Force fetching the files so that we can use the cache
										snake_case								=			'hf-internal-testing/tiny-random-bert'
										BertConfig.from_pretrained(lowerCAmelCase    )
										BertModel.from_pretrained(lowerCAmelCase    )
										BertTokenizer.from_pretrained(lowerCAmelCase    )
										pipeline(task='fill-mask'		,    model=lowerCAmelCase    )

										# baseline - just load from_pretrained with normal network
										snake_case								=			[sys.executable, '-c', '\n'.join([load, run, mock]    )]

										# should succeed
										snake_case								=			self.get_env()
										# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
										snake_case								=			'1'
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

										snake_case								=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

										snake_case								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

										# Force fetching the files so that we can use the cache
										snake_case								=			'hf-internal-testing/tiny-random-bert'
										BertConfig.from_pretrained(lowerCAmelCase    )
										BertModel.from_pretrained(lowerCAmelCase    )
										BertTokenizer.from_pretrained(lowerCAmelCase    )
										pipeline(task='fill-mask'		,    model=lowerCAmelCase    )

										# baseline - just load from_pretrained with normal network
										snake_case								=			[sys.executable, '-c', '\n'.join([load, run, mock]    )]

										# should succeed
										snake_case								=			self.get_env()
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

										snake_case								=			'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

										snake_case								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

										# baseline - just load from_pretrained with normal network
										snake_case								=			[sys.executable, '-c', '\n'.join([load, run]    )]

										# should succeed
										snake_case								=			self.get_env()
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )

										# next emulate no network
										snake_case								=			[sys.executable, '-c', '\n'.join([load, mock, run]    )]

										# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
										# env["TRANSFORMERS_OFFLINE"] = "0"
										# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
										# self.assertEqual(result.returncode, 1, result.stderr)

										# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
										snake_case								=			'1'
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			'\nfrom transformers import pipeline\n        '
										snake_case								=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

										snake_case								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
										snake_case								=			self.get_env()
										snake_case								=			'1'
										snake_case								=			[sys.executable, '-c', '\n'.join([load, mock, run]    )]
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    1		,    result.stderr    )
										self.assertIn(
										    'You cannot infer task automatically within `pipeline` when using offline mode'		,    result.stderr.decode().replace('\n'		,    ''    )		,    )

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			'\nfrom transformers import AutoModel\n        '
										snake_case								=			'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

										# baseline - just load from_pretrained with normal network
										snake_case								=			[sys.executable, '-c', '\n'.join([load, run]    )]

										# should succeed
										snake_case								=			self.get_env()
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )

										# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
										snake_case								=			'1'
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )


