





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


_UpperCamelCase       = logging.get_logger(__name__)

_UpperCamelCase       = {
    '''source_spm''': '''source.spm''',
    '''target_spm''': '''target.spm''',
    '''vocab''': '''vocab.json''',
    '''target_vocab_file''': '''target_vocab.json''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

_UpperCamelCase       = {
    '''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'''
        )
    },
}

_UpperCamelCase       = {'''Helsinki-NLP/opus-mt-en-de''': 512}
_UpperCamelCase       = {}




class 				_lowerCamelCase (    a    ):



      """simple docstring"""






      UpperCAmelCase_						:      Union[str, Any]       =VOCAB_FILES_NAMES
      UpperCAmelCase_						:      List[str]       =PRETRAINED_VOCAB_FILES_MAP
      UpperCAmelCase_						:      Dict       =PRETRAINED_INIT_CONFIGURATION
      UpperCAmelCase_						:      Dict       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      UpperCAmelCase_						:      Optional[int]       =["input_ids", "attention_mask"]
      UpperCAmelCase_						:      Optional[int]       =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'''






        __snake_case   :				Tuple        =       {} 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}"""

        __snake_case   :				Dict        =       separate_vocabs
        __snake_case   :				List[str]        =       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:
          __snake_case   :				Optional[int]        =       load_json(UpperCAmelCase   )
          __snake_case   :				Optional[Any]        =       {v: k for k, v in self.target_encoder.items()}
          __snake_case   :				Dict        =       []
        else:
          __snake_case   :				List[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[str]        =       source_lang
        __snake_case   :				Union[str, Any]        =       target_lang
        __snake_case   :				int        =       [source_spm, target_spm]

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

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

        self._setup_normalizer()







      def 						UpperCAmelCase		( self   )       ->   List[str]:






        '''simple docstring'''






        try:
          from sacremoses import MosesPunctNormalizer

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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   str:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   Optional[Any]:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   int:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   List[str]:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   str:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase				,						**UpperCAmelCase   )       ->   Any:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase				,						**UpperCAmelCase   )       ->   Dict:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   str:






        '''simple docstring'''






        __snake_case   :				str        =       self.spm_source if self._decode_use_source_tokenizer else self.spm_target
        __snake_case   :				List[Any]        =       []
        __snake_case   :				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(UpperCAmelCase   ) + token + " "
            __snake_case   :				Tuple        =       []
          else:
            current_sub_tokens.append(UpperCAmelCase   )
        out_string += sp_model.decode_pieces(UpperCAmelCase   )
        return out_string.strip()







      def 						UpperCAmelCase		( 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 						UpperCAmelCase		( self   )       ->   str:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self   )       ->   List[Any]:






        '''simple docstring'''






        __snake_case   :				str        =       self.spm_target
        if self.separate_vocabs:
          __snake_case   :				Any        =       self.target_encoder







      @property
      def 						UpperCAmelCase		( self   )       ->   int:






        '''simple docstring'''






        return len(self.encoder   )







      def 						UpperCAmelCase		( 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
        __snake_case   :				List[Any]        =       []

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

        return tuple(UpperCAmelCase   )







      def 						UpperCAmelCase		( self   )       ->   Dict:






        '''simple docstring'''






        return self.get_src_vocab()







      def 						UpperCAmelCase		( self   )       ->   List[Any]:






        '''simple docstring'''






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







      def 						UpperCAmelCase		( self   )       ->   List[Any]:






        '''simple docstring'''






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







      def __getstate__( self   )       ->   Dict:






        '''simple docstring'''






        __snake_case   :				Union[str, Any]        =       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'''






        __snake_case   :				str        =       d

        # for backward compatibility
        if not hasattr(self				,						"sp_model_kwargs"   ):
          __snake_case   :				List[str]        =       {}

        __snake_case    ,   __snake_case   :				List[str]        =       (load_spm(UpperCAmelCase				,						self.sp_model_kwargs   ) for f in self.spm_files)
        __snake_case   :				Dict        =       self.spm_source
        self._setup_normalizer()







      def 						UpperCAmelCase		( self				,						*UpperCAmelCase				,						**UpperCAmelCase   )       ->   Optional[int]:






        '''simple docstring'''






        return 1







      def 						UpperCAmelCase		( self				,						UpperCAmelCase   )       ->   Dict:






        '''simple docstring'''






        __snake_case   :				List[str]        =       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 						UpperCAmelCase		( 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__(	lowercase       :    str   ,     lowercase       :    Dict[str, Any]				) ->    sentencepiece.SentencePieceProcessor:
  __snake_case   :				Tuple        =       sentencepiece.SentencePieceProcessor(**lowercase				)
  spm.Load(lowercase				)
  return spm







def       lowerCAmelCase__(	lowercase       :    Optional[Any]   ,     lowercase       :    str				) ->    None:
  with open(lowercase   ,     "w"				) as f:
    json.dump(lowercase   ,     lowercase   ,     indent=2				)







def       lowerCAmelCase__(	lowercase       :    str				) ->    Union[Dict, List]:
  with open(lowercase   ,     "r"				) as f:
    return json.load(lowercase				)








from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


_UpperCamelCase       = {
    '''configuration_deta''': ['''DETA_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''DetaConfig'''],
}

try:
     if not is_vision_available():
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     pass
else:
     _UpperCamelCase       = ['''DetaImageProcessor''']

try:
     if not is_torch_available():
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     pass
else:
     _UpperCamelCase       = [
         '''DETA_PRETRAINED_MODEL_ARCHIVE_LIST''',
         '''DetaForObjectDetection''',
         '''DetaModel''',
         '''DetaPreTrainedModel''',
     ]


if TYPE_CHECKING:
     from .configuration_deta import DETA_PRETRAINED_CONFIG_ARCHIVE_MAP, DetaConfig

     try:
          if not is_vision_available():
               raise OptionalDependencyNotAvailable()
     except OptionalDependencyNotAvailable:
          pass
     else:
          from .image_processing_deta import DetaImageProcessor

     try:
          if not is_torch_available():
               raise OptionalDependencyNotAvailable()
     except OptionalDependencyNotAvailable:
          pass
     else:
          from .modeling_deta import (
              DETA_PRETRAINED_MODEL_ARCHIVE_LIST,
              DetaForObjectDetection,
              DetaModel,
              DetaPreTrainedModel,
          )

else:
     import sys

     _UpperCamelCase       = _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)

