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


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




   lowercase__	: List[str]     		=   pa.array(TypedSequence([1, 2, 3]       ,	type=Value("""int32""")))
   self.assertEqual(arr.type       ,	pa.intaa())


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




   lowercase__	: Optional[int]     		=   pa.array(TypedSequence([1, 2, 3]       ,	try_type=Value("""int32""")))
   self.assertEqual(arr.type       ,	pa.intaa())


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




   lowercase__	: List[str]     		=   pa.array(TypedSequence([[[1, 2, 3]]]       ,	type=ArrayaD((1, 3)       ,	"""int64""")))
   self.assertEqual(arr.type       ,	ArrayaDExtensionType((1, 3)       ,	"""int64"""))


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




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


 def   lowercase__  (       self):


   '''simple docstring'''




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



 @require_pil
 def   lowercase__  (       self):


   '''simple docstring'''




   import PIL.Image

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

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


  '''simple docstring'''






  lowercase__	: Any     		=   pa.BufferReader(SCREAMING_SNAKE_CASE__ ) if isinstance(SCREAMING_SNAKE_CASE__			,			pa.Buffer ) else pa.memory_map(SCREAMING_SNAKE_CASE__ )
  lowercase__	: str     		=   pa.ipc.open_stream(SCREAMING_SNAKE_CASE__ )
  lowercase__	: 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						(   lowercase_			,			lowercase_ )		->						Any:


  '''simple docstring'''






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

def 		UpperCamelCase						(   )		->						List[Any]:


  '''simple docstring'''






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

@pytest.mark.parametrize("""writer_batch_size"""			,			[None, 1, 10] )
def 		UpperCamelCase						(   lowercase_ )		->						List[str]:


  '''simple docstring'''






  lowercase__	: int     		=   pa.BufferOutputStream()
  with ArrowWriter(
      stream=SCREAMING_SNAKE_CASE__			,			writer_batch_size=SCREAMING_SNAKE_CASE__			,			hash_salt="""split_name"""			,			check_duplicates=SCREAMING_SNAKE_CASE__			,			) as writer:
    with pytest.raises(SCREAMING_SNAKE_CASE__ ):
      writer.write({"""col_1""": """foo""", """col_2""": 1}			,			key=[1, 2] )
      lowercase__	: Tuple     		=   writer.finalize()

@pytest.mark.parametrize("""writer_batch_size"""			,			[None, 2, 10] )
def 		UpperCamelCase						(   lowercase_ )		->						int:


  '''simple docstring'''






  lowercase__	: Tuple     		=   pa.BufferOutputStream()
  with ArrowWriter(
      stream=SCREAMING_SNAKE_CASE__			,			writer_batch_size=SCREAMING_SNAKE_CASE__			,			hash_salt="""split_name"""			,			check_duplicates=SCREAMING_SNAKE_CASE__			,			) as writer:
    with pytest.raises(SCREAMING_SNAKE_CASE__ ):
      writer.write({"""col_1""": """foo""", """col_2""": 1}			,			key=10 )
      writer.write({"""col_1""": """bar""", """col_2""": 2}			,			key=10 )
      lowercase__	: Any     		=   writer.finalize()

@pytest.mark.parametrize("""writer_batch_size"""			,			[None, 2, 10] )
def 		UpperCamelCase						(   lowercase_ )		->						Dict:


  '''simple docstring'''






  lowercase__	: str     		=   pa.BufferOutputStream()
  with ArrowWriter(
      stream=SCREAMING_SNAKE_CASE__			,			writer_batch_size=SCREAMING_SNAKE_CASE__			,			hash_salt="""split_name"""			,			check_duplicates=SCREAMING_SNAKE_CASE__			,			) as writer:
    writer.write({"""col_1""": """foo""", """col_2""": 1}			,			key=1 )
    writer.write({"""col_1""": """bar""", """col_2""": 2}			,			key=2 )
    lowercase__	: Optional[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						(   lowercase_			,			lowercase_ )		->						str:


  '''simple docstring'''






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


  '''simple docstring'''






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


  '''simple docstring'''






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

def 		UpperCamelCase						(   )		->						Optional[Any]:


  '''simple docstring'''






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

def 		UpperCamelCase						(   lowercase_ )		->						Dict:


  '''simple docstring'''






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

def 		UpperCamelCase						(   lowercase_			,			lowercase_ )		->						str:


  '''simple docstring'''






  if isinstance(lst[0]			,			SCREAMING_SNAKE_CASE__ ):
    change_first_primitive_element_in_list(lst[0]			,			SCREAMING_SNAKE_CASE__ )
  else:
    lowercase__	: 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 		UpperCamelCase						(   lowercase_			,			lowercase_			,			lowercase_ )		->						Tuple:


  '''simple docstring'''






  lowercase__	: Tuple     		=   pa.array(TypedSequence(SCREAMING_SNAKE_CASE__			,			optimized_int_type=SCREAMING_SNAKE_CASE__ ) )
  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						(   lowercase_			,			lowercase_			,			lowercase_ )		->						str:


  '''simple docstring'''






  lowercase__	: Tuple     		=   pa.array(OptimizedTypedSequence(SCREAMING_SNAKE_CASE__			,			col=SCREAMING_SNAKE_CASE__ ) )
  assert get_base_dtype(arr.type ) == expected_dtype

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

@pytest.mark.parametrize("""raise_exception"""			,			[False, True] )
def 		UpperCamelCase						(   lowercase_			,			lowercase_ )		->						Union[str, Any]:


  '''simple docstring'''






  lowercase__	: Optional[int]     		=   str(tmp_path / """dataset-train.arrow""" )
  try:
    with ArrowWriter(path=SCREAMING_SNAKE_CASE__ ) 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						(   lowercase_ )		->						int:


  '''simple docstring'''






  lowercase__	: Optional[int]     		=   """mock://dataset-train.arrow"""
  with ArrowWriter(path=SCREAMING_SNAKE_CASE__			,			storage_options=mockfs.storage_options ) as writer:
    assert isinstance(writer._fs			,			type(SCREAMING_SNAKE_CASE__ ) )
    assert writer._fs.storage_options == mockfs.storage_options
    writer.write({"""col_1""": """foo""", """col_2""": 1} )
    writer.write({"""col_1""": """bar""", """col_2""": 2} )
    lowercase__	: int     		=   writer.finalize()
  assert num_examples == 2
  assert num_bytes > 0
  assert mockfs.exists(SCREAMING_SNAKE_CASE__ )

def 		UpperCamelCase						(   )		->						str:


  '''simple docstring'''






  lowercase__	: Dict     		=   pa.BufferOutputStream()
  with ParquetWriter(stream=SCREAMING_SNAKE_CASE__ ) as writer:
    writer.write({"""col_1""": """foo""", """col_2""": 1} )
    writer.write({"""col_1""": """bar""", """col_2""": 2} )
    lowercase__	: Optional[int]     		=   writer.finalize()
  assert num_examples == 2
  assert num_bytes > 0
  lowercase__	: str     		=   pa.BufferReader(output.getvalue() )
  lowercase__	: pa.Table     		=   pq.read_table(SCREAMING_SNAKE_CASE__ )
  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						(   lowercase_			,			lowercase_ )		->						Tuple:


  '''simple docstring'''






  import PIL.Image

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


  '''simple docstring'''






  lowercase__	: List[Any]     		=   pa.schema([pa.field("""col_1"""			,			pa.string()			,			nullable=SCREAMING_SNAKE_CASE__ )] )
  lowercase__	: Any     		=   pa.BufferOutputStream()
  with ArrowWriter(stream=SCREAMING_SNAKE_CASE__ ) as writer:
    writer._build_writer(inferred_schema=SCREAMING_SNAKE_CASE__ )
    assert writer._schema == pa.schema([pa.field("""col_1"""			,			pa.string() )] )










'''simple docstring'''


def      snake_case_  (    SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	):







				"""simple docstring"""





				if b == 0:
								return 1
				if (b % 2) == 0:
								return actual_power(SCREAMING_SNAKE_CASE__				, int(b / 2	)	) * actual_power(SCREAMING_SNAKE_CASE__				, int(b / 2	)	)
				else:
								return a * actual_power(SCREAMING_SNAKE_CASE__				, int(b / 2	)	) * actual_power(SCREAMING_SNAKE_CASE__				, int(b / 2	)	)







def      snake_case_  (    SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	):







				"""simple docstring"""





				if b < 0:
								return 1 / actual_power(SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	)
				return actual_power(SCREAMING_SNAKE_CASE__				, SCREAMING_SNAKE_CASE__	)


if __name__ == "__main__":
				print(power(-2, -3))




