






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__   :      Optional[Any]				=		False, False, False
@dataclass
class     lowerCamelCase_					:
				a__       							=	None
				a__       							=	True
				a__       							=	True
				a__       							=	None
				# Automatically constructed
				a__       							=	"dict"
				a__       							=	pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}			)
				a__       							=	field(default='''Audio'''      ,					init=lowerCamelCase      ,					repr=lowerCamelCase			)







				def __call__(    self  ):
								"""simple docstring"""
								return self.pa_type







				def       A						(    self							,       __lowerCAmelCase  ):
								"""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
												__magic_name__		:Tuple   =   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!)
																				__magic_name__		:int   =   np.frombuffer(value['''bytes''']							,       dtype=np.intaa  ).astype(np.floataa  ) / 3_2_7_6_7
																else:
																				__magic_name__		:Optional[int]   =   np.memmap(value['''path''']							,       dtype='''h'''							,       mode='''r'''  ).astype(np.floataa  ) / 3_2_7_6_7

																__magic_name__		:Tuple   =   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       A						(    self							,       __lowerCAmelCase							,       __lowerCAmelCase = None  ):
								"""simple docstring"""
								if not self.decode:
												raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''  )

								__magic_name__  ,					__magic_name__		:Optional[int]   =   (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

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

												with xopen(__lowerCAmelCase							,       '''rb'''							,       use_auth_token=__lowerCAmelCase  ) as f:
																__magic_name__  ,					__magic_name__		:List[str]   =   sf.read(__lowerCAmelCase  )

								else:
												__magic_name__  ,					__magic_name__		:Union[str, Any]   =   sf.read(__lowerCAmelCase  )

								__magic_name__		:Dict   =   array.T
								if self.mono:
												__magic_name__		:Tuple   =   librosa.to_mono(__lowerCAmelCase  )
								if self.sampling_rate and self.sampling_rate != sampling_rate:
												__magic_name__		:Any   =   librosa.resample(__lowerCAmelCase							,       orig_sr=__lowerCAmelCase							,       target_sr=self.sampling_rate  )
												__magic_name__		:Union[str, Any]   =   self.sampling_rate

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







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







				def       A						(    self							,       __lowerCAmelCase  ):
								"""simple docstring"""

								@no_op_if_value_is_null
								def path_to_bytes(__lowerCAmelCase  ):
												with xopen(__lowerCAmelCase							,       '''rb'''  ) as f:
																__magic_name__		:Any   =   f.read()
												return bytes_

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














import pytest
import requests

from datasets.utils.file_utils import http_head

from .utils import OfflineSimulationMode, RequestWouldHangIndefinitelyError, offline

@pytest.mark.integration
def   __lowercase       (			):







				"""simple docstring"""






				with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT			):
								with pytest.raises(snake_case			):
												requests.request('''GET''',	'''https://huggingface.co'''			)
								with pytest.raises(requests.exceptions.ConnectTimeout			):
												requests.request('''GET''',	'''https://huggingface.co''',	timeout=1.0			)

@pytest.mark.integration
def   __lowercase       (			):







				"""simple docstring"""






				with offline(OfflineSimulationMode.CONNECTION_FAILS			):
								with pytest.raises(requests.exceptions.ConnectionError			):
												requests.request('''GET''',	'''https://huggingface.co'''			)

def   __lowercase       (			):







				"""simple docstring"""






				with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1			):
								with pytest.raises(snake_case			):
												http_head('''https://huggingface.co'''			)






