
"""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


__A						,							__A						,							__A   :							Optional[int]										=   False, False, False
@dataclass
class 			_a :
			"""simple docstring"""
			UpperCamelCase__			      =	None
			UpperCamelCase__			      =	True
			UpperCamelCase__			      =	True
			UpperCamelCase__			      =	None
			# Automatically constructed
			UpperCamelCase__			      =	"dict"
			UpperCamelCase__			      =	pa.struct({"""bytes""": pa.binary(), """path""": pa.string()})
			UpperCamelCase__			      =	field(default="""Audio"""			,							init=__SCREAMING_SNAKE_CASE			,							repr=__SCREAMING_SNAKE_CASE)



			def __call__(		self     :    List[Any]   )->Tuple:
							return self.pa_type



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

															_UpperCAmelCase					=    BytesIO(bytes()   )
															sf.write(_a				,   _a				,   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     :    int				,   __UpperCamelCase     :    Dict				,   __UpperCamelCase     :    Optional[Any] = None   )->str:
							if not self.decode:
											raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''   )

							_UpperCAmelCase		,						_UpperCAmelCase					=    (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

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

											with xopen(_a				,   '''rb'''				,   use_auth_token=_a   ) as f:
															_UpperCAmelCase		,						_UpperCAmelCase					=    sf.read(_a   )

							else:
											_UpperCAmelCase		,						_UpperCAmelCase					=    sf.read(_a   )

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

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



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



			def  lowercase__						(		self     :    Dict				,   __UpperCamelCase     :    Optional[Any]   )->List[Any]:

							@no_op_if_value_is_null
							def path_to_bytes(__UpperCamelCase     :    Optional[int]   ):
											with xopen(_a				,   '''rb'''   ) as f:
															_UpperCAmelCase					=    f.read()
											return bytes_

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








"""simple docstring"""


class 			_a :
			"""simple docstring"""



			def __init__(		self     :    Tuple				,   __UpperCamelCase     :    list[int]   )->None:
							_UpperCAmelCase					=    len(__UpperCamelCase   )
							_UpperCAmelCase					=    [0] * len_array

							if len_array > 0:
											_UpperCAmelCase					=    array[0]

							for i in range(1				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    self.prefix_sum[i - 1] + array[i]



			def  lowercase__						(		self     :    str				,   __UpperCamelCase     :    int				,   __UpperCamelCase     :    int   )->int:
							if start == 0:
											return self.prefix_sum[end]

							return self.prefix_sum[end] - self.prefix_sum[start - 1]



			def  lowercase__						(		self     :    Any				,   __UpperCamelCase     :    int   )->bool:
							_UpperCAmelCase					=    {0}
							for sum_item in self.prefix_sum:
											if sum_item - target_sum in sums:
															return True

											sums.add(__UpperCamelCase   )

							return False


if __name__ == "__main__":
					import doctest

					doctest.testmod()



