
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      UpperCamelCase (				snake_case__ ):







		"""simple docstring"""



		def       A(  self		:       Any		)    ->   Dict:


						'''simple docstring'''
						A	 =      pa.array(TypedSequence([1, 2, 3]		)		)
						self.assertEqual(arr.type						,pa.intaa()		)
		def       A(  self		:       Union[str, Any]		)    ->   Optional[int]:


						'''simple docstring'''
						with self.assertRaises(_SCREAMING_SNAKE_CASE		):
										A	 =      pa.array(TypedSequence([1, 2, 3]		)						,type=pa.intaa()		)
		def       A(  self		:       Optional[Any]		)    ->   List[Any]:


						'''simple docstring'''
						with self.assertRaises(_SCREAMING_SNAKE_CASE		):
										A	 =      pa.array(TypedSequence([1, 2, 3]						,try_type=Value('bool'		)						,type=Value('int64'		)		)		)
		def       A(  self		:       List[Any]		)    ->   Tuple:


						'''simple docstring'''
						A	 =      pa.array(TypedSequence([1, 2, 3]						,type=Value('int32'		)		)		)
						self.assertEqual(arr.type						,pa.intaa()		)
		def       A(  self		:       Tuple		)    ->   int:


						'''simple docstring'''
						with self.assertRaises((TypeError, pa.lib.ArrowInvalid)		):
										A	 =      pa.array(TypedSequence(['foo', 'bar']						,type=Value('int64'		)		)		)
		def       A(  self		:       Optional[int]		)    ->   str:


						'''simple docstring'''
						A	 =      pa.array(TypedSequence([1, 2, 3]						,try_type=Value('int32'		)		)		)
						self.assertEqual(arr.type						,pa.intaa()		)
		def       A(  self		:       int		)    ->   Any:


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


						'''simple docstring'''
						A	 =      pa.array(TypedSequence([[[1, 2, 3]]]						,type=ArrayaD((1, 3)						,'int64'		)		)		)
						self.assertEqual(arr.type						,ArrayaDExtensionType((1, 3)						,'int64'		)		)
		def       A(  self		:       int		)    ->   List[str]:


						'''simple docstring'''
						with self.assertRaises((TypeError, pa.lib.ArrowInvalid)		):
										A	 =      pa.array(TypedSequence(['foo', 'bar']						,type=ArrayaD((1, 3)						,'int64'		)		)		)
		def       A(  self		:       List[str]		)    ->   Tuple:


						'''simple docstring'''
						A	 =      pa.array(TypedSequence([[[1, 2, 3]]]						,try_type=ArrayaD((1, 3)						,'int64'		)		)		)
						self.assertEqual(arr.type						,ArrayaDExtensionType((1, 3)						,'int64'		)		)
		def       A(  self		:       Union[str, Any]		)    ->   int:


						'''simple docstring'''
						A	 =      pa.array(TypedSequence(['foo', 'bar']						,try_type=ArrayaD((1, 3)						,'int64'		)		)		)
						self.assertEqual(arr.type						,pa.string()		)







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


						'''simple docstring'''
						import PIL.Image

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







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







@pytest.mark.parametrize('writer_batch_size',						[None, 2, 10]   )
def      snake_case		(							UpperCAmelCase	:       Any   ):
				A	 =      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   )
												A ,		A	 =      writer.finalize()







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







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







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







def      snake_case		(							):
				A	 =      pa.BufferOutputStream()
				with ParquetWriter(stream=UpperCAmelCase   ) as writer:
								writer.write({'col_1': 'foo', 'col_2': 1}   )
								writer.write({'col_1': 'bar', 'col_2': 2}   )
								A ,		A	 =      writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				A	 =      pa.BufferReader(output.getvalue()   )
				A	 =      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      snake_case		(							UpperCAmelCase	:       Dict,						UpperCAmelCase	:       Optional[int]   ):
				import PIL.Image

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







def      snake_case		(							UpperCAmelCase	:       tuple   ):
				if num[3] % 2 != 0:
								return False

				if (num[2] + num[3] + num[4]) % 3 != 0:
								return False

				if num[5] % 5 != 0:
								return False

				A	 =      [7, 11, 13, 17]
				for i, test in enumerate(UpperCAmelCase   ):
								if (num[i + 4] * 1_00 + num[i + 5] * 10 + num[i + 6]) % test != 0:
												return False
				return True







def      snake_case		(							UpperCAmelCase	:       int = 10   ):
				return sum(
				    int(''.join(map(UpperCAmelCase,						UpperCAmelCase   )   )   )
				    for num in permutations(range(UpperCAmelCase   )   )
				    if is_substring_divisible(UpperCAmelCase   )   )


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





