





'''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_       :
							lowerCamelCase_           =					None
							lowerCamelCase_           =					True
							lowerCamelCase_           =					True
							lowerCamelCase_           =					None
							# Automatically constructed
							lowerCamelCase_           =					"dict"
							lowerCamelCase_           =					pa.struct({"bytes": pa.binary(), "path": pa.string()}    )
							lowerCamelCase_           =					field(default="Audio"				,  init=_snake_case				,  repr=_snake_case    )


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



									"""simple docstring"""



									return self.pa_type


							def      _snake_case  (							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(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
											SCREAMING_SNAKE_CASE__			=      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!)
															SCREAMING_SNAKE_CASE__			=      np.frombuffer(value["""bytes"""]							,				dtype=np.intaa		).astype(np.floataa		) / 3_2767
													else:
															SCREAMING_SNAKE_CASE__			=      np.memmap(value["""path"""]							,				dtype="""h"""							,				mode="""r"""		).astype(np.floataa		) / 3_2767

													SCREAMING_SNAKE_CASE__			=      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      _snake_case  (							self    :Any							,				__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."""		)

									SCREAMING_SNAKE_CASE__	,						SCREAMING_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

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

											with xopen(lowerCAmelCase__							,				"""rb"""							,				use_auth_token=lowerCAmelCase__		) as f:
													SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      sf.read(lowerCAmelCase__		)

									else:
											SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      sf.read(lowerCAmelCase__		)

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

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


							def      _snake_case  (							self    :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      _snake_case  (							self    :Dict							,				__A    :Union[pa.StringArray, pa.StructArray]		)   ->   pa.StructArray:



									"""simple docstring"""



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


							def      _snake_case  (							self    :Dict							,				__A    :pa.StructArray		)   ->   pa.StructArray:



									"""simple docstring"""



									@no_op_if_value_is_null
									def path_to_bytes(__A    :List[Any]		):
											with xopen(lowerCAmelCase__							,				"""rb"""		) as f:
													SCREAMING_SNAKE_CASE__			=      f.read()
											return bytes_

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






from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


_lowerCamelCase   				=			logging.get_logger(__name__)

_lowerCamelCase   				=			{
    'shi-labs/nat-mini-in1k-224': 'https://huggingface.co/shi-labs/nat-mini-in1k-224/resolve/main/config.json',
    # See all Nat models at https://huggingface.co/models?filter=nat
}




class   UpperCamelCase_       (     UpperCamelCase__				,  UpperCamelCase__    ):
							lowerCamelCase_           =					"nat"

							lowerCamelCase_           =					{
							    "num_attention_heads": "num_heads",
							    "num_hidden_layers": "num_layers",
							}


							def __init__(							self    :List[Any]							,				__A    :Optional[Any]=4							,				__A    :Any=3							,				__A    :Optional[int]=64							,				__A    :Optional[int]=[3, 4, 6, 5]							,				__A    :Union[str, Any]=[2, 4, 8, 16]							,				__A    :Optional[Any]=7							,				__A    :Optional[Any]=3.0							,				__A    :List[Any]=True							,				__A    :int=0.0							,				__A    :Dict=0.0							,				__A    :Optional[Any]=0.1							,				__A    :str="gelu"							,				__A    :Optional[Any]=0.0_2							,				__A    :Optional[int]=1E-5							,				__A    :Optional[int]=0.0							,				__A    :Optional[Any]=None							,				__A    :Union[str, Any]=None							,				**__A    :Union[str, Any]							,				)   ->   Optional[int]:



									"""simple docstring"""



									super().__init__(**__A		)

									SCREAMING_SNAKE_CASE__			=      patch_size
									SCREAMING_SNAKE_CASE__			=      num_channels
									SCREAMING_SNAKE_CASE__			=      embed_dim
									SCREAMING_SNAKE_CASE__			=      depths
									SCREAMING_SNAKE_CASE__			=      len(__A		)
									SCREAMING_SNAKE_CASE__			=      num_heads
									SCREAMING_SNAKE_CASE__			=      kernel_size
									SCREAMING_SNAKE_CASE__			=      mlp_ratio
									SCREAMING_SNAKE_CASE__			=      qkv_bias
									SCREAMING_SNAKE_CASE__			=      hidden_dropout_prob
									SCREAMING_SNAKE_CASE__			=      attention_probs_dropout_prob
									SCREAMING_SNAKE_CASE__			=      drop_path_rate
									SCREAMING_SNAKE_CASE__			=      hidden_act
									SCREAMING_SNAKE_CASE__			=      layer_norm_eps
									SCREAMING_SNAKE_CASE__			=      initializer_range
									# we set the hidden_size attribute in order to make Nat work with VisionEncoderDecoderModel
									# this indicates the channel dimension after the last stage of the model
									SCREAMING_SNAKE_CASE__			=      int(embed_dim * 2 ** (len(__A		) - 1)		)
									SCREAMING_SNAKE_CASE__			=      layer_scale_init_value
									SCREAMING_SNAKE_CASE__			=      ["""stem"""] + [f'''stage{idx}''' for idx in range(1							,				len(__A		) + 1		)]
									SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      get_aligned_output_features_output_indices(
									    out_features=__A							,				out_indices=__A							,				stage_names=self.stage_names		)