





import copy
import os
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest

from datasets.arrow_writer import ArrowWriter, OptimizedTypedSequence, ParquetWriter, TypedSequence
from datasets.features import ArrayaD, ClassLabel, Features, Image, Value
from datasets.features.features import ArrayaDExtensionType, cast_to_python_objects
from datasets.keyhash import DuplicatedKeysError, InvalidKeyError

from .utils import require_pil
class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									lowercase_	:   Any  				=     pa.array(TypedSequence([1, 2, 3]  )  )
									self.assertEqual(arr.type,   pa.intaa()  )






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									with self.assertRaises(snake_case__  ):
																lowercase_	:   Union[str, Any]  				=     pa.array(TypedSequence([1, 2, 3]  ),   type=pa.intaa()  )






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									with self.assertRaises(snake_case__  ):
																lowercase_	:   List[str]  				=     pa.array(TypedSequence([1, 2, 3],   try_type=Value("""bool"""  ),   type=Value("""int64"""  )  )  )






		def 						snake_case__				(							self  )       -> Tuple:






									"""simple docstring"""
									lowercase_	:   Optional[int]  				=     pa.array(TypedSequence([1, 2, 3],   type=Value("""int32"""  )  )  )
									self.assertEqual(arr.type,   pa.intaa()  )






		def 						snake_case__				(							self  )       -> Optional[int]:






									"""simple docstring"""
									with self.assertRaises((TypeError, pa.lib.ArrowInvalid)  ):
																lowercase_	:   Any  				=     pa.array(TypedSequence(["""foo""", """bar"""],   type=Value("""int64"""  )  )  )






		def 						snake_case__				(							self  )       -> Tuple:






									"""simple docstring"""
									lowercase_	:   Union[str, Any]  				=     pa.array(TypedSequence([1, 2, 3],   try_type=Value("""int32"""  )  )  )
									self.assertEqual(arr.type,   pa.intaa()  )






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									lowercase_	:   Tuple  				=     pa.array(TypedSequence(["""foo""", """bar"""],   try_type=Value("""int64"""  )  )  )
									self.assertEqual(arr.type,   pa.string()  )






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     pa.array(TypedSequence([[[1, 2, 3]]],   type=ArrayaD((1, 3),   """int64"""  )  )  )
									self.assertEqual(arr.type,   ArrayaDExtensionType((1, 3),   """int64"""  )  )






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									with self.assertRaises((TypeError, pa.lib.ArrowInvalid)  ):
																lowercase_	:   Tuple  				=     pa.array(TypedSequence(["""foo""", """bar"""],   type=ArrayaD((1, 3),   """int64"""  )  )  )






		def 						snake_case__				(							self  )       -> int:






									"""simple docstring"""
									lowercase_	:   Dict  				=     pa.array(TypedSequence([[[1, 2, 3]]],   try_type=ArrayaD((1, 3),   """int64"""  )  )  )
									self.assertEqual(arr.type,   ArrayaDExtensionType((1, 3),   """int64"""  )  )






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									lowercase_	:   int  				=     pa.array(TypedSequence(["""foo""", """bar"""],   try_type=ArrayaD((1, 3),   """int64"""  )  )  )
									self.assertEqual(arr.type,   pa.string()  )






		@require_pil
		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									import PIL.Image

									lowercase_	:   Tuple  				=     PIL.Image.fromarray(np.arange(10,   dtype=np.uinta  ).reshape(2,   5  )  )
									with patch(
									    """datasets.arrow_writer.cast_to_python_objects""",   side_effect=snake_case__  ) as mock_cast_to_python_objects:
																lowercase_	:   List[str]  				=     pa.array(TypedSequence([{"""path""": None, """bytes""": B"""image_bytes"""}, pil_image],   type=Image()  )  )
																lowercase_				,		lowercase_	:   Optional[Any]  				=     mock_cast_to_python_objects.call_args_list[-1]
																self.assertIn("""optimize_list_casting""",   snake_case__  )
																self.assertFalse(kwargs["""optimize_list_casting"""]  )





def  __magic_name__  (				lowercase				,       lowercase				)				-> Optional[int]:

							"""simple docstring"""



							lowercase_	:   List[str]  				=     pa.BufferReader(lowercase				) if isinstance(lowercase				,       pa.Buffer				) else pa.memory_map(lowercase				)
							lowercase_	:   Optional[Any]  				=     pa.ipc.open_stream(lowercase				)
							lowercase_	:   pa.Table  				=     f.read_all()
							assert len(pa_table.to_batches()				) == expected_num_chunks
							assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}
							del pa_table





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 1, 10]				)
@pytest.mark.parametrize(
    """fields"""				,       [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]				)
def  __magic_name__  (				lowercase				,       lowercase				)				-> Union[str, Any]:

							"""simple docstring"""



							lowercase_	:   Dict  				=     pa.BufferOutputStream()
							lowercase_	:   Optional[int]  				=     pa.schema(lowercase				) if fields else None
							with ArrowWriter(stream=lowercase				,       schema=lowercase				,       writer_batch_size=lowercase				) as writer:
														writer.write({"""col_1""": """foo""", """col_2""": 1}				)
														writer.write({"""col_1""": """bar""", """col_2""": 2}				)
														lowercase_				,		lowercase_	:   Any  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowercase_	:   Union[str, Any]  				=     {"""col_1""": pa.string(), """col_2""": pa.intaa()}
							assert writer._schema == pa.schema(lowercase				,       metadata=writer._schema.metadata				)
							_check_output(output.getvalue()				,       expected_num_chunks=num_examples if writer_batch_size == 1 else 1				)





def  __magic_name__  (				)				-> Dict:

							"""simple docstring"""



							lowercase_	:   Any  				=     pa.BufferOutputStream()
							lowercase_	:   Optional[int]  				=     Features({"""labels""": ClassLabel(names=["""neg""", """pos"""]				)}				)
							with ArrowWriter(stream=lowercase				,       features=lowercase				) as writer:
														writer.write({"""labels""": 0}				)
														writer.write({"""labels""": 1}				)
														lowercase_				,		lowercase_	:   Tuple  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							assert writer._schema == features.arrow_schema
							assert writer._schema.metadata == features.arrow_schema.metadata
							lowercase_	:   Any  				=     pa.BufferReader(output.getvalue()				)
							lowercase_	:   Tuple  				=     pa.ipc.open_stream(lowercase				)
							lowercase_	:   pa.Table  				=     f.read_all()
							lowercase_	:   List[Any]  				=     pa_table.schema
							assert pa_table.num_rows == 2
							assert schema == features.arrow_schema
							assert schema.metadata == features.arrow_schema.metadata
							assert features == Features.from_arrow_schema(lowercase				)





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 1, 10]				)
def  __magic_name__  (				lowercase				)				-> Tuple:

							"""simple docstring"""



							lowercase_	:   Tuple  				=     pa.BufferOutputStream()
							with ArrowWriter(
							    stream=lowercase				,       writer_batch_size=lowercase				,       hash_salt="""split_name"""				,       check_duplicates=lowercase				,       ) as writer:
														with pytest.raises(lowercase				):
																					writer.write({"""col_1""": """foo""", """col_2""": 1}				,       key=[1, 2]				)
																					lowercase_				,		lowercase_	:   str  				=     writer.finalize()





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 2, 10]				)
def  __magic_name__  (				lowercase				)				-> int:

							"""simple docstring"""



							lowercase_	:   Optional[Any]  				=     pa.BufferOutputStream()
							with ArrowWriter(
							    stream=lowercase				,       writer_batch_size=lowercase				,       hash_salt="""split_name"""				,       check_duplicates=lowercase				,       ) as writer:
														with pytest.raises(lowercase				):
																					writer.write({"""col_1""": """foo""", """col_2""": 1}				,       key=10				)
																					writer.write({"""col_1""": """bar""", """col_2""": 2}				,       key=10				)
																					lowercase_				,		lowercase_	:   Any  				=     writer.finalize()





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 2, 10]				)
def  __magic_name__  (				lowercase				)				-> List[Any]:

							"""simple docstring"""



							lowercase_	:   Union[str, Any]  				=     pa.BufferOutputStream()
							with ArrowWriter(
							    stream=lowercase				,       writer_batch_size=lowercase				,       hash_salt="""split_name"""				,       check_duplicates=lowercase				,       ) as writer:
														writer.write({"""col_1""": """foo""", """col_2""": 1}				,       key=1				)
														writer.write({"""col_1""": """bar""", """col_2""": 2}				,       key=2				)
														lowercase_				,		lowercase_	:   List[Any]  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							_check_output(output.getvalue()				,       expected_num_chunks=num_examples if writer_batch_size == 1 else 1				)





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 1, 10]				)
@pytest.mark.parametrize(
    """fields"""				,       [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]				)
def  __magic_name__  (				lowercase				,       lowercase				)				-> List[Any]:

							"""simple docstring"""



							lowercase_	:   Tuple  				=     pa.BufferOutputStream()
							lowercase_	:   List[str]  				=     pa.schema(lowercase				) if fields else None
							with ArrowWriter(stream=lowercase				,       schema=lowercase				,       writer_batch_size=lowercase				) as writer:
														writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}				)
														writer.write_batch({"""col_1""": [], """col_2""": []}				)
														lowercase_				,		lowercase_	:   List[str]  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowercase_	:   int  				=     {"""col_1""": pa.string(), """col_2""": pa.intaa()}
							assert writer._schema == pa.schema(lowercase				,       metadata=writer._schema.metadata				)
							_check_output(output.getvalue()				,       expected_num_chunks=num_examples if writer_batch_size == 1 else 1				)





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 1, 10]				)
@pytest.mark.parametrize(
    """fields"""				,       [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]				)
def  __magic_name__  (				lowercase				,       lowercase				)				-> Optional[int]:

							"""simple docstring"""



							lowercase_	:   Any  				=     pa.BufferOutputStream()
							lowercase_	:   Dict  				=     pa.schema(lowercase				) if fields else None
							with ArrowWriter(stream=lowercase				,       schema=lowercase				,       writer_batch_size=lowercase				) as writer:
														writer.write_table(pa.Table.from_pydict({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}				)				)
														lowercase_				,		lowercase_	:   Optional[Any]  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowercase_	:   Tuple  				=     {"""col_1""": pa.string(), """col_2""": pa.intaa()}
							assert writer._schema == pa.schema(lowercase				,       metadata=writer._schema.metadata				)
							_check_output(output.getvalue()				,       expected_num_chunks=num_examples if writer_batch_size == 1 else 1				)





@pytest.mark.parametrize("""writer_batch_size"""				,       [None, 1, 10]				)
@pytest.mark.parametrize(
    """fields"""				,       [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]				)
def  __magic_name__  (				lowercase				,       lowercase				)				-> Union[str, Any]:

							"""simple docstring"""



							lowercase_	:   Dict  				=     pa.BufferOutputStream()
							lowercase_	:   Any  				=     pa.schema(lowercase				) if fields else None
							with ArrowWriter(stream=lowercase				,       schema=lowercase				,       writer_batch_size=lowercase				) as writer:
														writer.write_row(pa.Table.from_pydict({"""col_1""": ["""foo"""], """col_2""": [1]}				)				)
														writer.write_row(pa.Table.from_pydict({"""col_1""": ["""bar"""], """col_2""": [2]}				)				)
														lowercase_				,		lowercase_	:   str  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowercase_	:   Tuple  				=     {"""col_1""": pa.string(), """col_2""": pa.intaa()}
							assert writer._schema == pa.schema(lowercase				,       metadata=writer._schema.metadata				)
							_check_output(output.getvalue()				,       expected_num_chunks=num_examples if writer_batch_size == 1 else 1				)





def  __magic_name__  (				)				-> Tuple:

							"""simple docstring"""



							with tempfile.TemporaryDirectory() as tmp_dir:
														lowercase_	:   Tuple  				=     {"""col_1""": pa.string(), """col_2""": pa.intaa()}
														lowercase_	:   Dict  				=     os.path.join(lowercase				,       """test.arrow"""				)
														with ArrowWriter(path=lowercase				,       schema=pa.schema(lowercase				)				) as writer:
																					writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}				)
																					lowercase_				,		lowercase_	:   Tuple  				=     writer.finalize()
														assert num_examples == 2
														assert num_bytes > 0
														assert writer._schema == pa.schema(lowercase				,       metadata=writer._schema.metadata				)
														_check_output(lowercase				,       1				)





def  __magic_name__  (				lowercase				)				-> str:

							"""simple docstring"""



							if pa.types.is_list(lowercase				):
														return get_base_dtype(arr_type.value_type				)
							else:
														return arr_type





def  __magic_name__  (				lowercase				,       lowercase				)				-> Union[str, Any]:

							"""simple docstring"""



							if isinstance(lst[0]				,       lowercase				):
														change_first_primitive_element_in_list(lst[0]				,       lowercase				)
							else:
														lowercase_	:   Optional[Any]  				=     value





@pytest.mark.parametrize("""optimized_int_type, expected_dtype"""				,       [(None, pa.intaa()), (Value("""int32"""				), pa.intaa())]				)
@pytest.mark.parametrize("""sequence"""				,       [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]				)
def  __magic_name__  (				lowercase				,       lowercase				,       lowercase				)				-> Tuple:

							"""simple docstring"""



							lowercase_	:   Tuple  				=     pa.array(TypedSequence(lowercase				,       optimized_int_type=lowercase				)				)
							assert get_base_dtype(arr.type				) == expected_dtype





@pytest.mark.parametrize(
    """col, expected_dtype"""				,       [
        ("""attention_mask""", pa.inta()),
        ("""special_tokens_mask""", pa.inta()),
        ("""token_type_ids""", pa.inta()),
        ("""input_ids""", pa.intaa()),
        ("""other""", pa.intaa()),
    ]				,       )
@pytest.mark.parametrize("""sequence"""				,       [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]				)
def  __magic_name__  (				lowercase				,       lowercase				,       lowercase				)				-> Tuple:

							"""simple docstring"""



							lowercase_	:   Dict  				=     pa.array(OptimizedTypedSequence(lowercase				,       col=lowercase				)				)
							assert get_base_dtype(arr.type				) == expected_dtype

							# not in range
							if col != "other":
														# avoids errors due to in-place modifications
														lowercase_	:   List[str]  				=     copy.deepcopy(lowercase				)
														lowercase_	:   Optional[Any]  				=     np.iinfo(expected_dtype.to_pandas_dtype()				).max + 1
														change_first_primitive_element_in_list(lowercase				,       lowercase				)
														lowercase_	:   Optional[Any]  				=     pa.array(OptimizedTypedSequence(lowercase				,       col=lowercase				)				)
														assert get_base_dtype(arr.type				) == pa.intaa()





@pytest.mark.parametrize("""raise_exception"""				,       [False, True]				)
def  __magic_name__  (				lowercase				,       lowercase				)				-> Any:

							"""simple docstring"""



							lowercase_	:   List[str]  				=     str(tmp_path / """dataset-train.arrow"""				)
							try:
														with ArrowWriter(path=lowercase				) as writer:
																					if raise_exception:
																												raise pa.lib.ArrowInvalid()
																					else:
																												writer.stream.close()
							except pa.lib.ArrowInvalid:
														pass
							finally:
														assert writer.stream.closed





def  __magic_name__  (				lowercase				)				-> List[str]:

							"""simple docstring"""



							lowercase_	:   Optional[Any]  				=     """mock://dataset-train.arrow"""
							with ArrowWriter(path=lowercase				,       storage_options=mockfs.storage_options				) as writer:
														assert isinstance(writer._fs				,       type(lowercase				)				)
														assert writer._fs.storage_options == mockfs.storage_options
														writer.write({"""col_1""": """foo""", """col_2""": 1}				)
														writer.write({"""col_1""": """bar""", """col_2""": 2}				)
														lowercase_				,		lowercase_	:   str  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							assert mockfs.exists(lowercase				)





def  __magic_name__  (				)				-> Optional[int]:

							"""simple docstring"""



							lowercase_	:   int  				=     pa.BufferOutputStream()
							with ParquetWriter(stream=lowercase				) as writer:
														writer.write({"""col_1""": """foo""", """col_2""": 1}				)
														writer.write({"""col_1""": """bar""", """col_2""": 2}				)
														lowercase_				,		lowercase_	:   Dict  				=     writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							lowercase_	:   Tuple  				=     pa.BufferReader(output.getvalue()				)
							lowercase_	:   pa.Table  				=     pq.read_table(lowercase				)
							assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}





@require_pil
@pytest.mark.parametrize("""embed_local_files"""				,       [False, True]				)
def  __magic_name__  (				lowercase				,       lowercase				)				-> Optional[int]:

							"""simple docstring"""



							import PIL.Image

							lowercase_	:   List[str]  				=     str(tmp_path / """test_image_rgb.jpg"""				)
							PIL.Image.fromarray(np.zeros((5, 5)				,       dtype=np.uinta				)				).save(lowercase				,       format="""png"""				)
							lowercase_	:   Any  				=     pa.BufferOutputStream()
							with ParquetWriter(
							    stream=lowercase				,       features=Features({"""image""": Image()}				)				,       embed_local_files=lowercase				) as writer:
														writer.write({"""image""": image_path}				)
														writer.finalize()
							lowercase_	:   str  				=     pa.BufferReader(output.getvalue()				)
							lowercase_	:   pa.Table  				=     pq.read_table(lowercase				)
							lowercase_	:   Optional[int]  				=     pa_table.to_pydict()
							if embed_local_files:
														assert isinstance(out["""image"""][0]["""path"""]				,       lowercase				)
														with open(lowercase				,       """rb"""				) as f:
																					assert out["image"][0]["bytes"] == f.read()
							else:
														assert out["image"][0]["path"] == image_path
														assert out["image"][0]["bytes"] is None





def  __magic_name__  (				)				-> Tuple:

							"""simple docstring"""



							lowercase_	:   Union[str, Any]  				=     pa.schema([pa.field("""col_1"""				,       pa.string()				,       nullable=lowercase				)]				)
							lowercase_	:   List[str]  				=     pa.BufferOutputStream()
							with ArrowWriter(stream=lowercase				) as writer:
														writer._build_writer(inferred_schema=lowercase				)
														assert writer._schema == pa.schema([pa.field("""col_1"""				,       pa.string()				)]				)






UpperCAmelCase_							=				8.31_44_62  # Unit - J mol-1 K-1





def  __magic_name__  (				lowercase				,       lowercase				,       lowercase				)				-> float:

							"""simple docstring"""



							if moles < 0 or kelvin < 0 or volume < 0:
														raise ValueError("""Invalid inputs. Enter positive value."""				)
							return moles * kelvin * UNIVERSAL_GAS_CONSTANT / volume





def  __magic_name__  (				lowercase				,       lowercase				,       lowercase				)				-> float:

							"""simple docstring"""



							if moles < 0 or kelvin < 0 or pressure < 0:
														raise ValueError("""Invalid inputs. Enter positive value."""				)
							return moles * kelvin * UNIVERSAL_GAS_CONSTANT / pressure


if __name__ == "__main__":
							from doctest import testmod

							testmod()