




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


lowerCAmelCase_						=			logging.get_logger(__name__)

lowerCAmelCase_						=			'▁'

lowerCAmelCase_						=			{
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
}

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

lowerCAmelCase_						=			{
    'facebook/s2t-small-librispeech-asr': 10_24,
}

lowerCAmelCase_						=			['pt', 'fr', 'ru', 'nl', 'ro', 'it', 'es', 'de']

lowerCAmelCase_						=			{'mustc': MUSTC_LANGS}
class 						_A (  _lowerCamelCase				):
  _UpperCamelCase       :      List[str]			=      VOCAB_FILES_NAMES
  _UpperCamelCase       :      List[Any]			=      PRETRAINED_VOCAB_FILES_MAP
  _UpperCamelCase       :      Optional[int]			=      MAX_MODEL_INPUT_SIZES
  _UpperCamelCase       :      Optional[int]			=      ['''input_ids''', '''attention_mask''']

  _UpperCamelCase       :      List[int]			=      []






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






       """simple docstring"""







       lowercase :       List[str]       =			{} if sp_model_kwargs is None else sp_model_kwargs

       super().__init__(
           bos_token=_A				,						eos_token=_A				,						unk_token=_A				,						pad_token=_A				,						do_upper_case=_A				,						do_lower_case=_A				,						tgt_lang=_A				,						lang_codes=_A				,						sp_model_kwargs=self.sp_model_kwargs				,						**_A				,						)
       lowercase :       List[str]       =			do_upper_case
       lowercase :       List[Any]       =			do_lower_case

       lowercase :       Union[str, Any]       =			load_json(_A				)
       lowercase :       List[Any]       =			{v: k for k, v in self.encoder.items()}
       lowercase :       Optional[int]       =			spm_file
       lowercase :       Optional[int]       =			load_spm(_A				,						self.sp_model_kwargs				)

       if lang_codes is not None:
            lowercase :       Dict       =			lang_codes
            lowercase :       List[Any]       =			LANGUAGES[lang_codes]
            lowercase :       Union[str, Any]       =			[f"""<lang:{lang}>""" for lang in self.langs]
            lowercase :       List[str]       =			{lang: self.sp_model.PieceToId(f"""<lang:{lang}>"""				) for lang in self.langs}

            lowercase :       List[Any]       =			self.lang_tokens
            lowercase :       Tuple       =			tgt_lang if tgt_lang is not None else self.langs[0]

            self.set_tgt_lang_special_tokens(self._tgt_lang				)
       else:
            lowercase :       Union[str, Any]       =			{}






  @property
  def  __a      (  self		:     List[str]				)		->   int:






       """simple docstring"""







       return len(self.encoder				)






  @property
  def  __a      (  self		:     Tuple				)		->   str:






       """simple docstring"""







       return self._tgt_lang






  @tgt_lang.setter
  def  __a      (  self		:     List[str]				,						_A		:     Any				)		->   None:






       """simple docstring"""







       lowercase :       Optional[int]       =			new_tgt_lang
       self.set_tgt_lang_special_tokens(_A				)






  def  __a      (  self		:     List[Any]				,						_A		:     str				)		->   None:






       """simple docstring"""







       lowercase :       List[str]       =			self.lang_code_to_id[tgt_lang]
       lowercase :       List[str]       =			[lang_code_id]






  def  __a      (  self		:     Optional[int]				,						_A		:     str				)		->   List[str]:






       """simple docstring"""







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






  def  __a      (  self		:     Tuple				,						_A		:     Dict				)		->   int:






       """simple docstring"""







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






  def  __a      (  self		:     str				,						_A		:     int				)		->   str:






       """simple docstring"""







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






  def  __a      (  self		:     Tuple				,						_A		:     List[str]				)		->   str:






       """simple docstring"""







       lowercase :       Tuple       =			[]
       lowercase :       int       =			''''''
       for token in tokens:
            # make sure that special tokens are not decoded using sentencepiece model
            if token in self.all_special_tokens:
                 lowercase :       Dict       =			self.sp_model.decode(_A				)
                 out_string += (decoded.upper() if self.do_upper_case else decoded) + token + " "
                 lowercase :       Tuple       =			[]
            else:
                 current_sub_tokens.append(_A				)
       lowercase :       List[Any]       =			self.sp_model.decode(_A				)
       out_string += decoded.upper() if self.do_upper_case else decoded
       return out_string.strip()






  def  __a      (  self		:     List[str]				,						_A		:     List[str]				,						_A		:     List[str]=None				)		->   List[int]:






       """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  __a      (  self		:     List[str]				,						_A		:     List[int]				,						_A		:     Optional[List[int]] = None				,						_A		:     bool = False				)		->   List[int]:






       """simple docstring"""







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

       lowercase :       Dict       =			[1] * len(self.prefix_tokens				)
       lowercase :       Union[str, Any]       =			[1]
       if token_ids_a is None:
            return prefix_ones + ([0] * len(_A				)) + suffix_ones
       return prefix_ones + ([0] * len(_A				)) + ([0] * len(_A				)) + suffix_ones






  def  __a      (  self		:     Optional[int]				)		->   Dict:






       """simple docstring"""







       lowercase :       Optional[Any]       =			self.encoder.copy()
       vocab.update(self.added_tokens_encoder				)
       return vocab






  def __getstate__(  self		:     Union[str, Any]				)		->   Dict:






       """simple docstring"""







       lowercase :       Any       =			self.__dict__.copy()
       lowercase :       int       =			None
       return state






  def __setstate__(  self		:     List[Any]				,						_A		:     Dict				)		->   None:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			d

       # for backward compatibility
       if not hasattr(self				,						'''sp_model_kwargs'''				):
            lowercase :       Optional[Any]       =			{}

       lowercase :       List[Any]       =			load_spm(self.spm_file				,						self.sp_model_kwargs				)






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






       """simple docstring"""







       lowercase :       Optional[Any]       =			Path(_A				)
       assert save_dir.is_dir(), f"""{save_directory} should be a directory"""
       lowercase :       Union[str, Any]       =			save_dir / (
           (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
       )
       lowercase :       Dict       =			save_dir / (
           (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
       )

       save_json(self.encoder				,						_A				)

       if os.path.abspath(self.spm_file				) != os.path.abspath(_A				) and os.path.isfile(self.spm_file				):
            copyfile(self.spm_file				,						_A				)
       elif not os.path.isfile(self.spm_file				):
            with open(_A				,						'''wb'''				) as fi:
                 lowercase :       Dict       =			self.sp_model.serialized_model_proto()
                 fi.write(_A				)

       return (str(_A				), str(_A				))






def  snake_case(    __magic_name__			,		__magic_name__   )			->							sentencepiece.SentencePieceProcessor:





     '''simple docstring'''


     lowercase :       List[Any]       =			sentencepiece.SentencePieceProcessor(**__magic_name__   )
     spm.Load(str(__magic_name__   )   )
     return spm






def  snake_case(    __magic_name__   )			->							Union[Dict, List]:





     '''simple docstring'''


     with open(__magic_name__			,		'''r'''   ) as f:
          return json.load(__magic_name__   )






def  snake_case(    __magic_name__			,		__magic_name__   )			->							None:





     '''simple docstring'''


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





import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


lowerCAmelCase_						=			'\\n    Text data.\n    Second line of data.'

lowerCAmelCase_						=			'file'






@pytest.fixture(scope='''session'''   )
def  snake_case(    __magic_name__   )			->							List[str]:





     '''simple docstring'''


     lowercase :       Any       =			tmp_path_factory.mktemp('''data'''   ) / (FILE_PATH + '''.zstd''')
     lowercase :       List[Any]       =			bytes(__magic_name__			,		'''utf-8'''   )
     with zstd.open(__magic_name__			,		'''wb'''   ) as f:
          f.write(__magic_name__   )
     return path






@pytest.fixture
def  snake_case(    __magic_name__   )			->							List[Any]:





     '''simple docstring'''


     with open(os.path.join(tmpfs.local_root_dir			,		__magic_name__   )			,		'''w'''   ) as f:
          f.write(__magic_name__   )
     return FILE_PATH






@pytest.mark.parametrize('''compression_format'''			,		['''gzip''', '''xz''', '''zstd''']   )
def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__   )			->							List[str]:





     '''simple docstring'''


     lowercase :       Optional[int]       =			{'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
     lowercase :       int       =			input_paths[compression_format]
     lowercase :       Any       =			tmp_path / '''cache'''
     lowercase :       int       =			DownloadConfig(cache_dir=__magic_name__			,		extract_compressed_file=__magic_name__   )
     lowercase :       Optional[Any]       =			cached_path(__magic_name__			,		download_config=__magic_name__   )
     with open(__magic_name__   ) as f:
          lowercase :       Optional[int]       =			f.read()
     with open(__magic_name__   ) as f:
          lowercase :       str       =			f.read()
     assert extracted_file_content == expected_file_content






@pytest.mark.parametrize('''default_extracted'''			,		[True, False]   )
@pytest.mark.parametrize('''default_cache_dir'''			,		[True, False]   )
def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__   )			->							List[str]:





     '''simple docstring'''


     lowercase :       Dict       =			'''custom_cache'''
     lowercase :       Union[str, Any]       =			'''custom_extracted_dir'''
     lowercase :       str       =			tmp_path / '''custom_extracted_path'''
     if default_extracted:
          lowercase :       Tuple       =			('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
     else:
          monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''			,		__magic_name__   )
          monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''			,		str(__magic_name__   )   )
          lowercase :       int       =			custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

     lowercase :       List[str]       =			xz_file
     lowercase :       Any       =			(
         DownloadConfig(extract_compressed_file=__magic_name__   )
         if default_cache_dir
         else DownloadConfig(cache_dir=tmp_path / custom_cache_dir			,		extract_compressed_file=__magic_name__   )
     )
     lowercase :       Optional[int]       =			cached_path(__magic_name__			,		download_config=__magic_name__   )
     assert Path(__magic_name__   ).parent.parts[-2:] == expected






def  snake_case(    __magic_name__   )			->							List[Any]:





     '''simple docstring'''


     lowercase :       Any       =			str(Path(__magic_name__   ).resolve()   )
     assert cached_path(__magic_name__   ) == text_file
     # relative path
     lowercase :       Union[str, Any]       =			str(Path(__magic_name__   ).resolve().relative_to(Path(os.getcwd()   )   )   )
     assert cached_path(__magic_name__   ) == text_file






def  snake_case(    __magic_name__   )			->							Union[str, Any]:





     '''simple docstring'''


     lowercase :       List[Any]       =			str(tmp_path.resolve() / '''__missing_file__.txt'''   )
     with pytest.raises(__magic_name__   ):
          cached_path(__magic_name__   )
     # relative path
     lowercase :       Optional[int]       =			'''./__missing_file__.txt'''
     with pytest.raises(__magic_name__   ):
          cached_path(__magic_name__   )






def  snake_case(    __magic_name__   )			->							List[str]:





     '''simple docstring'''


     lowercase :       List[str]       =			get_from_cache(F"""tmp://{tmpfs_file}"""   )
     with open(__magic_name__   ) as f:
          lowercase :       List[Any]       =			f.read()
     assert output_file_content == FILE_CONTENT






@patch('''datasets.config.HF_DATASETS_OFFLINE'''			,		__magic_name__   )
def  snake_case(    )			->							List[Any]:





     '''simple docstring'''


     with pytest.raises(__magic_name__   ):
          cached_path('''https://huggingface.co'''   )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''			,		__magic_name__   )
def  snake_case(    __magic_name__   )			->							Tuple:





     '''simple docstring'''


     lowercase :       Any       =			tmp_path_factory.mktemp('''data'''   ) / '''file.html'''
     with pytest.raises(__magic_name__   ):
          http_get('''https://huggingface.co'''			,		temp_file=__magic_name__   )
     with pytest.raises(__magic_name__   ):
          http_head('''https://huggingface.co'''   )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''			,		__magic_name__   )
def  snake_case(    __magic_name__   )			->							Optional[Any]:





     '''simple docstring'''


     lowercase :       Optional[Any]       =			tmp_path_factory.mktemp('''data'''   ) / '''file.html'''
     with pytest.raises(__magic_name__   ):
          ftp_get('''ftp://huggingface.co'''			,		temp_file=__magic_name__   )
     with pytest.raises(__magic_name__   ):
          ftp_head('''ftp://huggingface.co'''   )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''			,		__magic_name__   )
def  snake_case(    __magic_name__   )			->							List[str]:





     '''simple docstring'''


     lowercase :       str       =			tmp_path_factory.mktemp('''data'''   ) / '''file.html'''
     with pytest.raises(__magic_name__   ):
          fsspec_get('''s3://huggingface.co'''			,		temp_file=__magic_name__   )
     with pytest.raises(__magic_name__   ):
          fsspec_head('''s3://huggingface.co'''   )