






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


UpperCAmelCase__			,							UpperCAmelCase__			,							UpperCAmelCase__      			=						False, False, False




@dataclass
class 							snake_case_       :

						"""simple docstring"""



						snake_case__        =      None
						snake_case__        =      True
						snake_case__        =      True
						snake_case__        =      None
						# Automatically constructed
						snake_case__        =      "dict"
						snake_case__        =      pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}      )
						snake_case__        =      field(default="""Audio"""		,							init=__UpperCamelCase		,							repr=__UpperCamelCase      )



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




											'''simple docstring'''



											return self.pa_type



						def        UpperCAmelCase__					(self:					List[Any] ,					__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
																__a					:						int      =					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!)
																										__a					:						Optional[Any]      =					np.frombuffer(value["bytes"] ,					dtype=np.intaa   ).astype(np.floataa   ) / 32767
																					else:
																										__a					:						Dict      =					np.memmap(value["path"] ,					dtype="h" ,					mode="r"   ).astype(np.floataa   ) / 32767

																					__a					:						str      =					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        UpperCAmelCase__					(self:					List[str] ,					__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."   )

											__a      ,			__a					:						List[Any]      =					(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					:						Union[str, Any]      =					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:
																__a					:						Union[str, Any]      =					token_per_repo_id or {}
																__a					:						str      =					path.split("::"   )[-1]
																try:
																					__a					:						str      =					string_to_dict(__UpperCAmelCase ,					config.HUB_DATASETS_URL   )["repo_id"]
																					__a					:						int      =					token_per_repo_id[repo_id]
																except (ValueError, KeyError):
																					__a					:						Union[str, Any]      =					None

																with xopen(__UpperCAmelCase ,					"rb" ,					use_auth_token=__UpperCAmelCase   ) as f:
																					__a      ,			__a					:						List[Any]      =					sf.read(__UpperCAmelCase   )

											else:
																__a      ,			__a					:						Union[str, Any]      =					sf.read(__UpperCAmelCase   )

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

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



						def        UpperCAmelCase__					(self:					Tuple   )    ->     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        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					Union[pa.StringArray, pa.StructArray]   )    ->     pa.StructArray:




											'''simple docstring'''



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



						def        UpperCAmelCase__					(self:					Tuple ,					__UpperCAmelCase:					pa.StructArray   )    ->     pa.StructArray:




											'''simple docstring'''



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

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













from abc import ABC, abstractmethod
from argparse import ArgumentParser




class 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						@staticmethod
						@abstractmethod
						def        UpperCAmelCase__					(__UpperCAmelCase:					ArgumentParser   )    ->     Tuple:




											'''simple docstring'''



											raise NotImplementedError()



						@abstractmethod
						def        UpperCAmelCase__					(self:					List[str]   )    ->     List[str]:




											'''simple docstring'''



											raise NotImplementedError()





