





'''simple docstring'''






__lowerCamelCase						: str		=   'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'


def 					_a   (__SCREAMING_SNAKE_CASE   ):


		"""simple docstring"""

		if not isinstance(__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ):
				_UpperCamelCase       			=f'''a bytes-like object is required, not \'{data.__class__.__name__}\''''
				raise TypeError(__SCREAMING_SNAKE_CASE   )

		_UpperCamelCase       			=''''''.join(bin(__SCREAMING_SNAKE_CASE   )[2:].zfill(8   ) for byte in data   )

		_UpperCamelCase       			=len(__SCREAMING_SNAKE_CASE   ) % 6 != 0

		if padding_needed:
				# The padding that will be added later
				_UpperCamelCase       			=b'''=''' * ((6 - len(__SCREAMING_SNAKE_CASE   ) % 6) // 2)

				# Append binary_stream with arbitrary binary digits (0's by default) to make its
				# length a multiple of 6.
				binary_stream += "0" * (6 - len(__SCREAMING_SNAKE_CASE   ) % 6)
		else:
				_UpperCamelCase       			=b''''''

		# Encode every 6 binary digits to their corresponding Base64 character
		return (
		    "".join(
		        B64_CHARSET[int(binary_stream[index : index + 6]  ,      2   )]
		        for index in range(0  ,      len(__SCREAMING_SNAKE_CASE   )  ,      6   )   ).encode()
		    + padding
		)


def 					_a   (__SCREAMING_SNAKE_CASE   ):


		"""simple docstring"""

		if not isinstance(__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ) and not isinstance(__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ):
				_UpperCamelCase       			=(
				    '''argument should be a bytes-like object or ASCII string, '''
				    f'''not \'{encoded_data.__class__.__name__}\''''
				)
				raise TypeError(__SCREAMING_SNAKE_CASE   )

		# In case encoded_data is a bytes-like object, make sure it contains only
		# ASCII characters so we convert it to a string object
		if isinstance(__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ):
				try:
						_UpperCamelCase       			=encoded_data.decode('''utf-8'''   )
				except UnicodeDecodeError:
						raise ValueError('''base64 encoded data should only contain ASCII characters'''   )

		_UpperCamelCase       			=encoded_data.count('''='''   )

		# Check if the encoded string contains non base64 characters
		if padding:
				assert all(
				    char in B64_CHARSET for char in encoded_data[:-padding]   ), "Invalid base64 character(s) found."
		else:
				assert all(
				    char in B64_CHARSET for char in encoded_data   ), "Invalid base64 character(s) found."

		# Check the padding
		assert len(__SCREAMING_SNAKE_CASE   ) % 4 == 0 and padding < 3, "Incorrect padding"

		if padding:
				# Remove padding if there is one
				_UpperCamelCase       			=encoded_data[:-padding]

				_UpperCamelCase       			=''''''.join(
				    bin(B64_CHARSET.index(__SCREAMING_SNAKE_CASE   )   )[2:].zfill(6   ) for char in encoded_data   )[: -padding * 2]
		else:
				_UpperCamelCase       			=''''''.join(
				    bin(B64_CHARSET.index(__SCREAMING_SNAKE_CASE   )   )[2:].zfill(6   ) for char in encoded_data   )

		_UpperCamelCase       			=[
		    int(binary_stream[index : index + 8]  ,      2   )
		    for index in range(0  ,      len(__SCREAMING_SNAKE_CASE   )  ,      8   )
		]

		return bytes(__SCREAMING_SNAKE_CASE   )


if __name__ == "__main__":
						import doctest

						doctest.testmod()













'''simple docstring'''






import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


__lowerCamelCase						: Union[str, Any]		=   logging.get_logger(__name__)







class 		UpperCAmelCase   :


  """simple docstring"""







  lowerCAmelCase_					=				None




@experimental
def 					_a   (__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ):


  """simple docstring"""

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

  return _map_with_joblib(__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   )


def 					_a   (__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ):


  """simple docstring"""

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

  if len(__SCREAMING_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(__SCREAMING_SNAKE_CASE   )}, '''
        f'''length: {sum(len(i[1]   ) for i in split_kwds   )}'''   )

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

  return mapped


def 					_a   (__SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE  ,      __SCREAMING_SNAKE_CASE   ):


  """simple docstring"""

  import joblib

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


@experimental
@contextlib.contextmanager
def 					_a   (__SCREAMING_SNAKE_CASE   ):


  """simple docstring"""

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






