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 					__SCREAMING_SNAKE_CASE       (       lowerCamelCase_):






       """simple docstring"""

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







       def      lowercase_   ( self				):
          with self.assertRaises(snake_case__				):
             __snake_case   :			List[str]										=						pa.array(TypedSequence([1, 2, 3]				)			,		type=pa.intaa()				)







       def      lowercase_   ( self				):
          with self.assertRaises(snake_case__				):
             __snake_case   :			Tuple										=						pa.array(TypedSequence([1, 2, 3]			,		try_type=Value('bool'				)			,		type=Value('int64'				)				)				)







       def      lowercase_   ( self				):
          __snake_case   :			Any										=						pa.array(TypedSequence([1, 2, 3]			,		type=Value('int32'				)				)				)
          self.assertEqual(arr.type			,		pa.intaa()				)







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







       def      lowercase_   ( self				):
          __snake_case   :			List[Any]										=						pa.array(TypedSequence([1, 2, 3]			,		try_type=Value('int32'				)				)				)
          self.assertEqual(arr.type			,		pa.intaa()				)







       def      lowercase_   ( self				):
          __snake_case   :			Union[str, Any]										=						pa.array(TypedSequence(['foo', 'bar']			,		try_type=Value('int64'				)				)				)
          self.assertEqual(arr.type			,		pa.string()				)







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







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







       def      lowercase_   ( self				):
          __snake_case   :			Union[str, Any]										=						pa.array(TypedSequence([[[1, 2, 3]]]			,		try_type=ArrayaD((1, 3)			,		'int64'				)				)				)
          self.assertEqual(arr.type			,		ArrayaDExtensionType((1, 3)			,		'int64'				)				)







       def      lowercase_   ( self				):
          __snake_case   :			str										=						pa.array(TypedSequence(['foo', 'bar']			,		try_type=ArrayaD((1, 3)			,		'int64'				)				)				)
          self.assertEqual(arr.type			,		pa.string()				)







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

          __snake_case   :			Union[str, Any]										=						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:
             __snake_case   :			Optional[int]										=						pa.array(TypedSequence([{'path': None, 'bytes': b'image_bytes'}, pil_image]			,		type=Image()				)				)
             __snake_case   :			Dict										=						mock_cast_to_python_objects.call_args_list[-1]
             self.assertIn('optimize_list_casting'			,		snake_case__				)
             self.assertFalse(kwargs['optimize_list_casting']				)






def 							UpperCAmelCase__(					__UpperCAmelCase			:     Any     ,				__UpperCAmelCase			:     int       ):
   __snake_case   :			str										=						pa.BufferReader(lowerCamelCase_       ) if isinstance(lowerCamelCase_     ,				pa.Buffer       ) else pa.memory_map(lowerCamelCase_       )
   __snake_case   :			str										=						pa.ipc.open_stream(lowerCamelCase_       )
   __snake_case   :			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 							UpperCAmelCase__(					__UpperCAmelCase			:     Tuple     ,				__UpperCAmelCase			:     Dict       ):
   __snake_case   :			Any										=						pa.BufferOutputStream()
   __snake_case   :			Any										=						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}       )
      __snake_case   :			Optional[int]										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      __snake_case   :			str										=						{'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 							UpperCAmelCase__(					):
   __snake_case   :			Tuple										=						pa.BufferOutputStream()
   __snake_case   :			int										=						Features({'labels': ClassLabel(names=['neg', 'pos']       )}       )
   with ArrowWriter(stream=lowerCamelCase_     ,				features=lowerCamelCase_       ) as writer:
      writer.write({'labels': 0}       )
      writer.write({'labels': 1}       )
      __snake_case   :			str										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   assert writer._schema == features.arrow_schema
   assert writer._schema.metadata == features.arrow_schema.metadata
   __snake_case   :			List[str]										=						pa.BufferReader(output.getvalue()       )
   __snake_case   :			Tuple										=						pa.ipc.open_stream(lowerCamelCase_       )
   __snake_case   :			pa.Table										=						f.read_all()
   __snake_case   :			int										=						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 							UpperCAmelCase__(					__UpperCAmelCase			:     List[Any]       ):
   __snake_case   :			Tuple										=						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]       )
         __snake_case   :			Optional[Any]										=						writer.finalize()






@pytest.mark.parametrize('writer_batch_size'     ,				[None, 2, 10]       )
def 							UpperCAmelCase__(					__UpperCAmelCase			:     Optional[int]       ):
   __snake_case   :			Union[str, Any]										=						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       )
         __snake_case   :			Tuple										=						writer.finalize()






@pytest.mark.parametrize('writer_batch_size'     ,				[None, 2, 10]       )
def 							UpperCAmelCase__(					__UpperCAmelCase			:     Any       ):
   __snake_case   :			int										=						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       )
      __snake_case   :			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 							UpperCAmelCase__(					__UpperCAmelCase			:     Optional[Any]     ,				__UpperCAmelCase			:     str       ):
   __snake_case   :			List[Any]										=						pa.BufferOutputStream()
   __snake_case   :			int										=						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': []}       )
      __snake_case   :			str										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      __snake_case   :			str										=						{'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 							UpperCAmelCase__(					__UpperCAmelCase			:     str     ,				__UpperCAmelCase			:     Optional[int]       ):
   __snake_case   :			Dict										=						pa.BufferOutputStream()
   __snake_case   :			int										=						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]}       )       )
      __snake_case   :			int										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      __snake_case   :			Union[str, Any]										=						{'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 							UpperCAmelCase__(					__UpperCAmelCase			:     List[Any]     ,				__UpperCAmelCase			:     List[str]       ):
   __snake_case   :			List[Any]										=						pa.BufferOutputStream()
   __snake_case   :			Optional[int]										=						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]}       )       )
      __snake_case   :			int										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      __snake_case   :			Any										=						{'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 							UpperCAmelCase__(					):





   with tempfile.TemporaryDirectory() as tmp_dir:
      __snake_case   :			Optional[int]										=						{'col_1': pa.string(), 'col_2': pa.intaa()}
      __snake_case   :			Any										=						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]}       )
         __snake_case   :			Any										=						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 							UpperCAmelCase__(					__UpperCAmelCase			:     Union[str, Any]       ):





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






def 							UpperCAmelCase__(					__UpperCAmelCase			:     Optional[int]     ,				__UpperCAmelCase			:     Dict       ):





   if isinstance(lst[0]     ,				lowerCamelCase_       ):
      change_first_primitive_element_in_list(lst[0]     ,				lowerCamelCase_       )
   else:
      __snake_case   :			int										=						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 							UpperCAmelCase__(					__UpperCAmelCase			:     Tuple     ,				__UpperCAmelCase			:     Any     ,				__UpperCAmelCase			:     Tuple       ):
   __snake_case   :			List[Any]										=						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 							UpperCAmelCase__(					__UpperCAmelCase			:     Optional[Any]     ,				__UpperCAmelCase			:     int     ,				__UpperCAmelCase			:     Union[str, Any]       ):
   __snake_case   :			List[str]										=						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
      __snake_case   :			Union[str, Any]										=						copy.deepcopy(lowerCamelCase_       )
      __snake_case   :			int										=						np.iinfo(expected_dtype.to_pandas_dtype()       ).max + 1
      change_first_primitive_element_in_list(lowerCamelCase_     ,				lowerCamelCase_       )
      __snake_case   :			Optional[Any]										=						pa.array(OptimizedTypedSequence(lowerCamelCase_     ,				col=lowerCamelCase_       )       )
      assert get_base_dtype(arr.type       ) == pa.intaa()






@pytest.mark.parametrize('raise_exception'     ,				[False, True]       )
def 							UpperCAmelCase__(					__UpperCAmelCase			:     Any     ,				__UpperCAmelCase			:     Any       ):
   __snake_case   :			List[str]										=						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 							UpperCAmelCase__(					__UpperCAmelCase			:     List[Any]       ):
   __snake_case   :			str										=						'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}       )
      __snake_case   :			Dict										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   assert mockfs.exists(lowerCamelCase_       )






def 							UpperCAmelCase__(					):
   __snake_case   :			Union[str, Any]										=						pa.BufferOutputStream()
   with ParquetWriter(stream=lowerCamelCase_       ) as writer:
      writer.write({'col_1': 'foo', 'col_2': 1}       )
      writer.write({'col_1': 'bar', 'col_2': 2}       )
      __snake_case   :			Dict										=						writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   __snake_case   :			Optional[Any]										=						pa.BufferReader(output.getvalue()       )
   __snake_case   :			pa.Table										=						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 							UpperCAmelCase__(					__UpperCAmelCase			:     Tuple     ,				__UpperCAmelCase			:     Any       ):
   import PIL.Image

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









from __future__ import annotations





def        __UpperCAmelCase						(							lowerCamelCase_ :     list[float]  )       ->  float:







			"""simple docstring"""





			SCREAMING_SNAKE_CASE_   : Tuple					  =      0.0_0
			SCREAMING_SNAKE_CASE_   : Tuple					  =      0
			for resistor in resistors:
						if resistor <= 0:
									SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      F'Resistor at index {index} has a negative or zero value!'
									raise ValueError(lowerCamelCase_  )
						first_sum += 1 / float(lowerCamelCase_  )
						index += 1
			return 1 / first_sum





def        __UpperCAmelCase						(							lowerCamelCase_ :     list[float]  )       ->  float:







			"""simple docstring"""





			SCREAMING_SNAKE_CASE_   : Dict					  =      0.0_0
			SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      0
			for resistor in resistors:
						sum_r += resistor
						if resistor < 0:
									SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      F'Resistor at index {index} has a negative value!'
									raise ValueError(lowerCamelCase_  )
						index += 1
			return sum_r


if __name__ == "__main__":
						import doctest

						doctest.testmod()



