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	:							List[Any]			)   ->						List[str]:

  """simple docstring"""





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

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

  """simple docstring"""





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

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

  """simple docstring"""





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

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

  """simple docstring"""





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

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

  """simple docstring"""





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

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

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Dict			)   ->						Any:

  """simple docstring"""





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

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

  """simple docstring"""





  UpperCamelCase_					:						List[Any] =					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[str]			)   ->						Optional[Any]:

  """simple docstring"""





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

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

  """simple docstring"""





  UpperCamelCase_					:						Dict =					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	:							Union[str, Any]			)   ->						int:

  """simple docstring"""





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


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

  """simple docstring"""





  import PIL.Image

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





def    __lowercase     (						lowerCamelCase					:     Optional[Any]  ,  lowerCamelCase					:     int	):
 UpperCamelCase_					:						Dict =					pa.BufferReader(lowerCamelCase	) if isinstance(lowerCamelCase  ,  pa.Buffer	) else pa.memory_map(lowerCamelCase	)
 UpperCamelCase_					:						Any =					pa.ipc.open_stream(lowerCamelCase	)
 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					:     Union[str, Any]  ,  lowerCamelCase					:     List[str]	):
 UpperCamelCase_					:						List[str] =					pa.BufferOutputStream()
 UpperCamelCase_					:						int =					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}	)
  UpperCamelCase_, UpperCamelCase_					:						Optional[Any] =					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(lowerCamelCase  ,  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_					:						Tuple =					Features({'labels': ClassLabel(names=['neg', 'pos']	)}	)
 with ArrowWriter(stream=lowerCamelCase  ,  features=lowerCamelCase	) as writer:
  writer.write({'labels': 0}	)
  writer.write({'labels': 1}	)
  UpperCamelCase_, UpperCamelCase_					:						Tuple =					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_					:						List[Any] =					pa.BufferReader(output.getvalue()	)
 UpperCamelCase_					:						int =					pa.ipc.open_stream(lowerCamelCase	)
 UpperCamelCase_					:						pa.Table =					f.read_all()
 UpperCamelCase_					:						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(lowerCamelCase	)





@pytest.mark.parametrize('writer_batch_size'  ,  [None, 1, 10]	)
def    __lowercase     (						lowerCamelCase					:     Dict	):
 UpperCamelCase_					:						List[str] =					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]	)
   UpperCamelCase_, UpperCamelCase_					:						List[str] =					writer.finalize()





@pytest.mark.parametrize('writer_batch_size'  ,  [None, 2, 10]	)
def    __lowercase     (						lowerCamelCase					:     Tuple	):
 UpperCamelCase_					:						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	)
   UpperCamelCase_, UpperCamelCase_					:						Dict =					writer.finalize()





@pytest.mark.parametrize('writer_batch_size'  ,  [None, 2, 10]	)
def    __lowercase     (						lowerCamelCase					:     List[str]	):
 UpperCamelCase_					:						Union[str, Any] =					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	)
  UpperCamelCase_, UpperCamelCase_					:						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					:     Union[str, Any]  ,  lowerCamelCase					:     Union[str, Any]	):
 UpperCamelCase_					:						Tuple =					pa.BufferOutputStream()
 UpperCamelCase_					:						Any =					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': []}	)
  UpperCamelCase_, UpperCamelCase_					:						List[str] =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  UpperCamelCase_					:						Tuple =					{'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    __lowercase     (						lowerCamelCase					:     Optional[int]  ,  lowerCamelCase					:     str	):
 UpperCamelCase_					:						str =					pa.BufferOutputStream()
 UpperCamelCase_					:						List[Any] =					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]}	)	)
  UpperCamelCase_, UpperCamelCase_					:						List[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(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    __lowercase     (						lowerCamelCase					:     int  ,  lowerCamelCase					:     Union[str, Any]	):
 UpperCamelCase_					:						Dict =					pa.BufferOutputStream()
 UpperCamelCase_					:						Optional[Any] =					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]}	)	)
  UpperCamelCase_, UpperCamelCase_					:						Union[str, Any] =					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(lowerCamelCase  ,  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_					:						Optional[int] =					{'col_1': pa.string(), 'col_2': pa.intaa()}
  UpperCamelCase_					:						int =					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]}	)
   UpperCamelCase_, UpperCamelCase_					:						int =					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    __lowercase     (						lowerCamelCase					:     Union[str, Any]	):
 if pa.types.is_list(lowerCamelCase	):
  return get_base_dtype(arr_type.value_type	)
 else:
  return arr_type





def    __lowercase     (						lowerCamelCase					:     Optional[Any]  ,  lowerCamelCase					:     Dict	):
 if isinstance(lst[0]  ,  lowerCamelCase	):
  change_first_primitive_element_in_list(lst[0]  ,  lowerCamelCase	)
 else:
  UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     Dict  ,  lowerCamelCase					:     Tuple  ,  lowerCamelCase					:     List[Any]	):
 UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     Optional[int]  ,  lowerCamelCase					:     Union[str, Any]  ,  lowerCamelCase					:     Union[str, Any]	):
 # in range
 UpperCamelCase_					:						List[Any] =					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
  UpperCamelCase_					:						str =					copy.deepcopy(lowerCamelCase	)
  UpperCamelCase_					:						List[str] =					np.iinfo(expected_dtype.to_pandas_dtype()	).max + 1
  change_first_primitive_element_in_list(lowerCamelCase  ,  lowerCamelCase	)
  UpperCamelCase_					:						Optional[int] =					pa.array(OptimizedTypedSequence(lowerCamelCase  ,  col=lowerCamelCase	)	)
  assert get_base_dtype(arr.type	) == pa.intaa()





@pytest.mark.parametrize('raise_exception'  ,  [False, True]	)
def    __lowercase     (						lowerCamelCase					:     Optional[int]  ,  lowerCamelCase					:     Dict	):
 UpperCamelCase_					:						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    __lowercase     (						lowerCamelCase					:     Any	):
 UpperCamelCase_					:						Dict =					'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}	)
  UpperCamelCase_, UpperCamelCase_					:						Union[str, Any] =					writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 assert mockfs.exists(lowerCamelCase	)





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

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




def    __lowercase     (						lowerCamelCase					:     str  ,  lowerCamelCase					:     str	):

 def get_matched_characters(lowerCamelCase					:     str  ,  lowerCamelCase					:     str	) -> str:
  UpperCamelCase_					:						Tuple =					[]
  UpperCamelCase_					:						List[Any] =					min(len(_stra	)  ,  len(_stra	)	) // 2
  for i, l in enumerate(_stra	):
   UpperCamelCase_					:						int =					int(max(0  ,  i - limit	)	)
   UpperCamelCase_					:						Dict =					int(min(i + limit + 1  ,  len(_stra	)	)	)
   if l in _stra[left:right]:
    matched.append(lowerCamelCase	)
    UpperCamelCase_					:						Dict =					F"{_stra[0:_stra.index(lowerCamelCase	)]} {_stra[_stra.index(lowerCamelCase	) + 1:]}"

  return "".join(lowerCamelCase	)

 # matching characters
 UpperCamelCase_					:						str =					get_matched_characters(lowerCamelCase  ,  lowerCamelCase	)
 UpperCamelCase_					:						str =					get_matched_characters(lowerCamelCase  ,  lowerCamelCase	)
 UpperCamelCase_					:						Union[str, Any] =					len(lowerCamelCase	)

 # transposition
 UpperCamelCase_					:						int =					(
     len([(ca, ca) for ca, ca in zip(lowerCamelCase  ,  lowerCamelCase	) if ca != ca]	) // 2
 )

 if not match_count:
  UpperCamelCase_					:						Union[str, Any] =					0.0
 else:
  UpperCamelCase_					:						str =					(
      1
      / 3
      * (
          match_count / len(lowerCamelCase	)
          + match_count / len(lowerCamelCase	)
          + (match_count - transpositions) / match_count
      )
  )

 # common prefix up to 4 characters
 UpperCamelCase_					:						Dict =					0
 for ca, ca in zip(stra[:4]  ,  stra[:4]	):
  if ca == ca:
   prefix_len += 1
  else:
   break

 return jaro + 0.1 * prefix_len * (1 - jaro)


if __name__ == "__main__":
   import doctest

   doctest.testmod()
   print(jaro_winkler('hello', 'world'))



