


"""simple docstring"""







import multiprocessing
import os
from typing import BinaryIO, Optional, Union

import fsspec

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.json.json import Json
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader


class 				SCREAMING_SNAKE_CASE__    (		lowercase    ):
       """simple docstring"""


       def __init__(						self		,					snake_case__		,					snake_case__ = None		,					snake_case__ = None		,					snake_case__ = None		,					snake_case__ = False		,					snake_case__ = False		,					snake_case__ = None		,					snake_case__ = None		,					**snake_case__		,					):






         """simple docstring"""





         super().__init__(
             snake_case__		,					split=snake_case__		,					features=snake_case__		,					cache_dir=snake_case__		,					keep_in_memory=snake_case__		,					streaming=snake_case__		,					num_proc=snake_case__		,					**snake_case__		,					)
         lowerCAmelCase					:					List[Any]								=							field
         lowerCAmelCase					:					Tuple								=							path_or_paths if isinstance(snake_case__		,					snake_case__						) else {self.split: path_or_paths}
         lowerCAmelCase					:					List[Any]								=							Json(
             cache_dir=snake_case__		,					data_files=snake_case__		,					features=snake_case__		,					field=snake_case__		,					**snake_case__		,					)







       def 	lowercase__							(						self						):






         """simple docstring"""





         if self.streaming:
           lowerCAmelCase					:					Optional[Any]								=							self.builder.as_streaming_dataset(split=self.split						)
         # Build regular (map-style) dataset
         else:
           lowerCAmelCase					:					Union[str, Any]								=							None
           lowerCAmelCase					:					Tuple								=							None
           lowerCAmelCase					:					Any								=							None
           lowerCAmelCase					:					Tuple								=							None

           self.builder.download_and_prepare(
               download_config=snake_case__		,					download_mode=snake_case__		,					verification_mode=snake_case__		,					base_path=snake_case__		,					num_proc=self.num_proc		,					)
           lowerCAmelCase					:					Tuple								=							self.builder.as_dataset(
               split=self.split		,					verification_mode=snake_case__		,					in_memory=self.keep_in_memory						)
         return dataset


class 				SCREAMING_SNAKE_CASE__    :
       """simple docstring"""


       def __init__(						self		,					snake_case__		,					snake_case__		,					snake_case__ = None		,					snake_case__ = None		,					**snake_case__		,					):






         """simple docstring"""





         if num_proc is not None and num_proc <= 0:
           raise ValueError(f"""num_proc {num_proc} must be an integer > 0."""						)

         lowerCAmelCase					:					Any								=							dataset
         lowerCAmelCase					:					List[str]								=							path_or_buf
         lowerCAmelCase					:					Optional[int]								=							batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
         lowerCAmelCase					:					Optional[int]								=							num_proc
         lowerCAmelCase					:					Optional[Any]								=							"utf-8"
         lowerCAmelCase					:					Any								=							to_json_kwargs

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Dict								=							self.to_json_kwargs.pop("path_or_buf"		,					snake_case__						)
         lowerCAmelCase					:					Optional[Any]								=							self.to_json_kwargs.pop("orient"		,					"records"						)
         lowerCAmelCase					:					Dict								=							self.to_json_kwargs.pop("lines"		,					True if orient == "records" else False						)
         lowerCAmelCase					:					List[Any]								=							self.to_json_kwargs.pop("index"		,					False if orient in ["split", "table"] else True						)
         lowerCAmelCase					:					Tuple								=							self.to_json_kwargs.pop("compression"		,					snake_case__						)

         if compression not in [None, "infer", "gzip", "bz2", "xz"]:
           raise NotImplementedError(f"""`datasets` currently does not support {compression} compression"""						)

         if isinstance(self.path_or_buf		,					(str, bytes, os.PathLike)						):
           with fsspec.open(self.path_or_buf		,					"wb"		,					compression=snake_case__						) as buffer:
             lowerCAmelCase					:					Any								=							self._write(file_obj=snake_case__		,					orient=snake_case__		,					lines=snake_case__		,					index=snake_case__		,					**self.to_json_kwargs						)
         else:
           if compression:
             raise NotImplementedError(
                 f"""The compression parameter is not supported when writing to a buffer, but compression={compression}"""
                 " was passed. Please provide a local path instead."						)
           lowerCAmelCase					:					Optional[Any]								=							self._write(
               file_obj=self.path_or_buf		,					orient=snake_case__		,					lines=snake_case__		,					index=snake_case__		,					**self.to_json_kwargs						)
         return written

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         lowerCAmelCase						,  lowerCAmelCase						,  lowerCAmelCase						,  lowerCAmelCase						,  lowerCAmelCase					:					str								=							args

         lowerCAmelCase					:					Any								=							query_table(
             table=self.dataset.data		,					key=slice(snake_case__		,					offset + self.batch_size						)		,					indices=self.dataset._indices		,					)
         lowerCAmelCase					:					List[Any]								=							batch.to_pandas().to_json(
             path_or_buf=snake_case__		,					orient=snake_case__		,					lines=snake_case__		,					index=snake_case__		,					**snake_case__						)
         if not json_str.endswith("\n"						):
           json_str += "\n"
         return json_str.encode(self.encoding						)







       def 	lowercase__							(						self		,					snake_case__		,					snake_case__		,					snake_case__		,					snake_case__		,					**snake_case__		,					):






         """simple docstring"""





         lowerCAmelCase					:					Dict								=							0

         if self.num_proc is None or self.num_proc == 1:
           for offset in logging.tqdm(
               range(0		,					len(self.dataset						)		,					self.batch_size						)		,					unit="ba"		,					disable=not logging.is_progress_bar_enabled()		,					desc="Creating json from Arrow format"		,					):
             lowerCAmelCase					:					Union[str, Any]								=							self._batch_json((offset, orient, lines, index, to_json_kwargs)						)
             written += file_obj.write(snake_case__						)
         else:
           lowerCAmelCase						,  lowerCAmelCase					:					Optional[int]								=							len(self.dataset						), self.batch_size
           with multiprocessing.Pool(self.num_proc						) as pool:
             for json_str in logging.tqdm(
                 pool.imap(
                     self._batch_json		,					[(offset, orient, lines, index, to_json_kwargs) for offset in range(0		,					snake_case__		,					snake_case__						)]		,					)		,					total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size		,					unit="ba"		,					disable=not logging.is_progress_bar_enabled()		,					desc="Creating json from Arrow format"		,					):
               written += file_obj.write(snake_case__						)

         return written









"""simple docstring"""







import pytest

from datasets.parallel import ParallelBackendConfig, parallel_backend
from datasets.utils.py_utils import map_nested

from .utils import require_dill_gt_0_3_2, require_joblibspark, require_not_windows
def  a__      (    SCREAMING_SNAKE_CASE   :   str    ):  # picklable for multiprocessing


  '''simple docstring'''







  return i + 1
@require_dill_gt_0_3_2
@require_joblibspark
@require_not_windows
def  a__      (    ):


  '''simple docstring'''







  with parallel_backend("spark"    ):
    assert ParallelBackendConfig.backend_name == "spark"

  lowerCAmelCase					:					List[str]								=							[1, 2, 3]
  with pytest.raises(SCREAMING_SNAKE_CASE    ):
    with parallel_backend("unsupported backend"    ):
      map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=2    )

  with pytest.raises(SCREAMING_SNAKE_CASE    ):
    with parallel_backend("unsupported backend"    ):
      map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=-1    )





@require_dill_gt_0_3_2
@require_joblibspark
@require_not_windows
@pytest.mark.parametrize("num_proc"    ,				[2, -1]    )
def  a__      (    SCREAMING_SNAKE_CASE   :   Tuple    ):


  '''simple docstring'''







  lowerCAmelCase					:					Tuple								=							[1, 2]
  lowerCAmelCase					:					int								=							{"a": 1, "b": 2}
  lowerCAmelCase					:					List[str]								=							{"a": [1, 2], "b": [3, 4]}
  lowerCAmelCase					:					Dict								=							{"a": {"1": 1}, "b": 2}
  lowerCAmelCase					:					Tuple								=							{"a": 1, "b": 2, "c": 3, "d": 4}
  lowerCAmelCase					:					Any								=							[2, 3]
  lowerCAmelCase					:					Any								=							{"a": 2, "b": 3}
  lowerCAmelCase					:					Optional[int]								=							{"a": [2, 3], "b": [4, 5]}
  lowerCAmelCase					:					Optional[int]								=							{"a": {"1": 2}, "b": 3}
  lowerCAmelCase					:					str								=							{"a": 2, "b": 3, "c": 4, "d": 5}

  with parallel_backend("spark"    ):
    assert map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
    assert map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
    assert map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
    assert map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
    assert map_nested(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE    ,				num_proc=SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa





