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


__lowerCAmelCase							,    __lowerCAmelCase							,    __lowerCAmelCase												=	False, False, False


@dataclass
class       _lowerCAmelCase   :







							'''simple docstring'''

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

							def __call__(self  )					->       Any:
														return self.pa_type

							def 	lowercase							(self  ,  UpperCAmelCase  )					->       dict:
														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
																					_snake_case       =  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!)
																																			_snake_case       =  np.frombuffer(value["""bytes"""]  ,  dtype=np.intaa  ).astype(np.floataa  ) / 32767
																												else:
																																			_snake_case       =  np.memmap(value["""path"""]  ,  dtype="""h"""  ,  mode="""r"""  ).astype(np.floataa  ) / 32767

																												_snake_case       =  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 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase = None  )					->       dict:
														if not self.decode:
																					raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""  )

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

																					with xopen(UpperCAmelCase  ,  """rb"""  ,  use_auth_token=UpperCAmelCase  ) as f:
																												_snake_case,							_snake_case       =  sf.read(UpperCAmelCase  )

														else:
																					_snake_case,							_snake_case       =  sf.read(UpperCAmelCase  )

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

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

							def 	lowercase							(self  )					->       Union["FeatureType", Dict[str, "FeatureType"]]:
														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  ,  UpperCAmelCase  )					->       pa.StructArray:
														if pa.types.is_string(storage.type  ):
																					_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.binary()  )
																					_snake_case       =  pa.StructArray.from_arrays([bytes_array, storage]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()  )
														elif pa.types.is_binary(storage.type  ):
																					_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.string()  )
																					_snake_case       =  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       =  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:
																												_snake_case       =  storage.field("""bytes"""  )
																					else:
																												_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.binary()  )
																					if storage.type.get_field_index("""path"""  ) >= 0:
																												_snake_case       =  storage.field("""path"""  )
																					else:
																												_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.string()  )
																					_snake_case       =  pa.StructArray.from_arrays([bytes_array, path_array]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()  )
														return array_cast(UpperCAmelCase  ,  self.pa_type  )





							def 	lowercase							(self  ,  UpperCAmelCase  )					->       pa.StructArray:

														@no_op_if_value_is_null
														def path_to_bytes(UpperCAmelCase  ):
																					with xopen(UpperCAmelCase  ,  """rb"""  ) as f:
																												_snake_case       =  f.read()
																					return bytes_

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







'''simple docstring'''
from typing import TYPE_CHECKING

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


__lowerCAmelCase												=	{
    'configuration_encodec': [
        'ENCODEC_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'EncodecConfig',
    ],
    'feature_extraction_encodec': ['EncodecFeatureExtractor'],
}

try:
				if not is_torch_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				__lowerCAmelCase												=	[
				    '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

				__lowerCAmelCase												=	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)