






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       snake_case__ :

       '''simple docstring'''
       lowerCamelCase				:					Optional[int]								=			None
       lowerCamelCase				:					bool								=			True
       lowerCamelCase				:					bool								=			True
       lowerCamelCase				:					Optional[str]								=			None
       # Automatically constructed
       lowerCamelCase				:					ClassVar[str]								=			"dict"
       lowerCamelCase				:					ClassVar[Any]								=			pa.struct({"bytes": pa.binary(), "path": pa.string()})
       lowerCamelCase				:					str								=			field(default="Audio"						,     init=lowerCamelCase__						,     repr=lowerCamelCase__)



       def __call__(				self     )			->   Any:






            '''simple docstring'''





            return self.pa_type



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

                      __snake_case      :Any 							=    BytesIO(bytes()     )
                      sf.write(snake_case__      ,    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        __lowercase    (				self      ,    a__      ,    a__ = None     )			->   dict:






            '''simple docstring'''





            if not self.decode:
                 raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""     )

            __snake_case      :Union[str, 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

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

                 with xopen(snake_case__      ,    """rb"""      ,    use_auth_token=snake_case__     ) as f:
                      __snake_case      :Optional[int] 							=    sf.read(snake_case__     )

            else:
                 __snake_case      :Union[str, Any] 							=    sf.read(snake_case__     )

            __snake_case      :Dict 							=    array.T
            if self.mono:
                 __snake_case      :List[Any] 							=    librosa.to_mono(snake_case__     )
            if self.sampling_rate and self.sampling_rate != sampling_rate:
                 __snake_case      :Optional[Any] 							=    librosa.resample(snake_case__      ,    orig_sr=snake_case__      ,    target_sr=self.sampling_rate     )
                 __snake_case      :Dict 							=    self.sampling_rate

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



       def        __lowercase    (				self     )			->   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        __lowercase    (				self      ,    a__     )			->   pa.StructArray:






            '''simple docstring'''





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



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






            '''simple docstring'''





            @no_op_if_value_is_null
            def path_to_bytes(a__     ):
                 with xopen(snake_case__      ,    """rb"""     ) as f:
                      __snake_case      :Optional[int] 							=    f.read()
                 return bytes_

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

							"""simple docstring"""



							with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT				):
														with pytest.raises(lowercase				):
																					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  __magic_name__  (				)				-> Union[str, Any]:

							"""simple docstring"""



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





def  __magic_name__  (				)				-> Dict:

							"""simple docstring"""



							with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1				):
														with pytest.raises(lowercase				):
																					http_head("""https://huggingface.co"""				)