




"""simple docstring"""





# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
			import torch



class      a__				(	TensorFormatter[Mapping, "torch.Tensor", Mapping]   ):


			def __init__(      self      :    Tuple      ,  UpperCamelCase_      :    Optional[int]=None      ,  **UpperCamelCase_      :    List[str]):


						"""simple docstring"""




						super().__init__(features=UpperCamelCase_)
						__UpperCAmelCase	: Tuple		=					torch_tensor_kwargs
						import torch  # noqa import torch at initialization


			def       a_     (      self      :    int      ,  UpperCamelCase_      :    Optional[int]):


						"""simple docstring"""




						import torch

						if isinstance(UpperCamelCase_      ,  UpperCamelCase_) and column:
									if all(
									    isinstance(UpperCamelCase_      ,  torch.Tensor) and x.shape == column[0].shape and x.dtype == column[0].dtype
									    for x in column):
												return torch.stack(UpperCamelCase_)
						return column


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    Union[str, Any]):


						"""simple docstring"""




						import torch

						if isinstance(UpperCamelCase_      ,  (str, bytes, type(UpperCamelCase_))):
									return value
						elif isinstance(UpperCamelCase_      ,  (np.character, np.ndarray)) and np.issubdtype(value.dtype      ,  np.character):
									return value.tolist()

						__UpperCAmelCase	: Union[str, Any]		=					{}

						if isinstance(UpperCamelCase_      ,  (np.number, np.ndarray)) and np.issubdtype(value.dtype      ,  np.integer):
									__UpperCAmelCase	: List[str]		=					{"dtype": torch.intaa}
						elif isinstance(UpperCamelCase_      ,  (np.number, np.ndarray)) and np.issubdtype(value.dtype      ,  np.floating):
									__UpperCAmelCase	: Optional[Any]		=					{"dtype": torch.floataa}
						elif config.PIL_AVAILABLE and "PIL" in sys.modules:
									import PIL.Image

									if isinstance(UpperCamelCase_      ,  PIL.Image.Image):
												__UpperCAmelCase	: Tuple		=					np.asarray(UpperCamelCase_)
						return torch.tensor(UpperCamelCase_      ,  **{**default_dtype, **self.torch_tensor_kwargs})


			def       a_     (      self      :    List[Any]      ,  UpperCamelCase_      :    List[str]):


						"""simple docstring"""




						import torch

						# support for torch, tf, jax etc.
						if hasattr(UpperCamelCase_      ,  "__array__") and not isinstance(UpperCamelCase_      ,  torch.Tensor):
									__UpperCAmelCase	: List[str]		=					data_struct.__array__()
						# support for nested types like struct of list of struct
						if isinstance(UpperCamelCase_      ,  np.ndarray):
									if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
												return self._consolidate([self.recursive_tensorize(UpperCamelCase_) for substruct in data_struct])
						elif isinstance(UpperCamelCase_      ,  (list, tuple)):
									return self._consolidate([self.recursive_tensorize(UpperCamelCase_) for substruct in data_struct])
						return self._tensorize(UpperCamelCase_)


			def       a_     (      self      :    Tuple      ,  UpperCamelCase_      :    dict):


						"""simple docstring"""




						return map_nested(self._recursive_tensorize      ,  UpperCamelCase_      ,  map_list=UpperCamelCase_)


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    pa.Table):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					self.numpy_arrow_extractor().extract_row(UpperCamelCase_)
						__UpperCAmelCase	: Union[str, Any]		=					self.python_features_decoder.decode_row(UpperCamelCase_)
						return self.recursive_tensorize(UpperCamelCase_)


			def       a_     (      self      :    Optional[int]      ,  UpperCamelCase_      :    pa.Table):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					self.numpy_arrow_extractor().extract_column(UpperCamelCase_)
						__UpperCAmelCase	: str		=					self.python_features_decoder.decode_column(UpperCamelCase_      ,  pa_table.column_names[0])
						__UpperCAmelCase	: Optional[int]		=					self.recursive_tensorize(UpperCamelCase_)
						__UpperCAmelCase	: Tuple		=					self._consolidate(UpperCamelCase_)
						return column


			def       a_     (      self      :    Dict      ,  UpperCamelCase_      :    pa.Table):


						"""simple docstring"""




						__UpperCAmelCase	: Tuple		=					self.numpy_arrow_extractor().extract_batch(UpperCamelCase_)
						__UpperCAmelCase	: List[Any]		=					self.python_features_decoder.decode_batch(UpperCamelCase_)
						__UpperCAmelCase	: Any		=					self.recursive_tensorize(UpperCamelCase_)
						for column_name in batch:
									__UpperCAmelCase	: List[Any]		=					self._consolidate(batch[column_name])
						return batch








"""simple docstring"""





from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader



class      a__				(	__magic_name__   ):


			def __init__(      self      :    List[Any]      ,  UpperCamelCase_      :    pyspark.sql.DataFrame      ,  UpperCamelCase_      :    Optional[NamedSplit] = None      ,  UpperCamelCase_      :    Optional[Features] = None      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    str = None      ,  UpperCamelCase_      :    bool = False      ,  UpperCamelCase_      :    str = None      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    str = "arrow"      ,  **UpperCamelCase_      :    Tuple      ,  ):


						"""simple docstring"""




						super().__init__(
						    split=UpperCamelCase_      ,  features=UpperCamelCase_      ,  cache_dir=UpperCamelCase_      ,  keep_in_memory=UpperCamelCase_      ,  streaming=UpperCamelCase_      ,  **UpperCamelCase_      ,  )
						__UpperCAmelCase	: List[Any]		=					load_from_cache_file
						__UpperCAmelCase	: List[Any]		=					file_format
						__UpperCAmelCase	: str		=					Spark(
						    df=UpperCamelCase_      ,  features=UpperCamelCase_      ,  cache_dir=UpperCamelCase_      ,  working_dir=UpperCamelCase_      ,  **UpperCamelCase_      ,  )


			def       a_     (      self      :    Union[str, Any]):


						"""simple docstring"""




						if self.streaming:
									return self.builder.as_streaming_dataset(split=self.split)
						__UpperCAmelCase	: List[Any]		=					None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
						self.builder.download_and_prepare(
						    download_mode=UpperCamelCase_      ,  file_format=self._file_format      ,  )
						return self.builder.as_dataset(split=self.split)


