





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       (  lowerCamelCase__      ):


				def _snake_case    (						self							):
								lowerCamelCase			=pa.array(TypedSequence([1, 2, 3]							)							)
								self.assertEqual(arr.type				,	pa.intaa()							)


				def _snake_case    (						self							):






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


				def _snake_case    (						self							):






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


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


				def _snake_case    (						self							):






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


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


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


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


				def _snake_case    (						self							):






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


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


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


				@require_pil
				def _snake_case    (						self							):
								import PIL.Image

								lowerCamelCase			=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:
												lowerCamelCase			=pa.array(TypedSequence([{"""path""": None, """bytes""": B"""image_bytes"""}, pil_image]				,	type=Image()							)							)
												lowerCamelCase			=mock_cast_to_python_objects.call_args_list[-1]
												self.assertIn("""optimize_list_casting"""				,	__lowerCamelCase							)
												self.assertFalse(kwargs["""optimize_list_casting"""]							)





def        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					int:
				lowerCamelCase			=pa.BufferReader(lowerCamelCase_			) if isinstance(lowerCamelCase_     , pa.Buffer			) else pa.memory_map(lowerCamelCase_			)
				lowerCamelCase			=pa.ipc.open_stream(lowerCamelCase_			)
				lowerCamelCase			=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        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					List[Any]:
				lowerCamelCase			=pa.BufferOutputStream()
				lowerCamelCase			=pa.schema(lowerCamelCase_			) if fields else None
				with ArrowWriter(stream=lowerCamelCase_     , schema=lowerCamelCase_     , writer_batch_size=lowerCamelCase_			) as writer:
								writer.write({"""col_1""": """foo""", """col_2""": 1}			)
								writer.write({"""col_1""": """bar""", """col_2""": 2}			)
								lowerCamelCase			=writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				if not fields:
								lowerCamelCase			={'''col_1''': pa.string(), '''col_2''': pa.intaa()}
				assert writer._schema == pa.schema(lowerCamelCase_     , metadata=writer._schema.metadata			)
				_check_output(output.getvalue()     , expected_num_chunks=num_examples if writer_batch_size == 1 else 1			)





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





@pytest.mark.parametrize("""writer_batch_size"""     , [None, 1, 10]			)
def        _lowercase   (  _UpperCAmelCase			)  ->					List[Any]:
				lowerCamelCase			=pa.BufferOutputStream()
				with ArrowWriter(
				    stream=lowerCamelCase_     , writer_batch_size=lowerCamelCase_     , hash_salt="""split_name"""     , check_duplicates=lowerCamelCase_     , ) as writer:
								with pytest.raises(lowerCamelCase_			):
												writer.write({"""col_1""": """foo""", """col_2""": 1}     , key=[1, 2]			)
												lowerCamelCase			=writer.finalize()





@pytest.mark.parametrize("""writer_batch_size"""     , [None, 2, 10]			)
def        _lowercase   (  _UpperCAmelCase			)  ->					Tuple:
				lowerCamelCase			=pa.BufferOutputStream()
				with ArrowWriter(
				    stream=lowerCamelCase_     , writer_batch_size=lowerCamelCase_     , hash_salt="""split_name"""     , check_duplicates=lowerCamelCase_     , ) as writer:
								with pytest.raises(lowerCamelCase_			):
												writer.write({"""col_1""": """foo""", """col_2""": 1}     , key=10			)
												writer.write({"""col_1""": """bar""", """col_2""": 2}     , key=10			)
												lowerCamelCase			=writer.finalize()





@pytest.mark.parametrize("""writer_batch_size"""     , [None, 2, 10]			)
def        _lowercase   (  _UpperCAmelCase			)  ->					Optional[Any]:
				lowerCamelCase			=pa.BufferOutputStream()
				with ArrowWriter(
				    stream=lowerCamelCase_     , writer_batch_size=lowerCamelCase_     , hash_salt="""split_name"""     , check_duplicates=lowerCamelCase_     , ) as writer:
								writer.write({"""col_1""": """foo""", """col_2""": 1}     , key=1			)
								writer.write({"""col_1""": """bar""", """col_2""": 2}     , key=2			)
								lowerCamelCase			=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        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					List[str]:
				lowerCamelCase			=pa.BufferOutputStream()
				lowerCamelCase			=pa.schema(lowerCamelCase_			) if fields else None
				with ArrowWriter(stream=lowerCamelCase_     , schema=lowerCamelCase_     , writer_batch_size=lowerCamelCase_			) as writer:
								writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}			)
								writer.write_batch({"""col_1""": [], """col_2""": []}			)
								lowerCamelCase			=writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				if not fields:
								lowerCamelCase			={'''col_1''': pa.string(), '''col_2''': pa.intaa()}
				assert writer._schema == pa.schema(lowerCamelCase_     , 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        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					str:
				lowerCamelCase			=pa.BufferOutputStream()
				lowerCamelCase			=pa.schema(lowerCamelCase_			) if fields else None
				with ArrowWriter(stream=lowerCamelCase_     , schema=lowerCamelCase_     , writer_batch_size=lowerCamelCase_			) as writer:
								writer.write_table(pa.Table.from_pydict({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}			)			)
								lowerCamelCase			=writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				if not fields:
								lowerCamelCase			={'''col_1''': pa.string(), '''col_2''': pa.intaa()}
				assert writer._schema == pa.schema(lowerCamelCase_     , 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        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					Optional[int]:
				lowerCamelCase			=pa.BufferOutputStream()
				lowerCamelCase			=pa.schema(lowerCamelCase_			) if fields else None
				with ArrowWriter(stream=lowerCamelCase_     , schema=lowerCamelCase_     , writer_batch_size=lowerCamelCase_			) 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]}			)			)
								lowerCamelCase			=writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				if not fields:
								lowerCamelCase			={'''col_1''': pa.string(), '''col_2''': pa.intaa()}
				assert writer._schema == pa.schema(lowerCamelCase_     , metadata=writer._schema.metadata			)
				_check_output(output.getvalue()     , expected_num_chunks=num_examples if writer_batch_size == 1 else 1			)





def        _lowercase   (  )  ->					Optional[int]:

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





def        _lowercase   (  _UpperCAmelCase			)  ->					str:

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





def        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					List[Any]:

				if isinstance(lst[0]     , lowerCamelCase_			):
								change_first_primitive_element_in_list(lst[0]     , lowerCamelCase_			)
				else:
								lowerCamelCase			=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        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase     , _UpperCAmelCase			)  ->					Optional[Any]:
				lowerCamelCase			=pa.array(TypedSequence(lowerCamelCase_     , optimized_int_type=lowerCamelCase_			)			)
				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        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase     , _UpperCAmelCase			)  ->					List[Any]:
				lowerCamelCase			=pa.array(OptimizedTypedSequence(lowerCamelCase_     , col=lowerCamelCase_			)			)
				assert get_base_dtype(arr.type			) == expected_dtype

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





@pytest.mark.parametrize("""raise_exception"""     , [False, True]			)
def        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					int:
				lowerCamelCase			=str(tmp_path / """dataset-train.arrow"""			)
				try:
								with ArrowWriter(path=lowerCamelCase_			) as writer:
												if raise_exception:
																raise pa.lib.ArrowInvalid()
												else:
																writer.stream.close()
				except pa.lib.ArrowInvalid:
								pass
				finally:
								assert writer.stream.closed





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





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





@require_pil
@pytest.mark.parametrize("""embed_local_files"""     , [False, True]			)
def        _lowercase   (  _UpperCAmelCase     , _UpperCAmelCase			)  ->					Optional[int]:
				import PIL.Image

				lowerCamelCase			=str(tmp_path / """test_image_rgb.jpg"""			)
				PIL.Image.fromarray(np.zeros((5, 5)     , dtype=np.uinta			)			).save(lowerCamelCase_     , format="""png"""			)
				lowerCamelCase			=pa.BufferOutputStream()
				with ParquetWriter(
				    stream=lowerCamelCase_     , features=Features({"""image""": Image()}			)     , embed_local_files=lowerCamelCase_			) as writer:
								writer.write({"""image""": image_path}			)
								writer.finalize()
				lowerCamelCase			=pa.BufferReader(output.getvalue()			)
				lowerCamelCase			=pq.read_table(lowerCamelCase_			)
				lowerCamelCase			=pa_table.to_pydict()
				if embed_local_files:
								assert isinstance(out["""image"""][0]["""path"""]     , lowerCamelCase_			)
								with open(lowerCamelCase_     , """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        _lowercase   (  )  ->					int:
				lowerCamelCase			=pa.schema([pa.field("""col_1"""     , pa.string()     , nullable=lowerCamelCase_			)]			)
				lowerCamelCase			=pa.BufferOutputStream()
				with ArrowWriter(stream=lowerCamelCase_			) as writer:
								writer._build_writer(inferred_schema=lowerCamelCase_			)
								assert writer._schema == pa.schema([pa.field("""col_1"""     , pa.string()			)]			)












import inspect
import unittest

class  __A       (  unittest.TestCase      ):


				def _snake_case    (						self							):
								try:
												import diffusers  # noqa: F401
								except ImportError:
												assert False


				def _snake_case    (						self							):
								import diffusers
								from diffusers.dependency_versions_table import deps

								lowerCamelCase			=inspect.getmembers(UpperCAmelCase_				,	inspect.isclass							)

								for cls_name, cls_module in all_classes:
												if "dummy_" in cls_module.__module__:
																for backend in cls_module._backends:
																				if backend == "k_diffusion":
																								lowerCamelCase			="""k-diffusion"""
																				elif backend == "invisible_watermark":
																								lowerCamelCase			="""invisible-watermark"""
																				assert backend in deps, f"""{backend} is not in the deps table!"""





