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 		_lowercase						(					snake_case_       ):

 def 						SCREAMING_SNAKE_CASE__   ( self	:							int			)   ->						Dict:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Optional[int]			)   ->						int:

  """simple docstring"""





  with self.assertRaises(__A			):
   UpperCamelCase_					:						Any =					pa.array(TypedSequence([1, 2, 3]			)   ,      type=pa.intaa()			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Optional[Any]			)   ->						str:

  """simple docstring"""





  with self.assertRaises(__A			):
   UpperCamelCase_					:						Optional[int] =					pa.array(TypedSequence([1, 2, 3]   ,      try_type=Value('bool'			)   ,      type=Value('int64'			)			)			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Dict			)   ->						Optional[Any]:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Any			)   ->						List[str]:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Any			)   ->						int:

  """simple docstring"""





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

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

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Union[str, Any]			)   ->						Tuple:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							List[Any]			)   ->						Tuple:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Tuple			)   ->						Dict:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							List[str]			)   ->						List[Any]:

  """simple docstring"""





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


 @require_pil
 def 						SCREAMING_SNAKE_CASE__   ( self	:							List[Any]			)   ->						Union[str, Any]:

  """simple docstring"""





  import PIL.Image

  UpperCamelCase_					:						Union[str, Any] =					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:
   UpperCamelCase_					:						List[str] =					pa.array(TypedSequence([{'path': None, 'bytes': B'image_bytes'}, pil_image]   ,      type=Image()			)			)
   UpperCamelCase_					:						Tuple =					mock_cast_to_python_objects.call_args_list[-1]
   self.assertIn('optimize_list_casting'   ,      __A			)
   self.assertFalse(kwargs['optimize_list_casting']			)





def    __lowercase     (						lowerCamelCase					:     Optional[int]  ,  lowerCamelCase					:     str	):
 UpperCamelCase_					:						Any =					pa.BufferReader(SCREAMING_SNAKE_CASE_	) if isinstance(SCREAMING_SNAKE_CASE_  ,  pa.Buffer	) else pa.memory_map(SCREAMING_SNAKE_CASE_	)
 UpperCamelCase_					:						List[str] =					pa.ipc.open_stream(SCREAMING_SNAKE_CASE_	)
 UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     List[str]  ,  lowerCamelCase					:     Any	):
 UpperCamelCase_					:						List[Any] =					pa.BufferOutputStream()
 UpperCamelCase_					:						Optional[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}	)
  UpperCamelCase_					:						str =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  UpperCamelCase_					:						int =					{"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    __lowercase     (						):
 UpperCamelCase_					:						Any =					pa.BufferOutputStream()
 UpperCamelCase_					:						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}	)
  UpperCamelCase_					:						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
 UpperCamelCase_					:						Union[str, Any] =					pa.BufferReader(output.getvalue()	)
 UpperCamelCase_					:						List[str] =					pa.ipc.open_stream(SCREAMING_SNAKE_CASE_	)
 UpperCamelCase_					:						pa.Table =					f.read_all()
 UpperCamelCase_					:						str =					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    __lowercase     (						lowerCamelCase					:     List[Any]	):
 UpperCamelCase_					:						Dict =					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]	)
   UpperCamelCase_					:						Dict =					writer.finalize()





@pytest.mark.parametrize('writer_batch_size'  ,  [None, 2, 10]	)
def    __lowercase     (						lowerCamelCase					:     str	):
 UpperCamelCase_					:						Optional[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=10	)
   writer.write({'col_1': 'bar', 'col_2': 2}  ,  key=10	)
   UpperCamelCase_					:						List[Any] =					writer.finalize()





@pytest.mark.parametrize('writer_batch_size'  ,  [None, 2, 10]	)
def    __lowercase     (						lowerCamelCase					:     List[Any]	):
 UpperCamelCase_					:						Optional[Any] =					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	)
  UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     Any  ,  lowerCamelCase					:     int	):
 UpperCamelCase_					:						Any =					pa.BufferOutputStream()
 UpperCamelCase_					:						Optional[int] =					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': []}	)
  UpperCamelCase_					:						List[Any] =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     List[Any]  ,  lowerCamelCase					:     List[Any]	):
 UpperCamelCase_					:						Optional[int] =					pa.BufferOutputStream()
 UpperCamelCase_					:						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_table(pa.Table.from_pydict({'col_1': ['foo', 'bar'], 'col_2': [1, 2]}	)	)
  UpperCamelCase_					:						Union[str, Any] =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  UpperCamelCase_					:						Optional[int] =					{"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    __lowercase     (						lowerCamelCase					:     Any  ,  lowerCamelCase					:     Tuple	):
 UpperCamelCase_					:						List[str] =					pa.BufferOutputStream()
 UpperCamelCase_					:						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_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_					:						Dict =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  UpperCamelCase_					:						Optional[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	)





def    __lowercase     (						):

 with tempfile.TemporaryDirectory() as tmp_dir:
  UpperCamelCase_					:						Any =					{"col_1": pa.string(), "col_2": pa.intaa()}
  UpperCamelCase_					:						Any =					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]}	)
   UpperCamelCase_					:						int =					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    __lowercase     (						lowerCamelCase					:     Optional[Any]	):

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





def    __lowercase     (						lowerCamelCase					:     Any  ,  lowerCamelCase					:     Tuple	):

 if isinstance(lst[0]  ,  SCREAMING_SNAKE_CASE_	):
  change_first_primitive_element_in_list(lst[0]  ,  SCREAMING_SNAKE_CASE_	)
 else:
  UpperCamelCase_					:						List[str] =					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    __lowercase     (						lowerCamelCase					:     Optional[int]  ,  lowerCamelCase					:     Optional[Any]  ,  lowerCamelCase					:     Any	):
 UpperCamelCase_					:						Optional[Any] =					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    __lowercase     (						lowerCamelCase					:     List[Any]  ,  lowerCamelCase					:     Optional[Any]  ,  lowerCamelCase					:     List[str]	):
 UpperCamelCase_					:						Optional[Any] =					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
  UpperCamelCase_					:						Union[str, Any] =					copy.deepcopy(SCREAMING_SNAKE_CASE_	)
  UpperCamelCase_					:						Optional[Any] =					np.iinfo(expected_dtype.to_pandas_dtype()	).max + 1
  change_first_primitive_element_in_list(SCREAMING_SNAKE_CASE_  ,  SCREAMING_SNAKE_CASE_	)
  UpperCamelCase_					:						Optional[Any] =					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    __lowercase     (						lowerCamelCase					:     Tuple  ,  lowerCamelCase					:     Tuple	):
 UpperCamelCase_					:						Optional[Any] =					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    __lowercase     (						lowerCamelCase					:     Optional[Any]	):
 UpperCamelCase_					:						Dict =					"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}	)
  UpperCamelCase_					:						Optional[int] =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 assert mockfs.exists(SCREAMING_SNAKE_CASE_	)





def    __lowercase     (						):
 UpperCamelCase_					:						Union[str, Any] =					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}	)
  UpperCamelCase_					:						List[Any] =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 UpperCamelCase_					:						Dict =					pa.BufferReader(output.getvalue()	)
 UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     Dict  ,  lowerCamelCase					:     Any	):
 import PIL.Image

 UpperCamelCase_					:						Dict =					str(tmp_path / 'test_image_rgb.jpg'	)
 PIL.Image.fromarray(np.zeros((5, 5)  ,  dtype=np.uinta	)	).save(SCREAMING_SNAKE_CASE_  ,  format='png'	)
 UpperCamelCase_					:						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()
 UpperCamelCase_					:						Dict =					pa.BufferReader(output.getvalue()	)
 UpperCamelCase_					:						pa.Table =					pq.read_table(SCREAMING_SNAKE_CASE_	)
 UpperCamelCase_					:						Tuple =					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    __lowercase     (						):
 UpperCamelCase_					:						Tuple =					pa.schema([pa.field('col_1'  ,  pa.string()  ,  nullable=SCREAMING_SNAKE_CASE_	)]	)
 UpperCamelCase_					:						int =					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()	)]	)







import warnings

from ...utils import logging
from .image_processing_dpt import DPTImageProcessor


_snake_case             =     logging.get_logger(__name__)
class 	UpperCAmelCase_					(					UpperCamelCase    ):

   '''simple docstring'''




   def __init__(		self						,				*__A						,				**__A ):






     """simple docstring"""





     warnings.warn(
         "The class DPTFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please"
         " use DPTImageProcessor instead."						,				__A						,				)
     super().__init__(*__A						,				**__A )



