code: 
  
  
  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
  
  
  SCREAMING_SNAKE_CASE : int          =    logging.get_logger(__name__)
  
  SCREAMING_SNAKE_CASE : Dict          =    {
      "source_spm": "source.spm",
      "target_spm": "target.spm",
      "vocab": "vocab.json",
      "target_vocab_file": "target_vocab.json",
      "tokenizer_config_file": "tokenizer_config.json",
  }
  
  SCREAMING_SNAKE_CASE : Any          =    {
      "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"
          )
      },
  }
  
  SCREAMING_SNAKE_CASE : List[str]          =    {"Helsinki-NLP/opus-mt-en-de": 5_12}
  SCREAMING_SNAKE_CASE : str          =    {}
  
  
  
  
  
  class 					snake_case__  ( __A       ):
  	UpperCAmelCase   :							List[Any]         = VOCAB_FILES_NAMES
  	UpperCAmelCase   :							Optional[int]         = PRETRAINED_VOCAB_FILES_MAP
  	UpperCAmelCase   :							int         = PRETRAINED_INIT_CONFIGURATION
  	UpperCAmelCase   :							str         = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  	UpperCAmelCase   :							Union[str, Any]         = ["""input_ids""", """attention_mask"""]
  	UpperCAmelCase   :							Dict         = re.compile(""">>.+<<"""       )  # type: re.Pattern
  
  
  
  
  
  	def __init__(  self				,    UpperCamelCase_				,    UpperCamelCase_				,    UpperCamelCase_				,    UpperCamelCase_=None				,    UpperCamelCase_=None				,    UpperCamelCase_=None				,    UpperCamelCase_="<unk>"				,    UpperCamelCase_="</s>"				,    UpperCamelCase_="<pad>"				,    UpperCamelCase_=512				,    UpperCamelCase_ = None				,    UpperCamelCase_=False				,    **UpperCamelCase_				,    )		->			None:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   Union[str, Any]			=	{} 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=UpperCamelCase_				,    target_lang=UpperCamelCase_				,    unk_token=UpperCamelCase_				,    eos_token=UpperCamelCase_				,    pad_token=UpperCamelCase_				,    model_max_length=UpperCamelCase_				,    sp_model_kwargs=self.sp_model_kwargs				,    target_vocab_file=UpperCamelCase_				,    separate_vocabs=UpperCamelCase_				,    **UpperCamelCase_				,    )
  								assert Path(UpperCamelCase_							).exists(), f"""cannot find spm source {source_spm}"""
  
  								a_       :   int			=	separate_vocabs
  								a_       :   Any			=	load_json(UpperCamelCase_							)
  								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:
  															a_       :   int			=	load_json(UpperCamelCase_							)
  															a_       :   List[str]			=	{v: k for k, v in self.target_encoder.items()}
  															a_       :   Tuple			=	[]
  								else:
  															a_       :   Tuple			=	{v: k for k, v in self.encoder.items()}
  															a_       :   list			=	[k for k in self.encoder if k.startswith(""">>"""							) and k.endswith("""<<"""							)]
  
  								a_       :   str			=	source_lang
  								a_       :   Optional[int]			=	target_lang
  								a_       :   List[str]			=	[source_spm, target_spm]
  
  								# load SentencePiece model for pre-processing
  								a_       :   Dict			=	load_spm(UpperCamelCase_				,    self.sp_model_kwargs							)
  								a_       :   Tuple			=	load_spm(UpperCamelCase_				,    self.sp_model_kwargs							)
  								a_       :   List[Any]			=	self.spm_source
  								a_       :   Dict			=	self.encoder
  
  								# Multilingual target side: default to using first supported language code.
  
  								self._setup_normalizer()
  
  
  
  
  
  	def     A				(  self							)		->			Any:
  								"""simple docstring"""
  
  
  
  
  
  
  								try:
  															from sacremoses import MosesPunctNormalizer
  
  															a_       :   List[str]			=	MosesPunctNormalizer(self.source_lang							).normalize
  								except (ImportError, FileNotFoundError):
  															warnings.warn("""Recommended: pip install sacremoses."""							)
  															a_       :   Optional[int]			=	lambda UpperCamelCase_							: x
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			str:
  								"""simple docstring"""
  
  
  
  
  
  
  								return self.punc_normalizer(UpperCamelCase_							) if x else ""
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			Any:
  								"""simple docstring"""
  
  
  
  
  
  
  								return self.current_encoder.get(UpperCamelCase_				,    self.current_encoder[self.unk_token]							)
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			Optional[int]:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   int			=	self.language_code_re.match(UpperCamelCase_							)
  								a_       :   list			=	[match.group(0							)] if match else []
  								return code, self.language_code_re.sub(""""""				,    UpperCamelCase_							)
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			List[str]:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_		,				a_       :   Union[str, Any]			=	self.remove_language_code(UpperCamelCase_							)
  								a_       :   List[str]			=	self.current_spm.encode(UpperCamelCase_				,    out_type=UpperCamelCase_							)
  								return code + pieces
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			str:
  								"""simple docstring"""
  
  
  
  
  
  
  								return self.decoder.get(UpperCamelCase_				,    self.unk_token							)
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_				,    **UpperCamelCase_							)		->			List[Any]:
  								"""simple docstring"""
  
  
  
  
  
  
  								return super().batch_decode(UpperCamelCase_				,    **UpperCamelCase_							)
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_				,    **UpperCamelCase_							)		->			Tuple:
  								"""simple docstring"""
  
  
  
  
  
  
  								return super().decode(UpperCamelCase_				,    **UpperCamelCase_							)
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			str:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   List[str]			=	self.spm_source if self._decode_use_source_tokenizer else self.spm_target
  								a_       :   Optional[Any]			=	[]
  								a_       :   Optional[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(UpperCamelCase_							) + token + " "
  																						a_       :   List[Any]			=	[]
  															else:
  																						current_sub_tokens.append(UpperCamelCase_							)
  								out_string += sp_model.decode_pieces(UpperCamelCase_							)
  								return out_string.strip()
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_				,    UpperCamelCase_=None							)		->			List[int]:
  								"""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     A				(  self							)		->			int:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   List[Any]			=	self.spm_source
  								a_       :   Optional[int]			=	self.encoder
  
  
  
  
  
  	def     A				(  self							)		->			Any:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   int			=	self.spm_target
  								if self.separate_vocabs:
  															a_       :   Dict			=	self.target_encoder
  
  
  
  
  
  	@property
  	def     A				(  self							)		->			int:
  								"""simple docstring"""
  
  
  
  
  
  
  								return len(self.encoder							)
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_				,    UpperCamelCase_ = None							)		->			Tuple[str]:
  								"""simple docstring"""
  
  
  
  
  
  
  								if not os.path.isdir(UpperCamelCase_							):
  															logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""							)
  															return
  								a_       :   Tuple			=	[]
  
  								if self.separate_vocabs:
  															a_       :   int			=	os.path.join(
  															    UpperCamelCase_				,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab"""]				,    )
  															a_       :   int			=	os.path.join(
  															    UpperCamelCase_				,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""target_vocab_file"""]				,    )
  															save_json(self.encoder				,    UpperCamelCase_							)
  															save_json(self.target_encoder				,    UpperCamelCase_							)
  															saved_files.append(UpperCamelCase_							)
  															saved_files.append(UpperCamelCase_							)
  								else:
  															a_       :   Dict			=	os.path.join(
  															    UpperCamelCase_				,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab"""]							)
  															save_json(self.encoder				,    UpperCamelCase_							)
  															saved_files.append(UpperCamelCase_							)
  
  								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]				,    ):
  															a_       :   List[str]			=	os.path.join(
  															    UpperCamelCase_				,    (filename_prefix + """-""" if filename_prefix else """""") + spm_save_filename							)
  															if os.path.abspath(UpperCamelCase_							) != os.path.abspath(UpperCamelCase_							) and os.path.isfile(UpperCamelCase_							):
  																						copyfile(UpperCamelCase_				,    UpperCamelCase_							)
  																						saved_files.append(UpperCamelCase_							)
  															elif not os.path.isfile(UpperCamelCase_							):
  																						with open(UpperCamelCase_				,    """wb"""							) as fi:
  																													a_       :   str			=	spm_model.serialized_model_proto()
  																													fi.write(UpperCamelCase_							)
  																						saved_files.append(UpperCamelCase_							)
  
  								return tuple(UpperCamelCase_							)
  
  
  
  
  
  	def     A				(  self							)		->			Dict:
  								"""simple docstring"""
  
  
  
  
  
  
  								return self.get_src_vocab()
  
  
  
  
  
  	def     A				(  self							)		->			List[str]:
  								"""simple docstring"""
  
  
  
  
  
  
  								return dict(self.encoder				,    **self.added_tokens_encoder							)
  
  
  
  
  
  	def     A				(  self							)		->			int:
  								"""simple docstring"""
  
  
  
  
  
  
  								return dict(self.target_encoder				,    **self.added_tokens_decoder							)
  
  
  
  
  
  	def __getstate__(  self							)		->			Dict:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   Dict			=	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				,    UpperCamelCase_							)		->			None:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   Optional[int]			=	d
  
  								# for backward compatibility
  								if not hasattr(self				,    """sp_model_kwargs"""							):
  															a_       :   Union[str, Any]			=	{}
  
  								a_		,				a_       :   int			=	(load_spm(UpperCamelCase_				,    self.sp_model_kwargs							) for f in self.spm_files)
  								a_       :   List[str]			=	self.spm_source
  								self._setup_normalizer()
  
  
  
  
  
  	def     A				(  self				,    *UpperCamelCase_				,    **UpperCamelCase_							)		->			str:
  								"""simple docstring"""
  
  
  
  
  
  
  								return 1
  
  
  
  
  
  	def     A				(  self				,    UpperCamelCase_							)		->			List[Any]:
  								"""simple docstring"""
  
  
  
  
  
  
  								a_       :   Union[str, Any]			=	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     A				(  self				,    UpperCamelCase_				,    UpperCamelCase_ = None				,    UpperCamelCase_ = False							)		->			List[int]:
  								"""simple docstring"""
  
  
  
  
  
  
  								if already_has_special_tokens:
  															return self._special_token_mask(UpperCamelCase_							)
  								elif token_ids_a is None:
  															return self._special_token_mask(UpperCamelCase_							) + [1]
  								else:
  															return self._special_token_mask(token_ids_a + token_ids_a							) + [1]
  
  
  
  
  
  def 						_lowerCamelCase		(							SCREAMING_SNAKE_CASE_				: str	,	SCREAMING_SNAKE_CASE_				: Dict[str, Any]      ):
  
  
  
  
  
  
  							"""simple docstring"""
  
  
  
  							a_       :   Any			=	sentencepiece.SentencePieceProcessor(**SCREAMING_SNAKE_CASE_      )
  							spm.Load(SCREAMING_SNAKE_CASE_      )
  							return spm
  
  
  
  
  
  def 						_lowerCamelCase		(							SCREAMING_SNAKE_CASE_				: List[str]	,	SCREAMING_SNAKE_CASE_				: str      ):
  
  
  
  
  
  
  							"""simple docstring"""
  
  
  
  							with open(SCREAMING_SNAKE_CASE_	,	"""w"""      ) as f:
  														json.dump(SCREAMING_SNAKE_CASE_	,	SCREAMING_SNAKE_CASE_	,	indent=2      )
  
  
  
  
  
  def 						_lowerCamelCase		(							SCREAMING_SNAKE_CASE_				: str      ):
  
  
  
  
  
  
  							"""simple docstring"""
  
  
  
  							with open(SCREAMING_SNAKE_CASE_	,	"""r"""      ) as f:
  														return json.load(SCREAMING_SNAKE_CASE_      )
  
  
  
style_context: 
  
  
  def 						_lowerCamelCase		(							SCREAMING_SNAKE_CASE_				: list[int]      ):
  
  
  
  
  
  
  							"""simple docstring"""
  
  
  
  							a_       :   List[Any]			=	len(SCREAMING_SNAKE_CASE_      )
  							if n in (0, 1):
  														return 0
  							return sum(query * (n - i - 1) for i, query in enumerate(sorted(SCREAMING_SNAKE_CASE_      )      )      )
  
  
  if __name__ == "__main__":
  	import doctest
  
  	doctest.testmod()
  
  
  