
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


_A   =     None
_A   =     "<" 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
_A   =     [
    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 	_lowerCAmelCase					:
				_lowercase					=True
				_lowercase					=None
				# Automatically constructed
				_lowercase					="PIL.Image.Image"
				_lowercase					=pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}       )
				_lowercase					=field(default='''Image'''					,      init=__a					,      repr=__a       )




				def __call__(					self     )				->      Any:
							return self.pa_type




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

							if isinstance(_UpperCamelCase				,				_UpperCamelCase     ):
										lowerCAmelCase_						=	np.array(_UpperCamelCase     )

							if isinstance(_UpperCamelCase				,				_UpperCamelCase     ):
										return {"path": value, "bytes": None}
							elif isinstance(_UpperCamelCase				,				_UpperCamelCase     ):
										return {"path": None, "bytes": value}
							elif isinstance(_UpperCamelCase				,				np.ndarray     ):
										# convert the image array to PNG/TIFF bytes
										return encode_np_array(_UpperCamelCase     )
							elif isinstance(_UpperCamelCase				,				PIL.Image.Image     ):
										# convert the PIL image to bytes (default format is PNG/TIFF)
										return encode_pil_image(_UpperCamelCase     )
							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      __a						(					self				,				_UpperCamelCase				,				_UpperCamelCase=None     )				->      "PIL.Image.Image":
							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:
										lowerCAmelCase_						=	{}

							lowerCAmelCase_	,  lowerCAmelCase_						=	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(_UpperCamelCase     ):
																lowerCAmelCase_						=	PIL.Image.open(_UpperCamelCase     )
													else:
																lowerCAmelCase_						=	path.split("::"     )[-1]
																try:
																			lowerCAmelCase_						=	string_to_dict(_UpperCamelCase				,				config.HUB_DATASETS_URL     )["repo_id"]
																			lowerCAmelCase_						=	token_per_repo_id.get(_UpperCamelCase     )
																except ValueError:
																			lowerCAmelCase_						=	None
																with xopen(_UpperCamelCase				,				"rb"				,				use_auth_token=_UpperCamelCase     ) as f:
																			lowerCAmelCase_						=	BytesIO(f.read()     )
																lowerCAmelCase_						=	PIL.Image.open(bytes_     )
							else:
										lowerCAmelCase_						=	PIL.Image.open(BytesIO(bytes_     )     )
							image.load()  # to avoid "Too many open files" errors
							return image




				def      __a						(					self     )				->      Union["FeatureType", Dict[str, "FeatureType"]]:
							from .features import Value

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




				def      __a						(					self				,				_UpperCamelCase     )				->      pa.StructArray:
							if pa.types.is_string(storage.type     ):
										lowerCAmelCase_						=	pa.array([None] * len(_UpperCamelCase     )				,				type=pa.binary()     )
										lowerCAmelCase_						=	pa.StructArray.from_arrays([bytes_array, storage]				,				["bytes", "path"]				,				mask=storage.is_null()     )
							elif pa.types.is_binary(storage.type     ):
										lowerCAmelCase_						=	pa.array([None] * len(_UpperCamelCase     )				,				type=pa.string()     )
										lowerCAmelCase_						=	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:
													lowerCAmelCase_						=	storage.field("bytes"     )
										else:
													lowerCAmelCase_						=	pa.array([None] * len(_UpperCamelCase     )				,				type=pa.binary()     )
										if storage.type.get_field_index("path"     ) >= 0:
													lowerCAmelCase_						=	storage.field("path"     )
										else:
													lowerCAmelCase_						=	pa.array([None] * len(_UpperCamelCase     )				,				type=pa.string()     )
										lowerCAmelCase_						=	pa.StructArray.from_arrays([bytes_array, path_array]				,				["bytes", "path"]				,				mask=storage.is_null()     )
							elif pa.types.is_list(storage.type     ):
										lowerCAmelCase_						=	pa.array(
										    [encode_np_array(np.array(_UpperCamelCase     )     )["bytes"] if arr is not None else None for arr in storage.to_pylist()]				,				type=pa.binary()				,				)
										lowerCAmelCase_						=	pa.array([None] * len(_UpperCamelCase     )				,				type=pa.string()     )
										lowerCAmelCase_						=	pa.StructArray.from_arrays(
										    [bytes_array, path_array]				,				["bytes", "path"]				,				mask=bytes_array.is_null()     )
							return array_cast(_UpperCamelCase				,				self.pa_type     )




				def      __a						(					self				,				_UpperCamelCase     )				->      pa.StructArray:

							@no_op_if_value_is_null
							def path_to_bytes(_UpperCamelCase     ):
										with xopen(_UpperCamelCase				,				"rb"     ) as f:
													lowerCAmelCase_						=	f.read()
										return bytes_

							lowerCAmelCase_						=	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()				,				)
							lowerCAmelCase_						=	pa.array(
							    [os.path.basename(_UpperCamelCase     ) if path is not None else None for path in storage.field("path"     ).to_pylist()]				,				type=pa.string()				,				)
							lowerCAmelCase_						=	pa.StructArray.from_arrays([bytes_array, path_array]				,				["bytes", "path"]				,				mask=bytes_array.is_null()     )
							return array_cast(_UpperCamelCase				,				self.pa_type     )






def    lowerCamelCase__				(				):




			"""simple docstring"""


			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()
						lowerCAmelCase_						=	list(set(PIL.Image.OPEN.keys()				) & set(PIL.Image.SAVE.keys()				)				)
			return _IMAGE_COMPRESSION_FORMATS






def    lowerCamelCase__				(				__lowerCAmelCase  :  "PIL.Image.Image"				):




			"""simple docstring"""


			lowerCAmelCase_						=	BytesIO()
			if image.format in list_image_compression_formats():
						lowerCAmelCase_						=	image.format
			else:
						lowerCAmelCase_						=	"PNG" if image.mode in ["1", "L", "LA", "RGB", "RGBA"] else "TIFF"
			image.save(__lowerCAmelCase    ,							format=__lowerCAmelCase				)
			return buffer.getvalue()






def    lowerCamelCase__				(				__lowerCAmelCase  :  "PIL.Image.Image"				):




			"""simple docstring"""


			if hasattr(__lowerCAmelCase    ,							"filename"				) and image.filename != "":
						return {"path": image.filename, "bytes": None}
			else:
						return {"path": None, "bytes": image_to_bytes(__lowerCAmelCase				)}






def    lowerCamelCase__				(				__lowerCAmelCase  :  np.ndarray				):




			"""simple docstring"""


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

			lowerCAmelCase_						=	array.dtype
			lowerCAmelCase_						=	dtype.byteorder if dtype.byteorder != "=" else _NATIVE_BYTEORDER
			lowerCAmelCase_						=	dtype.kind
			lowerCAmelCase_						=	dtype.itemsize

			lowerCAmelCase_						=	None

			# Multi-channel array case (only np.dtype("|u1") is allowed)
			if array.shape[2:]:
						lowerCAmelCase_						=	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:
						lowerCAmelCase_						=	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:
									lowerCAmelCase_						=	dtype_byteorder + dtype_kind + str(__lowerCAmelCase				)
									lowerCAmelCase_						=	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}"""				)
			lowerCAmelCase_						=	PIL.Image.fromarray(array.astype(__lowerCAmelCase				)				)
			return {"path": None, "bytes": image_to_bytes(__lowerCAmelCase				)}






def    lowerCamelCase__				(				__lowerCAmelCase  :  Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]]				):




			"""simple docstring"""


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

			if objs:
						lowerCAmelCase_	,  lowerCAmelCase_						=	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				):
									lowerCAmelCase_						=	no_op_if_value_is_null(__lowerCAmelCase				)
									return [obj_to_image_dict_func(__lowerCAmelCase				) for obj in objs]
						elif isinstance(__lowerCAmelCase    ,							PIL.Image.Image				):
									lowerCAmelCase_						=	no_op_if_value_is_null(__lowerCAmelCase				)
									return [obj_to_image_dict_func(__lowerCAmelCase				) for obj in objs]
						else:
									return objs
			else:
						return objs





from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


_A   =     logging.get_logger(__name__)


_A   =     TypeVar("DatasetType", Dataset, IterableDataset)






def    lowerCamelCase__				(				__lowerCAmelCase  :  List[DatasetType]    ,							__lowerCAmelCase  :  Optional[List[float]] = None    ,							__lowerCAmelCase  :  Optional[int] = None    ,							__lowerCAmelCase  :  Optional[DatasetInfo] = None    ,							__lowerCAmelCase  :  Optional[NamedSplit] = None    ,							__lowerCAmelCase  :  Literal["first_exhausted", "all_exhausted"] = "first_exhausted"    ,							):




			"""simple docstring"""


			from .arrow_dataset import Dataset
			from .iterable_dataset import IterableDataset

			if not datasets:
						raise ValueError("Unable to interleave an empty list of datasets."				)
			for i, dataset in enumerate(__lowerCAmelCase				):
						if not isinstance(__lowerCAmelCase    ,							(Dataset, IterableDataset)				):
									if isinstance(__lowerCAmelCase    ,							(DatasetDict, IterableDatasetDict)				):
												if not dataset:
															raise ValueError(
															    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
															    "is an empty dataset dictionary."				)
												raise ValueError(
												    F"""Dataset at position {i} has at least one split: {list(__lowerCAmelCase				)}\n"""
												    F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__lowerCAmelCase				)				)}']"""				)
									raise ValueError(
									    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__lowerCAmelCase				).__name__}."""				)
						if i == 0:
									lowerCAmelCase_	,  lowerCAmelCase_						=	(
									    (Dataset, IterableDataset) if isinstance(__lowerCAmelCase    ,							__lowerCAmelCase				) else (IterableDataset, Dataset)
									)
						elif not isinstance(__lowerCAmelCase    ,							__lowerCAmelCase				):
									raise ValueError(
									    F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""				)
			if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
						raise ValueError(F"""{stopping_strategy} is not supported. Please enter a valid stopping_strategy."""				)
			if dataset_type is Dataset:
						return _interleave_map_style_datasets(
						    __lowerCAmelCase    ,							__lowerCAmelCase    ,							__lowerCAmelCase    ,							info=__lowerCAmelCase    ,							split=__lowerCAmelCase    ,							stopping_strategy=__lowerCAmelCase				)
			else:
						return _interleave_iterable_datasets(
						    __lowerCAmelCase    ,							__lowerCAmelCase    ,							__lowerCAmelCase    ,							info=__lowerCAmelCase    ,							split=__lowerCAmelCase    ,							stopping_strategy=__lowerCAmelCase				)






def    lowerCamelCase__				(				__lowerCAmelCase  :  List[DatasetType]    ,							__lowerCAmelCase  :  Optional[DatasetInfo] = None    ,							__lowerCAmelCase  :  Optional[NamedSplit] = None    ,							__lowerCAmelCase  :  int = 0    ,							):




			"""simple docstring"""


			if not dsets:
						raise ValueError("Unable to concatenate an empty list of datasets."				)
			for i, dataset in enumerate(__lowerCAmelCase				):
						if not isinstance(__lowerCAmelCase    ,							(Dataset, IterableDataset)				):
									if isinstance(__lowerCAmelCase    ,							(DatasetDict, IterableDatasetDict)				):
												if not dataset:
															raise ValueError(
															    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
															    "is an empty dataset dictionary."				)
												raise ValueError(
												    F"""Dataset at position {i} has at least one split: {list(__lowerCAmelCase				)}\n"""
												    F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__lowerCAmelCase				)				)}']"""				)
									raise ValueError(
									    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__lowerCAmelCase				).__name__}."""				)
						if i == 0:
									lowerCAmelCase_	,  lowerCAmelCase_						=	(
									    (Dataset, IterableDataset) if isinstance(__lowerCAmelCase    ,							__lowerCAmelCase				) else (IterableDataset, Dataset)
									)
						elif not isinstance(__lowerCAmelCase    ,							__lowerCAmelCase				):
									raise ValueError(
									    F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""				)
			if dataset_type is Dataset:
						return _concatenate_map_style_datasets(__lowerCAmelCase    ,							info=__lowerCAmelCase    ,							split=__lowerCAmelCase    ,							axis=__lowerCAmelCase				)
			else:
						return _concatenate_iterable_datasets(__lowerCAmelCase    ,							info=__lowerCAmelCase    ,							split=__lowerCAmelCase    ,							axis=__lowerCAmelCase				)



