





import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE

class      lowerCAmelCase							(      __UpperCamelCase    ):
 UpperCAmelCase__									=	""""""
 UpperCAmelCase__									=	(
     None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
 )
 UpperCAmelCase__									=	None  # compression type in fsspec. ex: "gzip"
 UpperCAmelCase__									=	None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz
 def __init__(   self						:   List[Any]    ,  UpperCAmelCase						:   str = ""    ,  UpperCAmelCase						:   Optional[str] = None    ,  UpperCAmelCase						:   Optional[dict] = None    ,  **UpperCAmelCase						:   Any			)    ->			List[Any]:
     super().__init__(self    ,  **UpperCAmelCase			)
     # always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
     lowerCamelCase__					:   List[str]      				=	fsspec.open(
         UpperCAmelCase    ,  mode='rb'    ,  protocol=UpperCAmelCase    ,  compression=self.compression    ,  client_kwargs={
             'requote_redirect_url': False,  # see https://github.com/huggingface/datasets/pull/5459
             'trust_env': True,  # Enable reading proxy env variables.
             **(target_options or {}).pop('client_kwargs'    ,  {}			),  # To avoid issues if it was already passed.
         }    ,  **(target_options or {})    ,  )
     lowerCamelCase__					:   List[str]      				=	os.path.basename(self.file.path.split('::'			)[0]			)
     lowerCamelCase__					:   Tuple      				=	(
         self.compressed_name[: self.compressed_name.rindex('.'			)]
         if '.' in self.compressed_name
         else self.compressed_name
     )
     lowerCamelCase__					:   List[Any]      				=	None
 @classmethod
 def   A_			(   cls						:   Union[str, Any]    ,  UpperCAmelCase						:   Tuple			)    ->			List[Any]:
     # compressed file paths are always relative to the archive root
     return super()._strip_protocol(UpperCAmelCase			).lstrip('/'			)
 def   A_			(   self						:   Union[str, Any]			)    ->			int:
     if self.dir_cache is None:
         lowerCamelCase__					:   Any      				=	{**self.file.fs.info(self.file.path			), 'name': self.uncompressed_name}
         lowerCamelCase__					:   Optional[Any]      				=	{f['name']: f}
 def   A_			(   self						:   Dict    ,  UpperCAmelCase						:   str			)    ->			Optional[Any]:
     return self.file.open().read()







 def   A_			(   self						:   Dict    ,  UpperCAmelCase						:   str    ,  UpperCAmelCase						:   str = "rb"    ,  UpperCAmelCase						:   List[Any]=None    ,  UpperCAmelCase						:   Union[str, Any]=True    ,  UpperCAmelCase						:   Optional[int]=None    ,  **UpperCAmelCase						:   List[str]    ,  )    ->			Union[str, Any]:
     lowerCamelCase__					:   Tuple      				=	self._strip_protocol(UpperCAmelCase			)
     if mode != "rb":
         raise ValueError(F"""Tried to read with mode {mode} on file {self.file.path} opened with mode 'rb'"""			)
     return self.file.open()


class      lowerCAmelCase							(      __UpperCamelCase    ):
 UpperCAmelCase__									=	"""bz2"""
 UpperCAmelCase__									=	"""bz2"""
 UpperCAmelCase__									=	""".bz2"""


class      lowerCAmelCase							(      __UpperCamelCase    ):
 UpperCAmelCase__									=	"""gzip"""
 UpperCAmelCase__									=	"""gzip"""
 UpperCAmelCase__									=	""".gz"""


class      lowerCAmelCase							(      __UpperCamelCase    ):
 UpperCAmelCase__									=	"""lz4"""
 UpperCAmelCase__									=	"""lz4"""
 UpperCAmelCase__									=	""".lz4"""


class      lowerCAmelCase							(      __UpperCamelCase    ):
 UpperCAmelCase__									=	"""xz"""
 UpperCAmelCase__									=	"""xz"""
 UpperCAmelCase__									=	""".xz"""


class      lowerCAmelCase							(      __UpperCamelCase    ):
 UpperCAmelCase__									=	"""zstd"""
 UpperCAmelCase__									=	"""zstd"""
 UpperCAmelCase__									=	""".zst"""
 def __init__(   self						:   int    ,  UpperCAmelCase						:   str    ,  UpperCAmelCase						:   str = "rb"    ,  UpperCAmelCase						:   Optional[str] = None    ,  UpperCAmelCase						:   Optional[dict] = None    ,  UpperCAmelCase						:   int = DEFAULT_BLOCK_SIZE    ,  **UpperCAmelCase						:   List[str]    ,  )    ->			Tuple:
     super().__init__(
         fo=UpperCAmelCase    ,  mode=UpperCAmelCase    ,  target_protocol=UpperCAmelCase    ,  target_options=UpperCAmelCase    ,  block_size=UpperCAmelCase    ,  **UpperCAmelCase    ,  )
     # We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
     #
     # File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
     #     out.close = close
     # AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
     #
     # see https://github.com/intake/filesystem_spec/issues/725
     lowerCamelCase__					:   Optional[Any]      				=	self.file.__enter__

     class      lowerCAmelCase							:
      def __init__(   self						:   str    ,  UpperCAmelCase						:   Optional[int]			)    ->			str:
          lowerCamelCase__					:   str      				=	file_
      def __enter__(   self						:   List[str]			)    ->			Any:
          self._file.__enter__()
          return self
      def __exit__(   self						:   str    ,  *UpperCAmelCase						:   Dict    ,  **UpperCAmelCase						:   Tuple			)    ->			Optional[int]:
          self._file.__exit__(*UpperCAmelCase    ,  **UpperCAmelCase			)
      def __iter__(   self						:   Any			)    ->			Optional[Any]:
          return iter(self._file			)
      def   A_			(   self						:   Tuple			)    ->			Tuple:
          return next(self._file			)







      def __getattr__(   self						:   Dict    ,  UpperCAmelCase						:   Any			)    ->			Dict:
          return getattr(self._file    ,  UpperCAmelCase			)

     def fixed_enter(*UpperCAmelCase						:   Union[str, Any]    ,  **UpperCAmelCase						:   str			):
         return WrappedFile(_enter(*UpperCAmelCase    ,  **UpperCAmelCase			)			)

     lowerCamelCase__					:   List[Any]      				=	fixed_enter












import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


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

class      lowerCAmelCase							:
 UpperCAmelCase__									=	None


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

    return mapped


def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase				)    ->  int:
    # 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=_UpperCAmelCase				):
        return joblib.Parallel()(
            joblib.delayed(_UpperCAmelCase				)((function, obj, types, None, True, None)				) for obj in iterable				)


@experimental
@contextlib.contextmanager
def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase				)    ->  Tuple:
    lowerCamelCase__					:   List[str]      				=	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__					:   str      				=	None





