


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							:					Tuple     						=     False, False, False






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


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




											'''simple docstring'''

											return self.pa_type


					def 							SCREAMING_SNAKE_CASE_  (						self	:		Union[str, Any]     ,    SCREAMING_SNAKE_CASE	:		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(SCREAMING_SNAKE_CASE     ,    SCREAMING_SNAKE_CASE						):
																	return {"bytes": None, "path": value}
											elif isinstance(SCREAMING_SNAKE_CASE     ,    SCREAMING_SNAKE_CASE						):
																	return {"bytes": value, "path": None}
											elif "array" in value:
																	# convert the audio array to wav bytes
																	__snake_case				=		BytesIO()
																	sf.write(SCREAMING_SNAKE_CASE     ,    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_2_7_6_7
																							else:
																													__snake_case				=		np.memmap(value["path"]     ,    dtype="h"     ,    mode="r"						).astype(np.floataa						) / 3_2_7_6_7

																							__snake_case				=		BytesIO(bytes()						)
																							sf.write(SCREAMING_SNAKE_CASE     ,    SCREAMING_SNAKE_CASE     ,    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	:		Optional[int]     ,    SCREAMING_SNAKE_CASE	:		dict     ,    SCREAMING_SNAKE_CASE	:		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(SCREAMING_SNAKE_CASE						)[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(SCREAMING_SNAKE_CASE     ,    config.HUB_DATASETS_URL						)["repo_id"]
																							__snake_case				=		token_per_repo_id[repo_id]
																	except (ValueError, KeyError):
																							__snake_case				=		None

																	with xopen(SCREAMING_SNAKE_CASE     ,    "rb"     ,    use_auth_token=SCREAMING_SNAKE_CASE						) as f:
																							__snake_case		,					__snake_case				=		sf.read(SCREAMING_SNAKE_CASE						)

											else:
																	__snake_case		,					__snake_case				=		sf.read(SCREAMING_SNAKE_CASE						)

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

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


					def 							SCREAMING_SNAKE_CASE_  (						self	:		List[Any]						)			->	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 							SCREAMING_SNAKE_CASE_  (						self	:		Tuple     ,    SCREAMING_SNAKE_CASE	:		Union[pa.StringArray, pa.StructArray]						)			->	pa.StructArray:




											'''simple docstring'''

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






					def 							SCREAMING_SNAKE_CASE_  (						self	:		Optional[Any]     ,    SCREAMING_SNAKE_CASE	:		pa.StructArray						)			->	pa.StructArray:




											'''simple docstring'''

											@no_op_if_value_is_null
											def path_to_bytes(SCREAMING_SNAKE_CASE	:		Union[str, Any]						):
																	with xopen(SCREAMING_SNAKE_CASE     ,    "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(SCREAMING_SNAKE_CASE						) 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(SCREAMING_SNAKE_CASE     ,    self.pa_type						)









from collections import OrderedDict
from typing import Mapping

from packaging import version

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


A							:					Union[str, Any]     						=     logging.get_logger(__name__)

A							:					Tuple     						=     {
    'facebook/data2vec-vision-base-ft': (
        'https://huggingface.co/facebook/data2vec-vision-base-ft/resolve/main/config.json'
    ),
}






class     UpperCamelCase(  _a   ):
					snake_case_		:				List[Any]										=	"""data2vec-vision"""


					def __init__(						self	:		Any     ,    SCREAMING_SNAKE_CASE	:		List[str]=7_6_8     ,    SCREAMING_SNAKE_CASE	:		Tuple=1_2     ,    SCREAMING_SNAKE_CASE	:		int=1_2     ,    SCREAMING_SNAKE_CASE	:		List[str]=3_0_7_2     ,    SCREAMING_SNAKE_CASE	:		Tuple="gelu"     ,    SCREAMING_SNAKE_CASE	:		Optional[Any]=0.0     ,    SCREAMING_SNAKE_CASE	:		List[Any]=0.0     ,    SCREAMING_SNAKE_CASE	:		Any=0.02     ,    SCREAMING_SNAKE_CASE	:		List[str]=1e-1_2     ,    SCREAMING_SNAKE_CASE	:		Optional[int]=2_2_4     ,    SCREAMING_SNAKE_CASE	:		Optional[int]=1_6     ,    SCREAMING_SNAKE_CASE	:		str=3     ,    SCREAMING_SNAKE_CASE	:		Optional[Any]=False     ,    SCREAMING_SNAKE_CASE	:		Any=False     ,    SCREAMING_SNAKE_CASE	:		Dict=False     ,    SCREAMING_SNAKE_CASE	:		int=False     ,    SCREAMING_SNAKE_CASE	:		Tuple=0.1     ,    SCREAMING_SNAKE_CASE	:		Any=0.1     ,    SCREAMING_SNAKE_CASE	:		Any=True     ,    SCREAMING_SNAKE_CASE	:		List[str]=[3, 5, 7, 1_1]     ,    SCREAMING_SNAKE_CASE	:		int=[1, 2, 3, 6]     ,    SCREAMING_SNAKE_CASE	:		List[str]=True     ,    SCREAMING_SNAKE_CASE	:		Tuple=0.4     ,    SCREAMING_SNAKE_CASE	:		Optional[Any]=2_5_6     ,    SCREAMING_SNAKE_CASE	:		Tuple=1     ,    SCREAMING_SNAKE_CASE	:		Optional[int]=False     ,    SCREAMING_SNAKE_CASE	:		List[Any]=2_5_5     ,    **SCREAMING_SNAKE_CASE	:		Any     ,    )			->	List[str]:




											'''simple docstring'''

											super().__init__(**SCREAMING_SNAKE_CASE						)

											__snake_case				=		hidden_size
											__snake_case				=		num_hidden_layers
											__snake_case				=		num_attention_heads
											__snake_case				=		intermediate_size
											__snake_case				=		hidden_act
											__snake_case				=		hidden_dropout_prob
											__snake_case				=		attention_probs_dropout_prob
											__snake_case				=		initializer_range
											__snake_case				=		layer_norm_eps

											__snake_case				=		image_size
											__snake_case				=		patch_size
											__snake_case				=		num_channels
											__snake_case				=		use_mask_token
											__snake_case				=		use_absolute_position_embeddings
											__snake_case				=		use_relative_position_bias
											__snake_case				=		use_shared_relative_position_bias
											__snake_case				=		layer_scale_init_value
											__snake_case				=		drop_path_rate
											__snake_case				=		use_mean_pooling
											# decode head attributes (semantic segmentation)
											__snake_case				=		out_indices
											__snake_case				=		pool_scales
											# auxiliary head attributes (semantic segmentation)
											__snake_case				=		use_auxiliary_head
											__snake_case				=		auxiliary_loss_weight
											__snake_case				=		auxiliary_channels
											__snake_case				=		auxiliary_num_convs
											__snake_case				=		auxiliary_concat_input
											__snake_case				=		semantic_loss_ignore_index






class     UpperCamelCase(  _a   ):
					snake_case_		:				Any										=	version.parse("""1.11"""   )


					@property
					def 							SCREAMING_SNAKE_CASE_  (						self	:		List[str]						)			->	Mapping[str, Mapping[int, str]]:




											'''simple docstring'''

											return OrderedDict(
											    [
											        ("pixel_values", {0: "batch", 1: "num_channels", 2: "height", 3: "width"}),
											    ]						)






					@property
					def 							SCREAMING_SNAKE_CASE_  (						self	:		Union[str, Any]						)			->	float:




											'''simple docstring'''

											return 1e-4





