
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


UpperCamelCase_						:     Optional[List[str]]	       =							None
UpperCamelCase_						:     int	       =							"""<""" 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
UpperCamelCase_						:     Union[str, Any]	       =							[
    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   __lowercase						:
	_A       =   True
	_A       =   None
	# Automatically constructed
	_A       =   "PIL.Image.Image"
	_A       =   pa.struct({"bytes": pa.binary(), "path": pa.string()}       )
	_A       =   field(default="Image"   , init=__snake_case   , repr=__snake_case       )






	def __call__(self				: Optional[Any]    ) ->			Optional[int]:
							return self.pa_type






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

							if isinstance(snake_case      ,	snake_case    ):
													_lowercase       :	Dict           =   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       _a(self				: str      ,	snake_case				: dict      ,	snake_case				: Optional[Any]=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:
													_lowercase       :	Optional[int]           =   {}

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






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

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






	def       _a(self				: int      ,	snake_case				: Union[pa.StringArray, pa.StructArray, pa.ListArray]    ) ->			pa.StructArray:
							if pa.types.is_string(storage.type    ):
													_lowercase       :	Tuple           =   pa.array([None] * len(snake_case    )      ,	type=pa.binary()    )
													_lowercase       :	List[str]           =   pa.StructArray.from_arrays([bytes_array, storage]      ,	["bytes", "path"]      ,	mask=storage.is_null()    )
							elif pa.types.is_binary(storage.type    ):
													_lowercase       :	Optional[int]           =   pa.array([None] * len(snake_case    )      ,	type=pa.string()    )
													_lowercase       :	Optional[int]           =   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:
																			_lowercase       :	Any           =   storage.field("bytes"    )
													else:
																			_lowercase       :	Tuple           =   pa.array([None] * len(snake_case    )      ,	type=pa.binary()    )
													if storage.type.get_field_index("path"    ) >= 0:
																			_lowercase       :	int           =   storage.field("path"    )
													else:
																			_lowercase       :	Optional[int]           =   pa.array([None] * len(snake_case    )      ,	type=pa.string()    )
													_lowercase       :	str           =   pa.StructArray.from_arrays([bytes_array, path_array]      ,	["bytes", "path"]      ,	mask=storage.is_null()    )
							elif pa.types.is_list(storage.type    ):
													_lowercase       :	List[str]           =   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()      ,	)
													_lowercase       :	int           =   pa.array([None] * len(snake_case    )      ,	type=pa.string()    )
													_lowercase       :	Tuple           =   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(self				: Dict      ,	snake_case				: pa.StructArray    ) ->			pa.StructArray:

							@no_op_if_value_is_null
							def path_to_bytes(snake_case				: List[str]    ):
													with xopen(snake_case      ,	"rb"    ) as f:
																			_lowercase       :	Union[str, Any]           =   f.read()
													return bytes_

							_lowercase       :	List[str]           =   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()      ,	)
							_lowercase       :	Tuple           =   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()      ,	)
							_lowercase       :	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 		UpperCamelCase     (       )   ->			List[str]:






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



def 		UpperCamelCase     (       _UpperCAmelCase  :   "PIL.Image.Image"  )   ->			bytes:






						'''simple docstring'''

						_lowercase       :	Any           =   BytesIO()
						if image.format in list_image_compression_formats():
												_lowercase       :	Optional[int]           =   image.format
						else:
												_lowercase       :	str           =   "PNG" if image.mode in ["1", "L", "LA", "RGB", "RGBA"] else "TIFF"
						image.save(_UpperCAmelCase ,			format=_UpperCAmelCase  )
						return buffer.getvalue()



def 		UpperCamelCase     (       _UpperCAmelCase  :   "PIL.Image.Image"  )   ->			dict:






						'''simple docstring'''

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



def 		UpperCamelCase     (       _UpperCAmelCase  :   np.ndarray  )   ->			dict:






						'''simple docstring'''

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

						_lowercase       :	List[str]           =   array.dtype
						_lowercase       :	int           =   dtype.byteorder if dtype.byteorder != "=" else _NATIVE_BYTEORDER
						_lowercase       :	Union[str, Any]           =   dtype.kind
						_lowercase       :	Any           =   dtype.itemsize

						_lowercase       :	List[Any]           =   None

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



def 		UpperCamelCase     (       _UpperCAmelCase  :   Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]]  )   ->			List[dict]:






						'''simple docstring'''

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

						if objs:
												_lowercase      ,    _lowercase       :	str           =   first_non_null_value(_UpperCAmelCase  )
												if isinstance(_UpperCAmelCase ,			_UpperCAmelCase  ):
																		return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
												if isinstance(_UpperCAmelCase ,			np.ndarray  ):
																		_lowercase       :	str           =   no_op_if_value_is_null(_UpperCAmelCase  )
																		return [obj_to_image_dict_func(_UpperCAmelCase  ) for obj in objs]
												elif isinstance(_UpperCAmelCase ,			PIL.Image.Image  ):
																		_lowercase       :	List[Any]           =   no_op_if_value_is_null(_UpperCAmelCase  )
																		return [obj_to_image_dict_func(_UpperCAmelCase  ) for obj in objs]
												else:
																		return objs
						else:
												return objs






import numpy as np
import torch
from torch.utils.data import Dataset, IterableDataset

from ..utils.generic import ModelOutput




class   __lowercase						(						__snake_case       ):






	def __init__(self				: List[Any]      ,	snake_case				: Optional[int]      ,	snake_case				: str      ,	snake_case				: str    ) ->			Dict:
							_lowercase       :	Tuple           =   dataset
							_lowercase       :	List[str]           =   process
							_lowercase       :	Any           =   params






	def __len__(self				: Optional[Any]    ) ->			Any:
							return len(self.dataset    )






	def __getitem__(self				: int      ,	snake_case				: Any    ) ->			int:
							_lowercase       :	Optional[Any]           =   self.dataset[i]
							_lowercase       :	Union[str, Any]           =   self.process(snake_case      ,	**self.params    )
							return processed




class   __lowercase						(						__snake_case       ):






	def __init__(self				: int      ,	snake_case				: List[Any]      ,	snake_case				: Tuple      ,	snake_case				: Tuple      ,	snake_case				: List[str]=None    ) ->			Optional[int]:
							_lowercase       :	List[str]           =   loader
							_lowercase       :	Optional[Any]           =   infer
							_lowercase       :	List[Any]           =   params
							if loader_batch_size == 1:
													# Let's spare some time by deactivating altogether
													_lowercase       :	str           =   None
							_lowercase       :	Optional[int]           =   loader_batch_size

							# Internal bookkeeping
							_lowercase       :	Dict           =   None
							_lowercase       :	Union[str, Any]           =   None






	def __len__(self				: Any    ) ->			Any:
							return len(self.loader    )






	def __iter__(self				: int    ) ->			Optional[Any]:
							_lowercase       :	List[str]           =   iter(self.loader    )
							return self






	def       _a(self				: int    ) ->			List[str]:
							if isinstance(self._loader_batch_data      ,	torch.Tensor    ):
													# Batch data is simple tensor, just fetch the slice
													_lowercase       :	Any           =   self._loader_batch_data[self._loader_batch_index]
							else:
													# Batch data is assumed to be BaseModelOutput (or dict)
													_lowercase       :	List[str]           =   {}
													for k, element in self._loader_batch_data.items():
																			if isinstance(snake_case      ,	snake_case    ):
																									# Convert ModelOutput to tuple first
																									_lowercase       :	Dict           =   element.to_tuple()
																									if isinstance(element[0]      ,	torch.Tensor    ):
																															_lowercase       :	Optional[int]           =   tuple(el[self._loader_batch_index].unsqueeze(0    ) for el in element    )
																									elif isinstance(element[0]      ,	np.ndarray    ):
																															_lowercase       :	Optional[Any]           =   tuple(np.expand_dims(el[self._loader_batch_index]      ,	0    ) for el in element    )
																									continue
																			if k in {"hidden_states", "past_key_values", "attentions"} and isinstance(snake_case      ,	snake_case    ):
																									# Those are stored as lists of tensors so need specific unbatching.
																									if isinstance(element[0]      ,	torch.Tensor    ):
																															_lowercase       :	Optional[int]           =   tuple(el[self._loader_batch_index].unsqueeze(0    ) for el in element    )
																									elif isinstance(element[0]      ,	np.ndarray    ):
																															_lowercase       :	List[str]           =   tuple(np.expand_dims(el[self._loader_batch_index]      ,	0    ) for el in element    )
																									continue
																			if element is None:
																									# This can happen for optional data that get passed around
																									_lowercase       :	Optional[Any]           =   None
																			elif isinstance(element[self._loader_batch_index]      ,	torch.Tensor    ):
																									# Take correct batch data, but make it looked like batch_size=1
																									# For compatibility with other methods within transformers

																									_lowercase       :	int           =   element[self._loader_batch_index].unsqueeze(0    )
																			elif isinstance(element[self._loader_batch_index]      ,	np.ndarray    ):
																									# Take correct batch data, but make it looked like batch_size=1
																									# For compatibility with other methods within transformers
																									_lowercase       :	Tuple           =   np.expand_dims(element[self._loader_batch_index]      ,	0    )
																			else:
																									# This is typically a list, so no need to `unsqueeze`.
																									_lowercase       :	Union[str, Any]           =   element[self._loader_batch_index]
            # Recreate the element by reusing the original class to make it look
            # batch_size=1
													_lowercase       :	int           =   self._loader_batch_data.__class__(snake_case    )
							self._loader_batch_index += 1
							return result






	def       _a(self				: Any    ) ->			str:
							if self._loader_batch_index is not None and self._loader_batch_index < self.loader_batch_size:
													# We are currently unrolling a batch so we just need to return
													# the current item within a batch
													return self.loader_batch_item()

							# We're out of items within a batch
							_lowercase       :	str           =   next(self.iterator    )
							_lowercase       :	Any           =   self.infer(snake_case      ,	**self.params    )
							# We now have a batch of "inferred things".
							if self.loader_batch_size is not None:
													# Try to infer the size of the batch
													if isinstance(snake_case      ,	torch.Tensor    ):
																			_lowercase       :	int           =   processed
													else:
																			_lowercase       :	List[str]           =   list(processed.keys()    )[0]
																			_lowercase       :	List[str]           =   processed[key]
													if isinstance(snake_case      ,	snake_case    ):
																			_lowercase       :	int           =   len(snake_case    )
													else:
																			_lowercase       :	Optional[int]           =   first_tensor.shape[0]
													if 0 < observed_batch_size < self.loader_batch_size:
																			# could be last batch so we can't unroll as many
																			# elements.
																			_lowercase       :	Union[str, Any]           =   observed_batch_size
													# Setting internal index to unwrap the batch
													_lowercase       :	Any           =   processed
													_lowercase       :	Dict           =   0
													return self.loader_batch_item()
							else:
													# We're not unrolling batches
													return processed




class   __lowercase						(						__snake_case       ):






	def __init__(self				: Union[str, Any]      ,	snake_case				: str      ,	snake_case				: Dict      ,	snake_case				: Optional[Any]      ,	snake_case				: List[str]=None    ) ->			List[str]:
							super().__init__(snake_case      ,	snake_case      ,	snake_case    )






	def __iter__(self				: List[Any]    ) ->			str:
							_lowercase       :	int           =   iter(self.loader    )
							_lowercase       :	Optional[int]           =   None
							return self






	def       _a(self				: Optional[Any]    ) ->			Optional[int]:
							if self.subiterator is None:
													_lowercase       :	List[str]           =   self.infer(next(self.iterator    )      ,	**self.params    )
							try:
													# Try to return next item
													_lowercase       :	str           =   next(self.subiterator    )
							except StopIteration:
													# When a preprocess iterator ends, we can start lookig at the next item
													# ChunkIterator will keep feeding until ALL elements of iterator
													# all have created their subiterator and have been iterating against.
													#
													# Another way to look at it, is we're basically flattening lists of lists
													# into a single list, but with generators
													_lowercase       :	Any           =   self.infer(next(self.iterator    )      ,	**self.params    )
													_lowercase       :	Optional[int]           =   next(self.subiterator    )
							return processed




class   __lowercase						(						__snake_case       ):






	def __iter__(self				: List[str]    ) ->			List[str]:
							_lowercase       :	Any           =   iter(self.loader    )
							return self






	def       _a(self				: List[Any]    ) ->			Union[str, Any]:
							# Extremely similar to PipelineIterator in its unpacking mechanism
							# BUT, we have an extra required item which is the presence of `is_last`
							# That is because everything is flattened by `PipelineChunkIterator` we
							# need to keep track of how to regroup here in the original `process`
							# boundaries so that `process` and `postprocess` see the same data.

							# This iterator accumulates items (possibly while unbatching) until it
							# its a `is_last` and then just passes it on to the caller.
							_lowercase       :	Dict           =   False
							_lowercase       :	Optional[int]           =   []
							if self._loader_batch_index is not None and self._loader_batch_index < self.loader_batch_size:
													while self._loader_batch_index < self.loader_batch_size:
																			_lowercase       :	Tuple           =   self.loader_batch_item()
																			_lowercase       :	List[Any]           =   item.pop("is_last"    )
																			accumulator.append(snake_case    )
																			if is_last:
																									return accumulator

							while not is_last:
													_lowercase       :	List[Any]           =   self.infer(next(self.iterator    )      ,	**self.params    )
													if self.loader_batch_size is not None:
																			if isinstance(snake_case      ,	torch.Tensor    ):
																									_lowercase       :	Optional[Any]           =   processed
																			else:
																									_lowercase       :	Tuple           =   list(processed.keys()    )[0]
																									_lowercase       :	int           =   processed[key]
																			if isinstance(snake_case      ,	snake_case    ):
																									_lowercase       :	str           =   len(snake_case    )
																			else:
																									_lowercase       :	List[str]           =   first_tensor.shape[0]
																			if 0 < observed_batch_size < self.loader_batch_size:
																									# could be last batch so we can't unroll as many
																									# elements.
																									_lowercase       :	Tuple           =   observed_batch_size
																			_lowercase       :	Any           =   processed
																			_lowercase       :	Dict           =   0
																			while self._loader_batch_index < self.loader_batch_size:
																									_lowercase       :	str           =   self.loader_batch_item()
																									_lowercase       :	int           =   item.pop("is_last"    )
																									accumulator.append(snake_case    )
																									if is_last:
																															return accumulator
													else:
																			_lowercase       :	str           =   processed
																			_lowercase       :	int           =   item.pop("is_last"    )
																			accumulator.append(snake_case    )
							return accumulator




class   __lowercase						(						__snake_case       ):






	def __init__(self				: int      ,	snake_case				: Dataset      ,	snake_case				: str    ) ->			List[Any]:
							_lowercase       :	Optional[Any]           =   dataset
							_lowercase       :	Any           =   key






	def __len__(self				: Any    ) ->			Union[str, Any]:
							return len(self.dataset    )






	def __getitem__(self				: int      ,	snake_case				: Any    ) ->			Any:
							return self.dataset[i][self.key]




class   __lowercase						(						__snake_case       ):






	def __init__(self				: int      ,	snake_case				: Dataset      ,	snake_case				: str      ,	snake_case				: str    ) ->			Dict:
							_lowercase       :	int           =   dataset
							_lowercase       :	Optional[Any]           =   keya
							_lowercase       :	Tuple           =   keya






	def __len__(self				: List[str]    ) ->			Union[str, Any]:
							return len(self.dataset    )






	def __getitem__(self				: Optional[Any]      ,	snake_case				: Dict    ) ->			int:
							return {"text": self.dataset[i][self.keya], "text_pair": self.dataset[i][self.keya]}




