


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    UpperCAmelCase					(       __snake_case    ):




			def __init__(						self							:		List[str]       ,       __magic_name__							:		NestedDataStructureLike[PathLike]       ,       __magic_name__							:		Optional[NamedSplit] = None       ,       __magic_name__							:		Optional[Features] = None       ,       __magic_name__							:		str = None       ,       __magic_name__							:		bool = False       ,       __magic_name__							:		bool = False       ,       __magic_name__							:		Optional[str] = None       ,       __magic_name__							:		Optional[int] = None       ,       **__magic_name__							:		List[Any]       ,       ):







						"""simple docstring"""







						super().__init__(
						    __magic_name__       ,       split=__magic_name__       ,       features=__magic_name__       ,       cache_dir=__magic_name__       ,       keep_in_memory=__magic_name__       ,       streaming=__magic_name__       ,       num_proc=__magic_name__       ,       **__magic_name__       ,       )
						UpperCamelCase         =					field
						UpperCamelCase         =					path_or_paths if isinstance(__magic_name__       ,       __magic_name__						) else {self.split: path_or_paths}
						UpperCamelCase         =					Json(
						    cache_dir=__magic_name__       ,       data_files=__magic_name__       ,       features=__magic_name__       ,       field=__magic_name__       ,       **__magic_name__       ,       )





			def   lowerCamelCase_					(						self							:		Any						):







						"""simple docstring"""







						if self.streaming:
									UpperCamelCase         =					self.builder.as_streaming_dataset(split=self.split						)
						# Build regular (map-style) dataset
						else:
									UpperCamelCase         =					None
									UpperCamelCase         =					None
									UpperCamelCase         =					None
									UpperCamelCase         =					None

									self.builder.download_and_prepare(
									    download_config=__magic_name__       ,       download_mode=__magic_name__       ,       verification_mode=__magic_name__       ,       base_path=__magic_name__       ,       num_proc=self.num_proc       ,       )
									UpperCamelCase         =					self.builder.as_dataset(
									    split=self.split       ,       verification_mode=__magic_name__       ,       in_memory=self.keep_in_memory						)
						return dataset




class    UpperCAmelCase					:




			def __init__(						self							:		Optional[int]       ,       __magic_name__							:		Dataset       ,       __magic_name__							:		Union[PathLike, BinaryIO]       ,       __magic_name__							:		Optional[int] = None       ,       __magic_name__							:		Optional[int] = None       ,       **__magic_name__							:		int       ,       ):







						"""simple docstring"""







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

						UpperCamelCase         =					dataset
						UpperCamelCase         =					path_or_buf
						UpperCamelCase         =					batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
						UpperCamelCase         =					num_proc
						UpperCamelCase         =					"""utf-8"""
						UpperCamelCase         =					to_json_kwargs




			def   lowerCamelCase_					(						self							:		Union[str, Any]						):







						"""simple docstring"""







						UpperCamelCase         =					self.to_json_kwargs.pop("""path_or_buf"""       ,       __magic_name__						)
						UpperCamelCase         =					self.to_json_kwargs.pop("""orient"""       ,       """records"""						)
						UpperCamelCase         =					self.to_json_kwargs.pop("""lines"""       ,       True if orient == """records""" else False						)
						UpperCamelCase         =					self.to_json_kwargs.pop("""index"""       ,       False if orient in ["""split""", """table"""] else True						)
						UpperCamelCase         =					self.to_json_kwargs.pop("""compression"""       ,       __magic_name__						)

						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=__magic_name__						) as buffer:
												UpperCamelCase         =					self._write(file_obj=__magic_name__       ,       orient=__magic_name__       ,       lines=__magic_name__       ,       index=__magic_name__       ,       **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."""						)
									UpperCamelCase         =					self._write(
									    file_obj=self.path_or_buf       ,       orient=__magic_name__       ,       lines=__magic_name__       ,       index=__magic_name__       ,       **self.to_json_kwargs						)
						return written




			def   lowerCamelCase_					(						self							:		Optional[Any]       ,       __magic_name__							:		Any						):







						"""simple docstring"""







						UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase         =					args

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





			def   lowerCamelCase_					(						self							:		Optional[Any]       ,       __magic_name__							:		BinaryIO       ,       __magic_name__							:		Dict       ,       __magic_name__							:		List[str]       ,       __magic_name__							:		Dict       ,       **__magic_name__							:		Dict       ,       ):







						"""simple docstring"""







						UpperCamelCase         =					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"""       ,       ):
												UpperCamelCase         =					self._batch_json((offset, orient, lines, index, to_json_kwargs)						)
												written += file_obj.write(__magic_name__						)
						else:
									UpperCamelCase  ,  UpperCamelCase         =					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       ,       __magic_name__       ,       __magic_name__						)]       ,       )       ,       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(__magic_name__						)

						return written








import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available






def 							_lowercase    (		*SCREAMING_SNAKE_CASE_   :		List[Any]	):





			"""simple docstring"""




			if not isinstance(SCREAMING_SNAKE_CASE_       ,  SCREAMING_SNAKE_CASE_	):
						UpperCamelCase         =					list(SCREAMING_SNAKE_CASE_	)
			for i in range(len(SCREAMING_SNAKE_CASE_	)	):
						UpperCamelCase         =					None
			gc.collect()
			if is_xpu_available():
						torch.xpu.empty_cache()
			elif is_npu_available():
						torch.npu.empty_cache()
			else:
						torch.cuda.empty_cache()
			return objects






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Exception	):





			"""simple docstring"""




			UpperCamelCase         =					[
			    """CUDA out of memory.""",  # CUDA OOM
			    """cuDNN error: CUDNN_STATUS_NOT_SUPPORTED.""",  # CUDNN SNAFU
			    """DefaultCPUAllocator: can't allocate memory""",  # CPU OOM
			]
			if isinstance(SCREAMING_SNAKE_CASE_       ,  SCREAMING_SNAKE_CASE_	) and len(exception.args	) == 1:
						return any(err in exception.args[0] for err in _statements	)
			return False






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		callable = None       ,  SCREAMING_SNAKE_CASE_   :		int = 128	):





			"""simple docstring"""




			if function is None:
						return functools.partial(SCREAMING_SNAKE_CASE_       ,  starting_batch_size=SCREAMING_SNAKE_CASE_	)

			UpperCamelCase         =					starting_batch_size

			def decorator(*SCREAMING_SNAKE_CASE_   :		Union[str, Any]       ,  **SCREAMING_SNAKE_CASE_   :		Tuple	):
						nonlocal batch_size
						gc.collect()
						if is_xpu_available():
									torch.xpu.empty_cache()
						elif is_npu_available():
									torch.npu.empty_cache()
						else:
									torch.cuda.empty_cache()
						UpperCamelCase         =					list(inspect.signature(SCREAMING_SNAKE_CASE_	).parameters.keys()	)
						# Guard against user error
						if len(SCREAMING_SNAKE_CASE_	) < (len(SCREAMING_SNAKE_CASE_	) + 1):
									UpperCamelCase         =					""", """.join([f'{arg}={value}' for arg, value in zip(params[1:]       ,  args[1:]	)]	)
									raise TypeError(
									    f'Batch size was passed into `{function.__name__}` as the first argument when called.'
									    f'Remove this as the decorator already does so: `{function.__name__}({arg_str})`'	)
						while True:
									if batch_size == 0:
												raise RuntimeError("""No executable batch size found, reached zero."""	)
									try:
												return function(SCREAMING_SNAKE_CASE_       ,  *SCREAMING_SNAKE_CASE_       ,  **SCREAMING_SNAKE_CASE_	)
									except Exception as e:
												if should_reduce_batch_size(SCREAMING_SNAKE_CASE_	):
															gc.collect()
															if is_xpu_available():
																		torch.xpu.empty_cache()
															elif is_npu_available():
																		torch.npu.empty_cache()
															else:
																		torch.cuda.empty_cache()
															batch_size //= 2
												else:
															raise

			return decorator




