


"""simple docstring"""

import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


UpperCAmelCase__  							=logging.get_logger(__name__)





class    lowerCamelCase__						:
      a :     str											=		None






@experimental
def lowerCAmelCase_      (				UpperCamelCase__	: List[Any]				,   UpperCamelCase__	: int				,   UpperCamelCase__	: Optional[Any]				,   UpperCamelCase__	: Optional[Any]				,   UpperCamelCase__	: List[str]				,   UpperCamelCase__	: Dict				,   UpperCamelCase__	: Union[str, Any] ):






 """simple docstring"""







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

 return _map_with_joblib(UpperCamelCase__				,   UpperCamelCase__				,   UpperCamelCase__				,   UpperCamelCase__				,   UpperCamelCase__				,   UpperCamelCase__				,   UpperCamelCase__ )


def lowerCAmelCase_      (				UpperCamelCase__	: Optional[int]				,   UpperCamelCase__	: Any				,   UpperCamelCase__	: Any				,   UpperCamelCase__	: Optional[int]				,   UpperCamelCase__	: List[Any]				,   UpperCamelCase__	: int				,   UpperCamelCase__	: int ):






 """simple docstring"""







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

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

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

 return mapped


def lowerCAmelCase_      (				UpperCamelCase__	: List[str]				,   UpperCamelCase__	: Optional[Any]				,   UpperCamelCase__	: List[str]				,   UpperCamelCase__	: List[str]				,   UpperCamelCase__	: Any				,   UpperCamelCase__	: Union[str, Any]				,   UpperCamelCase__	: Dict ):






 """simple docstring"""







 import joblib

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


@experimental
@contextlib.contextmanager
def lowerCAmelCase_      (				UpperCamelCase__	: int ):






 """simple docstring"""







 __lowercase									=	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:
  __lowercase									=	None




"""simple docstring"""

from typing import List

from .keymap import KEYMAP, get_character


def lowerCAmelCase_      (				UpperCamelCase__	: str ):






 """simple docstring"""







 def decorator(UpperCamelCase__	: Tuple ):
  __lowercase									=	getattr(UpperCamelCase__				,   """handle_key"""				,   [] )
  handle += [key]
  setattr(UpperCamelCase__				,   """handle_key"""				,   UpperCamelCase__ )
  return func

 return decorator


def lowerCAmelCase_      (				*UpperCamelCase__	: List[str] ):






 """simple docstring"""







 def decorator(UpperCamelCase__	: Tuple ):
  __lowercase									=	getattr(UpperCamelCase__				,   """handle_key"""				,   [] )
  handle += keys
  setattr(UpperCamelCase__				,   """handle_key"""				,   UpperCamelCase__ )
  return func

 return decorator





class    lowerCamelCase__						(		_a    ):
      def __new__(		cls	:	str      ,       A_	:	Optional[Any]      ,       A_	:	Union[str, Any]      ,       A_	:	int							):






       '''simple docstring'''


       __lowercase									=	super().__new__(cls      ,       A_      ,       A_      ,       A_							)
       if not hasattr(A_      ,       """key_handler"""							):
        setattr(A_      ,       """key_handler"""      ,       {}							)
       setattr(A_      ,       """handle_input"""      ,       KeyHandler.handle_input							)

       for value in attrs.values():
        __lowercase									=	getattr(A_      ,       """handle_key"""      ,       []							)
        for key in handled_keys:
         __lowercase									=	value
       return new_cls






      @staticmethod
      def SCREAMING_SNAKE_CASE_	(		cls	:	Dict							):






       '''simple docstring'''


       __lowercase									=	get_character()
       if char != KEYMAP["undefined"]:
        __lowercase									=	ord(A_							)
       __lowercase									=	cls.key_handler.get(A_							)
       if handler:
        __lowercase									=	char
        return handler(cls							)
       else:
        return None






def lowerCAmelCase_      (				cls	: int ):






 """simple docstring"""







 return KeyHandler(cls.__name__				,   cls.__bases__				,   cls.__dict__.copy() )
