





'''simple docstring'''






import numpy as np  # noqa: E402

from ...configuration_utils import ConfigMixin, register_to_config
from ...schedulers.scheduling_utils import SchedulerMixin


try:
	import librosa  # noqa: E402

	__UpperCamelCase      :   List[Any]								=     True
	__UpperCamelCase      :   str								=     ''''''
except Exception as e:
	__UpperCamelCase      :   Tuple								=     False
	__UpperCamelCase      :   Optional[Any]								=     (
	    f'Cannot import librosa because {e}. Make sure to correctly install librosa to be able to install it.'
	)


from PIL import Image  # noqa: E402



class  lowerCamelCase__				(						snake_case_  ,							snake_case_						):


	"""simple docstring"""





	__magic_name__							   =		"""mel_config.json"""




	@register_to_config
	def __init__(					self   ,     UpperCAmelCase__ = 2_5_6   ,     UpperCAmelCase__ = 2_5_6   ,     UpperCAmelCase__ = 2_2_0_5_0   ,     UpperCAmelCase__ = 2_0_4_8   ,     UpperCAmelCase__ = 5_1_2   ,     UpperCAmelCase__ = 8_0   ,     UpperCAmelCase__ = 3_2   ,     )       ->				Optional[Any]:
					_A :				List[Any]				=  hop_length
					_A :				Dict				=  sample_rate
					_A :				Optional[Any]				=  n_fft
					_A :				Tuple				=  top_db
					_A :				Optional[int]				=  n_iter
					self.set_resolution(UpperCAmelCase__   ,     UpperCAmelCase__			)
					_A :				str				=  None

					if not _librosa_can_be_imported:
									raise ValueError(_import_error			)




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__   ,     UpperCAmelCase__			)       ->				List[str]:
					_A :				List[str]				=  x_res
					_A :				Tuple				=  y_res
					_A :				str				=  self.y_res
					_A :				int				=  self.x_res * self.hop_length - 1




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None			)       ->				Optional[int]:
					if audio_file is not None:
									_A	,       _A :				List[str]				=  librosa.load(UpperCAmelCase__   ,     mono=UpperCAmelCase__   ,     sr=self.sr			)
					else:
									_A :				List[str]				=  raw_audio

					# Pad with silence if necessary.
					if len(self.audio			) < self.x_res * self.hop_length:
									_A :				Optional[Any]				=  np.concatenate([self.audio, np.zeros((self.x_res * self.hop_length - len(self.audio			),)			)]			)




	def 				_lowerCamelCase				(					self			)       ->				int:
					return len(self.audio			) // self.slice_size




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__ = 0			)       ->				np.ndarray:
					return self.audio[self.slice_size * slice : self.slice_size * (slice + 1)]




	def 				_lowerCamelCase				(					self			)       ->				int:
					return self.sr




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__			)       ->				Image.Image:
					_A :				str				=  librosa.feature.melspectrogram(
					    y=self.get_audio_slice(UpperCAmelCase__			)   ,     sr=self.sr   ,     n_fft=self.n_fft   ,     hop_length=self.hop_length   ,     n_mels=self.n_mels			)
					_A :				List[str]				=  librosa.power_to_db(UpperCAmelCase__   ,     ref=np.max   ,     top_db=self.top_db			)
					_A :				str				=  (((log_S + self.top_db) * 2_5_5 / self.top_db).clip(0   ,     2_5_5			) + 0.5).astype(np.uinta			)
					_A :				Optional[Any]				=  Image.fromarray(UpperCAmelCase__			)
					return image




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__			)       ->				np.ndarray:
					_A :				Dict				=  np.frombuffer(image.tobytes()   ,     dtype='''uint8'''			).reshape((image.height, image.width)			)
					_A :				Optional[Any]				=  bytedata.astype('''float'''			) * self.top_db / 2_5_5 - self.top_db
					_A :				int				=  librosa.db_to_power(UpperCAmelCase__			)
					_A :				Optional[Any]				=  librosa.feature.inverse.mel_to_audio(
					    UpperCAmelCase__   ,     sr=self.sr   ,     n_fft=self.n_fft   ,     hop_length=self.hop_length   ,     n_iter=self.n_iter			)
					return audio







'''simple docstring'''






import importlib
import inspect
from functools import wraps
from typing import TYPE_CHECKING, Optional, Union

from .download.streaming_download_manager import (
    xbasename,
    xdirname,
    xet_parse,
    xexists,
    xgetsize,
    xglob,
    xgzip_open,
    xisdir,
    xisfile,
    xjoin,
    xlistdir,
    xnumpy_load,
    xopen,
    xpandas_read_csv,
    xpandas_read_excel,
    xPath,
    xrelpath,
    xsio_loadmat,
    xsplit,
    xsplitext,
    xwalk,
    xxml_dom_minidom_parse,
)
from .utils.logging import get_logger
from .utils.patching import patch_submodule
from .utils.py_utils import get_imports


__UpperCamelCase      :   List[Any]								=     get_logger(__name__)


if TYPE_CHECKING:
	from .builder import DatasetBuilder





def      lowercase  (  lowerCAmelCase :						List[str]   , lowerCAmelCase :						Optional[Union[str, bool]] = None):



				"""simple docstring"""


				_A :				Union[str, Any]				=  importlib.import_module(lowerCAmelCase)

				# TODO(QL): always update the module to add subsequent new authentication
				if hasattr(lowerCAmelCase   , '''_patched_for_streaming''') and module._patched_for_streaming:
								return

				def wrap_auth(lowerCAmelCase :						Union[str, Any]):
								@wraps(lowerCAmelCase)
								def wrapper(*lowerCAmelCase :						Union[str, Any]   , **lowerCAmelCase :						int):
												return function(*lowerCAmelCase   , use_auth_token=lowerCAmelCase   , **lowerCAmelCase)

								_A :				Optional[Any]				=  '''wrap_auth'''
								return wrapper

				# open files in a streaming fashion
				patch_submodule(lowerCAmelCase   , '''open'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.listdir'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.walk'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''glob.glob'''   , wrap_auth(lowerCAmelCase)).start()
				# allow to navigate in remote zip files
				patch_submodule(lowerCAmelCase   , '''os.path.join'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.dirname'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.basename'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.relpath'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.split'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.splitext'''   , lowerCAmelCase).start()
				# allow checks on paths
				patch_submodule(lowerCAmelCase   , '''os.path.exists'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.path.isdir'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.path.isfile'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.path.getsize'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''pathlib.Path'''   , lowerCAmelCase).start()
				# file readers
				patch_submodule(lowerCAmelCase   , '''gzip.open'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''numpy.load'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''pandas.read_csv'''   , wrap_auth(lowerCAmelCase)   , attrs=['''__version__''']).start()
				patch_submodule(lowerCAmelCase   , '''pandas.read_excel'''   , wrap_auth(lowerCAmelCase)   , attrs=['''__version__''']).start()
				patch_submodule(lowerCAmelCase   , '''scipy.io.loadmat'''   , wrap_auth(lowerCAmelCase)   , attrs=['''__version__''']).start()
				patch_submodule(lowerCAmelCase   , '''xml.etree.ElementTree.parse'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''xml.dom.minidom.parse'''   , wrap_auth(lowerCAmelCase)).start()
				_A :				Optional[int]				=  True





def      lowercase  (  lowerCAmelCase :						"DatasetBuilder"):



				"""simple docstring"""


				extend_module_for_streaming(builder.__module__   , use_auth_token=builder.use_auth_token)
				# if needed, we also have to extend additional internal imports (like wmt14 -> wmt_utils)
				if not builder.__module__.startswith('''datasets.'''):  # check that it's not a packaged builder like csv
								for imports in get_imports(inspect.getfile(builder.__class__)):
												if imports[0] == "internal":
																_A :				Optional[int]				=  imports[1]
																_A :				str				=  '''.'''.join(builder.__module__.split('''.''')[:-1] + [internal_import_name])
																extend_module_for_streaming(lowerCAmelCase   , use_auth_token=builder.use_auth_token)

    # builders can inherit from other builders that might use streaming functionality
    # (for example, ImageFolder and AudioFolder inherit from FolderBuilder which implements examples generation)
    # but these parents builders are not patched automatically as they are not instantiated, so we patch them here
				from .builder import DatasetBuilder

				_A :				Tuple				=  [
				    cls.__module__
				    for cls in type(lowerCAmelCase).__mro__[1:]  # make sure it's not the same module we've already patched
				    if issubclass(cls   , lowerCAmelCase) and cls.__module__ != DatasetBuilder.__module__
				]  # check it's not a standard builder from datasets.builder
				for module in parent_builder_modules:
								extend_module_for_streaming(lowerCAmelCase   , use_auth_token=builder.use_auth_token)
