





# flake8: noqa
# Lint as: python3

from typing import Dict, List, Optional, Type

from .. import config
from ..utils import logging
from .formatting import (
    ArrowFormatter,
    CustomFormatter,
    Formatter,
    PandasFormatter,
    PythonFormatter,
    TensorFormatter,
    format_table,
    query_table,
)
from .np_formatter import NumpyFormatter


_a						=   logging.get_logger(__name__)

_a						=   {}
_a						=   {}
_a						=   {}





def lowerCAmelCase__(__snake_case ,__snake_case ,__snake_case = None ,)       ->   Union[str, Any]:







	'''simple docstring'''







	lowerCamelCase__              =	aliases if aliases is not None else []
	if format_type in _FORMAT_TYPES:
		logger.warning(
		    F'Overwriting format type \'{format_type}\' ({_FORMAT_TYPES[format_type].__name__} -> {formatter_cls.__name__})'							)
	lowerCamelCase__              =	formatter_cls
	for alias in set(aliases + [format_type]							):
		if alias in _FORMAT_TYPES_ALIASES:
			logger.warning(
			    F'Overwriting format type alias \'{alias}\' ({_FORMAT_TYPES_ALIASES[alias]} -> {format_type})'							)
		lowerCamelCase__              =	format_type





def lowerCAmelCase__(__snake_case ,__snake_case ,__snake_case = None							)       ->   List[str]:







	'''simple docstring'''







	lowerCamelCase__              =	aliases if aliases is not None else []
	for alias in set(aliases + [format_type]							):
		lowerCamelCase__              =	unavailable_error


# Here we define all the available formatting functions that can be used by `Dataset.set_format`
_register_formatter(PythonFormatter, None, aliases=["python"])
_register_formatter(ArrowFormatter, "arrow", aliases=["pa", "pyarrow"])
_register_formatter(NumpyFormatter, "numpy", aliases=["np"])
_register_formatter(PandasFormatter, "pandas", aliases=["pd"])
_register_formatter(CustomFormatter, "custom")

if config.TORCH_AVAILABLE:
			from .torch_formatter import TorchFormatter

			_register_formatter(TorchFormatter, "torch", aliases=["pt", "pytorch"])
else:
			_a						=   ValueError("PyTorch needs to be installed to be able to return PyTorch tensors.")
			_register_unavailable_formatter(_torch_error, "torch", aliases=["pt", "pytorch"])

if config.TF_AVAILABLE:
			from .tf_formatter import TFFormatter

			_register_formatter(TFFormatter, "tensorflow", aliases=["tf"])
else:
			_a						=   ValueError("Tensorflow needs to be installed to be able to return Tensorflow tensors.")
			_register_unavailable_formatter(_tf_error, "tensorflow", aliases=["tf"])

if config.JAX_AVAILABLE:
			from .jax_formatter import JaxFormatter

			_register_formatter(JaxFormatter, "jax", aliases=[])
else:
			_a						=   ValueError("JAX needs to be installed to be able to return JAX arrays.")
			_register_unavailable_formatter(_jax_error, "jax", aliases=[])





def lowerCAmelCase__(__snake_case							)       ->   Optional[str]:







	'''simple docstring'''







	if format_type in _FORMAT_TYPES_ALIASES:
		return _FORMAT_TYPES_ALIASES[format_type]
	else:
		return format_type





def lowerCAmelCase__(__snake_case ,**__snake_case							)       ->   Formatter:







	'''simple docstring'''







	lowerCamelCase__              =	get_format_type_from_alias(__snake_case							)
	if format_type in _FORMAT_TYPES:
		return _FORMAT_TYPES[format_type](**__snake_case							)
	if format_type in _FORMAT_TYPES_ALIASES_UNAVAILABLE:
		raise _FORMAT_TYPES_ALIASES_UNAVAILABLE[format_type]
	else:
		raise ValueError(
		    F'Return type should be None or selected in {list(type for type in _FORMAT_TYPES.keys() if type != None							)}, but got \'{format_type}\''							)









import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


_a						=   logging.get_logger(__name__)






class       __A					:

			'''simple docstring'''







			lowerCAmelCase_							  =		None





@experimental
def lowerCAmelCase__(__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case							)       ->   Tuple:







	'''simple docstring'''







	if ParallelBackendConfig.backend_name is None:
		return _map_with_multiprocessing_pool(
		    __snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case							)

	return _map_with_joblib(__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case							)





def lowerCAmelCase__(__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case							)       ->   int:







	'''simple docstring'''







	lowerCamelCase__              =	num_proc if num_proc <= len(__snake_case							) else len(__snake_case							)
	lowerCamelCase__              =	[]  # We organize the splits ourselve (contiguous splits)
	for index in range(__snake_case							):
		lowerCamelCase__              =	len(__snake_case							) // num_proc
		lowerCamelCase__              =	len(__snake_case							) % num_proc
		lowerCamelCase__              =	div * index + min(__snake_case ,__snake_case							)
		lowerCamelCase__              =	start + div + (1 if index < mod else 0)
		split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)							)

	if len(__snake_case							) != sum(len(i[1]							) for i in split_kwds							):
		raise ValueError(
		    F'Error dividing inputs iterable among processes. '
		    F'Total number of objects {len(__snake_case							)}, '
		    F'length: {sum(len(i[1]							) for i in split_kwds							)}'							)

	logger.info(
	    F'Spawning {num_proc} processes for {len(__snake_case							)} objects in slices of {[len(i[1]							) for i in split_kwds]}'							)
	lowerCamelCase__       ,			lowerCamelCase__              =	None, None
	if not disable_tqdm:
		lowerCamelCase__       ,			lowerCamelCase__              =	(RLock(),), tqdm.set_lock
	with Pool(__snake_case ,initargs=__snake_case ,initializer=__snake_case							) as pool:
		lowerCamelCase__              =	pool.map(__snake_case ,__snake_case							)
	logger.info(F'Finished {num_proc} processes'							)
	lowerCamelCase__              =	[obj for proc_res in mapped for obj in proc_res]
	logger.info(F'Unpacked {len(__snake_case							)} objects'							)

	return mapped





def lowerCAmelCase__(__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case ,__snake_case							)       ->   List[str]:







	'''simple docstring'''







	import joblib

	with joblib.parallel_backend(ParallelBackendConfig.backend_name ,n_jobs=__snake_case							):
		return joblib.Parallel()(
		    joblib.delayed(__snake_case							)((function, obj, types, None, True, None)							) for obj in iterable							)





@experimental
@contextlib.contextmanager
def lowerCAmelCase__(__snake_case							)       ->   int:







	'''simple docstring'''







	lowerCamelCase__              =	backend_name

	if backend_name == "spark":
		from joblibspark import register_spark

		register_spark()

		# TODO: call create_cache_and_write_probe if "download" in steps
		# TODO: raise NotImplementedError when Dataset.map etc is called

	try:
		yield
	finally:
		lowerCamelCase__              =	None



