






# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


a_							=    get_logger(__name__)


class      lowercase__     :
	a_         ="""dummy_data"""
	a_         ="""datasets"""
	a_         =False

	def __init__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase = None		,    __UpperCAmelCase = False		,    __UpperCAmelCase = True		,    __UpperCAmelCase = None		,    )->					List[Any]:







							'''simple docstring'''





							lowerCAmelCase__       =						0
							lowerCAmelCase__       =						dataset_name
							lowerCAmelCase__       =						cache_dir
							lowerCAmelCase__       =						use_local_dummy_data
							lowerCAmelCase__       =						config
							# download_callbacks take a single url as input
							lowerCAmelCase__       =						download_callbacks or []
							# if False, it doesn't load existing files and it returns the paths of the dummy files relative
							# to the dummy_data zip file root
							lowerCAmelCase__       =						load_existing_dummy_data

							# TODO(PVP, QL) might need to make this more general
							lowerCAmelCase__       =						str(__UpperCAmelCase       )
							# to be downloaded
							lowerCAmelCase__       =						None
							lowerCAmelCase__       =						None

	@property
	def 		UpperCAmelCase  (							self       )->					Union[str, Any]:







							'''simple docstring'''





							if self._dummy_file is None:
													lowerCAmelCase__       =						self.download_dummy_data()
							return self._dummy_file

	@property
	def 		UpperCAmelCase  (							self       )->					List[str]:







							'''simple docstring'''





							if self.config is not None:
													# structure is dummy / config_name / version_name
													return os.path.join("dummy"		,    self.config.name		,    self.version_name       )
							# structure is dummy / version_name
							return os.path.join("dummy"		,    self.version_name       )

	@property
	def 		UpperCAmelCase  (							self       )->					Optional[Any]:







							'''simple docstring'''





							return os.path.join(self.dummy_data_folder		,    "dummy_data.zip"       )

	def 		UpperCAmelCase  (							self       )->					Any:







							'''simple docstring'''





							lowerCAmelCase__       =						(
							    self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
							)

							lowerCAmelCase__       =						cached_path(
							    __UpperCAmelCase		,    cache_dir=self.cache_dir		,    extract_compressed_file=__UpperCAmelCase		,    force_extract=__UpperCAmelCase       )

							return os.path.join(__UpperCAmelCase		,    self.dummy_file_name       )

	@property
	def 		UpperCAmelCase  (							self       )->					str:







							'''simple docstring'''





							return os.path.join(self.datasets_scripts_dir		,    self.dataset_name		,    self.dummy_zip_file       )

	@property
	def 		UpperCAmelCase  (							self       )->					Dict:







							'''simple docstring'''





							if self._bucket_url is None:
													lowerCAmelCase__       =						hf_github_url(self.dataset_name		,    self.dummy_zip_file.replace(os.sep		,    "/"       )       )
							return self._bucket_url

	@property
	def 		UpperCAmelCase  (							self       )->					List[Any]:







							'''simple docstring'''





							if os.path.isdir(self.dummy_file       ):
													return self.dummy_file
							# else cut off path to file -> example `xsum`.
							return "/".join(self.dummy_file.replace(os.sep		,    "/"       ).split("/"       )[:-1]       )

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    *__UpperCAmelCase       )->					Dict:







							'''simple docstring'''





							if self.load_existing_dummy_data:
													# dummy data is downloaded and tested
													lowerCAmelCase__       =						self.dummy_file
							else:
													# dummy data cannot be downloaded and only the path to dummy file is returned
													lowerCAmelCase__       =						self.dummy_file_name

							# special case when data_url is a dict
							if isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ):
													return self.create_dummy_data_dict(__UpperCAmelCase		,    __UpperCAmelCase       )
							elif isinstance(__UpperCAmelCase		,    (list, tuple)       ):
													return self.create_dummy_data_list(__UpperCAmelCase		,    __UpperCAmelCase       )
							else:
													return self.create_dummy_data_single(__UpperCAmelCase		,    __UpperCAmelCase       )

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    *__UpperCAmelCase       )->					Tuple:







							'''simple docstring'''





							return self.download_and_extract(__UpperCAmelCase       )

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase       )->					Union[str, Any]:







							'''simple docstring'''





							return self.download_and_extract(__UpperCAmelCase       )

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    *__UpperCAmelCase		,    **__UpperCAmelCase       )->					Dict:







							'''simple docstring'''





							return path

	def 		UpperCAmelCase  (							self       )->					Dict:







							'''simple docstring'''





							return {}

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase       )->					int:







							'''simple docstring'''





							lowerCAmelCase__       =						{}
							for key, single_urls in data_url.items():
													for download_callback in self.download_callbacks:
																			if isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ):
																									for single_url in single_urls:
																															download_callback(__UpperCAmelCase       )
																			else:
																									lowerCAmelCase__       =						single_urls
																									download_callback(__UpperCAmelCase       )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
													if isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ):
																			lowerCAmelCase__       =						[os.path.join(__UpperCAmelCase		,    urllib.parse.quote_plus(Path(__UpperCAmelCase       ).name       )       ) for x in single_urls]
													else:
																			lowerCAmelCase__       =						single_urls
																			lowerCAmelCase__       =						os.path.join(__UpperCAmelCase		,    urllib.parse.quote_plus(Path(__UpperCAmelCase       ).name       )       )
													lowerCAmelCase__       =						value

							# make sure that values are unique
							if all(isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ) for i in dummy_data_dict.values()       ) and len(set(dummy_data_dict.values()       )       ) < len(
							    dummy_data_dict.values()       ):
													# append key to value to make its name unique
													lowerCAmelCase__       =						{key: value + key for key, value in dummy_data_dict.items()}

							return dummy_data_dict

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase       )->					Any:







							'''simple docstring'''





							lowerCAmelCase__       =						[]
							# trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
							lowerCAmelCase__       =						all(bool(re.findall("[0-9]{3,}-of-[0-9]{3,}"		,    __UpperCAmelCase       )       ) for url in data_url       )
							lowerCAmelCase__       =						all(
							    url.startswith("https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed"       ) for url in data_url       )
							if data_url and (is_tf_records or is_pubmed_records):
													lowerCAmelCase__       =						[data_url[0]] * len(__UpperCAmelCase       )
							for single_url in data_url:
													for download_callback in self.download_callbacks:
																			download_callback(__UpperCAmelCase       )
													# we force the name of each key to be the last file / folder name of the url path
													# if the url has arguments, we need to encode them with urllib.parse.quote_plus
													lowerCAmelCase__       =						os.path.join(__UpperCAmelCase		,    urllib.parse.quote_plus(single_url.split("/"       )[-1]       )       )
													dummy_data_list.append(__UpperCAmelCase       )
							return dummy_data_list

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase       )->					Any:







							'''simple docstring'''





							for download_callback in self.download_callbacks:
													download_callback(__UpperCAmelCase       )
							# we force the name of each key to be the last file / folder name of the url path
							# if the url has arguments, we need to encode them with urllib.parse.quote_plus
							lowerCAmelCase__       =						os.path.join(__UpperCAmelCase		,    urllib.parse.quote_plus(data_url.split("/"       )[-1]       )       )
							if os.path.exists(__UpperCAmelCase       ) or not self.load_existing_dummy_data:
													return value
							else:
													# Backward compatibility, maybe deprecate at one point.
													# For many datasets with single url calls to dl_manager.download_and_extract,
													# the dummy_data.zip file is actually the zipped downloaded file
													# while now we expected the dummy_data.zip file to be a directory containing
													# the downloaded file.
													return path_to_dummy_data

	def 		UpperCAmelCase  (							self       )->					Optional[int]:







							'''simple docstring'''





							pass

	def 		UpperCAmelCase  (							self       )->					List[str]:







							'''simple docstring'''





							pass

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase       )->					Optional[Any]:







							'''simple docstring'''





							def _iter_archive_members(__UpperCAmelCase       ):
													# this preserves the order of the members inside the ZIP archive
													lowerCAmelCase__       =						Path(self.dummy_file       ).parent
													lowerCAmelCase__       =						path.relative_to(__UpperCAmelCase       )
													with ZipFile(self.local_path_to_dummy_data       ) as zip_file:
																			lowerCAmelCase__       =						zip_file.namelist()
													for member in members:
																			if member.startswith(relative_path.as_posix()       ):
																									yield dummy_parent_path.joinpath(__UpperCAmelCase       )

							lowerCAmelCase__       =						Path(__UpperCAmelCase       )
							lowerCAmelCase__       =						_iter_archive_members(__UpperCAmelCase       ) if self.use_local_dummy_data else path.rglob("*"       )
							for file_path in file_paths:
													if file_path.is_file() and not file_path.name.startswith((".", "__")       ):
																			yield file_path.relative_to(__UpperCAmelCase       ).as_posix(), file_path.open("rb"       )


	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase       )->					int:







							'''simple docstring'''





							if not isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ):
													lowerCAmelCase__       =						[paths]
							for path in paths:
													if os.path.isfile(__UpperCAmelCase       ):
																			if os.path.basename(__UpperCAmelCase       ).startswith((".", "__")       ):
																									return
																			yield path
													else:
																			for dirpath, dirnames, filenames in os.walk(__UpperCAmelCase       ):
																									if os.path.basename(__UpperCAmelCase       ).startswith((".", "__")       ):
																															continue
																									dirnames.sort()
																									for filename in sorted(__UpperCAmelCase       ):
																															if filename.startswith((".", "__")       ):
																																					continue
																															yield os.path.join(__UpperCAmelCase		,    __UpperCAmelCase       )














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_							=    False, False, False


@dataclass
class      lowercase__     :
	a_         =None
	a_         =True
	a_         =True
	a_         =None
	# Automatically constructed
	a_         ="dict"
	a_         =pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}	)
	a_         =field(default="""Audio""",					init=_UpperCAmelCase,					repr=_UpperCAmelCase	)

	def __call__(							self       )->					Optional[int]:







							'''simple docstring'''





							return self.pa_type

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

																			lowerCAmelCase__       =						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 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase = None       )->					dict:







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

													with xopen(__UpperCAmelCase		,    "rb"		,    use_auth_token=__UpperCAmelCase       ) as f:
																			lowerCAmelCase__      ,							lowerCAmelCase__       =						sf.read(__UpperCAmelCase       )

							else:
													lowerCAmelCase__      ,							lowerCAmelCase__       =						sf.read(__UpperCAmelCase       )

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

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

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







							'''simple docstring'''





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


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







							'''simple docstring'''





							@no_op_if_value_is_null
							def path_to_bytes(__UpperCAmelCase       ):
													with xopen(__UpperCAmelCase		,    "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(__UpperCAmelCase       ) 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(__UpperCAmelCase		,    self.pa_type       )






