


"""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


__SCREAMING_SNAKE_CASE		,			__SCREAMING_SNAKE_CASE		,			__SCREAMING_SNAKE_CASE							:    Dict		=    False, False, False





@dataclass
class      __A     :





							'''simple docstring'''

							__lowercase:		Optional[int]							      =    None
							__lowercase:		bool							      =    True
							__lowercase:		bool							      =    True
							__lowercase:		Optional[str]							      =    None
							# Automatically constructed
							__lowercase:		ClassVar[str]							      =    "dict"
							__lowercase:		ClassVar[Any]							      =    pa.struct({"""bytes""": pa.binary(), """path""": pa.string()})
							__lowercase:		str							      =    field(default="""Audio"""						,		init=snake_case__						,		repr=snake_case__)



							def __call__(				self							:      Optional[Any]       )					->Optional[int]:



														"""simple docstring"""



														return self.pa_type



							def 		lowerCAmelCase       (				self							:      int		,       UpperCAmelCase_							:      Union[str, bytes, dict]       )					->dict:



														"""simple docstring"""



														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(UpperCAmelCase_		,       UpperCAmelCase_       ):
																					return {"bytes": None, "path": value}
														elif isinstance(UpperCAmelCase_		,       UpperCAmelCase_       ):
																					return {"bytes": value, "path": None}
														elif "array" in value:
																					# convert the audio array to wav bytes
																					snake_case_           =		BytesIO()
																					sf.write(UpperCAmelCase_		,       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       ) / 32_767
																												else:
																																			snake_case_           =		np.memmap(value["""path"""]		,       dtype="""h"""		,       mode="""r"""       ).astype(np.floataa       ) / 32_767

																												snake_case_           =		BytesIO(bytes()       )
																												sf.write(UpperCAmelCase_		,       UpperCAmelCase_		,       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 		lowerCAmelCase       (				self							:      Union[str, Any]		,       UpperCAmelCase_							:      dict		,       UpperCAmelCase_							:      Optional[Dict[str, Union[str, bool, None]]] = None       )					->dict:



														"""simple docstring"""



														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(UpperCAmelCase_       )[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(UpperCAmelCase_		,       config.HUB_DATASETS_URL       )["""repo_id"""]
																												snake_case_           =		token_per_repo_id[repo_id]
																					except (ValueError, KeyError):
																												snake_case_           =		None

																					with xopen(UpperCAmelCase_		,       """rb"""		,       use_auth_token=UpperCAmelCase_       ) as f:
																												snake_case_       ,						snake_case_           =		sf.read(UpperCAmelCase_       )

														else:
																					snake_case_       ,						snake_case_           =		sf.read(UpperCAmelCase_       )

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

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



							def 		lowerCAmelCase       (				self							:      Dict       )					->Union["FeatureType", Dict[str, "FeatureType"]]:



														"""simple docstring"""



														from .features import Value

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



							def 		lowerCAmelCase       (				self							:      Union[str, Any]		,       UpperCAmelCase_							:      Union[pa.StringArray, pa.StructArray]       )					->pa.StructArray:



														"""simple docstring"""



														if pa.types.is_string(storage.type       ):
																					snake_case_           =		pa.array([None] * len(UpperCAmelCase_       )		,       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(UpperCAmelCase_       )		,       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(UpperCAmelCase_       ) 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(UpperCAmelCase_       )		,       type=pa.binary()       )
																					if storage.type.get_field_index("""path"""       ) >= 0:
																												snake_case_           =		storage.field("""path"""       )
																					else:
																												snake_case_           =		pa.array([None] * len(UpperCAmelCase_       )		,       type=pa.string()       )
																					snake_case_           =		pa.StructArray.from_arrays([bytes_array, path_array]		,       ["""bytes""", """path"""]		,       mask=storage.is_null()       )
														return array_cast(UpperCAmelCase_		,       self.pa_type       )



							def 		lowerCAmelCase       (				self							:      Optional[Any]		,       UpperCAmelCase_							:      pa.StructArray       )					->pa.StructArray:



														"""simple docstring"""



														@no_op_if_value_is_null
														def path_to_bytes(UpperCAmelCase_							:      Dict       ):
																					with xopen(UpperCAmelCase_		,       """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(UpperCAmelCase_       ) 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(UpperCAmelCase_		,       self.pa_type       )







"""simple docstring"""
import os
import unicodedata
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


__SCREAMING_SNAKE_CASE							:    List[str]		=    logging.get_logger(__name__)
__SCREAMING_SNAKE_CASE							:    Optional[int]		=    {'vocab_file': 'spiece.model'}

__SCREAMING_SNAKE_CASE							:    List[str]		=    {
    'vocab_file': {
        'albert-base-v1': 'https://huggingface.co/albert-base-v1/resolve/main/spiece.model',
        'albert-large-v1': 'https://huggingface.co/albert-large-v1/resolve/main/spiece.model',
        'albert-xlarge-v1': 'https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model',
        'albert-xxlarge-v1': 'https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model',
        'albert-base-v2': 'https://huggingface.co/albert-base-v2/resolve/main/spiece.model',
        'albert-large-v2': 'https://huggingface.co/albert-large-v2/resolve/main/spiece.model',
        'albert-xlarge-v2': 'https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model',
        'albert-xxlarge-v2': 'https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model',
    }
}

__SCREAMING_SNAKE_CASE							:    List[str]		=    {
    'albert-base-v1': 512,
    'albert-large-v1': 512,
    'albert-xlarge-v1': 512,
    'albert-xxlarge-v1': 512,
    'albert-base-v2': 512,
    'albert-large-v2': 512,
    'albert-xlarge-v2': 512,
    'albert-xxlarge-v2': 512,
}

__SCREAMING_SNAKE_CASE							:    int		=    '▁'





class      __A     (snake_case__):





							'''simple docstring'''

							__lowercase:		Optional[Any]							      =    VOCAB_FILES_NAMES
							__lowercase:		Optional[int]							      =    PRETRAINED_VOCAB_FILES_MAP
							__lowercase:		Tuple							      =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES



							def __init__(				self							:      int		,       UpperCAmelCase_							:      Any		,       UpperCAmelCase_							:      Dict=True		,       UpperCAmelCase_							:      Dict=True		,       UpperCAmelCase_							:      List[str]=False		,       UpperCAmelCase_							:      List[Any]="[CLS]"		,       UpperCAmelCase_							:      Any="[SEP]"		,       UpperCAmelCase_							:      str="<unk>"		,       UpperCAmelCase_							:      str="[SEP]"		,       UpperCAmelCase_							:      Optional[Any]="<pad>"		,       UpperCAmelCase_							:      Optional[int]="[CLS]"		,       UpperCAmelCase_							:      int="[MASK]"		,       UpperCAmelCase_							:      Optional[Dict[str, Any]] = None		,       **UpperCAmelCase_							:      Union[str, Any]		,       )					->None:



														"""simple docstring"""



														snake_case_           =		(
														    AddedToken(UpperCAmelCase_		,       lstrip=UpperCAmelCase_		,       rstrip=UpperCAmelCase_		,       normalized=UpperCAmelCase_       )
														    if isinstance(UpperCAmelCase_		,       UpperCAmelCase_       )
														    else mask_token
														)

														snake_case_           =		{} if sp_model_kwargs is None else sp_model_kwargs

														super().__init__(
														    do_lower_case=UpperCAmelCase_		,       remove_space=UpperCAmelCase_		,       keep_accents=UpperCAmelCase_		,       bos_token=UpperCAmelCase_		,       eos_token=UpperCAmelCase_		,       unk_token=UpperCAmelCase_		,       sep_token=UpperCAmelCase_		,       pad_token=UpperCAmelCase_		,       cls_token=UpperCAmelCase_		,       mask_token=UpperCAmelCase_		,       sp_model_kwargs=self.sp_model_kwargs		,       **UpperCAmelCase_		,       )

														snake_case_           =		do_lower_case
														snake_case_           =		remove_space
														snake_case_           =		keep_accents
														snake_case_           =		vocab_file

														snake_case_           =		spm.SentencePieceProcessor(**self.sp_model_kwargs       )
														self.sp_model.Load(UpperCAmelCase_       )



							@property
							def 		lowerCAmelCase       (				self							:      List[Any]       )					->Dict:



														"""simple docstring"""



														return len(self.sp_model       )



							def 		lowerCAmelCase       (				self							:      str       )					->List[Any]:



														"""simple docstring"""



														snake_case_           =		{self.convert_ids_to_tokens(UpperCAmelCase_       ): i for i in range(self.vocab_size       )}
														vocab.update(self.added_tokens_encoder       )
														return vocab



							def __getstate__(				self							:      Dict       )					->List[str]:



														"""simple docstring"""



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



							def __setstate__(				self							:      Tuple		,       UpperCAmelCase_							:      Optional[int]       )					->Optional[int]:



														"""simple docstring"""



														snake_case_           =		d

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

														snake_case_           =		spm.SentencePieceProcessor(**self.sp_model_kwargs       )
														self.sp_model.Load(self.vocab_file       )



							def 		lowerCAmelCase       (				self							:      Tuple		,       UpperCAmelCase_							:      Any       )					->str:



														"""simple docstring"""



														if self.remove_space:
																					snake_case_           =		""" """.join(inputs.strip().split()       )
														else:
																					snake_case_           =		inputs
														snake_case_           =		outputs.replace("""``"""		,       """\""""       ).replace("""''"""		,       """\""""       )

														if not self.keep_accents:
																					snake_case_           =		unicodedata.normalize("""NFKD"""		,       UpperCAmelCase_       )
																					snake_case_           =		"""""".join([c for c in outputs if not unicodedata.combining(UpperCAmelCase_       )]       )
														if self.do_lower_case:
																					snake_case_           =		outputs.lower()

														return outputs



							def 		lowerCAmelCase       (				self							:      Dict		,       UpperCAmelCase_							:      str       )					->List[str]:



														"""simple docstring"""



														snake_case_           =		self.preprocess_text(UpperCAmelCase_       )
														snake_case_           =		self.sp_model.encode(UpperCAmelCase_		,       out_type=UpperCAmelCase_       )
														snake_case_           =		[]
														for piece in pieces:
																					if len(UpperCAmelCase_       ) > 1 and piece[-1] == str(""","""       ) and piece[-2].isdigit():
																												snake_case_           =		self.sp_model.EncodeAsPieces(piece[:-1].replace(UpperCAmelCase_		,       """"""       )       )
																												if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																																			if len(cur_pieces[0]       ) == 1:
																																										snake_case_           =		cur_pieces[1:]
																																			else:
																																										snake_case_           =		cur_pieces[0][1:]
																												cur_pieces.append(piece[-1]       )
																												new_pieces.extend(UpperCAmelCase_       )
																					else:
																												new_pieces.append(UpperCAmelCase_       )

														return new_pieces



							def 		lowerCAmelCase       (				self							:      Optional[int]		,       UpperCAmelCase_							:      Optional[int]       )					->Dict:



														"""simple docstring"""



														return self.sp_model.PieceToId(UpperCAmelCase_       )



							def 		lowerCAmelCase       (				self							:      str		,       UpperCAmelCase_							:      Union[str, Any]       )					->Optional[int]:



														"""simple docstring"""



														return self.sp_model.IdToPiece(UpperCAmelCase_       )



							def 		lowerCAmelCase       (				self							:      str		,       UpperCAmelCase_							:      Dict       )					->Any:



														"""simple docstring"""



														snake_case_           =		[]
														snake_case_           =		""""""
														snake_case_           =		False
														for token in tokens:
																					# make sure that special tokens are not decoded using sentencepiece model
																					if token in self.all_special_tokens:
																												if not prev_is_special:
																																			out_string += " "
																												out_string += self.sp_model.decode(UpperCAmelCase_       ) + token
																												snake_case_           =		True
																												snake_case_           =		[]
																					else:
																												current_sub_tokens.append(UpperCAmelCase_       )
																												snake_case_           =		False
														out_string += self.sp_model.decode(UpperCAmelCase_       )
														return out_string.strip()



							def 		lowerCAmelCase       (				self							:      Tuple		,       UpperCAmelCase_							:      List[int]		,       UpperCAmelCase_							:      Optional[List[int]] = None       )					->List[int]:



														"""simple docstring"""



														snake_case_           =		[self.sep_token_id]
														snake_case_           =		[self.cls_token_id]
														if token_ids_a is None:
																					return cls + token_ids_a + sep
														return cls + token_ids_a + sep + token_ids_a + sep



							def 		lowerCAmelCase       (				self							:      Tuple		,       UpperCAmelCase_							:      List[int]		,       UpperCAmelCase_							:      Optional[List[int]] = None		,       UpperCAmelCase_							:      bool = False       )					->List[int]:



														"""simple docstring"""



														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 not None:
																					return [1] + ([0] * len(UpperCAmelCase_       )) + [1] + ([0] * len(UpperCAmelCase_       )) + [1]
														return [1] + ([0] * len(UpperCAmelCase_       )) + [1]



							def 		lowerCAmelCase       (				self							:      Dict		,       UpperCAmelCase_							:      List[int]		,       UpperCAmelCase_							:      Optional[List[int]] = None       )					->List[int]:



														"""simple docstring"""



														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       ) * [0] + len(token_ids_a + sep       ) * [1]



							def 		lowerCAmelCase       (				self							:      str		,       UpperCAmelCase_							:      str		,       UpperCAmelCase_							:      Optional[str] = None       )					->Tuple[str]:



														"""simple docstring"""



														if not os.path.isdir(UpperCAmelCase_       ):
																					logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
																					return
														snake_case_           =		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:
																												snake_case_           =		self.sp_model.serialized_model_proto()
																												fi.write(UpperCAmelCase_       )

														return (out_vocab_file,)



