




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


UpperCamelCase__    ,			UpperCamelCase__    ,			UpperCamelCase__					    =    False, False, False
@dataclass
class     a__					:
						_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=snake_case__				,       repr=snake_case__   )






						def __call__(     self    ):




											"""simple docstring"""



											return self.pa_type






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""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(_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            = 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            = np.frombuffer(value["bytes"]	,    dtype=np.intaa    ).astype(np.floataa    ) / 3_2_7_6_7
																					else:
																										__lowerCAmelCase            = np.memmap(value["path"]	,    dtype="h"	,    mode="r"    ).astype(np.floataa    ) / 3_2_7_6_7

																					__lowerCAmelCase            = 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        __SCREAMING_SNAKE_CASE(     self	,    _A	,    _A = None    ):




											"""simple docstring"""



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

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

																with xopen(_A	,    "rb"	,    use_auth_token=_A    ) as f:
																					__lowerCAmelCase     ,							__lowerCAmelCase            = sf.read(_A    )

											else:
																__lowerCAmelCase     ,							__lowerCAmelCase            = sf.read(_A    )

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

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






						def        __SCREAMING_SNAKE_CASE(     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        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



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






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



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

											__lowerCAmelCase            = 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            = 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            = pa.StructArray.from_arrays([bytes_array, path_array]	,    ["bytes", "path"]	,    mask=bytes_array.is_null()    )
											return array_cast(_A	,    self.pa_type    )










import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
UpperCamelCase__					    =    """."""


if __name__ == "__main__":
					UpperCamelCase__					    =    os.path.join(REPO_PATH, """utils/documentation_tests.txt""")
					UpperCamelCase__					    =    []
					UpperCamelCase__					    =    []
					with open(doctest_file_path) as fp:
										for line in fp:
															UpperCamelCase__					    =    line.strip()
															UpperCamelCase__					    =    os.path.join(REPO_PATH, line)
															if not (os.path.isfile(path) or os.path.isdir(path)):
																				non_existent_paths.append(line)
															all_paths.append(path)
					if len(non_existent_paths) > 0:
										UpperCamelCase__					    =    """\n""".join(non_existent_paths)
										raise ValueError(f'''`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}''')
					if all_paths != sorted(all_paths):
										raise ValueError("""Files in `utils/documentation_tests.txt` are not in alphabetical order.""")




