





"""simple docstring"""






import io
import json

import fsspec
import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.json import JsonDatasetReader, JsonDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases

def      __A   (      a_    :int ,							a_    :Union[str, Any])		->		int:
				assert isinstance(a_ ,							a_)
				assert dataset.num_rows == 4
				assert dataset.num_columns == 3
				assert dataset.column_names == ["col_1", "col_2", "col_3"]
				for feature, expected_dtype in expected_features.items():
								assert dataset.features[feature].dtype == expected_dtype

@pytest.mark.parametrize('''keep_in_memory''' ,							[False, True])
def      __A   (      a_    :str ,							a_    :Tuple ,							a_    :Optional[int])		->		Optional[int]:
				__a			:    Union[str, Any]      							=	tmp_path / '''cache'''
				__a			:    Dict      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
								__a			:    Optional[int]      							=	JsonDatasetReader(a_ ,							cache_dir=a_ ,							keep_in_memory=a_).read()
				_check_json_dataset(a_ ,							a_)

@pytest.mark.parametrize(
    '''features''' ,							[
        None,
        {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''},
        {'''col_1''': '''string''', '''col_2''': '''string''', '''col_3''': '''string'''},
        {'''col_1''': '''int32''', '''col_2''': '''int32''', '''col_3''': '''int32'''},
        {'''col_1''': '''float32''', '''col_2''': '''float32''', '''col_3''': '''float32'''},
    ] ,							)
def      __A   (      a_    :List[str] ,							a_    :List[str] ,							a_    :Any)		->		Dict:
				__a			:    Optional[int]      							=	tmp_path / '''cache'''
				__a			:    str      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				__a			:    Tuple      							=	features.copy() if features else default_expected_features
				__a			:    Dict      							=	(
				    Features({feature: Value(a_) for feature, dtype in features.items()}) if features is not None else None
				)
				__a			:    int      							=	JsonDatasetReader(a_ ,							features=a_ ,							cache_dir=a_).read()
				_check_json_dataset(a_ ,							a_)

@pytest.mark.parametrize(
    '''features''' ,							[
        None,
        {'''col_3''': '''float64''', '''col_1''': '''string''', '''col_2''': '''int64'''},
    ] ,							)
def      __A   (      a_    :Any ,							a_    :List[Any] ,							a_    :Union[str, Any])		->		Optional[Any]:
				__a			:    List[str]      							=	tmp_path / '''cache'''
				__a			:    Tuple      							=	{'''col_3''': '''float64''', '''col_1''': '''string''', '''col_2''': '''int64'''}
				__a			:    str      							=	features.copy() if features else default_expected_features
				__a			:    Optional[int]      							=	(
				    Features({feature: Value(a_) for feature, dtype in features.items()}) if features is not None else None
				)
				__a			:    int      							=	JsonDatasetReader(a_ ,							features=a_ ,							cache_dir=a_).read()
				assert isinstance(a_ ,							a_)
				assert dataset.num_rows == 2
				assert dataset.num_columns == 3
				assert dataset.column_names == ["col_3", "col_1", "col_2"]
				for feature, expected_dtype in expected_features.items():
								assert dataset.features[feature].dtype == expected_dtype

def      __A   (      a_    :Optional[Any] ,							a_    :List[str])		->		Tuple:
				# jsonl_312_path features are {"col_3": "float64", "col_1": "string", "col_2": "int64"}
				__a			:    List[Any]      							=	{'''col_2''': '''int64''', '''col_3''': '''float64''', '''col_1''': '''string'''}
				__a			:    Dict      							=	features.copy()
				__a			:    List[Any]      							=	(
				    Features({feature: Value(a_) for feature, dtype in features.items()}) if features is not None else None
				)
				__a			:    List[Any]      							=	tmp_path / '''cache'''
				__a			:    Optional[Any]      							=	JsonDatasetReader(a_ ,							features=a_ ,							cache_dir=a_).read()
				assert isinstance(a_ ,							a_)
				assert dataset.num_rows == 2
				assert dataset.num_columns == 3
				assert dataset.column_names == ["col_2", "col_3", "col_1"]
				for feature, expected_dtype in expected_features.items():
								assert dataset.features[feature].dtype == expected_dtype

@pytest.mark.parametrize('''split''' ,							[None, NamedSplit('''train'''), '''train''', '''test'''])
def      __A   (      a_    :Tuple ,							a_    :Optional[int] ,							a_    :List[Any])		->		int:
				__a			:    Tuple      							=	tmp_path / '''cache'''
				__a			:    Any      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				__a			:    List[str]      							=	JsonDatasetReader(a_ ,							cache_dir=a_ ,							split=a_).read()
				_check_json_dataset(a_ ,							a_)
				assert dataset.split == split if split else "train"

@pytest.mark.parametrize('''path_type''' ,							[str, list])
def      __A   (      a_    :Tuple ,							a_    :Dict ,							a_    :Any)		->		Optional[int]:
				if issubclass(a_ ,							a_):
								__a			:    List[str]      							=	jsonl_path
				elif issubclass(a_ ,							a_):
								__a			:    str      							=	[jsonl_path]
				__a			:    Optional[int]      							=	tmp_path / '''cache'''
				__a			:    List[str]      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				__a			:    Dict      							=	JsonDatasetReader(a_ ,							cache_dir=a_).read()
				_check_json_dataset(a_ ,							a_)

def      __A   (      a_    :int ,							a_    :Union[str, Any] ,							a_    :Any=("train",))		->		List[str]:
				assert isinstance(a_ ,							a_)
				for split in splits:
								__a			:    str      							=	dataset_dict[split]
								assert dataset.num_rows == 4
								assert dataset.num_columns == 3
								assert dataset.column_names == ["col_1", "col_2", "col_3"]
								for feature, expected_dtype in expected_features.items():
												assert dataset.features[feature].dtype == expected_dtype

@pytest.mark.parametrize('''keep_in_memory''' ,							[False, True])
def      __A   (      a_    :int ,							a_    :List[Any] ,							a_    :Any)		->		Optional[Any]:
				__a			:    str      							=	tmp_path / '''cache'''
				__a			:    Union[str, Any]      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
								__a			:    List[Any]      							=	JsonDatasetReader({'''train''': jsonl_path} ,							cache_dir=a_ ,							keep_in_memory=a_).read()
				_check_json_datasetdict(a_ ,							a_)

@pytest.mark.parametrize(
    '''features''' ,							[
        None,
        {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''},
        {'''col_1''': '''string''', '''col_2''': '''string''', '''col_3''': '''string'''},
        {'''col_1''': '''int32''', '''col_2''': '''int32''', '''col_3''': '''int32'''},
        {'''col_1''': '''float32''', '''col_2''': '''float32''', '''col_3''': '''float32'''},
    ] ,							)
def      __A   (      a_    :Tuple ,							a_    :str ,							a_    :Dict)		->		List[str]:
				__a			:    List[str]      							=	tmp_path / '''cache'''
				__a			:    List[Any]      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				__a			:    List[str]      							=	features.copy() if features else default_expected_features
				__a			:    Optional[Any]      							=	(
				    Features({feature: Value(a_) for feature, dtype in features.items()}) if features is not None else None
				)
				__a			:    Optional[int]      							=	JsonDatasetReader({'''train''': jsonl_path} ,							features=a_ ,							cache_dir=a_).read()
				_check_json_datasetdict(a_ ,							a_)

@pytest.mark.parametrize('''split''' ,							[None, NamedSplit('''train'''), '''train''', '''test'''])
def      __A   (      a_    :Dict ,							a_    :Union[str, Any] ,							a_    :Optional[Any])		->		Any:
				if split:
								__a			:    List[Any]      							=	{split: jsonl_path}
				else:
								__a			:    int      							=	'''train'''
								__a			:    str      							=	{'''train''': jsonl_path, '''test''': jsonl_path}
				__a			:    int      							=	tmp_path / '''cache'''
				__a			:    List[Any]      							=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
				__a			:    int      							=	JsonDatasetReader(a_ ,							cache_dir=a_).read()
				_check_json_datasetdict(a_ ,							a_ ,							splits=list(path.keys()))
				assert all(dataset[split].split == split for split in path.keys())

def      __A   (      a_    :Tuple)		->		Optional[Any]:
				return json.load(a_)

def      __A   (      a_    :Optional[Any])		->		Any:
				return [json.loads(a_) for line in buffer]






class 					__lowercase      :







					'''simple docstring'''




					@pytest.mark.parametrize('''lines, load_json_function'''						,      [(True, load_json_lines), (False, load_json)]  )
					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									with io.BytesIO() as buffer:
													JsonDatasetWriter(_UpperCAmelCase						,      _UpperCAmelCase						,      lines=_UpperCAmelCase  ).write()
													buffer.seek(0  )
													__a			:    Dict      							=	load_json_function(_UpperCAmelCase  )
									assert isinstance(_UpperCAmelCase						,      _UpperCAmelCase  )
									assert isinstance(exported_content[0]						,      _UpperCAmelCase  )
									assert len(_UpperCAmelCase  ) == 10







					@pytest.mark.parametrize(
					    '''orient, container, keys, len_at'''						,      [
					        ('''records''', list, {'''tokens''', '''labels''', '''answers''', '''id'''}, None),
					        ('''split''', dict, {'''columns''', '''data'''}, '''data'''),
					        ('''index''', dict, set('''0123456789'''  ), None),
					        ('''columns''', dict, {'''tokens''', '''labels''', '''answers''', '''id'''}, '''tokens'''),
					        ('''values''', list, None, None),
					        ('''table''', dict, {'''schema''', '''data'''}, '''data'''),
					    ]						,      )
					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									with io.BytesIO() as buffer:
													JsonDatasetWriter(_UpperCAmelCase						,      _UpperCAmelCase						,      lines=_UpperCAmelCase						,      orient=_UpperCAmelCase  ).write()
													buffer.seek(0  )
													__a			:    int      							=	load_json(_UpperCAmelCase  )
									assert isinstance(_UpperCAmelCase						,      _UpperCAmelCase  )
									if keys:
													if container is dict:
																	assert exported_content.keys() == keys
													else:
																	assert exported_content[0].keys() == keys
									else:
													assert not hasattr(_UpperCAmelCase						,      '''keys'''  ) and not hasattr(exported_content[0]						,      '''keys'''  )
									if len_at:
													assert len(exported_content[len_at]  ) == 10
									else:
													assert len(_UpperCAmelCase  ) == 10







					@pytest.mark.parametrize('''lines, load_json_function'''						,      [(True, load_json_lines), (False, load_json)]  )
					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									with io.BytesIO() as buffer:
													JsonDatasetWriter(_UpperCAmelCase						,      _UpperCAmelCase						,      lines=_UpperCAmelCase						,      num_proc=2  ).write()
													buffer.seek(0  )
													__a			:    Tuple      							=	load_json_function(_UpperCAmelCase  )
									assert isinstance(_UpperCAmelCase						,      _UpperCAmelCase  )
									assert isinstance(exported_content[0]						,      _UpperCAmelCase  )
									assert len(_UpperCAmelCase  ) == 10







					@pytest.mark.parametrize(
					    '''orient, container, keys, len_at'''						,      [
					        ('''records''', list, {'''tokens''', '''labels''', '''answers''', '''id'''}, None),
					        ('''split''', dict, {'''columns''', '''data'''}, '''data'''),
					        ('''index''', dict, set('''0123456789'''  ), None),
					        ('''columns''', dict, {'''tokens''', '''labels''', '''answers''', '''id'''}, '''tokens'''),
					        ('''values''', list, None, None),
					        ('''table''', dict, {'''schema''', '''data'''}, '''data'''),
					    ]						,      )
					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									with io.BytesIO() as buffer:
													JsonDatasetWriter(_UpperCAmelCase						,      _UpperCAmelCase						,      lines=_UpperCAmelCase						,      orient=_UpperCAmelCase						,      num_proc=2  ).write()
													buffer.seek(0  )
													__a			:    List[str]      							=	load_json(_UpperCAmelCase  )
									assert isinstance(_UpperCAmelCase						,      _UpperCAmelCase  )
									if keys:
													if container is dict:
																	assert exported_content.keys() == keys
													else:
																	assert exported_content[0].keys() == keys
									else:
													assert not hasattr(_UpperCAmelCase						,      '''keys'''  ) and not hasattr(exported_content[0]						,      '''keys'''  )
									if len_at:
													assert len(exported_content[len_at]  ) == 10
									else:
													assert len(_UpperCAmelCase  ) == 10







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase  ):
									with pytest.raises(_UpperCAmelCase  ):
													with io.BytesIO() as buffer:
																	JsonDatasetWriter(_UpperCAmelCase						,      _UpperCAmelCase						,      num_proc=0  )







					@pytest.mark.parametrize('''compression, extension'''						,      [('''gzip''', '''gz'''), ('''bz2''', '''bz2'''), ('''xz''', '''xz''')]  )
					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    Any      							=	tmp_path_factory.mktemp('''data'''  ) / f"""test.json.{extension}"""
									__a			:    int      							=	str(shared_datadir / f"""test_file.json.{extension}"""  )
									JsonDatasetWriter(_UpperCAmelCase						,      _UpperCAmelCase						,      compression=_UpperCAmelCase  ).write()

									with fsspec.open(_UpperCAmelCase						,      '''rb'''						,      compression='''infer'''  ) as f:
													__a			:    List[Any]      							=	f.read()
									with fsspec.open(_UpperCAmelCase						,      '''rb'''						,      compression='''infer'''  ) as f:
													__a			:    Optional[Any]      							=	f.read()
									assert exported_content == original_content


# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
			import jax
			import jaxlib

a_      =     get_logger()

a_      =     None






class _UpperCamelCase				(				TensorFormatter[Mapping, 'jax.Array', Mapping] ):
			'''simple docstring'''







			def __init__(      self				:			Optional[Any]   ,   a				:			str=None   ,   a				:			List[Any]=None   ,   **a				:			Any						)				->							Optional[Any]:

				"""simple docstring"""

				super().__init__(features=a						)
				import jax
				from jaxlib.xla_client import Device

				if isinstance(a   ,   a						):
					raise ValueError(
					    F"Expected {device} to be a `str` not {type(a						)}, as `jaxlib.xla_extension.Device` "
					    "is not serializable neither with `pickle` nor with `dill`. Instead you can surround "
					    "the device with `str()` to get its string identifier that will be internally mapped "
					    "to the actual `jaxlib.xla_extension.Device`."						)
				SCREAMING_SNAKE_CASE :   List[str]					  =							device if isinstance(a   ,   a						) else str(jax.devices()[0]						)
				# using global variable since `jaxlib.xla_extension.Device` is not serializable neither
				# with `pickle` nor with `dill`, so we need to use a global variable instead
				global DEVICE_MAPPING
				if DEVICE_MAPPING is None:
					SCREAMING_SNAKE_CASE :   str					  =							self._map_devices_to_str()
				if self.device not in list(DEVICE_MAPPING.keys()						):
					logger.warning(
					    F"Device with string identifier {self.device} not listed among the available "
					    F"devices: {list(DEVICE_MAPPING.keys()						)}, so falling back to the default "
					    F"device: {str(jax.devices()[0]						)}."						)
					SCREAMING_SNAKE_CASE :   Any					  =							str(jax.devices()[0]						)
				SCREAMING_SNAKE_CASE :   Any					  =							jnp_array_kwargs
			@staticmethod
			def 				__UpperCamelCase			(      )				->							Dict[str, "jaxlib.xla_extension.Device"]:

				"""simple docstring"""

				import jax

				return {str(a						): device for device in jax.devices()}
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			Tuple						)				->							str:

				"""simple docstring"""

				import jax
				import jax.numpy as jnp

				if isinstance(a   ,   a						) and column:
					if all(
					    isinstance(a   ,   jax.Array						) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column						):
						return jnp.stack(a   ,   axis=0						)
				return column
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			str						)				->							str:

				"""simple docstring"""

				import jax
				import jax.numpy as jnp

				if isinstance(a   ,   (str, bytes, type(a						))						):
					return value
				elif isinstance(a   ,   (np.character, np.ndarray)						) and np.issubdtype(value.dtype   ,   np.character						):
					return value.tolist()

				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							{}

				if isinstance(a   ,   (np.number, np.ndarray)						) and np.issubdtype(value.dtype   ,   np.integer						):
					# the default int precision depends on the jax config
					# see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
					if jax.config.jax_enable_xaa:
						SCREAMING_SNAKE_CASE :   Dict					  =							{"dtype": jnp.intaa}
					else:
						SCREAMING_SNAKE_CASE :   str					  =							{"dtype": jnp.intaa}
				elif isinstance(a   ,   (np.number, np.ndarray)						) and np.issubdtype(value.dtype   ,   np.floating						):
					SCREAMING_SNAKE_CASE :   int					  =							{"dtype": jnp.floataa}
				elif config.PIL_AVAILABLE and "PIL" in sys.modules:
					import PIL.Image

					if isinstance(a   ,   PIL.Image.Image						):
						SCREAMING_SNAKE_CASE :   Dict					  =							np.asarray(a						)

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
				global DEVICE_MAPPING
				if DEVICE_MAPPING is None:
					SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self._map_devices_to_str()

				with jax.default_device(DEVICE_MAPPING[self.device]						):
					# calling jnp.array on a np.ndarray does copy the data
					# see https://github.com/google/jax/issues/4486
					return jnp.array(a   ,   **{**default_dtype, **self.jnp_array_kwargs}						)
			def 				__UpperCamelCase			(      self				:			Any   ,   a				:			List[str]						)				->							Dict:

				"""simple docstring"""

				import jax

				# support for torch, tf, jax etc.
				if config.TORCH_AVAILABLE and "torch" in sys.modules:
					import torch

					if isinstance(a   ,   torch.Tensor						):
						return self._tensorize(data_struct.detach().cpu().numpy()[()]						)
				if hasattr(a   ,   "__array__"						) and not isinstance(a   ,   jax.Array						):
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							data_struct.__array__()
				# support for nested types like struct of list of struct
				if isinstance(a   ,   np.ndarray						):
					if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
						return self._consolidate([self.recursive_tensorize(a						) for substruct in data_struct]						)
				elif isinstance(a   ,   (list, tuple)						):
					return self._consolidate([self.recursive_tensorize(a						) for substruct in data_struct]						)
				return self._tensorize(a						)
			def 				__UpperCamelCase			(      self				:			Optional[Any]   ,   a				:			dict						)				->							Dict:

				"""simple docstring"""

				return map_nested(self._recursive_tensorize   ,   a   ,   map_list=a						)
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			pa.Table						)				->							Mapping:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self.numpy_arrow_extractor().extract_row(a						)
				SCREAMING_SNAKE_CASE :   List[Any]					  =							self.python_features_decoder.decode_row(a						)
				return self.recursive_tensorize(a						)
			def 				__UpperCamelCase			(      self				:			Optional[int]   ,   a				:			pa.Table						)				->							"jax.Array":

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[int]					  =							self.numpy_arrow_extractor().extract_column(a						)
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self.python_features_decoder.decode_column(a   ,   pa_table.column_names[0]						)
				SCREAMING_SNAKE_CASE :   Tuple					  =							self.recursive_tensorize(a						)
				SCREAMING_SNAKE_CASE :   Optional[int]					  =							self._consolidate(a						)
				return column






			def 				__UpperCamelCase			(      self				:			List[Any]   ,   a				:			pa.Table						)				->							Mapping:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   str					  =							self.numpy_arrow_extractor().extract_batch(a						)
				SCREAMING_SNAKE_CASE :   str					  =							self.python_features_decoder.decode_batch(a						)
				SCREAMING_SNAKE_CASE :   List[Any]					  =							self.recursive_tensorize(a						)
				for column_name in batch:
					SCREAMING_SNAKE_CASE :   List[Any]					  =							self._consolidate(batch[column_name]						)
				return batch