





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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

            with self.assertRaises(A      ):
                   _UpperCamelCase              =						pa.array(TypedSequence([1, 2, 3],       try_type=Value('''bool'''      ),       type=Value('''int64'''      )      )      )

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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

     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

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



     @require_pil
     def 	_UpperCamelCase     (      self      ):





            """simple docstring"""

            import PIL.Image

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






def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						pa.BufferReader(lowerCAmelCase					) if isinstance(lowerCAmelCase    ,       pa.Buffer					) else pa.memory_map(lowerCAmelCase					)
       _UpperCamelCase              =						pa.ipc.open_stream(lowerCAmelCase					)
       _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						pa.BufferOutputStream()
       _UpperCamelCase              =						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}					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       if not fields:
              _UpperCamelCase              =						{'''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     SCREAMING_SNAKE_CASE			(      ):
       _UpperCamelCase              =						pa.BufferOutputStream()
       _UpperCamelCase              =						Features({'''labels''': ClassLabel(names=['''neg''', '''pos''']					)}					)
       with ArrowWriter(stream=lowerCAmelCase    ,       features=lowerCAmelCase					) as writer:
              writer.write({'''labels''': 0}					)
              writer.write({'''labels''': 1}					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       assert writer._schema == features.arrow_schema
       assert writer._schema.metadata == features.arrow_schema.metadata
       _UpperCamelCase              =						pa.BufferReader(output.getvalue()					)
       _UpperCamelCase              =						pa.ipc.open_stream(lowerCAmelCase					)
       _UpperCamelCase              =						f.read_all()
       _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase					):
       _UpperCamelCase              =						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]					)
                     _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()


@pytest.mark.parametrize('''writer_batch_size'''    ,       [None, 2, 10]					)
def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase					):
       _UpperCamelCase              =						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					)
                     _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()


@pytest.mark.parametrize('''writer_batch_size'''    ,       [None, 2, 10]					)
def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase					):
       _UpperCamelCase              =						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					)
              _UpperCamelCase       ,		_UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						pa.BufferOutputStream()
       _UpperCamelCase              =						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''': []}					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       if not fields:
              _UpperCamelCase              =						{'''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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						pa.BufferOutputStream()
       _UpperCamelCase              =						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]}					)					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       if not fields:
              _UpperCamelCase              =						{'''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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						pa.BufferOutputStream()
       _UpperCamelCase              =						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]}					)					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       if not fields:
              _UpperCamelCase              =						{'''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     SCREAMING_SNAKE_CASE			(      ):
       with tempfile.TemporaryDirectory() as tmp_dir:
              _UpperCamelCase              =						{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
              _UpperCamelCase              =						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]}					)
                     _UpperCamelCase       ,		_UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase					):
       if pa.types.is_list(lowerCAmelCase					):
              return get_base_dtype(arr_type.value_type					)
       else:
              return arr_type


def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       if isinstance(lst[0]    ,       lowerCAmelCase					):
              change_first_primitive_element_in_list(lst[0]    ,       lowerCAmelCase					)
       else:
              _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase    ,       lowerCAmelCase					):
       # in range
       _UpperCamelCase              =						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
              _UpperCamelCase              =						copy.deepcopy(lowerCAmelCase					)
              _UpperCamelCase              =						np.iinfo(expected_dtype.to_pandas_dtype()					).max + 1
              change_first_primitive_element_in_list(lowerCAmelCase    ,       lowerCAmelCase					)
              _UpperCamelCase              =						pa.array(OptimizedTypedSequence(lowerCAmelCase    ,       col=lowerCAmelCase					)					)
              assert get_base_dtype(arr.type					) == pa.intaa()


@pytest.mark.parametrize('''raise_exception'''    ,       [False, True]					)
def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase					):
       _UpperCamelCase              =						'''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}					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       assert mockfs.exists(lowerCAmelCase					)


def     SCREAMING_SNAKE_CASE			(      ):
       _UpperCamelCase              =						pa.BufferOutputStream()
       with ParquetWriter(stream=lowerCAmelCase					) as writer:
              writer.write({'''col_1''': '''foo''', '''col_2''': 1}					)
              writer.write({'''col_1''': '''bar''', '''col_2''': 2}					)
              _UpperCamelCase       ,		_UpperCamelCase              =						writer.finalize()
       assert num_examples == 2
       assert num_bytes > 0
       _UpperCamelCase              =						pa.BufferReader(output.getvalue()					)
       _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       import PIL.Image

       _UpperCamelCase              =						str(tmp_path / '''test_image_rgb.jpg'''					)
       PIL.Image.fromarray(np.zeros((5, 5)    ,       dtype=np.uinta					)					).save(lowerCAmelCase    ,       format='''png'''					)
       _UpperCamelCase              =						pa.BufferOutputStream()
       with ParquetWriter(
           stream=lowerCAmelCase    ,       features=Features({'''image''': Image()}					)    ,       embed_local_files=lowerCAmelCase					) as writer:
              writer.write({'''image''': image_path}					)
              writer.finalize()
       _UpperCamelCase              =						pa.BufferReader(output.getvalue()					)
       _UpperCamelCase              =						pq.read_table(lowerCAmelCase					)
       _UpperCamelCase              =						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     SCREAMING_SNAKE_CASE			(      ):
       _UpperCamelCase              =						pa.schema([pa.field('''col_1'''    ,       pa.string()    ,       nullable=lowerCAmelCase					)]					)
       _UpperCamelCase              =						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()					)]					)













def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase					):
       _UpperCamelCase              =						0
       for ch in input_str:
              _UpperCamelCase              =						ord(lowerCAmelCase					)
              _UpperCamelCase              =						pow(2    ,       lowerCAmelCase					)

              # If we already turned on bit for current character's unicode
              if bitmap >> ch_unicode & 1 == 1:
                     return False
              bitmap |= ch_bit_index_on
       return True


if __name__ == "__main__":
       import doctest

       doctest.testmod()






