


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

SCREAMING_SNAKE_CASE	:Optional[int]  		=     {
    '''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	:List[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	:str  		=     {'''Helsinki-NLP/opus-mt-en-de''': 5_12}
SCREAMING_SNAKE_CASE	:Optional[int]  		=     {}


class     __lowerCAmelCase     ( a					):






  """simple docstring"""





  _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       :       int    ,				_lowerCAmelCase       :       Any    ,				_lowerCAmelCase       :       Optional[int]    ,				_lowerCAmelCase       :       List[Any]    ,				_lowerCAmelCase       :       Dict=None    ,				_lowerCAmelCase       :       Dict=None    ,				_lowerCAmelCase       :       Dict=None    ,				_lowerCAmelCase       :       Optional[Any]="<unk>"    ,				_lowerCAmelCase       :       str="</s>"    ,				_lowerCAmelCase       :       Optional[int]="<pad>"    ,				_lowerCAmelCase       :       Optional[Any]=5_1_2    ,				_lowerCAmelCase       :       Optional[Dict[str, Any]] = None    ,				_lowerCAmelCase       :       Tuple=False    ,				**_lowerCAmelCase       :       Any    ,				)   -> None:




     """simple docstring"""
     snake_case_					    =					{} 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}'''

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

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

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

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

     self._setup_normalizer()



  def 					lowerCAmelCase__  (     self       :       int  )   -> Tuple:




     """simple docstring"""
     try:
        from sacremoses import MosesPunctNormalizer

        snake_case_					    =					MosesPunctNormalizer(self.source_lang  ).normalize
     except (ImportError, FileNotFoundError):
        warnings.warn("Recommended: pip install sacremoses."  )
        snake_case_					    =					lambda _lowerCAmelCase  : x



  def 					lowerCAmelCase__  (     self       :       Union[str, Any]    ,				_lowerCAmelCase       :       str  )   -> str:




     """simple docstring"""
     return self.punc_normalizer(_lowerCAmelCase  ) if x else ""



  def 					lowerCAmelCase__  (     self       :       Optional[int]    ,				_lowerCAmelCase       :       List[str]  )   -> Optional[int]:




     """simple docstring"""
     return self.current_encoder.get(_lowerCAmelCase    ,				self.current_encoder[self.unk_token]  )



  def 					lowerCAmelCase__  (     self       :       Any    ,				_lowerCAmelCase       :       str  )   -> Union[str, Any]:




     """simple docstring"""
     snake_case_					    =					self.language_code_re.match(_lowerCAmelCase  )
     snake_case_					    =					[match.group(0  )] if match else []
     return code, self.language_code_re.sub(""    ,				_lowerCAmelCase  )



  def 					lowerCAmelCase__  (     self       :       Optional[Any]    ,				_lowerCAmelCase       :       str  )   -> List[str]:




     """simple docstring"""
     snake_case_    ,    snake_case_					    =					self.remove_language_code(_lowerCAmelCase  )
     snake_case_					    =					self.current_spm.encode(_lowerCAmelCase    ,				out_type=_lowerCAmelCase  )
     return code + pieces



  def 					lowerCAmelCase__  (     self       :       int    ,				_lowerCAmelCase       :       int  )   -> str:




     """simple docstring"""
     return self.decoder.get(_lowerCAmelCase    ,				self.unk_token  )



  def 					lowerCAmelCase__  (     self       :       str    ,				_lowerCAmelCase       :       List[str]    ,				**_lowerCAmelCase       :       Union[str, Any]  )   -> int:




     """simple docstring"""
     return super().batch_decode(_lowerCAmelCase    ,				**_lowerCAmelCase  )



  def 					lowerCAmelCase__  (     self       :       Optional[Any]    ,				_lowerCAmelCase       :       int    ,				**_lowerCAmelCase       :       List[str]  )   -> List[Any]:




     """simple docstring"""
     return super().decode(_lowerCAmelCase    ,				**_lowerCAmelCase  )



  def 					lowerCAmelCase__  (     self       :       Optional[Any]    ,				_lowerCAmelCase       :       List[str]  )   -> str:




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



  def 					lowerCAmelCase__  (     self       :       List[str]    ,				_lowerCAmelCase       :       List[Any]    ,				_lowerCAmelCase       :       Dict=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 					lowerCAmelCase__  (     self       :       Dict  )   -> Tuple:




     """simple docstring"""
     snake_case_					    =					self.spm_source
     snake_case_					    =					self.encoder



  def 					lowerCAmelCase__  (     self       :       List[str]  )   -> Tuple:




     """simple docstring"""
     snake_case_					    =					self.spm_target
     if self.separate_vocabs:
        snake_case_					    =					self.target_encoder



  @property
  def 					lowerCAmelCase__  (     self       :       List[Any]  )   -> int:




     """simple docstring"""
     return len(self.encoder  )



  def 					lowerCAmelCase__  (     self       :       Optional[int]    ,				_lowerCAmelCase       :       str    ,				_lowerCAmelCase       :       Optional[str] = None  )   -> Tuple[str]:




     """simple docstring"""
     if not os.path.isdir(_lowerCAmelCase  ):
        logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
        return
     snake_case_					    =					[]

     if self.separate_vocabs:
        snake_case_					    =					os.path.join(
            _lowerCAmelCase    ,				(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab"]    ,				)
        snake_case_					    =					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:
        snake_case_					    =					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]    ,				):
        snake_case_					    =					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:
              snake_case_					    =					spm_model.serialized_model_proto()
              fi.write(_lowerCAmelCase  )
           saved_files.append(_lowerCAmelCase  )

     return tuple(_lowerCAmelCase  )



  def 					lowerCAmelCase__  (     self       :       Dict  )   -> Dict:




     """simple docstring"""
     return self.get_src_vocab()



  def 					lowerCAmelCase__  (     self       :       Tuple  )   -> str:




     """simple docstring"""
     return dict(self.encoder    ,				**self.added_tokens_encoder  )



  def 					lowerCAmelCase__  (     self       :       str  )   -> int:




     """simple docstring"""
     return dict(self.target_encoder    ,				**self.added_tokens_decoder  )



  def __getstate__(     self       :       int  )   -> Dict:




     """simple docstring"""
     snake_case_					    =					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       :       Optional[Any]    ,				_lowerCAmelCase       :       Dict  )   -> None:




     """simple docstring"""
     snake_case_					    =					d

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

     snake_case_    ,    snake_case_					    =					(load_spm(_lowerCAmelCase    ,				self.sp_model_kwargs  ) for f in self.spm_files)
     snake_case_					    =					self.spm_source
     self._setup_normalizer()



  def 					lowerCAmelCase__  (     self       :       Tuple    ,				*_lowerCAmelCase       :       Union[str, Any]    ,				**_lowerCAmelCase       :       Optional[int]  )   -> List[Any]:




     """simple docstring"""
     return 1



  def 					lowerCAmelCase__  (     self       :       Any    ,				_lowerCAmelCase       :       Any  )   -> Union[str, Any]:




     """simple docstring"""
     snake_case_					    =					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 					lowerCAmelCase__  (     self       :       Optional[int]    ,				_lowerCAmelCase       :       List    ,				_lowerCAmelCase       :       Optional[List] = None    ,				_lowerCAmelCase       :       bool = False  )   -> List[int]:




     """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 			_lowerCAmelCase						(							lowerCAmelCase_							:str		,						lowerCAmelCase_							:Dict[str, Any]							)->sentencepiece.SentencePieceProcessor:
   '''simple docstring'''






   snake_case_					    =					sentencepiece.SentencePieceProcessor(**lowerCAmelCase_							)
   spm.Load(lowerCAmelCase_							)
   return spm



def 			_lowerCAmelCase						(							lowerCAmelCase_							:Optional[Any]		,						lowerCAmelCase_							:str							)->None:
   '''simple docstring'''






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



def 			_lowerCAmelCase						(							lowerCAmelCase_							:str							)->Union[Dict, List]:
   '''simple docstring'''






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








from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_speech_available,
    is_torch_available,
)


SCREAMING_SNAKE_CASE	:List[Any]  		=     {
    '''configuration_speech_to_text_2''': ['''SPEECH_TO_TEXT_2_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''Speech2Text2Config'''],
    '''processing_speech_to_text_2''': ['''Speech2Text2Processor'''],
    '''tokenization_speech_to_text_2''': ['''Speech2Text2Tokenizer'''],
}


try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   SCREAMING_SNAKE_CASE	:int  		=     [
       '''SPEECH_TO_TEXT_2_PRETRAINED_MODEL_ARCHIVE_LIST''',
       '''Speech2Text2ForCausalLM''',
       '''Speech2Text2PreTrainedModel''',
   ]


if TYPE_CHECKING:
   from .configuration_speech_to_text_a import SPEECH_TO_TEXT_2_PRETRAINED_CONFIG_ARCHIVE_MAP, SpeechaTextaConfig
   from .processing_speech_to_text_a import SpeechaTextaProcessor
   from .tokenization_speech_to_text_a import SpeechaTextaTokenizer

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_speech_to_text_a import (
          SPEECH_TO_TEXT_2_PRETRAINED_MODEL_ARCHIVE_LIST,
          SpeechaTextaForCausalLM,
          SpeechaTextaPreTrainedModel,
      )

else:
   import sys

   SCREAMING_SNAKE_CASE	:str  		=     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




