





import json
import os
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


_snake_case           =			logging.get_logger(__name__)

_snake_case           =			"""▁"""

_snake_case           =			{
    """vocab_file""": """vocab.json""",
    """spm_file""": """sentencepiece.bpe.model""",
}

_snake_case           =			{
    """vocab_file""": {
        """facebook/s2t-small-librispeech-asr""": (
            """https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/vocab.json"""
        ),
    },
    """spm_file""": {
        """facebook/s2t-small-librispeech-asr""": (
            """https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/sentencepiece.bpe.model"""
        )
    },
}

_snake_case           =			{
    """facebook/s2t-small-librispeech-asr""": 1024,
}

_snake_case           =			["""pt""", """fr""", """ru""", """nl""", """ro""", """it""", """es""", """de"""]

_snake_case           =			{"""mustc""": MUSTC_LANGS}





class  lowerCAmelCase							(       lowercase_ ):
  __lowerCamelCase		=      VOCAB_FILES_NAMES
  __lowerCamelCase		=      PRETRAINED_VOCAB_FILES_MAP
  __lowerCamelCase		=      MAX_MODEL_INPUT_SIZES
  __lowerCamelCase		=      ['input_ids', 'attention_mask']

  __lowerCamelCase		=      []







  def __init__(      self  :List[str]   ,							_lowercase  :List[str]   ,							_lowercase  :Tuple   ,							_lowercase  :Optional[int]="<s>"   ,							_lowercase  :Dict="</s>"   ,							_lowercase  :Dict="<pad>"   ,							_lowercase  :Any="<unk>"   ,							_lowercase  :Dict=False   ,							_lowercase  :Optional[int]=False   ,							_lowercase  :Tuple=None   ,							_lowercase  :List[str]=None   ,							_lowercase  :Optional[Dict[str, Any]] = None   ,							**_lowercase  :Optional[Any]   ,							):






     '''simple docstring'''





     lowercase__			       =  {} if sp_model_kwargs is None else sp_model_kwargs

     super().__init__(
         bos_token=_lowercase   ,							eos_token=_lowercase   ,							unk_token=_lowercase   ,							pad_token=_lowercase   ,							do_upper_case=_lowercase   ,							do_lower_case=_lowercase   ,							tgt_lang=_lowercase   ,							lang_codes=_lowercase   ,							sp_model_kwargs=self.sp_model_kwargs   ,							**_lowercase   ,							)
     lowercase__			       =  do_upper_case
     lowercase__			       =  do_lower_case

     lowercase__			       =  load_json(_lowercase )
     lowercase__			       =  {v: k for k, v in self.encoder.items()}
     lowercase__			       =  spm_file
     lowercase__			       =  load_spm(_lowercase   ,							self.sp_model_kwargs )

     if lang_codes is not None:
        lowercase__			       =  lang_codes
        lowercase__			       =  LANGUAGES[lang_codes]
        lowercase__			       =  [f'''<lang:{lang}>''' for lang in self.langs]
        lowercase__			       =  {lang: self.sp_model.PieceToId(f'''<lang:{lang}>''' ) for lang in self.langs}

        lowercase__			       =  self.lang_tokens
        lowercase__			       =  tgt_lang if tgt_lang is not None else self.langs[0]

        self.set_tgt_lang_special_tokens(self._tgt_lang )
     else:
        lowercase__			       =  {}







  @property
  def   UpperCAmelCase					(      self  :List[Any] ):






     '''simple docstring'''





     return len(self.encoder )







  @property
  def   UpperCAmelCase					(      self  :Optional[Any] ):






     '''simple docstring'''





     return self._tgt_lang







  @tgt_lang.setter
  def   UpperCAmelCase					(      self  :str   ,							_lowercase  :Optional[int] ):






     '''simple docstring'''





     lowercase__			       =  new_tgt_lang
     self.set_tgt_lang_special_tokens(_lowercase )







  def   UpperCAmelCase					(      self  :List[str]   ,							_lowercase  :str ):






     '''simple docstring'''





     lowercase__			       =  self.lang_code_to_id[tgt_lang]
     lowercase__			       =  [lang_code_id]







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :str ):






     '''simple docstring'''





     return self.sp_model.encode(_lowercase   ,							out_type=_lowercase )







  def   UpperCAmelCase					(      self  :Dict   ,							_lowercase  :Any ):






     '''simple docstring'''





     return self.encoder.get(_lowercase   ,							self.encoder[self.unk_token] )







  def   UpperCAmelCase					(      self  :List[str]   ,							_lowercase  :int ):






     '''simple docstring'''





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







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :List[str] ):






     '''simple docstring'''





     lowercase__			       =  []
     lowercase__			       =  ""
     for token in tokens:
        # make sure that special tokens are not decoded using sentencepiece model
        if token in self.all_special_tokens:
           lowercase__			       =  self.sp_model.decode(_lowercase )
           out_string += (decoded.upper() if self.do_upper_case else decoded) + token + " "
           lowercase__			       =  []
        else:
           current_sub_tokens.append(_lowercase )
     lowercase__			       =  self.sp_model.decode(_lowercase )
     out_string += decoded.upper() if self.do_upper_case else decoded
     return out_string.strip()







  def   UpperCAmelCase					(      self  :str   ,							_lowercase  :Optional[Any]   ,							_lowercase  :Union[str, Any]=None ):






     '''simple docstring'''





     if token_ids_a is None:
        return self.prefix_tokens + token_ids_a + [self.eos_token_id]
     # We don't expect to process pairs, but leave the pair logic for API consistency
     return self.prefix_tokens + token_ids_a + token_ids_a + [self.eos_token_id]







  def   UpperCAmelCase					(      self  :int   ,							_lowercase  :List[int]   ,							_lowercase  :Optional[List[int]] = None   ,							_lowercase  :bool = False ):






     '''simple docstring'''





     if already_has_special_tokens:
        return super().get_special_tokens_mask(
            token_ids_a=_lowercase   ,							token_ids_a=_lowercase   ,							already_has_special_tokens=_lowercase )

     lowercase__			       =  [1] * len(self.prefix_tokens )
     lowercase__			       =  [1]
     if token_ids_a is None:
        return prefix_ones + ([0] * len(_lowercase )) + suffix_ones
     return prefix_ones + ([0] * len(_lowercase )) + ([0] * len(_lowercase )) + suffix_ones







  def   UpperCAmelCase					(      self  :Tuple ):






     '''simple docstring'''





     lowercase__			       =  self.encoder.copy()
     vocab.update(self.added_tokens_encoder )
     return vocab







  def __getstate__(      self  :Optional[int] ):






     '''simple docstring'''





     lowercase__			       =  self.__dict__.copy()
     lowercase__			       =  None
     return state







  def __setstate__(      self  :Tuple   ,							_lowercase  :Dict ):






     '''simple docstring'''





     lowercase__			       =  d

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

     lowercase__			       =  load_spm(self.spm_file   ,							self.sp_model_kwargs )







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :str   ,							_lowercase  :Optional[str] = None ):






     '''simple docstring'''





     lowercase__			       =  Path(_lowercase )
     assert save_dir.is_dir(), f'''{save_directory} should be a directory'''
     lowercase__			       =  save_dir / (
         (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["vocab_file"]
     )
     lowercase__			       =  save_dir / (
         (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["spm_file"]
     )

     save_json(self.encoder   ,							_lowercase )

     if os.path.abspath(self.spm_file ) != os.path.abspath(_lowercase ) and os.path.isfile(self.spm_file ):
        copyfile(self.spm_file   ,							_lowercase )
     elif not os.path.isfile(self.spm_file ):
        with open(_lowercase   ,							"wb" ) as fi:
           lowercase__			       =  self.sp_model.serialized_model_proto()
           fi.write(_lowercase )

     return (str(_lowercase ), str(_lowercase ))


def _A       (   __magic_name__  , __magic_name__	):
   lowercase__			       =  sentencepiece.SentencePieceProcessor(**__magic_name__	)
   spm.Load(str(__magic_name__	)	)
   return spm


def _A       (   __magic_name__	):
   with open(__magic_name__  , "r"	) as f:
      return json.load(__magic_name__	)







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













import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
  from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
  import torch_xla.core.xla_model as xm


def _A       (   __magic_name__	):
   if is_torch_version("<"  , "2.0.0"	) or not hasattr(__magic_name__  , "_dynamo"	):
      return False
   return isinstance(__magic_name__  , torch._dynamo.eval_frame.OptimizedModule	)


def _A       (   __magic_name__  , __magic_name__ = True	):
   lowercase__			       =  (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

   lowercase__			       =  is_compiled_module(__magic_name__	)
   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  model._orig_mod

   if is_deepspeed_available():
      options += (DeepSpeedEngine,)

   while isinstance(__magic_name__  , __magic_name__	):
      lowercase__			       =  model.module

   if not keep_fpaa_wrapper:
      lowercase__			       =  getattr(__magic_name__  , "forward"	)
      lowercase__			       =  model.__dict__.pop("_original_forward"  , __magic_name__	)
      if original_forward is not None:
         while hasattr(__magic_name__  , "__wrapped__"	):
            lowercase__			       =  forward.__wrapped__
            if forward == original_forward:
               break
         lowercase__			       =  forward
      if getattr(__magic_name__  , "_converted_to_transformer_engine"  , __magic_name__	):
         convert_model(__magic_name__  , to_transformer_engine=__magic_name__	)

   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  compiled_model

   return model


def _A       (   ):
   PartialState().wait_for_everyone()


def _A       (   __magic_name__  , __magic_name__	):
   if PartialState().distributed_type == DistributedType.TPU:
      xm.save(__magic_name__  , __magic_name__	)
   elif PartialState().local_process_index == 0:
      torch.save(__magic_name__  , __magic_name__	)


@contextmanager
def _A       (   **__magic_name__	):
   for key, value in kwargs.items():
      lowercase__			       =  str(__magic_name__	)

   yield

   for key in kwargs:
      if key.upper() in os.environ:
         del os.environ[key.upper()]


def _A       (   __magic_name__	):
   if not hasattr(__magic_name__  , "__qualname__"	) and not hasattr(__magic_name__  , "__name__"	):
      lowercase__			       =  getattr(__magic_name__  , "__class__"  , __magic_name__	)
   if hasattr(__magic_name__  , "__qualname__"	):
      return obj.__qualname__
   if hasattr(__magic_name__  , "__name__"	):
      return obj.__name__
   return str(__magic_name__	)


def _A       (   __magic_name__  , __magic_name__	):
   for key, value in source.items():
      if isinstance(__magic_name__  , __magic_name__	):
         lowercase__			       =  destination.setdefault(__magic_name__  , {}	)
         merge_dicts(__magic_name__  , __magic_name__	)
      else:
         lowercase__			       =  value

   return destination







def _A       (   __magic_name__ = None	):
   if port is None:
      lowercase__			       =  2_9500
   with socket.socket(socket.AF_INET  , socket.SOCK_STREAM	) as s:
      return s.connect_ex(("localhost", port)	) == 0






