




import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
  from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
  import torch_xla.core.xla_model as xm



def 						lowerCAmelCase		(		_lowerCAmelCase							:   List[str]	):




 """simple docstring"""







 if is_torch_version("<"						,							"2.0.0"	) or not hasattr(_lowerCAmelCase						,							"_dynamo"	):
  return False
 return isinstance(_lowerCAmelCase						,							torch._dynamo.eval_frame.OptimizedModule	)



def 						lowerCAmelCase		(		_lowerCAmelCase							:   List[str]						,							_lowerCAmelCase							:   bool = True	):




 """simple docstring"""







 UpperCAmelCase__							 =						(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

 UpperCAmelCase__							 =						is_compiled_module(_lowerCAmelCase	)
 if is_compiled:
  UpperCAmelCase__							 =						model
  UpperCAmelCase__							 =						model._orig_mod

 if is_deepspeed_available():
  options += (DeepSpeedEngine,)

 while isinstance(_lowerCAmelCase						,							_lowerCAmelCase	):
  UpperCAmelCase__							 =						model.module

 if not keep_fpaa_wrapper:
  UpperCAmelCase__							 =						getattr(_lowerCAmelCase						,							"forward"	)
  UpperCAmelCase__							 =						model.__dict__.pop("_original_forward"						,							_lowerCAmelCase	)
  if original_forward is not None:
   while hasattr(_lowerCAmelCase						,							"__wrapped__"	):
    UpperCAmelCase__							 =						forward.__wrapped__
    if forward == original_forward:
     break
   UpperCAmelCase__							 =						forward
  if getattr(_lowerCAmelCase						,							"_converted_to_transformer_engine"						,							_lowerCAmelCase	):
   convert_model(_lowerCAmelCase						,							to_transformer_engine=_lowerCAmelCase	)

 if is_compiled:
  UpperCAmelCase__							 =						model
  UpperCAmelCase__							 =						compiled_model

 return model



def 						lowerCAmelCase		(		):




 """simple docstring"""







 PartialState().wait_for_everyone()



def 						lowerCAmelCase		(		_lowerCAmelCase							:   Any						,							_lowerCAmelCase							:   Union[str, Any]	):




 """simple docstring"""







 if PartialState().distributed_type == DistributedType.TPU:
  xm.save(_lowerCAmelCase						,							_lowerCAmelCase	)
 elif PartialState().local_process_index == 0:
  torch.save(_lowerCAmelCase						,							_lowerCAmelCase	)



@contextmanager
def 						lowerCAmelCase		(		**_lowerCAmelCase							:   Optional[int]	):




 """simple docstring"""







 for key, value in kwargs.items():
  UpperCAmelCase__							 =						str(_lowerCAmelCase	)

 yield

 for key in kwargs:
  if key.upper() in os.environ:
   del os.environ[key.upper()]



def 						lowerCAmelCase		(		_lowerCAmelCase							:   Union[str, Any]	):




 """simple docstring"""







 if not hasattr(_lowerCAmelCase						,							"__qualname__"	) and not hasattr(_lowerCAmelCase						,							"__name__"	):
  UpperCAmelCase__							 =						getattr(_lowerCAmelCase						,							"__class__"						,							_lowerCAmelCase	)
 if hasattr(_lowerCAmelCase						,							"__qualname__"	):
  return obj.__qualname__
 if hasattr(_lowerCAmelCase						,							"__name__"	):
  return obj.__name__
 return str(_lowerCAmelCase	)



def 						lowerCAmelCase		(		_lowerCAmelCase							:   Optional[Any]						,							_lowerCAmelCase							:   str	):




 """simple docstring"""







 for key, value in source.items():
  if isinstance(_lowerCAmelCase						,							_lowerCAmelCase	):
   UpperCAmelCase__							 =						destination.setdefault(_lowerCAmelCase						,							{}	)
   merge_dicts(_lowerCAmelCase						,							_lowerCAmelCase	)
  else:
   UpperCAmelCase__							 =						value

 return destination



def 						lowerCAmelCase		(		_lowerCAmelCase							:   int = None	):




 """simple docstring"""







 if port is None:
  UpperCAmelCase__							 =						2_9500
 with socket.socket(socket.AF_INET						,							socket.SOCK_STREAM	) as s:
  return s.connect_ex(("localhost", port)	) == 0











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


_lowerCAmelCase						:   Tuple		=   logging.get_logger(__name__)


_lowerCAmelCase						:   Union[str, Any]		=   {"vocab_file": "vocab.json", "merges_file": "merges.txt", "tokenizer_file": "tokenizer.json"}

_lowerCAmelCase						:   str		=   {
    "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						:   List[Any]		=   {
    "allenai/led-base-16384": 1_6_3_8_4,
}
class 			_UpperCamelCase      (				lowerCAmelCase		):
      UpperCAmelCase_        =		VOCAB_FILES_NAMES
      UpperCAmelCase_        =		PRETRAINED_VOCAB_FILES_MAP
      UpperCAmelCase_        =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      UpperCAmelCase_        =		LEDTokenizer
      UpperCAmelCase_        =		["""input_ids""", """attention_mask"""]
      def __init__(			self   :Dict ,      lowerCamelCase   :Any=None ,      lowerCamelCase   :Dict=None ,      lowerCamelCase   :Dict=None ,      lowerCamelCase   :int="replace" ,      lowerCamelCase   :List[Any]="<s>" ,      lowerCamelCase   :Optional[Any]="</s>" ,      lowerCamelCase   :Optional[Any]="</s>" ,      lowerCamelCase   :int="<s>" ,      lowerCamelCase   :Optional[Any]="<unk>" ,      lowerCamelCase   :Dict="<pad>" ,      lowerCamelCase   :Tuple="<mask>" ,      lowerCamelCase   :str=False ,      lowerCamelCase   :Union[str, Any]=True ,      **lowerCamelCase   :List[str] ,      )     ->			Tuple:
       super().__init__(
           lowerCamelCase ,      lowerCamelCase ,      tokenizer_file=lowerCamelCase ,      errors=lowerCamelCase ,      bos_token=lowerCamelCase ,      eos_token=lowerCamelCase ,      sep_token=lowerCamelCase ,      cls_token=lowerCamelCase ,      unk_token=lowerCamelCase ,      pad_token=lowerCamelCase ,      mask_token=lowerCamelCase ,      add_prefix_space=lowerCamelCase ,      trim_offsets=lowerCamelCase ,      **lowerCamelCase ,      )

       UpperCAmelCase__							 =						json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()      )
       if pre_tok_state.get("add_prefix_space" ,      lowerCamelCase      ) != add_prefix_space:
        UpperCAmelCase__							 =						getattr(lowerCamelCase ,      pre_tok_state.pop("type"      )      )
        UpperCAmelCase__							 =						add_prefix_space
        UpperCAmelCase__							 =						pre_tok_class(**lowerCamelCase      )

       UpperCAmelCase__							 =						add_prefix_space

       # the pre_tokenizer is already updated in the GPT2TokenizerFast `__init__`
       UpperCAmelCase__							 =						"post_processor"
       UpperCAmelCase__							 =						getattr(self.backend_tokenizer ,      lowerCamelCase ,      lowerCamelCase      )
       if tokenizer_component_instance:
        UpperCAmelCase__							 =						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:
         UpperCAmelCase__							 =						tuple(state["sep"]      )
        if "cls" in state:
         UpperCAmelCase__							 =						tuple(state["cls"]      )

        UpperCAmelCase__							 =						False

        if state.get("add_prefix_space" ,      lowerCamelCase      ) != add_prefix_space:
         UpperCAmelCase__							 =						add_prefix_space
         UpperCAmelCase__							 =						True

        if state.get("trim_offsets" ,      lowerCamelCase      ) != trim_offsets:
         UpperCAmelCase__							 =						trim_offsets
         UpperCAmelCase__							 =						True

        if changes_to_apply:
         UpperCAmelCase__							 =						getattr(lowerCamelCase ,      state.pop("type"      )      )
         UpperCAmelCase__							 =						component_class(**lowerCamelCase      )
         setattr(self.backend_tokenizer ,      lowerCamelCase ,      lowerCamelCase      )
      @property
      # Copied from transformers.models.bart.tokenization_bart_fast.BartTokenizerFast.mask_token with BART->LED
      def 	UpperCAmelCase_						(			self   :List[str]      )     ->			str:
       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 	UpperCAmelCase_						(			self   :Union[str, Any] ,      lowerCamelCase   :List[str]      )     ->			Dict:
       UpperCAmelCase__							 =						AddedToken(lowerCamelCase ,      lstrip=lowerCamelCase ,      rstrip=lowerCamelCase      ) if isinstance(lowerCamelCase ,      lowerCamelCase      ) else value
       UpperCAmelCase__							 =						value
      def 	UpperCAmelCase_						(			self   :Optional[Any] ,      *lowerCamelCase   :Any ,      **lowerCamelCase   :int      )     ->			BatchEncoding:
       UpperCAmelCase__							 =						kwargs.get("is_split_into_words" ,      lowerCamelCase      )

       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(*lowerCamelCase ,      **lowerCamelCase      )
      def 	UpperCAmelCase_						(			self   :int ,      *lowerCamelCase   :Optional[Any] ,      **lowerCamelCase   :Optional[int]      )     ->			BatchEncoding:
       UpperCAmelCase__							 =						kwargs.get("is_split_into_words" ,      lowerCamelCase      )

       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(*lowerCamelCase ,      **lowerCamelCase      )
      def 	UpperCAmelCase_						(			self   :Any ,      lowerCamelCase   :str ,      lowerCamelCase   :Optional[str] = None      )     ->			Tuple[str]:
       UpperCAmelCase__							 =						self._tokenizer.model.save(lowerCamelCase ,      name=lowerCamelCase      )
       return tuple(lowerCamelCase      )
      def 	UpperCAmelCase_						(			self   :Optional[Any] ,      lowerCamelCase   :Optional[int] ,      lowerCamelCase   :Optional[Any]=None      )     ->			Dict:
       UpperCAmelCase__							 =						[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 	UpperCAmelCase_						(			self   :Optional[int] ,      lowerCamelCase   :List[int] ,      lowerCamelCase   :Optional[List[int]] = None      )     ->			List[int]:
       UpperCAmelCase__							 =						[self.sep_token_id]
       UpperCAmelCase__							 =						[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 	UpperCAmelCase_						(			self   :List[str] ,      lowerCamelCase   :Union[Dict[str, EncodedInput], BatchEncoding] ,      lowerCamelCase   :Optional[int] = None ,      lowerCamelCase   :PaddingStrategy = PaddingStrategy.DO_NOT_PAD ,      lowerCamelCase   :Optional[int] = None ,      lowerCamelCase   :Optional[bool] = None ,      )     ->			dict:
       UpperCAmelCase__							 =						super()._pad(
           encoded_inputs=lowerCamelCase ,      max_length=lowerCamelCase ,      padding_strategy=lowerCamelCase ,      pad_to_multiple_of=lowerCamelCase ,      return_attention_mask=lowerCamelCase ,      )

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

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

        if needs_to_be_padded:
         UpperCAmelCase__							 =						len(lowerCamelCase      ) - 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`
          UpperCAmelCase__							 =						(
              encoded_inputs["global_attention_mask"] + [-1] * difference
          )
         elif self.padding_side == "left":
          UpperCAmelCase__							 =						[-1] * difference + encoded_inputs[
              "global_attention_mask"
          ]
         else:
          raise ValueError("Invalid padding strategy:" + str(self.padding_side      )      )

       return encoded_inputs





