




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_				:    List[str]										=     False, False, False

@dataclass
class __lowercase:




					'''simple docstring'''


					__a						:						Optional[int]     		= None
					__a						:						bool     		= True
					__a						:						bool     		= True
					__a						:						Optional[str]     		= None
					# Automatically constructed
					__a						:						ClassVar[str]     		= "dict"
					__a						:						ClassVar[Any]     		= pa.struct({'bytes': pa.binary(), 'path': pa.string()}				)
					__a						:						str     		= field(default='Audio'	,	init=lowercase__	,	repr=lowercase__				)
					def __call__(    self     ):
								return self.pa_type
					def     snake_case_				(    self						,	__a     ):







								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
											__lowerCamelCase					:	Union[str, Any]	       =				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!)
																	__lowerCamelCase					:	int	       =				np.frombuffer(value['bytes']						,	dtype=np.intaa     ).astype(np.floataa     ) / 32767
														else:
																	__lowerCamelCase					:	Dict	       =				np.memmap(value['path']						,	dtype='h'						,	mode='r'     ).astype(np.floataa     ) / 32767

														__lowerCamelCase					:	Tuple	       =				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     snake_case_				(    self						,	__a						,	__a = None     ):







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

								__lowerCamelCase       ,      __lowerCamelCase					:	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

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

											with xopen(__a						,	'rb'						,	use_auth_token=__a     ) as f:
														__lowerCamelCase       ,      __lowerCamelCase					:	Dict	       =				sf.read(__a     )

								else:
											__lowerCamelCase       ,      __lowerCamelCase					:	Optional[Any]	       =				sf.read(__a     )

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

								return {"path": path, "array": array, "sampling_rate": sampling_rate}
					def     snake_case_				(    self     ):
								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						,	__a     ):







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





					def     snake_case_				(    self						,	__a     ):







								@no_op_if_value_is_null
								def path_to_bytes(__a     ):
											with xopen(__a						,	'rb'     ) as f:
														__lowerCamelCase					:	Tuple	       =				f.read()
											return bytes_

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









"""simple docstring"""
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


a_				:    List[Any]										=     {
    '''configuration_encodec''': [
        '''ENCODEC_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''EncodecConfig''',
    ],
    '''feature_extraction_encodec''': ['''EncodecFeatureExtractor'''],
}

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					a_				:    int										=     [
					    '''ENCODEC_PRETRAINED_MODEL_ARCHIVE_LIST''',
					    '''EncodecModel''',
					    '''EncodecPreTrainedModel''',
					]

if TYPE_CHECKING:
					from .configuration_encodec import (
					    ENCODEC_PRETRAINED_CONFIG_ARCHIVE_MAP,
					    EncodecConfig,
					)
					from .feature_extraction_encodec import EncodecFeatureExtractor

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_encodec import (
										    ENCODEC_PRETRAINED_MODEL_ARCHIVE_LIST,
										    EncodecModel,
										    EncodecPreTrainedModel,
										)

else:
					import sys

					a_				:    Tuple										=     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)



