





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   A			(SCREAMING_SNAKE_CASE    ):
			'''simple docstring'''




			def    a_       (			self   :    Optional[int]   )	->	Dict:







					"""simple docstring"""






					A__        =   pa.array(TypedSequence([1, 2, 3]   )   )
					self.assertEqual(arr.type							,	pa.intaa()   )






			def    a_       (			self   :    Dict   )	->	Union[str, Any]:







					"""simple docstring"""






					with self.assertRaises(__lowerCAmelCase   ):
							A__        =   pa.array(TypedSequence([1, 2, 3]   )							,	type=pa.intaa()   )






			def    a_       (			self   :    int   )	->	Any:







					"""simple docstring"""






					with self.assertRaises(__lowerCAmelCase   ):
							A__        =   pa.array(TypedSequence([1, 2, 3]							,	try_type=Value("""bool"""   )							,	type=Value("""int64"""   )   )   )






			def    a_       (			self   :    List[Any]   )	->	Optional[int]:







					"""simple docstring"""






					A__        =   pa.array(TypedSequence([1, 2, 3]							,	type=Value("""int32"""   )   )   )
					self.assertEqual(arr.type							,	pa.intaa()   )






			def    a_       (			self   :    Dict   )	->	str:







					"""simple docstring"""






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






			def    a_       (			self   :    Dict   )	->	Dict:







					"""simple docstring"""






					A__        =   pa.array(TypedSequence([1, 2, 3]							,	try_type=Value("""int32"""   )   )   )
					self.assertEqual(arr.type							,	pa.intaa()   )






			def    a_       (			self   :    Union[str, Any]   )	->	Any:







					"""simple docstring"""






					A__        =   pa.array(TypedSequence(["""foo""", """bar"""]							,	try_type=Value("""int64"""   )   )   )
					self.assertEqual(arr.type							,	pa.string()   )






			def    a_       (			self   :    Dict   )	->	Dict:







					"""simple docstring"""






					A__        =   pa.array(TypedSequence([[[1, 2, 3]]]							,	type=ArrayaD((1, 3)							,	"""int64"""   )   )   )
					self.assertEqual(arr.type							,	ArrayaDExtensionType((1, 3)							,	"""int64"""   )   )






			def    a_       (			self   :    Dict   )	->	str:







					"""simple docstring"""






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






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







					"""simple docstring"""






					A__        =   pa.array(TypedSequence([[[1, 2, 3]]]							,	try_type=ArrayaD((1, 3)							,	"""int64"""   )   )   )
					self.assertEqual(arr.type							,	ArrayaDExtensionType((1, 3)							,	"""int64"""   )   )






			def    a_       (			self   :    List[str]   )	->	List[Any]:







					"""simple docstring"""






					A__        =   pa.array(TypedSequence(["""foo""", """bar"""]							,	try_type=ArrayaD((1, 3)							,	"""int64"""   )   )   )
					self.assertEqual(arr.type							,	pa.string()   )







			@require_pil
			def    a_       (			self   :    str   )	->	str:







					"""simple docstring"""






					import PIL.Image

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







def    __lowerCamelCase					(							__a   :Optional[Any]    ,     __a   :int    )		->				Optional[Any]:






		"""simple docstring"""

		A__        =   pa.BufferReader(__a    ) if isinstance(__a    ,     pa.Buffer    ) else pa.memory_map(__a    )
		A__        =   pa.ipc.open_stream(__a    )
		A__        =   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, 1_0]    )
@pytest.mark.parametrize(
    """fields"""    ,     [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]    )
def    __lowerCamelCase					(							__a   :Dict    ,     __a   :Optional[Any]    )		->				List[str]:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		A__        =   pa.schema(__a    ) if fields else None
		with ArrowWriter(stream=__a    ,     schema=__a    ,     writer_batch_size=__a    ) as writer:
				writer.write({"""col_1""": """foo""", """col_2""": 1}    )
				writer.write({"""col_1""": """bar""", """col_2""": 2}    )
				A__      ,			A__        =   writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				A__        =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
		assert writer._schema == pa.schema(__a    ,     metadata=writer._schema.metadata    )
		_check_output(output.getvalue()    ,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1    )







def    __lowerCamelCase					(							)		->				int:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		A__        =   Features({"""labels""": ClassLabel(names=["""neg""", """pos"""]    )}    )
		with ArrowWriter(stream=__a    ,     features=__a    ) as writer:
				writer.write({"""labels""": 0}    )
				writer.write({"""labels""": 1}    )
				A__      ,			A__        =   writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		assert writer._schema == features.arrow_schema
		assert writer._schema.metadata == features.arrow_schema.metadata
		A__        =   pa.BufferReader(output.getvalue()    )
		A__        =   pa.ipc.open_stream(__a    )
		A__        =   f.read_all()
		A__        =   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(__a    )







@pytest.mark.parametrize("""writer_batch_size"""    ,     [None, 1, 1_0]    )
def    __lowerCamelCase					(							__a   :int    )		->				Dict:






		"""simple docstring"""

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







@pytest.mark.parametrize("""writer_batch_size"""    ,     [None, 2, 1_0]    )
def    __lowerCamelCase					(							__a   :str    )		->				Optional[int]:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		with ArrowWriter(
		    stream=__a    ,     writer_batch_size=__a    ,     hash_salt="""split_name"""    ,     check_duplicates=__a    ,     ) as writer:
				with pytest.raises(__a    ):
						writer.write({"""col_1""": """foo""", """col_2""": 1}    ,     key=1_0    )
						writer.write({"""col_1""": """bar""", """col_2""": 2}    ,     key=1_0    )
						A__      ,			A__        =   writer.finalize()







@pytest.mark.parametrize("""writer_batch_size"""    ,     [None, 2, 1_0]    )
def    __lowerCamelCase					(							__a   :Union[str, Any]    )		->				Optional[int]:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		with ArrowWriter(
		    stream=__a    ,     writer_batch_size=__a    ,     hash_salt="""split_name"""    ,     check_duplicates=__a    ,     ) as writer:
				writer.write({"""col_1""": """foo""", """col_2""": 1}    ,     key=1    )
				writer.write({"""col_1""": """bar""", """col_2""": 2}    ,     key=2    )
				A__      ,			A__        =   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, 1_0]    )
@pytest.mark.parametrize(
    """fields"""    ,     [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]    )
def    __lowerCamelCase					(							__a   :Any    ,     __a   :Any    )		->				Optional[int]:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		A__        =   pa.schema(__a    ) if fields else None
		with ArrowWriter(stream=__a    ,     schema=__a    ,     writer_batch_size=__a    ) as writer:
				writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}    )
				writer.write_batch({"""col_1""": [], """col_2""": []}    )
				A__      ,			A__        =   writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				A__        =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
		assert writer._schema == pa.schema(__a    ,     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, 1_0]    )
@pytest.mark.parametrize(
    """fields"""    ,     [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]    )
def    __lowerCamelCase					(							__a   :Tuple    ,     __a   :Optional[int]    )		->				Union[str, Any]:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		A__        =   pa.schema(__a    ) if fields else None
		with ArrowWriter(stream=__a    ,     schema=__a    ,     writer_batch_size=__a    ) as writer:
				writer.write_table(pa.Table.from_pydict({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}    )    )
				A__      ,			A__        =   writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				A__        =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
		assert writer._schema == pa.schema(__a    ,     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, 1_0]    )
@pytest.mark.parametrize(
    """fields"""    ,     [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]    )
def    __lowerCamelCase					(							__a   :List[str]    ,     __a   :str    )		->				Tuple:






		"""simple docstring"""

		A__        =   pa.BufferOutputStream()
		A__        =   pa.schema(__a    ) if fields else None
		with ArrowWriter(stream=__a    ,     schema=__a    ,     writer_batch_size=__a    ) 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]}    )    )
				A__      ,			A__        =   writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				A__        =   {"""col_1""": pa.string(), """col_2""": pa.intaa()}
		assert writer._schema == pa.schema(__a    ,     metadata=writer._schema.metadata    )
		_check_output(output.getvalue()    ,     expected_num_chunks=num_examples if writer_batch_size == 1 else 1    )







def    __lowerCamelCase					(							)		->				Tuple:






		"""simple docstring"""

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







def    __lowerCamelCase					(							__a   :Tuple    )		->				List[str]:






		"""simple docstring"""

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







def    __lowerCamelCase					(							__a   :Any    ,     __a   :Optional[int]    )		->				int:






		"""simple docstring"""

		if isinstance(lst[0]    ,     __a    ):
				change_first_primitive_element_in_list(lst[0]    ,     __a    )
		else:
				A__        =   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    __lowerCamelCase					(							__a   :str    ,     __a   :Dict    ,     __a   :Dict    )		->				Any:






		"""simple docstring"""

		A__        =   pa.array(TypedSequence(__a    ,     optimized_int_type=__a    )    )
		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    __lowerCamelCase					(							__a   :Tuple    ,     __a   :List[Any]    ,     __a   :Tuple    )		->				Tuple:






		"""simple docstring"""

		A__        =   pa.array(OptimizedTypedSequence(__a    ,     col=__a    )    )
		assert get_base_dtype(arr.type    ) == expected_dtype

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







@pytest.mark.parametrize("""raise_exception"""    ,     [False, True]    )
def    __lowerCamelCase					(							__a   :Optional[Any]    ,     __a   :int    )		->				List[str]:






		"""simple docstring"""

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







def    __lowerCamelCase					(							__a   :Tuple    )		->				List[str]:






		"""simple docstring"""

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







def    __lowerCamelCase					(							)		->				int:






		"""simple docstring"""

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







@require_pil
@pytest.mark.parametrize("""embed_local_files"""    ,     [False, True]    )
def    __lowerCamelCase					(							__a   :Dict    ,     __a   :Dict    )		->				Union[str, Any]:






		"""simple docstring"""

		import PIL.Image

		A__        =   str(tmp_path / """test_image_rgb.jpg"""    )
		PIL.Image.fromarray(np.zeros((5, 5)    ,     dtype=np.uinta    )    ).save(__a    ,     format="""png"""    )
		A__        =   pa.BufferOutputStream()
		with ParquetWriter(
		    stream=__a    ,     features=Features({"""image""": Image()}    )    ,     embed_local_files=__a    ) as writer:
				writer.write({"""image""": image_path}    )
				writer.finalize()
		A__        =   pa.BufferReader(output.getvalue()    )
		A__        =   pq.read_table(__a    )
		A__        =   pa_table.to_pydict()
		if embed_local_files:
				assert isinstance(out["""image"""][0]["""path"""]    ,     __a    )
				with open(__a    ,     """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    __lowerCamelCase					(							)		->				Optional[Any]:






		"""simple docstring"""

		A__        =   pa.schema([pa.field("""col_1"""    ,     pa.string()    ,     nullable=__a    )]    )
		A__        =   pa.BufferOutputStream()
		with ArrowWriter(stream=__a    ) as writer:
				writer._build_writer(inferred_schema=__a    )
				assert writer._schema == pa.schema([pa.field("""col_1"""    ,     pa.string()    )]    )










from multiprocessing import Lock, Pipe, Process

# lock used to ensure that two processes do not access a pipe at the same time
A    :					Dict =      Lock()







def    __lowerCamelCase					(							__a   :Dict    ,     __a   :List[str]    ,     __a   :Optional[int]    ,     __a   :Optional[int]    ,     __a   :Optional[Any]    ,     __a   :Optional[int]    ,     __a   :int    )		->				Dict:






		"""simple docstring"""

		global process_lock

		# we perform n swaps since after n swaps we know we are sorted
		# we *could* stop early if we are sorted already, but it takes as long to
		# find out we are sorted as it does to sort the list with this algorithm
		for i in range(0    ,     1_0    ):
				if (i + position) % 2 == 0 and r_send is not None:
						# send your value to your right neighbor
						process_lock.acquire()
						r_send[1].send(__a    )
						process_lock.release()

						# receive your right neighbor's value
						process_lock.acquire()
						A__        =   rr_cv[0].recv()
						process_lock.release()

						# take the lower value since you are on the left
						A__        =   min(__a    ,     __a    )
				elif (i + position) % 2 != 0 and l_send is not None:
						# send your value to your left neighbor
						process_lock.acquire()
						l_send[1].send(__a    )
						process_lock.release()

						# receive your left neighbor's value
						process_lock.acquire()
						A__        =   lr_cv[0].recv()
						process_lock.release()

						# take the higher value since you are on the right
						A__        =   max(__a    ,     __a    )
    # after all swaps are performed, send the values back to main
		result_pipe[1].send(__a    )







def    __lowerCamelCase					(							__a   :List[str]    )		->				int:






		"""simple docstring"""

		A__        =   []
		A__        =   []
		# initialize the list of pipes where the values will be retrieved
		for _ in arr:
				result_pipe.append(Pipe()    )
		# creates the processes
		# the first and last process only have one neighbor so they are made outside
		# of the loop
		A__        =   Pipe()
		A__        =   Pipe()
		process_array_.append(
		    Process(
		        target=__a    ,     args=(0, arr[0], None, temp_rs, None, temp_rr, result_pipe[0])    ,     )    )
		A__        =   temp_rs
		A__        =   temp_rr

		for i in range(1    ,     len(__a    ) - 1    ):
				A__        =   Pipe()
				A__        =   Pipe()
				process_array_.append(
				    Process(
				        target=__a    ,     args=(i, arr[i], temp_ls, temp_rs, temp_lr, temp_rr, result_pipe[i])    ,     )    )
				A__        =   temp_rs
				A__        =   temp_rr

		process_array_.append(
		    Process(
		        target=__a    ,     args=(
		            len(__a    ) - 1,
		            arr[len(__a    ) - 1],
		            temp_ls,
		            None,
		            temp_lr,
		            None,
		            result_pipe[len(__a    ) - 1],
		        )    ,     )    )

		# start the processes
		for p in process_array_:
				p.start()

		# wait for the processes to end and write their values to the list
		for p in range(0    ,     len(__a    )    ):
				A__        =   result_pipe[p][0].recv()
				process_array_[p].join()
		return arr







def    __lowerCamelCase					(							)		->				str:






		"""simple docstring"""

		A__        =   list(range(1_0    ,     0    ,     -1    )    )
		print("""Initial List"""    )
		print(*__a    )
		A__        =   odd_even_transposition(__a    )
		print("""Sorted List\n"""    )
		print(*__a    )


if __name__ == "__main__":
					main()



