





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				(		_UpperCAmelCase  ):
      def 		_lowerCamelCase   (		self							):
         A_  :					Union[str, Any]         =						pa.array(TypedSequence([1, 2, 3]							)							)
         self.assertEqual(arr.type   ,							pa.intaa()							)
      def 		_lowerCamelCase   (		self							):






         with self.assertRaises(__UpperCamelCase							):
            A_  :					int         =						pa.array(TypedSequence([1, 2, 3]							)   ,							type=pa.intaa()							)
      def 		_lowerCamelCase   (		self							):






         with self.assertRaises(__UpperCamelCase							):
            A_  :					int         =						pa.array(TypedSequence([1, 2, 3]   ,							try_type=Value("""bool"""							)   ,							type=Value("""int64"""							)							)							)
      def 		_lowerCamelCase   (		self							):
         A_  :					Union[str, Any]         =						pa.array(TypedSequence([1, 2, 3]   ,							type=Value("""int32"""							)							)							)
         self.assertEqual(arr.type   ,							pa.intaa()							)
      def 		_lowerCamelCase   (		self							):






         with self.assertRaises((TypeError, pa.lib.ArrowInvalid)							):
            A_  :					Optional[Any]         =						pa.array(TypedSequence(["""foo""", """bar"""]   ,							type=Value("""int64"""							)							)							)
      def 		_lowerCamelCase   (		self							):
         A_  :					Optional[Any]         =						pa.array(TypedSequence([1, 2, 3]   ,							try_type=Value("""int32"""							)							)							)
         self.assertEqual(arr.type   ,							pa.intaa()							)
      def 		_lowerCamelCase   (		self							):
         A_  :					Tuple         =						pa.array(TypedSequence(["""foo""", """bar"""]   ,							try_type=Value("""int64"""							)							)							)
         self.assertEqual(arr.type   ,							pa.string()							)
      def 		_lowerCamelCase   (		self							):
         A_  :					Optional[Any]         =						pa.array(TypedSequence([[[1, 2, 3]]]   ,							type=ArrayaD((1, 3)   ,							"""int64"""							)							)							)
         self.assertEqual(arr.type   ,							ArrayaDExtensionType((1, 3)   ,							"""int64"""							)							)
      def 		_lowerCamelCase   (		self							):






         with self.assertRaises((TypeError, pa.lib.ArrowInvalid)							):
            A_  :					int         =						pa.array(TypedSequence(["""foo""", """bar"""]   ,							type=ArrayaD((1, 3)   ,							"""int64"""							)							)							)
      def 		_lowerCamelCase   (		self							):
         A_  :					List[str]         =						pa.array(TypedSequence([[[1, 2, 3]]]   ,							try_type=ArrayaD((1, 3)   ,							"""int64"""							)							)							)
         self.assertEqual(arr.type   ,							ArrayaDExtensionType((1, 3)   ,							"""int64"""							)							)
      def 		_lowerCamelCase   (		self							):
         A_  :					str         =						pa.array(TypedSequence(["""foo""", """bar"""]   ,							try_type=ArrayaD((1, 3)   ,							"""int64"""							)							)							)
         self.assertEqual(arr.type   ,							pa.string()							)






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

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






def      _lowerCAmelCase		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					int         =						pa.BufferReader(_A  ) if isinstance(_A	,pa.Buffer  ) else pa.memory_map(_A  )
   A_  :					Optional[int]         =						pa.ipc.open_stream(_A  )
   A_  :					Optional[Any]         =						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		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Any         =						pa.BufferOutputStream()
   A_  :					Optional[Any]         =						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_  :					Dict         =						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      A_  :					List[Any]         =						{"""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		(	):
   '''simple docstring'''


   A_  :					Dict         =						pa.BufferOutputStream()
   A_  :					Any         =						Features({"""labels""": ClassLabel(names=["""neg""", """pos"""]  )}  )
   with ArrowWriter(stream=_A	,features=_A  ) as writer:
      writer.write({"""labels""": 0}  )
      writer.write({"""labels""": 1}  )
      A_   ,					A_  :					Optional[int]         =						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_  :					int         =						pa.BufferReader(output.getvalue()  )
   A_  :					Dict         =						pa.ipc.open_stream(_A  )
   A_  :					Tuple         =						f.read_all()
   A_  :					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(_A  )



@pytest.mark.parametrize("""writer_batch_size"""	,[None, 1, 1_0]  )
def      _lowerCAmelCase		(	_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Tuple         =						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_  :					int         =						writer.finalize()



@pytest.mark.parametrize("""writer_batch_size"""	,[None, 2, 1_0]  )
def      _lowerCAmelCase		(	_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Dict         =						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_  :					Optional[Any]         =						writer.finalize()



@pytest.mark.parametrize("""writer_batch_size"""	,[None, 2, 1_0]  )
def      _lowerCAmelCase		(	_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					List[Any]         =						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_  :					Union[str, 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, 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		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					str         =						pa.BufferOutputStream()
   A_  :					List[Any]         =						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_  :					Union[str, Any]         =						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      A_  :					List[Any]         =						{"""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		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Optional[int]         =						pa.BufferOutputStream()
   A_  :					Tuple         =						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_  :					int         =						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      A_  :					int         =						{"""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		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					List[Any]         =						pa.BufferOutputStream()
   A_  :					Union[str, Any]         =						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_  :					str         =						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      A_  :					str         =						{"""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		(	):
   '''simple docstring'''


   with tempfile.TemporaryDirectory() as tmp_dir:
      A_  :					Union[str, Any]         =						{"""col_1""": pa.string(), """col_2""": pa.intaa()}
      A_  :					List[str]         =						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_  :					List[Any]         =						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		(	_lowerCAmelCase  ):
   '''simple docstring'''


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



def      _lowerCAmelCase		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   if isinstance(lst[0]	,_A  ):
      change_first_primitive_element_in_list(lst[0]	,_A  )
   else:
      A_  :					Dict         =						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		(	_lowerCAmelCase	,_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Optional[Any]         =						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		(	_lowerCAmelCase	,_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Tuple         =						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_  :					Optional[Any]         =						copy.deepcopy(_A  )
      A_  :					Optional[int]         =						np.iinfo(expected_dtype.to_pandas_dtype()  ).max + 1
      change_first_primitive_element_in_list(_A	,_A  )
      A_  :					Union[str, Any]         =						pa.array(OptimizedTypedSequence(_A	,col=_A  )  )
      assert get_base_dtype(arr.type  ) == pa.intaa()



@pytest.mark.parametrize("""raise_exception"""	,[False, True]  )
def      _lowerCAmelCase		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Optional[Any]         =						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		(	_lowerCAmelCase  ):
   '''simple docstring'''


   A_  :					Union[str, Any]         =						"""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_  :					List[str]         =						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   assert mockfs.exists(_A  )



def      _lowerCAmelCase		(	):
   '''simple docstring'''


   A_  :					Optional[int]         =						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_  :					Tuple         =						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   A_  :					Dict         =						pa.BufferReader(output.getvalue()  )
   A_  :					Tuple         =						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		(	_lowerCAmelCase	,_lowerCAmelCase  ):
   '''simple docstring'''


   import PIL.Image

   A_  :					Union[str, Any]         =						str(tmp_path / """test_image_rgb.jpg"""  )
   PIL.Image.fromarray(np.zeros((5, 5)	,dtype=np.uinta  )  ).save(_A	,format="""png"""  )
   A_  :					Optional[Any]         =						pa.BufferOutputStream()
   with ParquetWriter(
       stream=_A	,features=Features({"""image""": Image()}  )	,embed_local_files=_A  ) as writer:
      writer.write({"""image""": image_path}  )
      writer.finalize()
   A_  :					List[str]         =						pa.BufferReader(output.getvalue()  )
   A_  :					Tuple         =						pq.read_table(_A  )
   A_  :					List[Any]         =						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		(	):
   '''simple docstring'''


   A_  :					int         =						pa.schema([pa.field("""col_1"""	,pa.string()	,nullable=_A  )]  )
   A_  :					Tuple         =						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()  )]  )








"""simple docstring"""



def 			_UpperCamelCase						(    _A   )     ->	int:





    """simple docstring"""

    if not isinstance(_A       ,   _A   ) or number < 0:
        raise ValueError("""Input must be a non-negative integer"""   )

    _UpperCAmelCase        = 0
    while number:
        # This way we arrive at next set bit (next 1) instead of looping
        # through each bit and checking for 1s hence the
        # loop won't run 32 times it will only run the number of `1` times
        number &= number - 1
        count += 1
    return count


if __name__ == "__main__":
 import doctest

 doctest.testmod()