


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[int]          =							logging.get_logger(__name__)

_A		:					Dict          =							{
    "source_spm": "source.spm",
    "target_spm": "target.spm",
    "vocab": "vocab.json",
    "target_vocab_file": "target_vocab.json",
    "tokenizer_config_file": "tokenizer_config.json",
}

_A		:					int          =							{
    "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		:					Dict          =							{"Helsinki-NLP/opus-mt-en-de": 512}
_A		:					Union[str, Any]          =							{}



class 							a       (      _UpperCAmelCase ):
     UpperCAmelCase__			:	Dict       =						VOCAB_FILES_NAMES
     UpperCAmelCase__			:	str       =						PRETRAINED_VOCAB_FILES_MAP
     UpperCAmelCase__			:	Optional[int]       =						PRETRAINED_INIT_CONFIGURATION
     UpperCAmelCase__			:	Union[str, Any]       =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     UpperCAmelCase__			:	Dict       =						['''input_ids''', '''attention_mask''']
     UpperCAmelCase__			:	Tuple       =						re.compile(">>.+<<" )  # type: re.Pattern





     def __init__(     self       :					Tuple     ,						SCREAMING_SNAKE_CASE_       :					Optional[int]     ,						SCREAMING_SNAKE_CASE_       :					List[str]     ,						SCREAMING_SNAKE_CASE_       :					Dict     ,						SCREAMING_SNAKE_CASE_       :					Any=None     ,						SCREAMING_SNAKE_CASE_       :					Tuple=None     ,						SCREAMING_SNAKE_CASE_       :					List[str]=None     ,						SCREAMING_SNAKE_CASE_       :					List[Any]="<unk>"     ,						SCREAMING_SNAKE_CASE_       :					List[Any]="</s>"     ,						SCREAMING_SNAKE_CASE_       :					Optional[Any]="<pad>"     ,						SCREAMING_SNAKE_CASE_       :					Union[str, Any]=512     ,						SCREAMING_SNAKE_CASE_       :					Optional[Dict[str, Any]] = None     ,						SCREAMING_SNAKE_CASE_       :					Any=False     ,						**SCREAMING_SNAKE_CASE_       :					List[str]     ,						):
        __lowerCamelCase:					Optional[int]	 =						{} 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}'

        __lowerCamelCase:					Tuple	 =						separate_vocabs
        __lowerCamelCase:					Dict	 =						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:
           __lowerCamelCase:					int	 =						load_json(_snake_case				)
           __lowerCamelCase:					List[str]	 =						{v: k for k, v in self.target_encoder.items()}
           __lowerCamelCase:					Dict	 =						[]
        else:
           __lowerCamelCase:					Optional[Any]	 =						{v: k for k, v in self.encoder.items()}
           __lowerCamelCase:					int	 =						[k for k in self.encoder if k.startswith(""">>"""				) and k.endswith("""<<"""				)]

        __lowerCamelCase:					str	 =						source_lang
        __lowerCamelCase:					Union[str, Any]	 =						target_lang
        __lowerCamelCase:					str	 =						[source_spm, target_spm]

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

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

        self._setup_normalizer()





     def 			SCREAMING_SNAKE_CASE__							(     self       :					List[Any]				):


        try:
           from sacremoses import MosesPunctNormalizer

           __lowerCamelCase:					Union[str, Any]	 =						MosesPunctNormalizer(self.source_lang				).normalize
        except (ImportError, FileNotFoundError):
           warnings.warn("""Recommended: pip install sacremoses."""				)
           __lowerCamelCase:					List[Any]	 =						lambda SCREAMING_SNAKE_CASE_				: x





     def 			SCREAMING_SNAKE_CASE__							(     self       :					int     ,						SCREAMING_SNAKE_CASE_       :					str				):
        return self.punc_normalizer(_snake_case				) if x else ""





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Optional[int]     ,						SCREAMING_SNAKE_CASE_       :					Dict				):
        return self.current_encoder.get(_snake_case     ,						self.current_encoder[self.unk_token]				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Union[str, Any]     ,						SCREAMING_SNAKE_CASE_       :					str				):
        __lowerCamelCase:					str	 =						self.language_code_re.match(_snake_case				)
        __lowerCamelCase:					List[Any]	 =						[match.group(0				)] if match else []
        return code, self.language_code_re.sub(""""""     ,						_snake_case				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Tuple     ,						SCREAMING_SNAKE_CASE_       :					str				):
        __lowerCamelCase ,  __lowerCamelCase:					Any	 =						self.remove_language_code(_snake_case				)
        __lowerCamelCase:					Optional[Any]	 =						self.current_spm.encode(_snake_case     ,						out_type=_snake_case				)
        return code + pieces





     def 			SCREAMING_SNAKE_CASE__							(     self       :					str     ,						SCREAMING_SNAKE_CASE_       :					int				):
        return self.decoder.get(_snake_case     ,						self.unk_token				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					str     ,						SCREAMING_SNAKE_CASE_       :					Dict     ,						**SCREAMING_SNAKE_CASE_       :					Optional[Any]				):
        return super().batch_decode(_snake_case     ,						**_snake_case				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					List[Any]     ,						SCREAMING_SNAKE_CASE_       :					Optional[Any]     ,						**SCREAMING_SNAKE_CASE_       :					Tuple				):
        return super().decode(_snake_case     ,						**_snake_case				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					str     ,						SCREAMING_SNAKE_CASE_       :					List[str]				):
        __lowerCamelCase:					Union[str, Any]	 =						self.spm_source if self._decode_use_source_tokenizer else self.spm_target
        __lowerCamelCase:					Tuple	 =						[]
        __lowerCamelCase:					Tuple	 =						""""""
        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 + " "
              __lowerCamelCase:					Optional[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       :					List[Any]     ,						SCREAMING_SNAKE_CASE_       :					Union[str, Any]     ,						SCREAMING_SNAKE_CASE_       :					List[Any]=None				):


        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       :					int				):
        __lowerCamelCase:					int	 =						self.spm_source
        __lowerCamelCase:					Any	 =						self.encoder





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Dict				):
        __lowerCamelCase:					Optional[int]	 =						self.spm_target
        if self.separate_vocabs:
           __lowerCamelCase:					Dict	 =						self.target_encoder





     @property
     def 			SCREAMING_SNAKE_CASE__							(     self       :					Optional[int]				):
        return len(self.encoder				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Optional[int]     ,						SCREAMING_SNAKE_CASE_       :					str     ,						SCREAMING_SNAKE_CASE_       :					Optional[str] = None				):


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

        if self.separate_vocabs:
           __lowerCamelCase:					Tuple	 =						os.path.join(
               _snake_case     ,						(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab"""]     ,						)
           __lowerCamelCase:					Optional[Any]	 =						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:
           __lowerCamelCase:					List[Any]	 =						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]     ,						):
           __lowerCamelCase:					Tuple	 =						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:
                 __lowerCamelCase:					Optional[int]	 =						spm_model.serialized_model_proto()
                 fi.write(_snake_case				)
              saved_files.append(_snake_case				)

        return tuple(_snake_case				)





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Optional[int]				):
        return self.get_src_vocab()





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





     def 			SCREAMING_SNAKE_CASE__							(     self       :					List[str]				):
        return dict(self.target_encoder     ,						**self.added_tokens_decoder				)





     def __getstate__(     self       :					int				):
        __lowerCamelCase:					Tuple	 =						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       :					List[Any]     ,						SCREAMING_SNAKE_CASE_       :					Dict				):
        __lowerCamelCase:					List[str]	 =						d

        # for backward compatibility
        if not hasattr(self     ,						"""sp_model_kwargs"""				):
           __lowerCamelCase:					Dict	 =						{}

        __lowerCamelCase ,  __lowerCamelCase:					Optional[Any]	 =						(load_spm(_snake_case     ,						self.sp_model_kwargs				) for f in self.spm_files)
        __lowerCamelCase:					Tuple	 =						self.spm_source
        self._setup_normalizer()





     def 			SCREAMING_SNAKE_CASE__							(     self       :					List[str]     ,						*SCREAMING_SNAKE_CASE_       :					Dict     ,						**SCREAMING_SNAKE_CASE_       :					List[str]				):
        return 1





     def 			SCREAMING_SNAKE_CASE__							(     self       :					Optional[Any]     ,						SCREAMING_SNAKE_CASE_       :					Tuple				):
        __lowerCamelCase:					List[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 			SCREAMING_SNAKE_CASE__							(     self       :					Dict     ,						SCREAMING_SNAKE_CASE_       :					List     ,						SCREAMING_SNAKE_CASE_       :					Optional[List] = None     ,						SCREAMING_SNAKE_CASE_       :					bool = False				):


        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      __lowerCAmelCase    (				snake_case						:     str    ,   snake_case						:     Dict[str, Any]			)							->				sentencepiece.SentencePieceProcessor:
   __lowerCamelCase:					Tuple	 =						sentencepiece.SentencePieceProcessor(**__A			)
   spm.Load(__A			)
   return spm



def      __lowerCAmelCase    (				snake_case						:     Dict    ,   snake_case						:     str			)							->				None:


   with open(__A    ,   """w"""			) as f:
      json.dump(__A    ,   __A    ,   indent=2			)







def      __lowerCAmelCase    (				snake_case						:     str			)							->				Union[Dict, List]:


   with open(__A    ,   """r"""			) as f:
      return json.load(__A			)








import os
from typing import List, Union

import tensorflow as tf
from tensorflow_text import BertTokenizer as BertTokenizerLayer
from tensorflow_text import FastBertTokenizer, ShrinkLongestTrimmer, case_fold_utfa, combine_segments, pad_model_inputs

from .tokenization_bert import BertTokenizer







class __snake_case     (     tf.keras.layers.Layer						):







   def __init__( self						:		Optional[int]   ,		_snake_case						:		List   ,		_snake_case						:		bool   ,		_snake_case						:		int = None   ,		_snake_case						:		int = None   ,		_snake_case						:		int = None   ,		_snake_case						:		str = "longest"   ,		_snake_case						:		bool = True   ,		_snake_case						:		int = 512   ,		_snake_case						:		int = None   ,		_snake_case						:		bool = True   ,		_snake_case						:		bool = True   ,		_snake_case						:		bool = True   ,		**_snake_case						:		Any   ,		):

        """simple docstring"""


        super().__init__()
        if use_fast_bert_tokenizer:
             UpperCAmelCase_		=				FastBertTokenizer(
                 _snake_case   ,		token_out_type=tf.intaa   ,		lower_case_nfd_strip_accents=_snake_case   ,		**_snake_case)
        else:
             UpperCAmelCase_		=				tf.lookup.StaticVocabularyTable(
                 tf.lookup.KeyValueTensorInitializer(
                     keys=_snake_case   ,		key_dtype=tf.string   ,		values=tf.range(tf.size(_snake_case   ,		out_type=tf.intaa)   ,		dtype=tf.intaa)   ,		value_dtype=tf.intaa   ,		)   ,		num_oov_buckets=1   ,		)
             UpperCAmelCase_		=				BertTokenizerLayer(
                 _snake_case   ,		token_out_type=tf.intaa   ,		lower_case=_snake_case   ,		**_snake_case)

        UpperCAmelCase_		=				vocab_list
        UpperCAmelCase_		=				do_lower_case
        UpperCAmelCase_		=				cls_token_id or vocab_list.index('''[CLS]''')
        UpperCAmelCase_		=				sep_token_id or vocab_list.index('''[SEP]''')
        UpperCAmelCase_		=				pad_token_id or vocab_list.index('''[PAD]''')
        UpperCAmelCase_		=				ShrinkLongestTrimmer(max_length - 3   ,		axis=1)  # Allow room for special tokens
        UpperCAmelCase_		=				max_length
        UpperCAmelCase_		=				padding
        UpperCAmelCase_		=				truncation
        UpperCAmelCase_		=				pad_to_multiple_of
        UpperCAmelCase_		=				return_token_type_ids
        UpperCAmelCase_		=				return_attention_mask







   @classmethod
   def lowerCamelCase		( cls						:		Optional[Any]   ,		_snake_case						:		"PreTrainedTokenizerBase"   ,		**_snake_case						:		str):  # noqa: F821

        """simple docstring"""


        UpperCAmelCase_		=				kwargs.pop('''do_lower_case'''   ,		_snake_case)
        UpperCAmelCase_		=				tokenizer.do_lower_case if do_lower_case is None else do_lower_case
        UpperCAmelCase_		=				kwargs.pop('''cls_token_id'''   ,		_snake_case)
        UpperCAmelCase_		=				tokenizer.cls_token_id if cls_token_id is None else cls_token_id
        UpperCAmelCase_		=				kwargs.pop('''sep_token_id'''   ,		_snake_case)
        UpperCAmelCase_		=				tokenizer.sep_token_id if sep_token_id is None else sep_token_id
        UpperCAmelCase_		=				kwargs.pop('''pad_token_id'''   ,		_snake_case)
        UpperCAmelCase_		=				tokenizer.pad_token_id if pad_token_id is None else pad_token_id

        UpperCAmelCase_		=				tokenizer.get_vocab()
        UpperCAmelCase_		=				sorted([(wordpiece, idx) for wordpiece, idx in vocab.items()]   ,		key=lambda _snake_case: x[1])
        UpperCAmelCase_		=				[entry[0] for entry in vocab]
        return cls(
            vocab_list=_snake_case   ,		do_lower_case=_snake_case   ,		cls_token_id=_snake_case   ,		sep_token_id=_snake_case   ,		pad_token_id=_snake_case   ,		**_snake_case   ,		)







   @classmethod
   def lowerCamelCase		( cls						:		List[str]   ,		_snake_case						:		Union[str, os.PathLike]   ,		*_snake_case						:		Union[str, Any]   ,		**_snake_case						:		Union[str, Any]):

        """simple docstring"""


        try:
             UpperCAmelCase_		=				BertTokenizer.from_pretrained(_snake_case   ,		*_snake_case   ,		**_snake_case)
        except:  # noqa: E722
             from .tokenization_bert_fast import BertTokenizerFast

             UpperCAmelCase_		=				BertTokenizerFast.from_pretrained(_snake_case   ,		*_snake_case   ,		**_snake_case)
        return cls.from_tokenizer(_snake_case   ,		**_snake_case)







   def lowerCamelCase		( self						:		int   ,		_snake_case						:		str):

        """simple docstring"""


        if self.do_lower_case:
             UpperCAmelCase_		=				case_fold_utfa(_snake_case)
        UpperCAmelCase_		=				self.tf_tokenizer.tokenize(_snake_case)
        return tokens.merge_dims(1   ,		-1)







   def lowerCamelCase		( self						:		int   ,		_snake_case						:		str   ,		_snake_case						:		str=None   ,		_snake_case						:		Union[str, Any]=None   ,		_snake_case						:		str=None   ,		_snake_case						:		int=None   ,		_snake_case						:		Optional[int]=None   ,		_snake_case						:		Optional[Any]=None   ,		_snake_case						:		List[str]=None   ,		):

        """simple docstring"""


        if padding is None:
             UpperCAmelCase_		=				self.padding
        if padding not in ("longest", "max_length"):
             raise ValueError('''Padding must be either \'longest\' or \'max_length\'!''')
        if max_length is not None and text_pair is not None:
             # Because we have to instantiate a Trimmer to do it properly
             raise ValueError('''max_length cannot be overridden at call time when truncating paired texts!''')
        if max_length is None:
             UpperCAmelCase_		=				self.max_length
        if truncation is None:
             UpperCAmelCase_		=				self.truncation
        if pad_to_multiple_of is None:
             UpperCAmelCase_		=				self.pad_to_multiple_of
        if return_token_type_ids is None:
             UpperCAmelCase_		=				self.return_token_type_ids
        if return_attention_mask is None:
             UpperCAmelCase_		=				self.return_attention_mask
        if not isinstance(_snake_case   ,		tf.Tensor):
             UpperCAmelCase_		=				tf.convert_to_tensor(_snake_case)
        if text_pair is not None and not isinstance(_snake_case   ,		tf.Tensor):
             UpperCAmelCase_		=				tf.convert_to_tensor(_snake_case)
        if text_pair is not None:
             if text.shape.rank > 1:
                  raise ValueError('''text argument should not be multidimensional when a text pair is supplied!''')
             if text_pair.shape.rank > 1:
                  raise ValueError('''text_pair should not be multidimensional!''')
        if text.shape.rank == 2:
             UpperCAmelCase_	,			UpperCAmelCase_		=				text[:, 0], text[:, 1]
        UpperCAmelCase_		=				self.unpaired_tokenize(_snake_case)
        if text_pair is None:  # Unpaired text
             if truncation:
                  UpperCAmelCase_		=				text[:, : max_length - 2]  # Allow room for special tokens
             UpperCAmelCase_	,			UpperCAmelCase_		=				combine_segments(
                 (text,)   ,		start_of_sequence_id=self.cls_token_id   ,		end_of_segment_id=self.sep_token_id)
        else:  # Paired text
             UpperCAmelCase_		=				self.unpaired_tokenize(_snake_case)
             if truncation:
                  UpperCAmelCase_	,			UpperCAmelCase_		=				self.paired_trimmer.trim([text, text_pair])
             UpperCAmelCase_	,			UpperCAmelCase_		=				combine_segments(
                 (text, text_pair)   ,		start_of_sequence_id=self.cls_token_id   ,		end_of_segment_id=self.sep_token_id)
        if padding == "longest":
             UpperCAmelCase_		=				input_ids.bounding_shape(axis=1)
             if pad_to_multiple_of is not None:
                  # No ceiling division in tensorflow, so we negate floordiv instead
                  UpperCAmelCase_		=				pad_to_multiple_of * (-tf.math.floordiv(-pad_length   ,		_snake_case))
        else:
             UpperCAmelCase_		=				max_length

        UpperCAmelCase_	,			UpperCAmelCase_		=				pad_model_inputs(_snake_case   ,		max_seq_length=_snake_case   ,		pad_value=self.pad_token_id)
        UpperCAmelCase_		=				{'''input_ids''': input_ids}
        if return_attention_mask:
             UpperCAmelCase_		=				attention_mask
        if return_token_type_ids:
             UpperCAmelCase_	,			UpperCAmelCase_		=				pad_model_inputs(
                 _snake_case   ,		max_seq_length=_snake_case   ,		pad_value=self.pad_token_id)
             UpperCAmelCase_		=				token_type_ids
        return output







   def lowerCamelCase		( self						:		Optional[Any]):

        """simple docstring"""


        return {
            "vocab_list": self.vocab_list,
            "do_lower_case": self.do_lower_case,
            "cls_token_id": self.cls_token_id,
            "sep_token_id": self.sep_token_id,
            "pad_token_id": self.pad_token_id,
        }






