





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 						__lowerCamelCase   (				self					:     Optional[int]      )       ->Optional[int]:
          lowerCamelCase__       :						Any        =							pa.array(TypedSequence([1, 2, 3]      )      )
          self.assertEqual(arr.type		, pa.intaa()      )







       def 						__lowerCamelCase   (				self					:     Optional[Any]      )       ->Dict:
          with self.assertRaises(A      ):
             lowerCamelCase__       :						Optional[int]        =							pa.array(TypedSequence([1, 2, 3]      )		, type=pa.intaa()      )







       def 						__lowerCamelCase   (				self					:     Union[str, Any]      )       ->Optional[int]:
          with self.assertRaises(A      ):
             lowerCamelCase__       :						Tuple        =							pa.array(TypedSequence([1, 2, 3]		, try_type=Value('''bool'''      )		, type=Value('''int64'''      )      )      )







       def 						__lowerCamelCase   (				self					:     str      )       ->Optional[Any]:
          lowerCamelCase__       :						Union[str, Any]        =							pa.array(TypedSequence([1, 2, 3]		, type=Value('''int32'''      )      )      )
          self.assertEqual(arr.type		, pa.intaa()      )







       def 						__lowerCamelCase   (				self					:     List[Any]      )       ->int:
          with self.assertRaises((TypeError, pa.lib.ArrowInvalid)      ):
             lowerCamelCase__       :						Dict        =							pa.array(TypedSequence(['''foo''', '''bar''']		, type=Value('''int64'''      )      )      )







       def 						__lowerCamelCase   (				self					:     List[str]      )       ->str:
          lowerCamelCase__       :						List[Any]        =							pa.array(TypedSequence([1, 2, 3]		, try_type=Value('''int32'''      )      )      )
          self.assertEqual(arr.type		, pa.intaa()      )







       def 						__lowerCamelCase   (				self					:     Any      )       ->List[str]:
          lowerCamelCase__       :						Optional[Any]        =							pa.array(TypedSequence(['''foo''', '''bar''']		, try_type=Value('''int64'''      )      )      )
          self.assertEqual(arr.type		, pa.string()      )







       def 						__lowerCamelCase   (				self					:     Optional[int]      )       ->int:
          lowerCamelCase__       :						int        =							pa.array(TypedSequence([[[1, 2, 3]]]		, type=ArrayaD((1, 3)		, '''int64'''      )      )      )
          self.assertEqual(arr.type		, ArrayaDExtensionType((1, 3)		, '''int64'''      )      )







       def 						__lowerCamelCase   (				self					:     Dict      )       ->Dict:
          with self.assertRaises((TypeError, pa.lib.ArrowInvalid)      ):
             lowerCamelCase__       :						Optional[int]        =							pa.array(TypedSequence(['''foo''', '''bar''']		, type=ArrayaD((1, 3)		, '''int64'''      )      )      )







       def 						__lowerCamelCase   (				self					:     Tuple      )       ->Any:
          lowerCamelCase__       :						Optional[Any]        =							pa.array(TypedSequence([[[1, 2, 3]]]		, try_type=ArrayaD((1, 3)		, '''int64'''      )      )      )
          self.assertEqual(arr.type		, ArrayaDExtensionType((1, 3)		, '''int64'''      )      )







       def 						__lowerCamelCase   (				self					:     str      )       ->Optional[int]:
          lowerCamelCase__       :						Any        =							pa.array(TypedSequence(['''foo''', '''bar''']		, try_type=ArrayaD((1, 3)		, '''int64'''      )      )      )
          self.assertEqual(arr.type		, pa.string()      )







       @require_pil
       def 						__lowerCamelCase   (				self					:     Optional[int]      )       ->Optional[Any]:
          import PIL.Image

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



def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    Any:







   """simple docstring"""
   lowerCamelCase__       :						int        =							pa.BufferReader(UpperCAmelCase					) if isinstance(UpperCAmelCase      ,  pa.Buffer					) else pa.memory_map(UpperCAmelCase					)
   lowerCamelCase__       :						str        =							pa.ipc.open_stream(UpperCAmelCase					)
   lowerCamelCase__       :						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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    Optional[int]:







   """simple docstring"""
   lowerCamelCase__       :						str        =							pa.BufferOutputStream()
   lowerCamelCase__       :						List[str]        =							pa.schema(UpperCAmelCase					) if fields else None
   with ArrowWriter(stream=UpperCAmelCase      ,  schema=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase					) as writer:
      writer.write({'''col_1''': '''foo''', '''col_2''': 1}					)
      writer.write({'''col_1''': '''bar''', '''col_2''': 2}					)
      lowerCamelCase__       :						Tuple        =							writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      lowerCamelCase__       :						Optional[int]        =							{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
   assert writer._schema == pa.schema(UpperCAmelCase      ,  metadata=writer._schema.metadata					)
   _check_output(output.getvalue()      ,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)



def 	_a						(   )    ->    int:







   """simple docstring"""
   lowerCamelCase__       :						Union[str, Any]        =							pa.BufferOutputStream()
   lowerCamelCase__       :						Optional[int]        =							Features({'''labels''': ClassLabel(names=['''neg''', '''pos''']					)}					)
   with ArrowWriter(stream=UpperCAmelCase      ,  features=UpperCAmelCase					) as writer:
      writer.write({'''labels''': 0}					)
      writer.write({'''labels''': 1}					)
      lowerCamelCase__       :						Union[str, Any]        =							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__       :						Tuple        =							pa.BufferReader(output.getvalue()					)
   lowerCamelCase__       :						Union[str, Any]        =							pa.ipc.open_stream(UpperCAmelCase					)
   lowerCamelCase__       :						pa.Table        =							f.read_all()
   lowerCamelCase__       :						Union[str, 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(UpperCAmelCase					)



@pytest.mark.parametrize('''writer_batch_size'''      ,  [None, 1, 10]					)
def 	_a						(   UpperCAmelCase					)    ->    str:







   """simple docstring"""
   lowerCamelCase__       :						List[str]        =							pa.BufferOutputStream()
   with ArrowWriter(
       stream=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase      ,  hash_salt='''split_name'''      ,  check_duplicates=UpperCAmelCase      ,  ) as writer:
      with pytest.raises(UpperCAmelCase					):
         writer.write({'''col_1''': '''foo''', '''col_2''': 1}      ,  key=[1, 2]					)
         lowerCamelCase__       :						int        =							writer.finalize()



@pytest.mark.parametrize('''writer_batch_size'''      ,  [None, 2, 10]					)
def 	_a						(   UpperCAmelCase					)    ->    int:







   """simple docstring"""
   lowerCamelCase__       :						Optional[int]        =							pa.BufferOutputStream()
   with ArrowWriter(
       stream=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase      ,  hash_salt='''split_name'''      ,  check_duplicates=UpperCAmelCase      ,  ) as writer:
      with pytest.raises(UpperCAmelCase					):
         writer.write({'''col_1''': '''foo''', '''col_2''': 1}      ,  key=10					)
         writer.write({'''col_1''': '''bar''', '''col_2''': 2}      ,  key=10					)
         lowerCamelCase__       :						Tuple        =							writer.finalize()



@pytest.mark.parametrize('''writer_batch_size'''      ,  [None, 2, 10]					)
def 	_a						(   UpperCAmelCase					)    ->    Optional[int]:







   """simple docstring"""
   lowerCamelCase__       :						Union[str, Any]        =							pa.BufferOutputStream()
   with ArrowWriter(
       stream=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase      ,  hash_salt='''split_name'''      ,  check_duplicates=UpperCAmelCase      ,  ) as writer:
      writer.write({'''col_1''': '''foo''', '''col_2''': 1}      ,  key=1					)
      writer.write({'''col_1''': '''bar''', '''col_2''': 2}      ,  key=2					)
      lowerCamelCase__       :						List[str]        =							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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    List[Any]:







   """simple docstring"""
   lowerCamelCase__       :						List[Any]        =							pa.BufferOutputStream()
   lowerCamelCase__       :						List[str]        =							pa.schema(UpperCAmelCase					) if fields else None
   with ArrowWriter(stream=UpperCAmelCase      ,  schema=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase					) as writer:
      writer.write_batch({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}					)
      writer.write_batch({'''col_1''': [], '''col_2''': []}					)
      lowerCamelCase__       :						Tuple        =							writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      lowerCamelCase__       :						Optional[int]        =							{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
   assert writer._schema == pa.schema(UpperCAmelCase      ,  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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    Union[str, Any]:







   """simple docstring"""
   lowerCamelCase__       :						List[Any]        =							pa.BufferOutputStream()
   lowerCamelCase__       :						Any        =							pa.schema(UpperCAmelCase					) if fields else None
   with ArrowWriter(stream=UpperCAmelCase      ,  schema=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase					) as writer:
      writer.write_table(pa.Table.from_pydict({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}					)					)
      lowerCamelCase__       :						Union[str, Any]        =							writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      lowerCamelCase__       :						List[str]        =							{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
   assert writer._schema == pa.schema(UpperCAmelCase      ,  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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    Dict:







   """simple docstring"""
   lowerCamelCase__       :						Optional[int]        =							pa.BufferOutputStream()
   lowerCamelCase__       :						List[str]        =							pa.schema(UpperCAmelCase					) if fields else None
   with ArrowWriter(stream=UpperCAmelCase      ,  schema=UpperCAmelCase      ,  writer_batch_size=UpperCAmelCase					) 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__       :						Tuple        =							writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   if not fields:
      lowerCamelCase__       :						Any        =							{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
   assert writer._schema == pa.schema(UpperCAmelCase      ,  metadata=writer._schema.metadata					)
   _check_output(output.getvalue()      ,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)



def 	_a						(   )    ->    Optional[Any]:







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



def 	_a						(   UpperCAmelCase					)    ->    List[str]:







   """simple docstring"""
   if pa.types.is_list(UpperCAmelCase					):
      return get_base_dtype(arr_type.value_type					)
   else:
      return arr_type



def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    int:







   """simple docstring"""
   if isinstance(lst[0]      ,  UpperCAmelCase					):
      change_first_primitive_element_in_list(lst[0]      ,  UpperCAmelCase					)
   else:
      lowerCamelCase__       :						Optional[Any]        =							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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase					)    ->    Any:







   """simple docstring"""
   lowerCamelCase__       :						Dict        =							pa.array(TypedSequence(UpperCAmelCase      ,  optimized_int_type=UpperCAmelCase					)					)
   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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase					)    ->    int:







   """simple docstring"""
   lowerCamelCase__       :						Tuple        =							pa.array(OptimizedTypedSequence(UpperCAmelCase      ,  col=UpperCAmelCase					)					)
   assert get_base_dtype(arr.type					) == expected_dtype

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



@pytest.mark.parametrize('''raise_exception'''      ,  [False, True]					)
def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    Union[str, Any]:







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



def 	_a						(   UpperCAmelCase					)    ->    Union[str, Any]:







   """simple docstring"""
   lowerCamelCase__       :						Optional[int]        =							'''mock://dataset-train.arrow'''
   with ArrowWriter(path=UpperCAmelCase      ,  storage_options=mockfs.storage_options					) as writer:
      assert isinstance(writer._fs      ,  type(UpperCAmelCase					)					)
      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__       :						Union[str, Any]        =							writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   assert mockfs.exists(UpperCAmelCase					)



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







   """simple docstring"""
   lowerCamelCase__       :						Union[str, Any]        =							pa.BufferOutputStream()
   with ParquetWriter(stream=UpperCAmelCase					) as writer:
      writer.write({'''col_1''': '''foo''', '''col_2''': 1}					)
      writer.write({'''col_1''': '''bar''', '''col_2''': 2}					)
      lowerCamelCase__       :						List[str]        =							writer.finalize()
   assert num_examples == 2
   assert num_bytes > 0
   lowerCamelCase__       :						Dict        =							pa.BufferReader(output.getvalue()					)
   lowerCamelCase__       :						pa.Table        =							pq.read_table(UpperCAmelCase					)
   assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}



@require_pil
@pytest.mark.parametrize('''embed_local_files'''      ,  [False, True]					)
def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    Optional[Any]:







   """simple docstring"""
   import PIL.Image

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







   """simple docstring"""
   lowerCamelCase__       :						Optional[Any]        =							pa.schema([pa.field('''col_1'''      ,  pa.string()      ,  nullable=UpperCAmelCase					)]					)
   lowerCamelCase__       :						int        =							pa.BufferOutputStream()
   with ArrowWriter(stream=UpperCAmelCase					) as writer:
      writer._build_writer(inferred_schema=UpperCAmelCase					)
      assert writer._schema == pa.schema([pa.field('''col_1'''      ,  pa.string()					)]					)










from math import ceil, sqrt



def 	_a						(   UpperCAmelCase = 1000000					)    ->    int:







   """simple docstring"""
   lowerCamelCase__       :						Any        =							0

   for outer_width in range(3      ,  (limit // 4) + 2					):
      if outer_width**2 > limit:
         lowerCamelCase__       :						List[Any]        =							max(ceil(sqrt(outer_width**2 - limit					)					)      ,  1					)
      else:
         lowerCamelCase__       :						Union[str, Any]        =							1
      if (outer_width - hole_width_lower_bound) % 2:
         hole_width_lower_bound += 1

      answer += (outer_width - hole_width_lower_bound - 2) // 2 + 1

   return answer


if __name__ == "__main__":
       print(F'''{solution() = }''')



