





import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


_lowerCAmelCase 				=     logging.get_logger(__name__)

class    _UpperCAmelCase							:
      a						       =							None






@experimental
def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase     ):
   '''simple docstring'''


   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  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase     ):
   '''simple docstring'''


   A_     :				Optional[Any]			      =	num_proc if num_proc <= len(_lowerCAmelCase     ) else len(_lowerCAmelCase     )
   A_     :				Optional[int]			      =	[]  # We organize the splits ourselve (contiguous splits)
   for index in range(_lowerCAmelCase     ):
      A_     :				Tuple			      =	len(_lowerCAmelCase     ) // num_proc
      A_     :				List[Any]			      =	len(_lowerCAmelCase     ) % num_proc
      A_     :				int			      =	div * index + min(_lowerCAmelCase  ,_lowerCAmelCase     )
      A_     :				Tuple			      =	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]}"""     )
   A_      ,				A_     :				Optional[int]			      =	None, None
   if not disable_tqdm:
      A_      ,				A_     :				Optional[int]			      =	(RLock(),), tqdm.set_lock
   with Pool(_lowerCAmelCase  ,initargs=_lowerCAmelCase  ,initializer=_lowerCAmelCase     ) as pool:
      A_     :				Any			      =	pool.map(_lowerCAmelCase  ,_lowerCAmelCase     )
   logger.info(f"""Finished {num_proc} processes"""     )
   A_     :				List[Any]			      =	[obj for proc_res in mapped for obj in proc_res]
   logger.info(f"""Unpacked {len(_lowerCAmelCase     )} objects"""     )

   return mapped



def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase  ,_lowerCAmelCase     ):
   '''simple docstring'''


   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     ):
   '''simple docstring'''


   A_     :				Union[str, 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:
      A_     :				Dict			      =	None













from typing import List

from .keymap import KEYMAP, get_character



def     _lowerCAmelCase		(  _lowerCAmelCase     ):
   '''simple docstring'''


   def decorator(_lowerCAmelCase     ):
      A_     :				List[Any]			      =	getattr(_lowerCAmelCase  ,"""handle_key"""  ,[]     )
      handle += [key]
      setattr(_lowerCAmelCase  ,"""handle_key"""  ,_lowerCAmelCase     )
      return func

   return decorator



def     _lowerCAmelCase		(  *_lowerCAmelCase     ):
   '''simple docstring'''


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

   return decorator

class    _UpperCAmelCase							(		_lowerCamelCase     ):
      def __new__(		cls      ,       a__      ,       a__      ,       a__       ):
         A_     :				Any			      =	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():
            A_     :				Tuple			      =	getattr(a__      ,       """handle_key"""      ,       []       )
            for key in handled_keys:
               A_     :				Optional[Any]			      =	value
         return new_cls






      @staticmethod
      def 		_lowerCamelCase      (		cls       ):
         A_     :				List[str]			      =	get_character()
         if char != KEYMAP["undefined"]:
            A_     :				str			      =	ord(a__       )
         A_     :				List[str]			      =	cls.key_handler.get(a__       )
         if handler:
            A_     :				Optional[int]			      =	char
            return handler(cls       )
         else:
            return None






def     _lowerCAmelCase		(  cls     ):
   '''simple docstring'''


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






