



'''simple docstring'''




import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


UpperCamelCase__       :	int    = logging.get_logger(__name__)






class 						_UpperCamelCase  :


			'''simple docstring'''

			_A      :		Tuple			    =		None


@experimental
def 				lowerCAmelCase_   (	_lowerCamelCase:   int       ,      _lowerCamelCase:   Optional[Any]       ,      _lowerCamelCase:   str       ,      _lowerCamelCase:   Optional[int]       ,      _lowerCamelCase:   Optional[Any]       ,      _lowerCamelCase:   int       ,      _lowerCamelCase:   List[str]					):
		if ParallelBackendConfig.backend_name is None:
				return _map_with_multiprocessing_pool(
				    _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase					)

		return _map_with_joblib(_lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase					)


def 				lowerCAmelCase_   (	_lowerCamelCase:   Optional[int]       ,      _lowerCamelCase:   str       ,      _lowerCamelCase:   int       ,      _lowerCamelCase:   Optional[Any]       ,      _lowerCamelCase:   List[Any]       ,      _lowerCamelCase:   Dict       ,      _lowerCamelCase:   Optional[int]					):
		__SCREAMING_SNAKE_CASE       :		Optional[int]           =		num_proc if num_proc <= len(_lowerCamelCase					) else len(_lowerCamelCase					)
		__SCREAMING_SNAKE_CASE       :		List[Any]           =		[]  # We organize the splits ourselve (contiguous splits)
		for index in range(_lowerCamelCase					):
				__SCREAMING_SNAKE_CASE       :		Any           =		len(_lowerCamelCase					) // num_proc
				__SCREAMING_SNAKE_CASE       :		int           =		len(_lowerCamelCase					) % num_proc
				__SCREAMING_SNAKE_CASE       :		int           =		div * index + min(_lowerCamelCase       ,      _lowerCamelCase					)
				__SCREAMING_SNAKE_CASE       :		Optional[int]           =		start + div + (1 if index < mod else 0)
				split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)					)

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

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

		return mapped


def 				lowerCAmelCase_   (	_lowerCamelCase:   Tuple       ,      _lowerCamelCase:   str       ,      _lowerCamelCase:   Union[str, Any]       ,      _lowerCamelCase:   Union[str, Any]       ,      _lowerCamelCase:   List[Any]       ,      _lowerCamelCase:   Union[str, Any]       ,      _lowerCamelCase:   List[Any]					):
		# progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
		# and it requires monkey-patching joblib internal classes which is subject to change
		import joblib

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


@experimental
@contextlib.contextmanager
def 				lowerCAmelCase_   (	_lowerCamelCase:   str					):
		__SCREAMING_SNAKE_CASE       :		Any           =		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:
				__SCREAMING_SNAKE_CASE       :		Optional[Any]           =		None




'''simple docstring'''




def 				lowerCAmelCase_   (	_lowerCamelCase:   int					):
		assert (
		    isinstance(_lowerCamelCase       ,      _lowerCamelCase					) and number_of_steps > 0
		), F"number_of_steps needs to be positive integer, your input {number_of_steps}"
		if number_of_steps == 1:
				return 1
		__SCREAMING_SNAKE_CASE       ,						__SCREAMING_SNAKE_CASE       :		Dict           =		1, 1
		for _ in range(number_of_steps - 1					):
				__SCREAMING_SNAKE_CASE       ,						__SCREAMING_SNAKE_CASE       :		Tuple           =		current + previous, current
		return current


if __name__ == "__main__":
						import doctest

						doctest.testmod()