




import importlib.util
import weakref
from functools import partialmethod

from .dependency_versions_check import dep_version_check
from .utils import is_accelerate_available, is_torch_available, logging


if is_torch_available():
 import torch

a_ 		=  logging.get_logger(__name__)




def        __SCREAMING_SNAKE_CASE (	)							->						str:

    """simple docstring"""


    return importlib.util.find_spec('''deepspeed'''   ) is not None


if is_accelerate_available() and is_deepspeed_available():
 from accelerate.utils.deepspeed import HfDeepSpeedConfig as DeepSpeedConfig
else:
 # Inherits from a dummy `object` if accelerate is not available, so that python succeeds to import this file.
 # Deepspeed glue code will never inherit this dummy object as it checks if accelerate is available.
 from builtins import object as DeepSpeedConfig

class 			_lowerCamelCase    (      _SCREAMING_SNAKE_CASE ):

      """simple docstring"""

      def __init__(  self       :			Optional[int]			,							snake_case       :			Optional[Any]					):
          # set global weakref object
          set_hf_deepspeed_config(self					)
          dep_version_check('''accelerate'''					)
          dep_version_check('''deepspeed'''					)
          super().__init__(snake_case					)







class 			_lowerCamelCase    (      _SCREAMING_SNAKE_CASE ):

      """simple docstring"""

      def __init__(  self       :			List[Any]			,							snake_case       :			Dict					):
          super().__init__(snake_case					)
          __UpperCamelCase			=    None
          __UpperCamelCase			=    []







      def    snake_case						(  self       :			Dict					):
          if self._dtype is None:
              raise ValueError('''trainer_config_process() wasn\'t called yet to tell dtype'''					)
          return self._dtype







      def    snake_case						(  self       :			str			,							snake_case       :			int					):
          __UpperCamelCase			=    self.get_value(snake_case					)
          if val is None:
              return False
          else:
              return val == "auto"







      def    snake_case						(  self       :			str			,							snake_case       :			str			,							snake_case       :			Optional[int]			,							snake_case       :			Optional[int]=None			,							snake_case       :			Any=True					):
          __UpperCamelCase		,   __UpperCamelCase			=    self.find_config_node(snake_case					)
          if config is None:
              return

          if config.get(snake_case					) == "auto":
              __UpperCamelCase			=    hf_val
              return

          if not must_match:
              return

          __UpperCamelCase			=    config.get(snake_case					)
          if ds_val is not None and ds_val != hf_val:
              self.mismatches.append(F"- ds {ds_key_long}={ds_val} vs hf {hf_key}={hf_val}"					)

      lowerCAmelCase__      :			List[str]     =       partialmethod(_SCREAMING_SNAKE_CASE				,      must_match=_SCREAMING_SNAKE_CASE )







      def    snake_case						(  self       :			List[Any]			,							snake_case       :			Optional[Any]					):
          __UpperCamelCase			=    args.world_size * args.per_device_train_batch_size * args.gradient_accumulation_steps
          self.fill_match(
              '''train_micro_batch_size_per_gpu'''			,							args.per_device_train_batch_size			,							'''per_device_train_batch_size'''					)
          self.fill_match('''gradient_accumulation_steps'''			,							args.gradient_accumulation_steps			,							'''gradient_accumulation_steps'''					)
          self.fill_match('''train_batch_size'''			,							snake_case			,							'''train_batch_size (calculated)'''					)
          self.fill_match('''gradient_clipping'''			,							args.max_grad_norm			,							'''max_grad_norm'''					)

          self.fill_match('''optimizer.params.lr'''			,							args.learning_rate			,							'''learning_rate'''					)
          self.fill_match('''optimizer.params.betas'''			,							[args.adam_betaa, args.adam_betaa]			,							'''adam_beta1+adam_beta2'''					)
          self.fill_match('''optimizer.params.eps'''			,							args.adam_epsilon			,							'''adam_epsilon'''					)
          self.fill_match('''optimizer.params.weight_decay'''			,							args.weight_decay			,							'''weight_decay'''					)

          self.fill_only('''scheduler.params.warmup_min_lr'''			,							0					)  # not a trainer arg
          self.fill_match('''scheduler.params.warmup_max_lr'''			,							args.learning_rate			,							'''learning_rate'''					)
          # total_num_steps - will get set in trainer_config_finalize

          # fp16
          if args.fpaa or args.fpaa_full_eval:
              __UpperCamelCase			=    '''apex''' if args.fpaa_backend == '''apex''' else '''amp'''
          else:
              __UpperCamelCase			=    None

          if args.save_on_each_node:
              # deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
              __UpperCamelCase			=    self.config.get('''checkpoint'''			,							{}					)
              __UpperCamelCase			=    args.save_on_each_node

          # amp: similar to the pytorch native amp - it has a bunch of optional params but we won't set
          # any here unless the user did the work
          self.fill_match(
              '''fp16.enabled'''			,							((args.fpaa or args.fpaa_full_eval) and fpaa_backend == '''amp''')			,							'''fp16|fp16_full_eval+fp16_backend(amp)'''			,							)

          # apex: delegates amp work to apex (which needs to be available), but it cannot be used with any
          # ZeRO features
          self.fill_match('''amp.enabled'''			,							fpaa_backend == '''apex'''			,							'''fp16+fp16_backend(apex)'''					)
          self.fill_match('''amp.opt_level'''			,							args.fpaa_opt_level			,							'''fp16_opt_level'''					)

          self.fill_match('''bf16.enabled'''			,							(args.bfaa or args.bfaa_full_eval)			,							'''bf16|bf16_full_eval'''					)

          # deepspeed's default mode is fp16 unless there is a config that says differently
          if self.is_true('''bf16.enabled'''					):
              __UpperCamelCase			=    torch.bfloataa
          elif self.is_false('''fp16.enabled'''					):
              __UpperCamelCase			=    torch.floataa
          else:
              __UpperCamelCase			=    torch.floataa







      def    snake_case						(  self       :			Dict			,							snake_case       :			Optional[Any]			,							snake_case       :			Optional[int]			,							snake_case       :			Optional[int]					):
          __UpperCamelCase			=    [
              '''zero_optimization.reduce_bucket_size''',
              '''zero_optimization.stage3_prefetch_bucket_size''',
              '''zero_optimization.stage3_param_persistence_threshold''',
          ]
          __UpperCamelCase			=    [x for x in hidden_size_based_keys if self.is_auto(snake_case					)]

          if len(snake_case					) > 0:
              if hasattr(model.config			,							'''hidden_size'''					):
                  __UpperCamelCase			=    model.config.hidden_size
              elif hasattr(model.config			,							'''hidden_sizes'''					):
                  # if there are many hidden sizes pick the largest one
                  __UpperCamelCase			=    max(model.config.hidden_sizes					)
              else:
                  raise ValueError(
                      '''The model\'s config file has neither `hidden_size` nor `hidden_sizes` entry, '''
                      '''therefore it\'s not possible to automatically fill out the following `auto` entries '''
                      F"in the DeepSpeed config file: {hidden_size_auto_keys}. You can fix that by replacing "
                      '''`auto` values for these keys with an integer value of your choice.'''					)

              self.fill_only('''zero_optimization.reduce_bucket_size'''			,							hidden_size * hidden_size					)
              if self.is_zeroa():
                  # automatically assign the optimal config values based on model config
                  self.fill_only('''zero_optimization.stage3_prefetch_bucket_size'''			,							0.9 * hidden_size * hidden_size					)
                  self.fill_only('''zero_optimization.stage3_param_persistence_threshold'''			,							10 * hidden_size					)

        # scheduler
          self.fill_match('''scheduler.params.total_num_steps'''			,							snake_case			,							'''num_training_steps (calculated)'''					)
          self.fill_match('''scheduler.params.warmup_num_steps'''			,							args.get_warmup_steps(snake_case					)			,							'''warmup_steps'''					)

          if len(self.mismatches					) > 0:
              __UpperCamelCase			=    '''\n'''.join(self.mismatches					)
              raise ValueError(
                  '''Please correct the following DeepSpeed config values that mismatch TrainingArguments'''
                  F" values:\n{mismatches}\nThe easiest method is to set these DeepSpeed config values to 'auto'."					)







a_ 		=  None




def        __SCREAMING_SNAKE_CASE (	lowercase_   )							->						str:

    """simple docstring"""


    global _hf_deepspeed_config_weak_ref
    # will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
    __UpperCamelCase			=    weakref.ref(lowercase_   )




def        __SCREAMING_SNAKE_CASE (	)							->						Optional[int]:

    """simple docstring"""


    global _hf_deepspeed_config_weak_ref
    __UpperCamelCase			=    None




def        __SCREAMING_SNAKE_CASE (	)							->						Optional[Any]:

    """simple docstring"""


    if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
        return _hf_deepspeed_config_weak_ref().is_zeroa()
    else:
        return False




def        __SCREAMING_SNAKE_CASE (	)							->						Optional[Any]:

    """simple docstring"""


    if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
        return _hf_deepspeed_config_weak_ref().config
    else:
        return None




def        __SCREAMING_SNAKE_CASE (	lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_   )							->						int:

    """simple docstring"""


    from accelerate.utils import DummyOptim, DummyScheduler

    __UpperCamelCase			=    hf_deepspeed_config.config

    # Optimizer + Scheduler
    # Currently supported combos:
    # 1. DS scheduler + DS optimizer: Yes
    # 2. HF scheduler + HF optimizer: Yes
    # 3. DS scheduler + HF optimizer: Yes
    # 4. HF scheduler + DS optimizer: No
    #
    # Unless Offload is enabled in which case it's:
    # 1. DS scheduler + DS optimizer: Yes
    # 2. HF scheduler + HF optimizer: Mostly*
    # 3. DS scheduler + HF optimizer: Mostly*
    # 4. HF scheduler + DS optimizer: No
    #
    # Mostly*: All non-native DeepSpeed optimizers that have both CPU and GPU implementation should work (except LAMB)

    __UpperCamelCase			=    None
    if "optimizer" in config:
        if args.adafactor:
            raise ValueError(
                '''--adafactor was passed, but also found `optimizer` configured in the DeepSpeed config. '''
                '''Only one optimizer can be configured.'''   )
        __UpperCamelCase			=    DummyOptim(params=lowercase_   )
    else:
        if hf_deepspeed_config.is_offload():
            logger.info(
                '''Detected ZeRO Offload and non-DeepSpeed optimizers: This combination should work as long as the'''
                ''' custom optimizer has both CPU and GPU implementation (except LAMB)'''   )

        # ds supports Adam, OneBitAdam, and Lamb optimizers and can import other optimizers from torch.
        # But trainer uses AdamW by default.
        __UpperCamelCase			=    trainer.create_optimizer()
        # To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
        __UpperCamelCase			=    True

    __UpperCamelCase			=    None
    if "scheduler" in config:
        __UpperCamelCase			=    DummyScheduler(lowercase_   )
    else:
        if isinstance(lowercase_       ,  lowercase_   ):
            raise ValueError(
                '''Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. '''
                '''Please configure a scheduler in the DeepSpeed config.'''   )
        __UpperCamelCase			=    trainer.create_scheduler(num_training_steps=lowercase_       ,  optimizer=lowercase_   )

    return optimizer, lr_scheduler




def        __SCREAMING_SNAKE_CASE (	lowercase_       ,  lowercase_       ,  lowercase_=False   )							->						Tuple:

    """simple docstring"""


    from deepspeed.utils import logger as ds_logger

    __UpperCamelCase			=    trainer.model
    __UpperCamelCase			=    trainer.args

    __UpperCamelCase			=    trainer.accelerator.state.deepspeed_plugin.hf_ds_config

    # resume config update - some bits like `model` and `num_training_steps` only become available during train
    hf_deepspeed_config.trainer_config_finalize(lowercase_       ,  lowercase_       ,  lowercase_   )

    # set the Deepspeed log level consistent with the Trainer
    ds_logger.setLevel(args.get_process_log_level()   )

    if inference:
        # only Z3 makes sense for the inference
        if not hf_deepspeed_config.is_zeroa():
            raise ValueError('''ZeRO inference only makes sense with ZeRO Stage 3 - please adjust your config'''   )

        # in case the training config is re-used for inference
        hf_deepspeed_config.del_config_sub_tree('''optimizer'''   )
        hf_deepspeed_config.del_config_sub_tree('''lr_scheduler'''   )
        __UpperCamelCase		,   __UpperCamelCase			=    None, None
        __UpperCamelCase			=    None
    else:
        __UpperCamelCase			=    None  # important for when deepspeed_init is used as re-init
        __UpperCamelCase			=    list(filter(lambda lowercase_   : p.requires_grad       ,  model.parameters()   )   )
        __UpperCamelCase		,   __UpperCamelCase			=    deepspeed_optim_sched(
            lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_       ,  lowercase_   )

    # keep for quick debug:
    # from pprint import pprint; pprint(config)

    return optimizer, lr_scheduler




def        __SCREAMING_SNAKE_CASE (	lowercase_       ,  lowercase_   )							->						int:

    """simple docstring"""


    import glob

    __UpperCamelCase			=    sorted(glob.glob(F"{checkpoint_path}/global_step*"   )   )

    if len(lowercase_   ) > 0:
        logger.info(F"Attempting to resume from {checkpoint_path}"   )
        # this magically updates self.optimizer and self.lr_scheduler
        __UpperCamelCase		,   __UpperCamelCase			=    deepspeed_engine.load_checkpoint(
            lowercase_       ,  load_optimizer_states=lowercase_       ,  load_lr_scheduler_states=lowercase_   )
        if load_path is None:
            raise ValueError(F"[deepspeed] failed to resume from checkpoint {checkpoint_path}"   )
    else:
        raise ValueError(F"Can't find a valid checkpoint at {checkpoint_path}"   )













'''simple docstring'''






import json
import os
import re
import warnings
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


A :			Optional[Any]							   =      logging.get_logger(__name__)

A :			Optional[Any]							   =      {
    'source_spm': 'source.spm',
    'target_spm': 'target.spm',
    'vocab': 'vocab.json',
    'target_vocab_file': 'target_vocab.json',
    'tokenizer_config_file': 'tokenizer_config.json',
}

A :			Tuple							   =      {
    'source_spm': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/source.spm'
    },
    'target_spm': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/target.spm'
    },
    'vocab': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/vocab.json'
    },
    'tokenizer_config_file': {
        'Helsinki-NLP/opus-mt-en-de': (
            'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/tokenizer_config.json'
        )
    },
}

A :			Optional[int]							   =      {'Helsinki-NLP/opus-mt-en-de': 512}
A :			Any							   =      {}







class       lowerCamelCase       (    __UpperCAmelCase	):
					_SCREAMING_SNAKE_CASE				 =     VOCAB_FILES_NAMES
					_SCREAMING_SNAKE_CASE				 =     PRETRAINED_VOCAB_FILES_MAP
					_SCREAMING_SNAKE_CASE				 =     PRETRAINED_INIT_CONFIGURATION
					_SCREAMING_SNAKE_CASE				 =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					_SCREAMING_SNAKE_CASE				 =     ["input_ids", "attention_mask"]
					_SCREAMING_SNAKE_CASE				 =     re.compile(">>.+<<"	)  # type: re.Pattern


					def __init__(       self						:	Optional[Any]							,  __snake_case						:	Tuple							,  __snake_case						:	Optional[int]							,  __snake_case						:	Optional[Any]							,  __snake_case						:	List[Any]=None							,  __snake_case						:	int=None							,  __snake_case						:	Optional[Any]=None							,  __snake_case						:	Optional[int]="<unk>"							,  __snake_case						:	str="</s>"							,  __snake_case						:	Tuple="<pad>"							,  __snake_case						:	List[str]=5_12							,  __snake_case						:	Optional[Dict[str, Any]] = None							,  __snake_case						:	Union[str, Any]=False							,  **__snake_case						:	int							,  ):





											'''simple docstring'''





											_snake_case: List[str]	       =		{} if sp_model_kwargs is None else sp_model_kwargs

											super().__init__(
											    # bos_token=bos_token,  unused. Start decoding with config.decoder_start_token_id
											    source_lang=__snake_case							,  target_lang=__snake_case							,  unk_token=__snake_case							,  eos_token=__snake_case							,  pad_token=__snake_case							,  model_max_length=__snake_case							,  sp_model_kwargs=self.sp_model_kwargs							,  target_vocab_file=__snake_case							,  separate_vocabs=__snake_case							,  **__snake_case							,  )
											assert Path(__snake_case							).exists(), f'''cannot find spm source {source_spm}'''

											_snake_case: List[Any]	       =		separate_vocabs
											_snake_case: List[Any]	       =		load_json(__snake_case							)
											if self.unk_token not in self.encoder:
																	raise KeyError('<unk> token must be in vocab'							)
											assert self.pad_token in self.encoder

											if separate_vocabs:
																	_snake_case: List[str]	       =		load_json(__snake_case							)
																	_snake_case: Optional[int]	       =		{v: k for k, v in self.target_encoder.items()}
																	_snake_case: Tuple	       =		[]
											else:
																	_snake_case: Any	       =		{v: k for k, v in self.encoder.items()}
																	_snake_case: list	       =		[k for k in self.encoder if k.startswith('>>'							) and k.endswith('<<'							)]

											_snake_case: List[Any]	       =		source_lang
											_snake_case: List[str]	       =		target_lang
											_snake_case: List[str]	       =		[source_spm, target_spm]

											# load SentencePiece model for pre-processing
											_snake_case: Optional[Any]	       =		load_spm(__snake_case							,  self.sp_model_kwargs							)
											_snake_case: List[str]	       =		load_spm(__snake_case							,  self.sp_model_kwargs							)
											_snake_case: str	       =		self.spm_source
											_snake_case: int	       =		self.encoder

											# Multilingual target side: default to using first supported language code.

											self._setup_normalizer()


					def  SCREAMING_SNAKE_CASE_		(       self						:	str							):





											'''simple docstring'''





											try:
																	from sacremoses import MosesPunctNormalizer

																	_snake_case: List[Any]	       =		MosesPunctNormalizer(self.source_lang							).normalize
											except (ImportError, FileNotFoundError):
																	warnings.warn('Recommended: pip install sacremoses.'							)
																	_snake_case: List[Any]	       =		lambda __snake_case							: x


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[int]							,  __snake_case						:	str							):





											'''simple docstring'''





											return self.punc_normalizer(__snake_case							) if x else ""


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[int]							,  __snake_case						:	Tuple							):





											'''simple docstring'''





											return self.current_encoder.get(__snake_case							,  self.current_encoder[self.unk_token]							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[str]							,  __snake_case						:	str							):





											'''simple docstring'''





											_snake_case: List[Any]	       =		self.language_code_re.match(__snake_case							)
											_snake_case: list	       =		[match.group(0							)] if match else []
											return code, self.language_code_re.sub(''							,  __snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[int]							,  __snake_case						:	str							):





											'''simple docstring'''





											_snake_case       ,    _snake_case: Optional[Any]	       =		self.remove_language_code(__snake_case							)
											_snake_case: Union[str, Any]	       =		self.current_spm.encode(__snake_case							,  out_type=__snake_case							)
											return code + pieces


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							,  __snake_case						:	int							):





											'''simple docstring'''





											return self.decoder.get(__snake_case							,  self.unk_token							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							,  __snake_case						:	Tuple							,  **__snake_case						:	int							):





											'''simple docstring'''





											return super().batch_decode(__snake_case							,  **__snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							,  __snake_case						:	Any							,  **__snake_case						:	List[str]							):





											'''simple docstring'''





											return super().decode(__snake_case							,  **__snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							,  __snake_case						:	List[str]							):





											'''simple docstring'''





											_snake_case: Optional[int]	       =		self.spm_source if self._decode_use_source_tokenizer else self.spm_target
											_snake_case: List[Any]	       =		[]
											_snake_case: Any	       =		''
											for token in tokens:
																	# make sure that special tokens are not decoded using sentencepiece model
																	if token in self.all_special_tokens:
																							out_string += sp_model.decode_pieces(__snake_case							) + token + " "
																							_snake_case: Any	       =		[]
																	else:
																							current_sub_tokens.append(__snake_case							)
											out_string += sp_model.decode_pieces(__snake_case							)
											return out_string.strip()


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[Any]							,  __snake_case						:	Tuple							,  __snake_case						:	Union[str, Any]=None							):





											'''simple docstring'''





											if token_ids_a is None:
																	return token_ids_a + [self.eos_token_id]
											# We don't expect to process pairs, but leave the pair logic for API consistency
											return token_ids_a + token_ids_a + [self.eos_token_id]


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[Any]							):





											'''simple docstring'''





											_snake_case: List[Any]	       =		self.spm_source
											_snake_case: List[str]	       =		self.encoder


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											_snake_case: int	       =		self.spm_target
											if self.separate_vocabs:
																	_snake_case: Union[str, Any]	       =		self.target_encoder


					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[int]							):





											'''simple docstring'''





											return len(self.encoder							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							,  __snake_case						:	str							,  __snake_case						:	Optional[str] = None							):





											'''simple docstring'''





											if not os.path.isdir(__snake_case							):
																	logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''							)
																	return
											_snake_case: str	       =		[]

											if self.separate_vocabs:
																	_snake_case: Optional[int]	       =		os.path.join(
																	    __snake_case							,  (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab']							,  )
																	_snake_case: Optional[int]	       =		os.path.join(
																	    __snake_case							,  (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['target_vocab_file']							,  )
																	save_json(self.encoder							,  __snake_case							)
																	save_json(self.target_encoder							,  __snake_case							)
																	saved_files.append(__snake_case							)
																	saved_files.append(__snake_case							)
											else:
																	_snake_case: int	       =		os.path.join(
																	    __snake_case							,  (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab']							)
																	save_json(self.encoder							,  __snake_case							)
																	saved_files.append(__snake_case							)

											for spm_save_filename, spm_orig_path, spm_model in zip(
											    [VOCAB_FILES_NAMES['source_spm'], VOCAB_FILES_NAMES['target_spm']]							,  self.spm_files							,  [self.spm_source, self.spm_target]							,  ):
																	_snake_case: Any	       =		os.path.join(
																	    __snake_case							,  (filename_prefix + '-' if filename_prefix else '') + spm_save_filename							)
																	if os.path.abspath(__snake_case							) != os.path.abspath(__snake_case							) and os.path.isfile(__snake_case							):
																							copyfile(__snake_case							,  __snake_case							)
																							saved_files.append(__snake_case							)
																	elif not os.path.isfile(__snake_case							):
																							with open(__snake_case							,  'wb'							) as fi:
																													_snake_case: Any	       =		spm_model.serialized_model_proto()
																													fi.write(__snake_case							)
																							saved_files.append(__snake_case							)

											return tuple(__snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							):





											'''simple docstring'''





											return self.get_src_vocab()


					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							):





											'''simple docstring'''





											return dict(self.encoder							,  **self.added_tokens_encoder							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							):





											'''simple docstring'''





											return dict(self.target_encoder							,  **self.added_tokens_decoder							)


					def __getstate__(       self						:	Union[str, Any]							):





											'''simple docstring'''





											_snake_case: str	       =		self.__dict__.copy()
											state.update(
											    {k: None for k in ['spm_source', 'spm_target', 'current_spm', 'punc_normalizer', 'target_vocab_file']}							)
											return state


					def __setstate__(       self						:	str							,  __snake_case						:	Dict							):





											'''simple docstring'''





											_snake_case: str	       =		d

											# for backward compatibility
											if not hasattr(self							,  'sp_model_kwargs'							):
																	_snake_case: List[Any]	       =		{}

											_snake_case       ,    _snake_case: Dict	       =		(load_spm(__snake_case							,  self.sp_model_kwargs							) for f in self.spm_files)
											_snake_case: str	       =		self.spm_source
											self._setup_normalizer()


					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[int]							,  *__snake_case						:	Tuple							,  **__snake_case						:	Tuple							):





											'''simple docstring'''





											return 1


					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							,  __snake_case						:	Optional[int]							):





											'''simple docstring'''





											_snake_case: Dict	       =		set(self.all_special_ids							)  # call it once instead of inside list comp
											all_special_ids.remove(self.unk_token_id							)  # <unk> is only sometimes special
											return [1 if x in all_special_ids else 0 for x in seq]




					def  SCREAMING_SNAKE_CASE_		(       self						:	Optional[Any]							,  __snake_case						:	List							,  __snake_case						:	Optional[List] = None							,  __snake_case						:	bool = False							):





											'''simple docstring'''





											if already_has_special_tokens:
																	return self._special_token_mask(__snake_case							)
											elif token_ids_a is None:
																	return self._special_token_mask(__snake_case							) + [1]
											else:
																	return self._special_token_mask(token_ids_a + token_ids_a							) + [1]





def    lowercase_   (		lowercase__					,	lowercase__    ) ->sentencepiece.SentencePieceProcessor:
						_snake_case: Union[str, Any]	       =		sentencepiece.SentencePieceProcessor(**lowercase__    )
						spm.Load(lowercase__    )
						return spm





def    lowercase_   (		lowercase__					,	lowercase__    ) ->None:
						with open(lowercase__					,	'w'    ) as f:
												json.dump(lowercase__					,	lowercase__					,	indent=2    )





def    lowercase_   (		lowercase__    ) ->Union[Dict, List]:
						with open(lowercase__					,	'r'    ) as f:
												return json.load(lowercase__    )





