
import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
							import PIL.Image

							from .features import FeatureType


lowerCamelCase_  :	Optional[List[str]] 				=					None
lowerCamelCase_  :	Dict 				=					"""<""" if sys.byteorder == """little""" else """>"""
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
lowerCamelCase_  :	Tuple 				=					[
    np.dtype("""|b1"""),
    np.dtype("""|u1"""),
    np.dtype("""<u2"""),
    np.dtype(""">u2"""),
    np.dtype("""<i2"""),
    np.dtype(""">i2"""),
    np.dtype("""<u4"""),
    np.dtype(""">u4"""),
    np.dtype("""<i4"""),
    np.dtype(""">i4"""),
    np.dtype("""<f4"""),
    np.dtype(""">f4"""),
    np.dtype("""<f8"""),
    np.dtype(""">f8"""),
]

@dataclass
class     _UpperCamelCase			:






			'''simple docstring'''

			__UpperCamelCase :  bool    =	True
			__UpperCamelCase :  Optional[str]    =	None
			# Automatically constructed
			__UpperCamelCase :  ClassVar[str]    =	"PIL.Image.Image"
			__UpperCamelCase :  ClassVar[Any]    =	pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}					)
			__UpperCamelCase :  str    =	field(default="""Image"""       ,    init=_A       ,    repr=_A					)

			def __call__(    self	:    Dict ):
							return self.pa_type

			def  lowerCAmelCase__		(    self	:    Any  ,		snake_case_	:    Union[str, bytes, dict, np.ndarray, "PIL.Image.Image"] ):
							if config.PIL_AVAILABLE:
											import PIL.Image
							else:
											raise ImportError("""To support encoding images, please install 'Pillow'.""" )

							if isinstance(snake_case_  ,		snake_case_ ):
											UpperCamelCase_:							List[Any]		=  np.array(snake_case_ )

							if isinstance(snake_case_  ,		snake_case_ ):
											return {"path": value, "bytes": None}
							elif isinstance(snake_case_  ,		snake_case_ ):
											return {"path": None, "bytes": value}
							elif isinstance(snake_case_  ,		np.ndarray ):
											# convert the image array to PNG/TIFF bytes
											return encode_np_array(snake_case_ )
							elif isinstance(snake_case_  ,		PIL.Image.Image ):
											# convert the PIL image to bytes (default format is PNG/TIFF)
											return encode_pil_image(snake_case_ )
							elif value.get("""path""" ) is not None and os.path.isfile(value["""path"""] ):
											# we set "bytes": None to not duplicate the data if they're already available locally
											return {"bytes": None, "path": value.get("""path""" )}
							elif value.get("""bytes""" ) is not None or value.get("""path""" ) is not None:
											# store the image bytes, and path is used to infer the image format using the file extension
											return {"bytes": value.get("""bytes""" ), "path": value.get("""path""" )}
							else:
											raise ValueError(
											    f'''An image sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.''' )

			def  lowerCAmelCase__		(    self	:    Optional[Any]  ,		snake_case_	:    dict  ,		snake_case_	:    Optional[int]=None ):
							if not self.decode:
											raise RuntimeError("""Decoding is disabled for this feature. Please use Image(decode=True) instead.""" )

							if config.PIL_AVAILABLE:
											import PIL.Image
							else:
											raise ImportError("""To support decoding images, please install 'Pillow'.""" )

							if token_per_repo_id is None:
											UpperCamelCase_:							Union[str, Any]		=  {}

							UpperCamelCase_,				UpperCamelCase_:							List[str]		=  value["""path"""], value["""bytes"""]
							if bytes_ is None:
											if path is None:
															raise ValueError(f'''An image should have one of \'path\' or \'bytes\' but both are None in {value}.''' )
											else:
															if is_local_path(snake_case_ ):
																			UpperCamelCase_:							str		=  PIL.Image.open(snake_case_ )
															else:
																			UpperCamelCase_:							str		=  path.split("""::""" )[-1]
																			try:
																							UpperCamelCase_:							Tuple		=  string_to_dict(snake_case_  ,		config.HUB_DATASETS_URL )["""repo_id"""]
																							UpperCamelCase_:							Tuple		=  token_per_repo_id.get(snake_case_ )
																			except ValueError:
																							UpperCamelCase_:							Tuple		=  None
																			with xopen(snake_case_  ,		"""rb"""  ,		use_auth_token=snake_case_ ) as f:
																							UpperCamelCase_:							Tuple		=  BytesIO(f.read() )
																			UpperCamelCase_:							List[str]		=  PIL.Image.open(bytes_ )
							else:
											UpperCamelCase_:							Optional[int]		=  PIL.Image.open(BytesIO(bytes_ ) )
							image.load()  # to avoid "Too many open files" errors
							return image

			def  lowerCAmelCase__		(    self	:    List[str] ):
							from .features import Value

							return (
							    self
							    if self.decode
							    else {
							        "bytes": Value("""binary""" ),
							        "path": Value("""string""" ),
							    }
							)

			def  lowerCAmelCase__		(    self	:    Any  ,		snake_case_	:    Union[pa.StringArray, pa.StructArray, pa.ListArray] ):
							if pa.types.is_string(storage.type ):
											UpperCamelCase_:							Union[str, Any]		=  pa.array([None] * len(snake_case_ )  ,		type=pa.binary() )
											UpperCamelCase_:							Any		=  pa.StructArray.from_arrays([bytes_array, storage]  ,		["""bytes""", """path"""]  ,		mask=storage.is_null() )
							elif pa.types.is_binary(storage.type ):
											UpperCamelCase_:							List[Any]		=  pa.array([None] * len(snake_case_ )  ,		type=pa.string() )
											UpperCamelCase_:							List[Any]		=  pa.StructArray.from_arrays([storage, path_array]  ,		["""bytes""", """path"""]  ,		mask=storage.is_null() )
							elif pa.types.is_struct(storage.type ):
											if storage.type.get_field_index("""bytes""" ) >= 0:
															UpperCamelCase_:							Dict		=  storage.field("""bytes""" )
											else:
															UpperCamelCase_:							Tuple		=  pa.array([None] * len(snake_case_ )  ,		type=pa.binary() )
											if storage.type.get_field_index("""path""" ) >= 0:
															UpperCamelCase_:							Optional[Any]		=  storage.field("""path""" )
											else:
															UpperCamelCase_:							Any		=  pa.array([None] * len(snake_case_ )  ,		type=pa.string() )
											UpperCamelCase_:							Optional[int]		=  pa.StructArray.from_arrays([bytes_array, path_array]  ,		["""bytes""", """path"""]  ,		mask=storage.is_null() )
							elif pa.types.is_list(storage.type ):
											UpperCamelCase_:							Any		=  pa.array(
											    [encode_np_array(np.array(snake_case_ ) )["""bytes"""] if arr is not None else None for arr in storage.to_pylist()]  ,		type=pa.binary()  ,		)
											UpperCamelCase_:							int		=  pa.array([None] * len(snake_case_ )  ,		type=pa.string() )
											UpperCamelCase_:							Optional[Any]		=  pa.StructArray.from_arrays(
											    [bytes_array, path_array]  ,		["""bytes""", """path"""]  ,		mask=bytes_array.is_null() )
							return array_cast(snake_case_  ,		self.pa_type )




			def  lowerCAmelCase__		(    self	:    Optional[Any]  ,		snake_case_	:    pa.StructArray ):

							@no_op_if_value_is_null
							def path_to_bytes(snake_case_	:    Optional[int] ):
											with xopen(snake_case_  ,		"""rb""" ) as f:
															UpperCamelCase_:							Optional[int]		=  f.read()
											return bytes_

							UpperCamelCase_:							int		=  pa.array(
							    [
							        (path_to_bytes(x["""path"""] ) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
							        for x in storage.to_pylist()
							    ]  ,		type=pa.binary()  ,		)
							UpperCamelCase_:							str		=  pa.array(
							    [os.path.basename(snake_case_ ) if path is not None else None for path in storage.field("""path""" ).to_pylist()]  ,		type=pa.string()  ,		)
							UpperCamelCase_:							Optional[int]		=  pa.StructArray.from_arrays([bytes_array, path_array]  ,		["""bytes""", """path"""]  ,		mask=bytes_array.is_null() )
							return array_cast(snake_case_  ,		self.pa_type )







def     A__   (  )					->		List[str]:
				if config.PIL_AVAILABLE:
								import PIL.Image
				else:
								raise ImportError("""To support encoding images, please install 'Pillow'."""  )

				global _IMAGE_COMPRESSION_FORMATS
				if _IMAGE_COMPRESSION_FORMATS is None:
								PIL.Image.init()
								UpperCamelCase_:							Dict		=  list(set(PIL.Image.OPEN.keys()  ) & set(PIL.Image.SAVE.keys()  )  )
				return _IMAGE_COMPRESSION_FORMATS







def     A__   (  lowerCamelCase  )					->		bytes:
				UpperCamelCase_:							Any		=  BytesIO()
				if image.format in list_image_compression_formats():
								UpperCamelCase_:							Union[str, Any]		=  image.format
				else:
								UpperCamelCase_:							Any		=  """PNG""" if image.mode in ["""1""", """L""", """LA""", """RGB""", """RGBA"""] else """TIFF"""
				image.save(lowerCamelCase       ,      format=lowerCamelCase  )
				return buffer.getvalue()







def     A__   (  lowerCamelCase  )					->		dict:
				if hasattr(lowerCamelCase       ,      """filename"""  ) and image.filename != "":
								return {"path": image.filename, "bytes": None}
				else:
								return {"path": None, "bytes": image_to_bytes(lowerCamelCase  )}







def     A__   (  lowerCamelCase  )					->		dict:
				if config.PIL_AVAILABLE:
								import PIL.Image
				else:
								raise ImportError("""To support encoding images, please install 'Pillow'."""  )

				UpperCamelCase_:							Tuple		=  array.dtype
				UpperCamelCase_:							int		=  dtype.byteorder if dtype.byteorder != """=""" else _NATIVE_BYTEORDER
				UpperCamelCase_:							Any		=  dtype.kind
				UpperCamelCase_:							Any		=  dtype.itemsize

				UpperCamelCase_:							Any		=  None

				# Multi-channel array case (only np.dtype("|u1") is allowed)
				if array.shape[2:]:
								UpperCamelCase_:							int		=  np.dtype("""|u1"""  )
								if dtype_kind not in ["u", "i"]:
												raise TypeError(
												    F'''Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays.'''  )
								if dtype is not dest_dtype:
												warnings.warn(F'''Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\''''  )
    # Exact match
				elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
								UpperCamelCase_:							List[str]		=  dtype
				else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
								while dtype_itemsize >= 1:
												UpperCamelCase_:							List[str]		=  dtype_byteorder + dtype_kind + str(lowerCamelCase  )
												UpperCamelCase_:							Dict		=  np.dtype(lowerCamelCase  )
												if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
																warnings.warn(F'''Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\''''  )
																break
												else:
																dtype_itemsize //= 2

				if dest_dtype is None:
								raise TypeError(
								    F'''Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}'''  )
				UpperCamelCase_:							List[str]		=  PIL.Image.fromarray(array.astype(lowerCamelCase  )  )
				return {"path": None, "bytes": image_to_bytes(lowerCamelCase  )}







def     A__   (  lowerCamelCase  )					->		List[dict]:
				if config.PIL_AVAILABLE:
								import PIL.Image
				else:
								raise ImportError("""To support encoding images, please install 'Pillow'."""  )

				if objs:
								UpperCamelCase_,				UpperCamelCase_:							Optional[Any]		=  first_non_null_value(lowerCamelCase  )
								if isinstance(lowerCamelCase       ,      lowerCamelCase  ):
												return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
								if isinstance(lowerCamelCase       ,      np.ndarray  ):
												UpperCamelCase_:							Optional[int]		=  no_op_if_value_is_null(lowerCamelCase  )
												return [obj_to_image_dict_func(lowerCamelCase  ) for obj in objs]
								elif isinstance(lowerCamelCase       ,      PIL.Image.Image  ):
												UpperCamelCase_:							Tuple		=  no_op_if_value_is_null(lowerCamelCase  )
												return [obj_to_image_dict_func(lowerCamelCase  ) for obj in objs]
								else:
												return objs
				else:
								return objs








import io
import itertools
import json
from dataclasses import dataclass
from typing import Optional

import pyarrow as pa
import pyarrow.json as paj

import datasets
from datasets.table import table_cast
from datasets.utils.file_utils import readline


lowerCamelCase_  :	Optional[Any] 				=					datasets.utils.logging.get_logger(__name__)

@dataclass
class     _UpperCamelCase			(   datasets.BuilderConfig					):






			'''simple docstring'''

			__UpperCamelCase :  Optional[datasets.Features]    =	None
			__UpperCamelCase :  str    =	"utf-8"
			__UpperCamelCase :  Optional[str]    =	None
			__UpperCamelCase :  Optional[str]    =	None
			__UpperCamelCase :  bool    =	True  # deprecated
			__UpperCamelCase :  Optional[int]    =	None  # deprecated
			__UpperCamelCase :  int    =	10 << 20  # 10MB
			__UpperCamelCase :  Optional[bool]    =	None




class     _UpperCamelCase			(   datasets.ArrowBasedBuilder					):






			'''simple docstring'''

			__UpperCamelCase :  Tuple    =	JsonConfig

			def  lowerCAmelCase__		(    self	:    int ):
							if self.config.block_size is not None:
											logger.warning("""The JSON loader parameter `block_size` is deprecated. Please use `chunksize` instead""" )
											UpperCamelCase_:							List[str]		=  self.config.block_size
							if self.config.use_threads is not True:
											logger.warning(
											    """The JSON loader parameter `use_threads` is deprecated and doesn't have any effect anymore.""" )
							if self.config.newlines_in_values is not None:
											raise ValueError("""The JSON loader parameter `newlines_in_values` is no longer supported""" )
							return datasets.DatasetInfo(features=self.config.features )

			def  lowerCAmelCase__		(    self	:    Dict  ,		snake_case_	:    str ):
							if not self.config.data_files:
											raise ValueError(f'''At least one data file must be specified, but got data_files={self.config.data_files}''' )
							UpperCamelCase_:							Dict		=  dl_manager.download_and_extract(self.config.data_files )
							if isinstance(snake_case_  ,		(str, list, tuple) ):
											UpperCamelCase_:							List[Any]		=  data_files
											if isinstance(snake_case_  ,		snake_case_ ):
															UpperCamelCase_:							str		=  [files]
											UpperCamelCase_:							Any		=  [dl_manager.iter_files(snake_case_ ) for file in files]
											return [datasets.SplitGenerator(name=datasets.Split.TRAIN  ,		gen_kwargs={"""files""": files} )]
							UpperCamelCase_:							Dict		=  []
							for split_name, files in data_files.items():
											if isinstance(snake_case_  ,		snake_case_ ):
															UpperCamelCase_:							Tuple		=  [files]
											UpperCamelCase_:							Optional[int]		=  [dl_manager.iter_files(snake_case_ ) for file in files]
											splits.append(datasets.SplitGenerator(name=snake_case_  ,		gen_kwargs={"""files""": files} ) )
							return splits

			def  lowerCAmelCase__		(    self	:    str  ,		snake_case_	:    pa.Table ):
							if self.config.features is not None:
											# adding missing columns
											for column_name in set(self.config.features ) - set(pa_table.column_names ):
															UpperCamelCase_:							Union[str, Any]		=  self.config.features.arrow_schema.field(snake_case_ ).type
															UpperCamelCase_:							Tuple		=  pa_table.append_column(snake_case_  ,		pa.array([None] * len(snake_case_ )  ,		type=snake_case_ ) )
											# more expensive cast to support nested structures with keys in a different order
											# allows str <-> int/float or str to Audio for example
											UpperCamelCase_:							int		=  table_cast(snake_case_  ,		self.config.features.arrow_schema )
							return pa_table




			def  lowerCAmelCase__		(    self	:    Dict  ,		snake_case_	:    Optional[Any] ):
							for file_idx, file in enumerate(itertools.chain.from_iterable(snake_case_ ) ):
											# If the file is one json object and if we need to look at the list of items in one specific field
											if self.config.field is not None:
															with open(snake_case_  ,		encoding=self.config.encoding  ,		errors=self.config.encoding_errors ) as f:
																			UpperCamelCase_:							Dict		=  json.load(snake_case_ )

															# We keep only the field we are interested in
															UpperCamelCase_:							Optional[int]		=  dataset[self.config.field]

															# We accept two format: a list of dicts or a dict of lists
															if isinstance(snake_case_  ,		(list, tuple) ):
																			UpperCamelCase_:							Optional[int]		=  set().union(*[row.keys() for row in dataset] )
																			UpperCamelCase_:							int		=  {col: [row.get(snake_case_ ) for row in dataset] for col in keys}
															else:
																			UpperCamelCase_:							Optional[int]		=  dataset
															UpperCamelCase_:							List[str]		=  pa.Table.from_pydict(snake_case_ )
															yield file_idx, self._cast_table(snake_case_ )

											# If the file has one json object per line
											else:
															with open(snake_case_  ,		"""rb""" ) as f:
																			UpperCamelCase_:							Optional[int]		=  0
																			# Use block_size equal to the chunk size divided by 32 to leverage multithreading
																			# Set a default minimum value of 16kB if the chunk size is really small
																			UpperCamelCase_:							Optional[int]		=  max(self.config.chunksize // 32  ,		16 << 10 )
																			UpperCamelCase_:							Tuple		=  (
																			    self.config.encoding_errors if self.config.encoding_errors is not None else """strict"""
																			)
																			while True:
																							UpperCamelCase_:							int		=  f.read(self.config.chunksize )
																							if not batch:
																											break
																							# Finish current line
																							try:
																											batch += f.readline()
																							except (AttributeError, io.UnsupportedOperation):
																											batch += readline(snake_case_ )
																							# PyArrow only accepts utf-8 encoded bytes
																							if self.config.encoding != "utf-8":
																											UpperCamelCase_:							Tuple		=  batch.decode(self.config.encoding  ,		errors=snake_case_ ).encode("""utf-8""" )
																							try:
																											while True:
																															try:
																																			UpperCamelCase_:							Tuple		=  paj.read_json(
																																			    io.BytesIO(snake_case_ )  ,		read_options=paj.ReadOptions(block_size=snake_case_ ) )
																																			break
																															except (pa.ArrowInvalid, pa.ArrowNotImplementedError) as e:
																																			if (
																																			    isinstance(snake_case_  ,		pa.ArrowInvalid )
																																			    and "straddling" not in str(snake_case_ )
																																			    or block_size > len(snake_case_ )
																																			):
																																							raise
																																			else:
																																							# Increase the block size in case it was too small.
																																							# The block size will be reset for the next file.
																																							logger.debug(
																																							    f'''Batch of {len(snake_case_ )} bytes couldn\'t be parsed with block_size={block_size}. Retrying with block_size={block_size * 2}.''' )
																																							block_size *= 2
																							except pa.ArrowInvalid as e:
																											try:
																															with open(
																															    snake_case_  ,		encoding=self.config.encoding  ,		errors=self.config.encoding_errors ) as f:
																																			UpperCamelCase_:							Optional[Any]		=  json.load(snake_case_ )
																											except json.JSONDecodeError:
																															logger.error(f'''Failed to read file \'{file}\' with error {type(snake_case_ )}: {e}''' )
																															raise e
																											# If possible, parse the file as a list of json objects and exit the loop
																											if isinstance(snake_case_  ,		snake_case_ ):  # list is the only sequence type supported in JSON
																															try:
																																			UpperCamelCase_:							Any		=  set().union(*[row.keys() for row in dataset] )
																																			UpperCamelCase_:							List[str]		=  {col: [row.get(snake_case_ ) for row in dataset] for col in keys}
																																			UpperCamelCase_:							int		=  pa.Table.from_pydict(snake_case_ )
																															except (pa.ArrowInvalid, AttributeError) as e:
																																			logger.error(f'''Failed to read file \'{file}\' with error {type(snake_case_ )}: {e}''' )
																																			raise ValueError(f'''Not able to read records in the JSON file at {file}.''' ) from None
																															yield file_idx, self._cast_table(snake_case_ )
																															break
																											else:
																															logger.error(f'''Failed to read file \'{file}\' with error {type(snake_case_ )}: {e}''' )
																															raise ValueError(
																															    f'''Not able to read records in the JSON file at {file}. '''
																															    f'''You should probably indicate the field of the JSON file containing your records. '''
																															    f'''This JSON file contain the following fields: {str(list(dataset.keys() ) )}. '''
																															    f'''Select the correct one and provide it as `field=\'XXX\'` to the dataset loading method. ''' ) from None
                        # Uncomment for debugging (will print the Arrow table size and elements)
                        # logger.warning(f"pa_table: {pa_table} num rows: {pa_table.num_rows}")
                        # logger.warning('\n'.join(str(pa_table.slice(i, 1).to_pydict()) for i in range(pa_table.num_rows)))
																							yield (file_idx, batch_idx), self._cast_table(snake_case_ )
																							batch_idx += 1






