


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


a_							,     a_							,     a_       							=    False, False, False
@dataclass
class snake_case      :
		__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=_UpperCamelCase     ,				repr=_UpperCamelCase)







		def __call__(					self      :       int		)       ->						Any:





						'''simple docstring'''







						return self.pa_type







		def  a_ (					self      :       str    ,   a__      :       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(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
										_A					  =			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!)
																		_A					  =			np.frombuffer(value["bytes"]    ,   dtype=np.intaa		).astype(np.floataa		) / 3_27_67
														else:
																		_A					  =			np.memmap(value["path"]    ,   dtype="h"    ,   mode="r"		).astype(np.floataa		) / 3_27_67

														_A					  =			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  a_ (					self      :       Optional[int]    ,   a__      :       dict    ,   a__      :       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."		)

						_A  ,		_A					  =			(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

						_A					  =			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:
										_A					  =			token_per_repo_id or {}
										_A					  =			path.split("::"		)[-1]
										try:
														_A					  =			string_to_dict(a__    ,   config.HUB_DATASETS_URL		)["repo_id"]
														_A					  =			token_per_repo_id[repo_id]
										except (ValueError, KeyError):
														_A					  =			None

										with xopen(a__    ,   "rb"    ,   use_auth_token=a__		) as f:
														_A  ,		_A					  =			sf.read(a__		)

						else:
										_A  ,		_A					  =			sf.read(a__		)

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

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







		def  a_ (					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  a_ (					self      :       Union[str, Any]    ,   a__      :       Union[pa.StringArray, pa.StructArray]		)       ->						pa.StructArray:





						'''simple docstring'''







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







		def  a_ (					self      :       int    ,   a__      :       pa.StructArray		)       ->						pa.StructArray:





						'''simple docstring'''







						@no_op_if_value_is_null
						def path_to_bytes(a__      :       int		):
										with xopen(a__    ,   "rb"		) as f:
														_A					  =			f.read()
										return bytes_

						_A					  =			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()    ,   )
						_A					  =			pa.array(
						    [os.path.basename(a__		) if path is not None else None for path in storage.field("path"		).to_pylist()]    ,   type=pa.string()    ,   )
						_A					  =			pa.StructArray.from_arrays([bytes_array, path_array]    ,   ["bytes", "path"]    ,   mask=bytes_array.is_null()		)
						return array_cast(a__    ,   self.pa_type		)



"""simple docstring"""


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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer


if TYPE_CHECKING:
		from ...tokenization_utils_base import TextInput
from ...utils import logging


a_       							=    logging.get_logger(__name__)

a_       							=    {"vocab_file": "spiece.model"}

a_       							=    {
    "vocab_file": {
        "t5-small": "https://huggingface.co/t5-small/resolve/main/spiece.model",
        "t5-base": "https://huggingface.co/t5-base/resolve/main/spiece.model",
        "t5-large": "https://huggingface.co/t5-large/resolve/main/spiece.model",
        "t5-3b": "https://huggingface.co/t5-3b/resolve/main/spiece.model",
        "t5-11b": "https://huggingface.co/t5-11b/resolve/main/spiece.model",
    }
}


# TODO(PVP) - this should be removed in Transformers v5
a_       							=    {
    "t5-small": 5_12,
    "t5-base": 5_12,
    "t5-large": 5_12,
    "t5-3b": 5_12,
    "t5-11b": 5_12,
}

a_       							=    "▁"
class snake_case      (  _UpperCamelCase):
		__UpperCamelCase           =			VOCAB_FILES_NAMES
		__UpperCamelCase           =			PRETRAINED_VOCAB_FILES_MAP
		__UpperCamelCase           =			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		__UpperCamelCase           =			['input_ids', 'attention_mask']







		def __init__(					self      :       List[str]    ,   a__      :       Optional[int]    ,   a__      :       Union[str, Any]="</s>"    ,   a__      :       Union[str, Any]="<unk>"    ,   a__      :       str="<pad>"    ,   a__      :       Optional[int]=1_00    ,   a__      :       List[Any]=None    ,   a__      :       Optional[Dict[str, Any]] = None    ,   a__      :       Any=True    ,   **a__      :       Optional[int]    ,   )       ->						None:





						'''simple docstring'''







						if extra_ids > 0 and additional_special_tokens is None:
										_A					  =			[F"""<extra_id_{i}>""" for i in range(a__		)]
						elif extra_ids > 0 and additional_special_tokens is not None:
										# Check that we have the right number of extra_id special tokens
										_A					  =			len(set(filter(lambda a__		: bool("extra_id" in str(a__		)		)    ,   a__		)		)		)
										if extra_tokens != extra_ids:
														raise ValueError(
														    F"""Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"""
														    " provided to T5Tokenizer. In this case the additional_special_tokens must include the extra_ids"
														    " tokens"		)
						if legacy:
										logger.warning_once(
										    F"""You are using the legacy behaviour of the {self.__class__}. This means that tokens that come after special tokens will not be properly handled. We recommend you to"""
										    " read the related pull request available at https://github.com/huggingface/transformers/pull/24565"		)

						_A					  =			legacy
						_A					  =			{} if sp_model_kwargs is None else sp_model_kwargs

						super().__init__(
						    eos_token=a__    ,   unk_token=a__    ,   pad_token=a__    ,   extra_ids=a__    ,   additional_special_tokens=a__    ,   sp_model_kwargs=self.sp_model_kwargs    ,   legacy=a__    ,   **a__    ,   )

						_A					  =			vocab_file
						_A					  =			extra_ids

						_A					  =			spm.SentencePieceProcessor(**self.sp_model_kwargs		)
						self.sp_model.Load(a__		)







		@staticmethod
		def  a_ (					a__      :       List[str]    ,   a__      :       Optional[int]    ,   a__      :       Tuple		)       ->						Tuple:





						'''simple docstring'''







						if pretrained_model_name_or_path in TaTokenizer.max_model_input_sizes:
										_A					  =			TaTokenizer.max_model_input_sizes[pretrained_model_name_or_path]
										if init_max_model_length is not None and init_max_model_length != max_model_length:
														return init_max_model_length
										elif init_max_model_length is None:
														warnings.warn(
														    "This tokenizer was incorrectly instantiated with a model max length of"
														    F""" {deprecated_max_model_length} which will be corrected in Transformers v5.\nFor now, this"""
														    " behavior is kept to avoid breaking backwards compatibility when padding/encoding with"
														    " `truncation is True`.\n- Be aware that you SHOULD NOT rely on"
														    F""" {pretrained_model_name_or_path} automatically truncating your input to"""
														    F""" {deprecated_max_model_length} when padding/encoding.\n- If you want to encode/pad to sequences"""
														    F""" longer than {deprecated_max_model_length} you can either instantiate this tokenizer with"""
														    " `model_max_length` or pass `max_length` when encoding/padding.\n- To avoid this warning, please"
														    " instantiate this tokenizer with `model_max_length` set to your preferred value."    ,   a__    ,   )

						return max_model_length







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





						'''simple docstring'''







						return self.sp_model.get_piece_size() + self._extra_ids







		def  a_ (					self      :       Dict		)       ->						Optional[Any]:





						'''simple docstring'''







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







		def  a_ (					self      :       Optional[Any]    ,   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__		)

						# normal case: some special tokens
						if token_ids_a is None:
										return ([0] * len(a__		)) + [1]
						return ([0] * len(a__		)) + [1] + ([0] * len(a__		)) + [1]







		def  a_ (					self      :       List[str]		)       ->						List[str]:





						'''simple docstring'''







						return list(
						    set(filter(lambda a__		: bool(re.search(r"<extra_id_\d+>"    ,   a__		)		) is not None    ,   self.additional_special_tokens		)		)		)







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





						'''simple docstring'''







						return [self._convert_token_to_id(a__		) for token in self.get_sentinel_tokens()]







		def  a_ (					self      :       List[Any]    ,   a__      :       List[int]		)       ->						List[int]:





						'''simple docstring'''







						if len(a__		) > 0 and token_ids[-1] == self.eos_token_id:
										warnings.warn(
										    F"""This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated"""
										    " eos tokens being added."		)
										return token_ids
						else:
										return token_ids + [self.eos_token_id]







		def  a_ (					self      :       int    ,   a__      :       List[int]    ,   a__      :       Optional[List[int]] = None		)       ->						List[int]:





						'''simple docstring'''







						_A					  =			[self.eos_token_id]

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







		def  a_ (					self      :       Union[str, Any]    ,   a__      :       List[int]    ,   a__      :       Optional[List[int]] = None		)       ->						List[int]:





						'''simple docstring'''







						_A					  =			self._add_eos_if_not_present(a__		)
						if token_ids_a is None:
										return token_ids_a
						else:
										_A					  =			self._add_eos_if_not_present(a__		)
										return token_ids_a + token_ids_a







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





						'''simple docstring'''







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







		def __setstate__(					self      :       int    ,   a__      :       Optional[int]		)       ->						Union[str, Any]:





						'''simple docstring'''







						_A					  =			d

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

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







		def  a_ (					self      :       int    ,   a__      :       "TextInput"    ,   **a__      :       List[str]		)       ->						List[str]:





						'''simple docstring'''







						if not self.legacy:
										_A					  =			SPIECE_UNDERLINE + text.replace(a__    ,   " "		)
						return super().tokenize(a__    ,   **a__		)







		def  a_ (					self      :       str    ,   a__      :       Dict    ,   **a__      :       Optional[int]		)       ->						Any:





						'''simple docstring'''







						if not self.legacy:
										_A					  =			text.startswith(a__		)
										if is_first:
														_A					  =			text[1:]

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

						if not self.legacy and not is_first and not text.startswith(" "		) and tokens[0].startswith(a__		):
										_A					  =			([tokens[0][1:]] if len(tokens[0]		) > 1 else []) + tokens[1:]
						return tokens







		def  a_ (					self      :       int    ,   a__      :       List[Any]		)       ->						List[str]:





						'''simple docstring'''







						if token.startswith("<extra_id_"		):
										_A					  =			re.match(r"<extra_id_(\d+)>"    ,   a__		)
										_A					  =			int(match.group(1		)		)
										return self.vocab_size - num - 1
						return self.sp_model.piece_to_id(a__		)







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





						'''simple docstring'''







						if index < self.sp_model.get_piece_size():
										_A					  =			self.sp_model.IdToPiece(a__		)
						else:
										_A					  =			F"""<extra_id_{self.vocab_size - 1 - index}>"""
						return token







		def  a_ (					self      :       Optional[int]    ,   a__      :       Tuple		)       ->						List[str]:





						'''simple docstring'''







						_A					  =			[]
						_A					  =			""
						_A					  =			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(a__		) + token
														_A					  =			True
														_A					  =			[]
										else:
														current_sub_tokens.append(a__		)
														_A					  =			False
						out_string += self.sp_model.decode(a__		)
						return out_string.strip()







		def  a_ (					self      :       Dict    ,   a__      :       str    ,   a__      :       Optional[str] = None		)       ->						Tuple[str]:





						'''simple docstring'''







						if not os.path.isdir(a__		):
										logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""		)
										return
						_A					  =			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__		) and os.path.isfile(self.vocab_file		):
										copyfile(self.vocab_file    ,   a__		)
						elif not os.path.isfile(self.vocab_file		):
										with open(a__    ,   "wb"		) as fi:
														_A					  =			self.sp_model.serialized_model_proto()
														fi.write(a__		)

						return (out_vocab_file,)