





'''simple docstring'''







import os
from shutil import copyfile
from typing import List, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
   from .tokenization_nllb import NllbTokenizer
else:
   _lowercase       						=			None


_lowercase       						=			logging.get_logger(__name__)


_lowercase       						=			{'vocab_file': 'sentencepiece.bpe.model', 'tokenizer_file': 'tokenizer.json'}

_lowercase       						=			{
    'vocab_file': {
        'facebook/nllb-200-distilled-600M': (
            'https://huggingface.co/facebook/nllb-200-distilled-600M/resolve/main/sentencepiece.bpe.model'
        ),
    },
    'tokenizer_file': {
        'facebook/nllb-200-distilled-600M': (
            'https://huggingface.co/facebook/nllb-200-distilled-600M/resolve/main/tokenizer.json'
        ),
    },
}

_lowercase       						=			{
    'facebook/nllb-large-en-ro': 1_024,
    'facebook/nllb-200-distilled-600M': 1_024,
}

# fmt: off
_lowercase       						=			['ace_Arab', 'ace_Latn', 'acm_Arab', 'acq_Arab', 'aeb_Arab', 'afr_Latn', 'ajp_Arab', 'aka_Latn', 'amh_Ethi', 'apc_Arab', 'arb_Arab', 'ars_Arab', 'ary_Arab', 'arz_Arab', 'asm_Beng', 'ast_Latn', 'awa_Deva', 'ayr_Latn', 'azb_Arab', 'azj_Latn', 'bak_Cyrl', 'bam_Latn', 'ban_Latn', 'bel_Cyrl', 'bem_Latn', 'ben_Beng', 'bho_Deva', 'bjn_Arab', 'bjn_Latn', 'bod_Tibt', 'bos_Latn', 'bug_Latn', 'bul_Cyrl', 'cat_Latn', 'ceb_Latn', 'ces_Latn', 'cjk_Latn', 'ckb_Arab', 'crh_Latn', 'cym_Latn', 'dan_Latn', 'deu_Latn', 'dik_Latn', 'dyu_Latn', 'dzo_Tibt', 'ell_Grek', 'eng_Latn', 'epo_Latn', 'est_Latn', 'eus_Latn', 'ewe_Latn', 'fao_Latn', 'pes_Arab', 'fij_Latn', 'fin_Latn', 'fon_Latn', 'fra_Latn', 'fur_Latn', 'fuv_Latn', 'gla_Latn', 'gle_Latn', 'glg_Latn', 'grn_Latn', 'guj_Gujr', 'hat_Latn', 'hau_Latn', 'heb_Hebr', 'hin_Deva', 'hne_Deva', 'hrv_Latn', 'hun_Latn', 'hye_Armn', 'ibo_Latn', 'ilo_Latn', 'ind_Latn', 'isl_Latn', 'ita_Latn', 'jav_Latn', 'jpn_Jpan', 'kab_Latn', 'kac_Latn', 'kam_Latn', 'kan_Knda', 'kas_Arab', 'kas_Deva', 'kat_Geor', 'knc_Arab', 'knc_Latn', 'kaz_Cyrl', 'kbp_Latn', 'kea_Latn', 'khm_Khmr', 'kik_Latn', 'kin_Latn', 'kir_Cyrl', 'kmb_Latn', 'kon_Latn', 'kor_Hang', 'kmr_Latn', 'lao_Laoo', 'lvs_Latn', 'lij_Latn', 'lim_Latn', 'lin_Latn', 'lit_Latn', 'lmo_Latn', 'ltg_Latn', 'ltz_Latn', 'lua_Latn', 'lug_Latn', 'luo_Latn', 'lus_Latn', 'mag_Deva', 'mai_Deva', 'mal_Mlym', 'mar_Deva', 'min_Latn', 'mkd_Cyrl', 'plt_Latn', 'mlt_Latn', 'mni_Beng', 'khk_Cyrl', 'mos_Latn', 'mri_Latn', 'zsm_Latn', 'mya_Mymr', 'nld_Latn', 'nno_Latn', 'nob_Latn', 'npi_Deva', 'nso_Latn', 'nus_Latn', 'nya_Latn', 'oci_Latn', 'gaz_Latn', 'ory_Orya', 'pag_Latn', 'pan_Guru', 'pap_Latn', 'pol_Latn', 'por_Latn', 'prs_Arab', 'pbt_Arab', 'quy_Latn', 'ron_Latn', 'run_Latn', 'rus_Cyrl', 'sag_Latn', 'san_Deva', 'sat_Beng', 'scn_Latn', 'shn_Mymr', 'sin_Sinh', 'slk_Latn', 'slv_Latn', 'smo_Latn', 'sna_Latn', 'snd_Arab', 'som_Latn', 'sot_Latn', 'spa_Latn', 'als_Latn', 'srd_Latn', 'srp_Cyrl', 'ssw_Latn', 'sun_Latn', 'swe_Latn', 'swh_Latn', 'szl_Latn', 'tam_Taml', 'tat_Cyrl', 'tel_Telu', 'tgk_Cyrl', 'tgl_Latn', 'tha_Thai', 'tir_Ethi', 'taq_Latn', 'taq_Tfng', 'tpi_Latn', 'tsn_Latn', 'tso_Latn', 'tuk_Latn', 'tum_Latn', 'tur_Latn', 'twi_Latn', 'tzm_Tfng', 'uig_Arab', 'ukr_Cyrl', 'umb_Latn', 'urd_Arab', 'uzn_Latn', 'vec_Latn', 'vie_Latn', 'war_Latn', 'wol_Latn', 'xho_Latn', 'ydd_Hebr', 'yor_Latn', 'yue_Hant', 'zho_Hans', 'zho_Hant', 'zul_Latn']
class 			_lowercase						(       __a					):
     _UpperCAmelCase										=					VOCAB_FILES_NAMES
     _UpperCAmelCase										=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _UpperCAmelCase										=					PRETRAINED_VOCAB_FILES_MAP
     _UpperCAmelCase										=					['''input_ids''', '''attention_mask''']
     _UpperCAmelCase										=					NllbTokenizer

     _UpperCAmelCase										=					[]
     _UpperCAmelCase										=					[]







     def __init__( self    ,   A__=None    ,   A__=None    ,   A__="<s>"    ,   A__="</s>"    ,   A__="</s>"    ,   A__="<s>"    ,   A__="<unk>"    ,   A__="<pad>"    ,   A__="<mask>"    ,   A__=None    ,   A__=None    ,   A__=None    ,   A__=False    ,   **A__    ,   ) ->	List[Any]:
      # Mask token behave like a normal word, i.e. include the space before it
      snake_case				=			AddedToken(A__    ,   lstrip=A__    ,   rstrip=A__    ) if isinstance(A__    ,   A__    ) else mask_token
      snake_case				=			legacy_behaviour
      super().__init__(
          vocab_file=A__    ,   tokenizer_file=A__    ,   bos_token=A__    ,   eos_token=A__    ,   sep_token=A__    ,   cls_token=A__    ,   unk_token=A__    ,   pad_token=A__    ,   mask_token=A__    ,   src_lang=A__    ,   tgt_lang=A__    ,   additional_special_tokens=A__    ,   legacy_behaviour=A__    ,   **A__    ,   )

      snake_case				=			vocab_file
      snake_case				=			False if not self.vocab_file else True

      snake_case				=			FAIRSEQ_LANGUAGE_CODES.copy()

      if additional_special_tokens is not None:
       # Only add those special tokens if they are not already there.
       _additional_special_tokens.extend(
           [t for t in additional_special_tokens if t not in _additional_special_tokens]    )

      self.add_special_tokens({'''additional_special_tokens''': _additional_special_tokens}    )
      snake_case				=			{
          lang_code: self.convert_tokens_to_ids(A__    ) for lang_code in FAIRSEQ_LANGUAGE_CODES
      }

      snake_case				=			src_lang if src_lang is not None else '''eng_Latn'''
      snake_case				=			self.convert_tokens_to_ids(self._src_lang    )
      snake_case				=			tgt_lang
      self.set_src_lang_special_tokens(self._src_lang    )







     @property
     def     UpperCamelCase			( self    ) ->	str:
      return self._src_lang







     @src_lang.setter
     def     UpperCamelCase			( self    ,   A__    ) ->	None:
      snake_case				=			new_src_lang
      self.set_src_lang_special_tokens(self._src_lang    )







     def     UpperCamelCase			( self    ,   A__    ,   A__ = None    ) ->	List[int]:
      if token_ids_a is None:
       return self.prefix_tokens + token_ids_a + self.suffix_tokens
      # 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.suffix_tokens







     def     UpperCamelCase			( self    ,   A__    ,   A__ = None    ) ->	List[int]:
      snake_case				=			[self.sep_token_id]
      snake_case				=			[self.cls_token_id]

      if token_ids_a is None:
       return len(cls + token_ids_a + sep    ) * [0]
      return len(cls + token_ids_a + sep + sep + token_ids_a + sep    ) * [0]







     def     UpperCamelCase			( self    ,   A__    ,   A__    ,   A__    ,   A__    ,   **A__    ) ->	Union[str, Any]:
      if src_lang is None or tgt_lang is None:
       raise ValueError('''Translation requires a `src_lang` and a `tgt_lang` for this model'''    )
      snake_case				=			src_lang
      snake_case				=			self(A__    ,   add_special_tokens=A__    ,   return_tensors=A__    ,   **A__    )
      snake_case				=			self.convert_tokens_to_ids(A__    )
      snake_case				=			tgt_lang_id
      return inputs







     def     UpperCamelCase			( self    ,   A__    ,   A__ = "eng_Latn"    ,   A__ = None    ,   A__ = "fra_Latn"    ,   **A__    ,   ) ->	BatchEncoding:
      snake_case				=			src_lang
      snake_case				=			tgt_lang
      return super().prepare_seqaseq_batch(A__    ,   A__    ,   **A__    )







     def     UpperCamelCase			( self    ) ->	List[Any]:
      return self.set_src_lang_special_tokens(self.src_lang    )







     def     UpperCamelCase			( self    ) ->	Optional[Any]:
      return self.set_tgt_lang_special_tokens(self.tgt_lang    )







     def     UpperCamelCase			( self    ,   A__    ) ->	None:
      snake_case				=			self.convert_tokens_to_ids(A__    )

      if self.legacy_behaviour:
       snake_case				=			[]
       snake_case				=			[self.eos_token_id, self.cur_lang_code]
      else:
       snake_case				=			[self.cur_lang_code]
       snake_case				=			[self.eos_token_id]

      snake_case				=			self.convert_ids_to_tokens(self.prefix_tokens    )
      snake_case				=			self.convert_ids_to_tokens(self.suffix_tokens    )

      snake_case				=			processors.TemplateProcessing(
          single=prefix_tokens_str + ['''$A'''] + suffix_tokens_str    ,   pair=prefix_tokens_str + ['''$A''', '''$B'''] + suffix_tokens_str    ,   special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str    ,   self.prefix_tokens + self.suffix_tokens    )    )    ,   )







     def     UpperCamelCase			( self    ,   A__    ) ->	None:
      snake_case				=			self.convert_tokens_to_ids(A__    )
      if self.legacy_behaviour:
       snake_case				=			[]
       snake_case				=			[self.eos_token_id, self.cur_lang_code]
      else:
       snake_case				=			[self.cur_lang_code]
       snake_case				=			[self.eos_token_id]

      snake_case				=			self.convert_ids_to_tokens(self.prefix_tokens    )
      snake_case				=			self.convert_ids_to_tokens(self.suffix_tokens    )

      snake_case				=			processors.TemplateProcessing(
          single=prefix_tokens_str + ['''$A'''] + suffix_tokens_str    ,   pair=prefix_tokens_str + ['''$A''', '''$B'''] + suffix_tokens_str    ,   special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str    ,   self.prefix_tokens + self.suffix_tokens    )    )    ,   )







     def     UpperCamelCase			( self    ,   A__    ,   A__ = None    ) ->	Tuple[str]:
      if not self.can_save_slow_tokenizer:
       raise ValueError(
           '''Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '''
           '''tokenizer.'''    )

      if not os.path.isdir(A__    ):
       logger.error(F"""Vocabulary path ({save_directory}) should be a directory."""    )
       return
      snake_case				=			os.path.join(
          A__    ,   (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']    )

      if os.path.abspath(self.vocab_file    ) != os.path.abspath(A__    ):
       copyfile(self.vocab_file    ,   A__    )

      return (out_vocab_file,)








'''simple docstring'''







import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
   from .features import FeatureType


_lowercase						,				_lowercase						,				_lowercase       						=			False, False, False
@dataclass
class 			_lowercase						:
     _UpperCAmelCase										=					None
     _UpperCAmelCase										=					True
     _UpperCAmelCase										=					True
     _UpperCAmelCase										=					None
     # Automatically constructed
     _UpperCAmelCase										=					"dict"
     _UpperCAmelCase										=					pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}					)
     _UpperCAmelCase										=					field(default='''Audio'''				,		init=__a				,		repr=__a					)







     def __call__( self    ) ->	Optional[Any]:
      return self.pa_type







     def     UpperCamelCase			( self    ,   A__    ) ->	dict:
      try:
       import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
      except ImportError as err:
       raise ImportError('''To support encoding audio data, please install \'soundfile\'.'''    ) from err
      if isinstance(A__    ,   A__    ):
       return {"bytes": None, "path": value}
      elif isinstance(A__    ,   A__    ):
       return {"bytes": value, "path": None}
      elif "array" in value:
       # convert the audio array to wav bytes
       snake_case				=			BytesIO()
       sf.write(A__    ,   value['''array''']    ,   value['''sampling_rate''']    ,   format='''wav'''    )
       return {"bytes": buffer.getvalue(), "path": None}
      elif value.get('''path'''    ) is not None and os.path.isfile(value['''path''']    ):
       # we set "bytes": None to not duplicate the data if they're already available locally
       if value["path"].endswith('''pcm'''    ):
        # "PCM" only has raw audio bytes
        if value.get('''sampling_rate'''    ) is None:
         # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
         raise KeyError('''To use PCM files, please specify a \'sampling_rate\' in Audio object'''    )
        if value.get('''bytes'''    ):
         # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
         snake_case				=			np.frombuffer(value['''bytes''']    ,   dtype=np.intaa    ).astype(np.floataa    ) / 3_27_67
        else:
         snake_case				=			np.memmap(value['''path''']    ,   dtype='''h'''    ,   mode='''r'''    ).astype(np.floataa    ) / 3_27_67

        snake_case				=			BytesIO(bytes()    )
        sf.write(A__    ,   A__    ,   value['''sampling_rate''']    ,   format='''wav'''    )
        return {"bytes": buffer.getvalue(), "path": None}
       else:
        return {"bytes": None, "path": value.get('''path'''    )}
      elif value.get('''bytes'''    ) is not None or value.get('''path'''    ) is not None:
       # store the audio bytes, and path is used to infer the audio format using the file extension
       return {"bytes": value.get('''bytes'''    ), "path": value.get('''path'''    )}
      else:
       raise ValueError(
           F"""An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."""    )







     def     UpperCamelCase			( self    ,   A__    ,   A__ = None    ) ->	dict:
      if not self.decode:
       raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''    )

      snake_case				,		snake_case				=			(value['''path'''], BytesIO(value['''bytes''']    )) if value['''bytes'''] is not None else (value['''path'''], None)
      if path is None and file is None:
       raise ValueError(F"""An audio sample should have one of 'path' or 'bytes' but both are None in {value}."""    )

      try:
       import librosa
       import soundfile as sf
      except ImportError as err:
       raise ImportError('''To support decoding audio files, please install \'librosa\' and \'soundfile\'.'''    ) from err

      snake_case				=			xsplitext(A__    )[1][1:].lower() if path is not None else None
      if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
       raise RuntimeError(
           '''Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '''
           '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '''    )
      elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
       raise RuntimeError(
           '''Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '''
           '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '''    )

      if file is None:
       snake_case				=			token_per_repo_id or {}
       snake_case				=			path.split('''::'''    )[-1]
       try:
        snake_case				=			string_to_dict(A__    ,   config.HUB_DATASETS_URL    )['''repo_id''']
        snake_case				=			token_per_repo_id[repo_id]
       except (ValueError, KeyError):
        snake_case				=			None

       with xopen(A__    ,   '''rb'''    ,   use_auth_token=A__    ) as f:
        snake_case				,		snake_case				=			sf.read(A__    )

      else:
       snake_case				,		snake_case				=			sf.read(A__    )

      snake_case				=			array.T
      if self.mono:
       snake_case				=			librosa.to_mono(A__    )
      if self.sampling_rate and self.sampling_rate != sampling_rate:
       snake_case				=			librosa.resample(A__    ,   orig_sr=A__    ,   target_sr=self.sampling_rate    )
       snake_case				=			self.sampling_rate

      return {"path": path, "array": array, "sampling_rate": sampling_rate}







     def     UpperCamelCase			( self    ) ->	Union["FeatureType", Dict[str, "FeatureType"]]:
      from .features import Value

      if self.decode:
       raise ValueError('''Cannot flatten a decoded Audio feature.'''    )
      return {
          "bytes": Value('''binary'''    ),
          "path": Value('''string'''    ),
      }







     def     UpperCamelCase			( self    ,   A__    ) ->	pa.StructArray:
      if pa.types.is_string(storage.type    ):
       snake_case				=			pa.array([None] * len(A__    )    ,   type=pa.binary()    )
       snake_case				=			pa.StructArray.from_arrays([bytes_array, storage]    ,   ['''bytes''', '''path''']    ,   mask=storage.is_null()    )
      elif pa.types.is_binary(storage.type    ):
       snake_case				=			pa.array([None] * len(A__    )    ,   type=pa.string()    )
       snake_case				=			pa.StructArray.from_arrays([storage, path_array]    ,   ['''bytes''', '''path''']    ,   mask=storage.is_null()    )
      elif pa.types.is_struct(storage.type    ) and storage.type.get_all_field_indices('''array'''    ):
       snake_case				=			pa.array([Audio().encode_example(A__    ) if x is not None else None for x in storage.to_pylist()]    )
      elif pa.types.is_struct(storage.type    ):
       if storage.type.get_field_index('''bytes'''    ) >= 0:
        snake_case				=			storage.field('''bytes'''    )
       else:
        snake_case				=			pa.array([None] * len(A__    )    ,   type=pa.binary()    )
       if storage.type.get_field_index('''path'''    ) >= 0:
        snake_case				=			storage.field('''path'''    )
       else:
        snake_case				=			pa.array([None] * len(A__    )    ,   type=pa.string()    )
       snake_case				=			pa.StructArray.from_arrays([bytes_array, path_array]    ,   ['''bytes''', '''path''']    ,   mask=storage.is_null()    )
      return array_cast(A__    ,   self.pa_type    )







     def     UpperCamelCase			( self    ,   A__    ) ->	pa.StructArray:

      @no_op_if_value_is_null
      def path_to_bytes(A__    ):
       with xopen(A__    ,   '''rb'''    ) as f:
        snake_case				=			f.read()
       return bytes_

      snake_case				=			pa.array(
          [
              (path_to_bytes(x['''path''']    ) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
              for x in storage.to_pylist()
          ]    ,   type=pa.binary()    ,   )
      snake_case				=			pa.array(
          [os.path.basename(A__    ) if path is not None else None for path in storage.field('''path'''    ).to_pylist()]    ,   type=pa.string()    ,   )
      snake_case				=			pa.StructArray.from_arrays([bytes_array, path_array]    ,   ['''bytes''', '''path''']    ,   mask=bytes_array.is_null()    )
      return array_cast(A__    ,   self.pa_type    )

