




"""simple docstring"""




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

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


a_        =       logging.get_logger(__name__)

a_        =       '▁'

a_        =       {'vocab_file': 'sentencepiece.bpe.model'}

a_        =       {
    'vocab_file': {
        'xlm-roberta-base': 'https://huggingface.co/xlm-roberta-base/resolve/main/sentencepiece.bpe.model',
        'xlm-roberta-large': 'https://huggingface.co/xlm-roberta-large/resolve/main/sentencepiece.bpe.model',
        'xlm-roberta-large-finetuned-conll02-dutch': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll02-dutch/resolve/main/sentencepiece.bpe.model'
        ),
        'xlm-roberta-large-finetuned-conll02-spanish': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll02-spanish/resolve/main/sentencepiece.bpe.model'
        ),
        'xlm-roberta-large-finetuned-conll03-english': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll03-english/resolve/main/sentencepiece.bpe.model'
        ),
        'xlm-roberta-large-finetuned-conll03-german': (
            'https://huggingface.co/xlm-roberta-large-finetuned-conll03-german/resolve/main/sentencepiece.bpe.model'
        ),
    }
}

a_        =       {
    'xlm-roberta-base': 5_1_2,
    'xlm-roberta-large': 5_1_2,
    'xlm-roberta-large-finetuned-conll02-dutch': 5_1_2,
    'xlm-roberta-large-finetuned-conll02-spanish': 5_1_2,
    'xlm-roberta-large-finetuned-conll03-english': 5_1_2,
    'xlm-roberta-large-finetuned-conll03-german': 5_1_2,
}

class   UpperCAmelCase_    (			snake_case		):
		UpperCamelCase							=VOCAB_FILES_NAMES
		UpperCamelCase							=PRETRAINED_VOCAB_FILES_MAP
		UpperCamelCase							=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		UpperCamelCase							=["input_ids", "attention_mask"]
		def __init__(						self				,						UpperCamelCase_				,						UpperCamelCase_="<s>"				,						UpperCamelCase_="</s>"				,						UpperCamelCase_="</s>"				,						UpperCamelCase_="<s>"				,						UpperCamelCase_="<unk>"				,						UpperCamelCase_="<pad>"				,						UpperCamelCase_="<mask>"				,						UpperCamelCase_ = None				,						**UpperCamelCase_				,						)						->       None:
					# Mask token behave like a normal word, i.e. include the space before it
					__lowercase			:  List[Any]								=       AddedToken(UpperCamelCase_				,						lstrip=UpperCamelCase_				,						rstrip=UpperCamelCase_       ) if isinstance(UpperCamelCase_				,						UpperCamelCase_       ) else mask_token

					__lowercase			:  Dict								=       {} if sp_model_kwargs is None else sp_model_kwargs

					super().__init__(
					    bos_token=UpperCamelCase_				,						eos_token=UpperCamelCase_				,						unk_token=UpperCamelCase_				,						sep_token=UpperCamelCase_				,						cls_token=UpperCamelCase_				,						pad_token=UpperCamelCase_				,						mask_token=UpperCamelCase_				,						sp_model_kwargs=self.sp_model_kwargs				,						**UpperCamelCase_				,						)

					__lowercase			:  Optional[int]								=       spm.SentencePieceProcessor(**self.sp_model_kwargs       )
					self.sp_model.Load(str(UpperCamelCase_       )       )
					__lowercase			:  str								=       vocab_file

					# Original fairseq vocab and spm vocab must be "aligned":
					# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
					# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
					# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
					# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

					# Mimic fairseq token-to-id alignment for the first 4 token
					__lowercase			:  List[Any]								=       {'''<s>''': 0, '''<pad>''': 1, '''</s>''': 2, '''<unk>''': 3}

					# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
					__lowercase			:  Tuple								=       1

					__lowercase			:  Any								=       len(self.sp_model       ) + self.fairseq_offset
					__lowercase			:  str								=       {v: k for k, v in self.fairseq_tokens_to_ids.items()}
		def __getstate__(						self       )						->       Optional[Any]:
					__lowercase			:  int								=       self.__dict__.copy()
					__lowercase			:  int								=       None
					__lowercase			:  Optional[Any]								=       self.sp_model.serialized_model_proto()
					return state
		def __setstate__(						self				,						UpperCamelCase_       )						->       Tuple:
					__lowercase			:  List[str]								=       d

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

					__lowercase			:  Optional[int]								=       spm.SentencePieceProcessor(**self.sp_model_kwargs       )
					self.sp_model.LoadFromSerializedProto(self.sp_model_proto       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_ = None       )						->       List[int]:

					if token_ids_a is None:
								return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
					__lowercase			:  Dict								=       [self.cls_token_id]
					__lowercase			:  Union[str, Any]								=       [self.sep_token_id]
					return cls + token_ids_a + sep + sep + token_ids_a + sep
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_ = None				,						UpperCamelCase_ = False       )						->       List[int]:

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

					if token_ids_a is None:
								return [1] + ([0] * len(UpperCamelCase_       )) + [1]
					return [1] + ([0] * len(UpperCamelCase_       )) + [1, 1] + ([0] * len(UpperCamelCase_       )) + [1]
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_ = None       )						->       List[int]:
					__lowercase			:  Optional[Any]								=       [self.sep_token_id]
					__lowercase			:  Optional[Any]								=       [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]
		@property
		def 	_lowerCamelCase							(						self       )						->       Dict:
					return len(self.sp_model       ) + self.fairseq_offset + 1  # Add the <mask> token
		def 	_lowerCamelCase							(						self       )						->       str:
					__lowercase			:  List[str]								=       {self.convert_ids_to_tokens(UpperCamelCase_       ): i for i in range(self.vocab_size       )}
					vocab.update(self.added_tokens_encoder       )
					return vocab
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_       )						->       List[str]:
					return self.sp_model.encode(UpperCamelCase_				,						out_type=UpperCamelCase_       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_       )						->       str:
					if token in self.fairseq_tokens_to_ids:
								return self.fairseq_tokens_to_ids[token]
					__lowercase			:  Optional[Any]								=       self.sp_model.PieceToId(UpperCamelCase_       )

					# Need to return unknown token if the SP model returned 0
					return spm_id + self.fairseq_offset if spm_id else self.unk_token_id
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_       )						->       Tuple:
					if index in self.fairseq_ids_to_tokens:
								return self.fairseq_ids_to_tokens[index]
					return self.sp_model.IdToPiece(index - self.fairseq_offset       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_       )						->       Dict:
					__lowercase			:  Tuple								=       ''''''.join(UpperCamelCase_       ).replace(UpperCamelCase_				,						''' '''       ).strip()
					return out_string
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_ = None       )						->       Tuple[str]:
					if not os.path.isdir(UpperCamelCase_       ):
								logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
								return
					__lowercase			:  List[Any]								=       os.path.join(
					    UpperCamelCase_				,						(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']       )

					if os.path.abspath(self.vocab_file       ) != os.path.abspath(UpperCamelCase_       ) and os.path.isfile(self.vocab_file       ):
								copyfile(self.vocab_file				,						UpperCamelCase_       )
					elif not os.path.isfile(self.vocab_file       ):
								with open(UpperCamelCase_				,						'''wb'''       ) as fi:
											__lowercase			:  Optional[Any]								=       self.sp_model.serialized_model_proto()
											fi.write(UpperCamelCase_       )

					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


__lowerCamelCase					,		__lowerCamelCase					,		__lowerCamelCase												=						False, False, False




@dataclass
class 				UpperCAmelCase			:
 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=_snake_case						, repr=_snake_case						)


 def __call__(      self							:	Any ):
    return self.pa_type


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       __lowerCamelCase							:	Union[str, bytes, 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(__lowerCamelCase			,       __lowerCamelCase ):
       return {"bytes": None, "path": value}
    elif isinstance(__lowerCamelCase			,       __lowerCamelCase ):
       return {"bytes": value, "path": None}
    elif "array" in value:
       # convert the audio array to wav bytes
       UpperCAmelCase__	:int      =      BytesIO()
       sf.write(__lowerCamelCase			,       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!)
             UpperCAmelCase__	:List[Any]      =      np.frombuffer(value['''bytes''']			,       dtype=np.intaa ).astype(np.floataa ) / 3_2_7_6_7
          else:
             UpperCAmelCase__	:Optional[Any]      =      np.memmap(value['''path''']			,       dtype='''h'''			,       mode='''r''' ).astype(np.floataa ) / 3_2_7_6_7

          UpperCAmelCase__	:Optional[Any]      =      BytesIO(bytes() )
          sf.write(__lowerCamelCase			,       __lowerCamelCase			,       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        __SCREAMING_SNAKE_CASE					(      self							:	List[Any]			,       __lowerCamelCase							:	dict			,       __lowerCamelCase							:	Optional[Dict[str, Union[str, bool, None]]] = None ):
    if not self.decode:
       raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.''' )

    UpperCAmelCase__   ,						UpperCAmelCase__	:str      =      (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

    UpperCAmelCase__	:List[str]      =      xsplitext(__lowerCamelCase )[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:
       UpperCAmelCase__	:Optional[Any]      =      token_per_repo_id or {}
       UpperCAmelCase__	:str      =      path.split('''::''' )[-1]
       try:
          UpperCAmelCase__	:Tuple      =      string_to_dict(__lowerCamelCase			,       config.HUB_DATASETS_URL )['''repo_id''']
          UpperCAmelCase__	:str      =      token_per_repo_id[repo_id]
       except (ValueError, KeyError):
          UpperCAmelCase__	:Tuple      =      None

       with xopen(__lowerCamelCase			,       '''rb'''			,       use_auth_token=__lowerCamelCase ) as f:
          UpperCAmelCase__   ,						UpperCAmelCase__	:Union[str, Any]      =      sf.read(__lowerCamelCase )

    else:
       UpperCAmelCase__   ,						UpperCAmelCase__	:List[Any]      =      sf.read(__lowerCamelCase )

    UpperCAmelCase__	:Optional[int]      =      array.T
    if self.mono:
       UpperCAmelCase__	:Any      =      librosa.to_mono(__lowerCamelCase )
    if self.sampling_rate and self.sampling_rate != sampling_rate:
       UpperCAmelCase__	:Union[str, Any]      =      librosa.resample(__lowerCamelCase			,       orig_sr=__lowerCamelCase			,       target_sr=self.sampling_rate )
       UpperCAmelCase__	:List[str]      =      self.sampling_rate

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


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    from .features import Value

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


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any			,       __lowerCamelCase							:	Union[pa.StringArray, pa.StructArray] ):
    if pa.types.is_string(storage.type ):
       UpperCAmelCase__	:List[str]      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.binary() )
       UpperCAmelCase__	:Tuple      =      pa.StructArray.from_arrays([bytes_array, storage]			,       ['''bytes''', '''path''']			,       mask=storage.is_null() )
    elif pa.types.is_binary(storage.type ):
       UpperCAmelCase__	:str      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.string() )
       UpperCAmelCase__	:int      =      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''' ):
       UpperCAmelCase__	:Any      =      pa.array([Audio().encode_example(__lowerCamelCase ) 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:
          UpperCAmelCase__	:str      =      storage.field('''bytes''' )
       else:
          UpperCAmelCase__	:List[str]      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.binary() )
       if storage.type.get_field_index('''path''' ) >= 0:
          UpperCAmelCase__	:Optional[int]      =      storage.field('''path''' )
       else:
          UpperCAmelCase__	:Optional[int]      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.string() )
       UpperCAmelCase__	:List[str]      =      pa.StructArray.from_arrays([bytes_array, path_array]			,       ['''bytes''', '''path''']			,       mask=storage.is_null() )
    return array_cast(__lowerCamelCase			,       self.pa_type )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       __lowerCamelCase							:	pa.StructArray ):

    @no_op_if_value_is_null
    def path_to_bytes(__lowerCamelCase							:	Dict ):
       with xopen(__lowerCamelCase			,       '''rb''' ) as f:
          UpperCAmelCase__	:Any      =      f.read()
       return bytes_

    UpperCAmelCase__	:Union[str, Any]      =      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()			,       )
    UpperCAmelCase__	:Optional[int]      =      pa.array(
        [os.path.basename(__lowerCamelCase ) if path is not None else None for path in storage.field('''path''' ).to_pylist()]			,       type=pa.string()			,       )
    UpperCAmelCase__	:Optional[int]      =      pa.StructArray.from_arrays([bytes_array, path_array]			,       ['''bytes''', '''path''']			,       mask=bytes_array.is_null() )
    return array_cast(__lowerCamelCase			,       self.pa_type )



