
"""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


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

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

UpperCamelCase      :							Union[str, 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"
        )
    },
}

UpperCamelCase      :							Any          =					{"Helsinki-NLP/opus-mt-en-de": 5_1_2}
UpperCamelCase      :							str          =					{}

class __lowerCAmelCase						( lowerCAmelCase_     ):
     lowercase		  =		VOCAB_FILES_NAMES
     lowercase		  =		PRETRAINED_VOCAB_FILES_MAP
     lowercase		  =		PRETRAINED_INIT_CONFIGURATION
     lowercase		  =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     lowercase		  =		["""input_ids""", """attention_mask"""]
     lowercase		  =		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      ,	):

      '''simple docstring'''




      __UpperCamelCase							    =			{} 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=lowerCamelCase_      ,	target_lang=lowerCamelCase_      ,	unk_token=lowerCamelCase_      ,	eos_token=lowerCamelCase_      ,	pad_token=lowerCamelCase_      ,	model_max_length=lowerCamelCase_      ,	sp_model_kwargs=self.sp_model_kwargs      ,	target_vocab_file=lowerCamelCase_      ,	separate_vocabs=lowerCamelCase_      ,	**lowerCamelCase_      ,	)
      assert Path(lowerCamelCase_ ).exists(), F'cannot find spm source {source_spm}'

      __UpperCamelCase							    =			separate_vocabs
      __UpperCamelCase							    =			load_json(lowerCamelCase_ )
      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:
       __UpperCamelCase							    =			load_json(lowerCamelCase_ )
       __UpperCamelCase							    =			{v: k for k, v in self.target_encoder.items()}
       __UpperCamelCase							    =			[]
      else:
       __UpperCamelCase							    =			{v: k for k, v in self.encoder.items()}
       __UpperCamelCase							    =			[k for k in self.encoder if k.startswith('>>' ) and k.endswith('<<' )]

      __UpperCamelCase							    =			source_lang
      __UpperCamelCase							    =			target_lang
      __UpperCamelCase							    =			[source_spm, target_spm]

      # load SentencePiece model for pre-processing
      __UpperCamelCase							    =			load_spm(lowerCamelCase_      ,	self.sp_model_kwargs )
      __UpperCamelCase							    =			load_spm(lowerCamelCase_      ,	self.sp_model_kwargs )
      __UpperCamelCase							    =			self.spm_source
      __UpperCamelCase							    =			self.encoder

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

      self._setup_normalizer()





     def 		UpperCAmelCase   (      self ):

      '''simple docstring'''




      try:
       from sacremoses import MosesPunctNormalizer

       __UpperCamelCase							    =			MosesPunctNormalizer(self.source_lang ).normalize
      except (ImportError, FileNotFoundError):
       warnings.warn('Recommended: pip install sacremoses.' )
       __UpperCamelCase							    =			lambda __UpperCAmelCase : x





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




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





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




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





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




      __UpperCamelCase							    =			self.language_code_re.match(lowerCamelCase_ )
      __UpperCamelCase							    =			[match.group(0 )] if match else []
      return code, self.language_code_re.sub(''      ,	lowerCamelCase_ )





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




      __UpperCamelCase							    =			self.remove_language_code(lowerCamelCase_ )
      __UpperCamelCase							    =			self.current_spm.encode(lowerCamelCase_      ,	out_type=lowerCamelCase_ )
      return code + pieces





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




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





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase      ,	**__UpperCAmelCase ):

      '''simple docstring'''




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





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase      ,	**__UpperCAmelCase ):

      '''simple docstring'''




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





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




      __UpperCamelCase							    =			self.spm_source if self._decode_use_source_tokenizer else self.spm_target
      __UpperCamelCase							    =			[]
      __UpperCamelCase							    =			""""""
      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(lowerCamelCase_ ) + token + " "
        __UpperCamelCase							    =			[]
       else:
        current_sub_tokens.append(lowerCamelCase_ )
      out_string += sp_model.decode_pieces(lowerCamelCase_ )
      return out_string.strip()





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase      ,	__UpperCAmelCase=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 		UpperCAmelCase   (      self ):

      '''simple docstring'''




      __UpperCamelCase							    =			self.spm_source
      __UpperCamelCase							    =			self.encoder





     def 		UpperCAmelCase   (      self ):

      '''simple docstring'''




      __UpperCamelCase							    =			self.spm_target
      if self.separate_vocabs:
       __UpperCamelCase							    =			self.target_encoder





     @property
     def 		UpperCAmelCase   (      self ):

      '''simple docstring'''




      return len(self.encoder )





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase      ,	__UpperCAmelCase = None ):

      '''simple docstring'''




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

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

      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]      ,	):
       __UpperCamelCase							    =			os.path.join(
           lowerCamelCase_      ,	(filename_prefix + '-' if filename_prefix else '') + spm_save_filename )
       if os.path.abspath(lowerCamelCase_ ) != os.path.abspath(lowerCamelCase_ ) and os.path.isfile(lowerCamelCase_ ):
        copyfile(lowerCamelCase_      ,	lowerCamelCase_ )
        saved_files.append(lowerCamelCase_ )
       elif not os.path.isfile(lowerCamelCase_ ):
        with open(lowerCamelCase_      ,	'wb' ) as fi:
         __UpperCamelCase							    =			spm_model.serialized_model_proto()
         fi.write(lowerCamelCase_ )
        saved_files.append(lowerCamelCase_ )

      return tuple(lowerCamelCase_ )





     def 		UpperCAmelCase   (      self ):

      '''simple docstring'''




      return self.get_src_vocab()





     def 		UpperCAmelCase   (      self ):

      '''simple docstring'''




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





     def 		UpperCAmelCase   (      self ):

      '''simple docstring'''




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





     def __getstate__(      self ):

      '''simple docstring'''




      __UpperCamelCase							    =			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 ):

      '''simple docstring'''




      __UpperCamelCase							    =			d

      # for backward compatibility
      if not hasattr(self      ,	'sp_model_kwargs' ):
       __UpperCamelCase							    =			{}

      __UpperCamelCase							    =			(load_spm(lowerCamelCase_      ,	self.sp_model_kwargs ) for f in self.spm_files)
      __UpperCamelCase							    =			self.spm_source
      self._setup_normalizer()





     def 		UpperCAmelCase   (      self      ,	*__UpperCAmelCase      ,	**__UpperCAmelCase ):

      '''simple docstring'''




      return 1





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




      __UpperCamelCase							    =			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 ):

      '''simple docstring'''




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






def 				A		(							snake_case :str							,   snake_case :List[str]						) ->      List[str]:
 __UpperCamelCase							    =			sentencepiece.SentencePieceProcessor(**__UpperCamelCase						)
 spm.Load(__UpperCamelCase						)
 return spm






def 				A		(							snake_case :Any							,   snake_case :int						) ->      int:
 with open(__UpperCamelCase							,   'w'						) as f:
  json.dump(__UpperCamelCase							,   __UpperCamelCase							,   indent=2						)






def 				A		(							snake_case :int						) ->      str:
 with open(__UpperCamelCase							,   'r'						) as f:
  return json.load(__UpperCamelCase						)







# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

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


__lowerCamelCase   	=  {
    '''configuration_tvlt''': ['''TVLT_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''TvltConfig'''],
    '''feature_extraction_tvlt''': ['''TvltFeatureExtractor'''],
    '''processing_tvlt''': ['''TvltProcessor'''],
}

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       __lowerCamelCase   	=  [
           '''TVLT_PRETRAINED_MODEL_ARCHIVE_LIST''',
           '''TvltModel''',
           '''TvltForPreTraining''',
           '''TvltForAudioVisualClassification''',
           '''TvltPreTrainedModel''',
       ]

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


if TYPE_CHECKING:
       from .configuration_tvlt import TVLT_PRETRAINED_CONFIG_ARCHIVE_MAP, TvltConfig
       from .processing_tvlt import TvltProcessor
       from .feature_extraction_tvlt import TvltFeatureExtractor

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_tvlt import (
                  TVLT_PRETRAINED_MODEL_ARCHIVE_LIST,
                  TvltForAudioVisualClassification,
                  TvltForPreTraining,
                  TvltModel,
                  TvltPreTrainedModel,
              )

       try:
              if not is_vision_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .image_processing_tvlt import TvltImageProcessor


else:
       import sys

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



