



"""simple docstring"""






import collections.abc
from typing import Optional, Tuple, Union

import torch
import torch.utils.checkpoint
from torch import nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import BaseModelOutputWithNoAttention, ImageClassifierOutputWithNoAttention
from ...modeling_utils import PreTrainedModel
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_poolformer import PoolFormerConfig


__snake_case							:     Dict   =					logging.get_logger(__name__)

# General docstring
__snake_case							:     int   =					'PoolFormerConfig'

# Base docstring
__snake_case							:     int   =					'sail/poolformer_s12'
__snake_case							:     Any   =					[1, 512, 7, 7]

# Image classification docstring
__snake_case							:     Any   =					'sail/poolformer_s12'
__snake_case							:     List[str]   =					'tabby, tabby cat'

__snake_case							:     Optional[Any]   =					[
    'sail/poolformer_s12',
    # See all PoolFormer models at https://huggingface.co/models?filter=poolformer
]


def     _lowercase       (						__snake_case      ,__snake_case = 0.0      ,__snake_case = False			)		->     Any:
			if drop_prob == 0.0 or not training:
						return input
			__lowerCAmelCase   :						List[str]				= 1 - drop_prob
			__lowerCAmelCase   :						Optional[int]				= (input.shape[0],) + (1,) * (input.ndim - 1)  # work with diff dim tensors, not just 2D ConvNets
			__lowerCAmelCase   :						str				= keep_prob + torch.rand(snake_case_      ,dtype=input.dtype      ,device=input.device			)
			random_tensor.floor_()  # binarize
			__lowerCAmelCase   :						List[str]				= input.div(snake_case_			) * random_tensor
			return output



class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	List[Any] ,    _SCREAMING_SNAKE_CASE:	Dict = None)       -> None:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						List[str]				= drop_prob

	def 							_SCREAMING_SNAKE_CASE     (  self:	Dict ,    _SCREAMING_SNAKE_CASE:	str)       -> torch.Tensor:

				"""simple docstring"""

				return drop_path(_a ,    self.drop_prob ,    self.training)







	def 							_SCREAMING_SNAKE_CASE     (  self:	str)       -> str:

				"""simple docstring"""

				return "p={}".format(self.drop_prob)






class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	Optional[int] ,    _SCREAMING_SNAKE_CASE:	List[str] ,    _SCREAMING_SNAKE_CASE:	List[str] ,    _SCREAMING_SNAKE_CASE:	int ,    _SCREAMING_SNAKE_CASE:	int ,    _SCREAMING_SNAKE_CASE:	Union[str, Any] ,    _SCREAMING_SNAKE_CASE:	str=None)       -> List[str]:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						int				= patch_size if isinstance(_a ,    collections.abc.Iterable) else (patch_size, patch_size)
				__lowerCAmelCase   :						Optional[Any]				= stride if isinstance(_a ,    collections.abc.Iterable) else (stride, stride)
				__lowerCAmelCase   :						Dict				= padding if isinstance(_a ,    collections.abc.Iterable) else (padding, padding)

				__lowerCAmelCase   :						Optional[Any]				= nn.Convad(_a ,    _a ,    kernel_size=_a ,    stride=_a ,    padding=_a)
				__lowerCAmelCase   :						str				= norm_layer(_a) if norm_layer else nn.Identity()







	def 							_SCREAMING_SNAKE_CASE     (  self:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	Optional[Any])       -> List[Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						Any				= self.projection(_a)
				__lowerCAmelCase   :						List[Any]				= self.norm(_a)
				return embeddings






class     A__     (						nn.GroupNorm							):




	'''simple docstring'''





	def __init__(  self:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	Any ,    **_SCREAMING_SNAKE_CASE:	Optional[Any])       -> List[Any]:

				"""simple docstring"""

				super().__init__(1 ,    _a ,    **_a)






class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	Dict ,    _SCREAMING_SNAKE_CASE:	int)       -> int:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						Union[str, Any]				= nn.AvgPoolad(_a ,    stride=1 ,    padding=pool_size // 2 ,    count_include_pad=_a)







	def 							_SCREAMING_SNAKE_CASE     (  self:	str ,    _SCREAMING_SNAKE_CASE:	Optional[Any])       -> Tuple:

				"""simple docstring"""

				return self.pool(_a) - hidden_states






class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	Dict ,    _SCREAMING_SNAKE_CASE:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	Any ,    _SCREAMING_SNAKE_CASE:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	Optional[Any])       -> int:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						Tuple				= nn.Convad(_a ,    _a ,    1)
				__lowerCAmelCase   :						Tuple				= nn.Convad(_a ,    _a ,    1)
				__lowerCAmelCase   :						Any				= PoolFormerDropPath(_a)
				if isinstance(config.hidden_act ,    _a):
							__lowerCAmelCase   :						Optional[int]				= ACTaFN[config.hidden_act]
				else:
							__lowerCAmelCase   :						Optional[Any]				= config.hidden_act







	def 							_SCREAMING_SNAKE_CASE     (  self:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	Tuple)       -> Any:

				"""simple docstring"""

				__lowerCAmelCase   :						Tuple				= self.conva(_a)
				__lowerCAmelCase   :						int				= self.act_fn(_a)
				__lowerCAmelCase   :						List[str]				= self.drop(_a)
				__lowerCAmelCase   :						Union[str, Any]				= self.conva(_a)
				__lowerCAmelCase   :						Optional[int]				= self.drop(_a)

				return hidden_states






class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	Tuple ,    _SCREAMING_SNAKE_CASE:	Optional[int] ,    _SCREAMING_SNAKE_CASE:	str ,    _SCREAMING_SNAKE_CASE:	List[str] ,    _SCREAMING_SNAKE_CASE:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	Dict ,    _SCREAMING_SNAKE_CASE:	Tuple)       -> Union[str, Any]:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						List[str]				= PoolFormerPooling(_a)
				__lowerCAmelCase   :						int				= PoolFormerOutput(_a ,    _a ,    _a ,    _a)
				__lowerCAmelCase   :						Optional[Any]				= PoolFormerGroupNorm(_a)
				__lowerCAmelCase   :						Optional[Any]				= PoolFormerGroupNorm(_a)

				# Useful for training neural nets
				__lowerCAmelCase   :						List[Any]				= PoolFormerDropPath(_a) if drop_path > 0.0 else nn.Identity()
				__lowerCAmelCase   :						List[str]				= config.use_layer_scale
				if config.use_layer_scale:
							__lowerCAmelCase   :						Tuple				= nn.Parameter(
							    config.layer_scale_init_value * torch.ones((_a)) ,    requires_grad=_a)
							__lowerCAmelCase   :						Tuple				= nn.Parameter(
							    config.layer_scale_init_value * torch.ones((_a)) ,    requires_grad=_a)







	def 							_SCREAMING_SNAKE_CASE     (  self:	Tuple ,    _SCREAMING_SNAKE_CASE:	Tuple)       -> Union[str, Any]:

				"""simple docstring"""

				if self.use_layer_scale:
							__lowerCAmelCase   :						Optional[Any]				= self.pooling(self.before_norm(_a))
							__lowerCAmelCase   :						Union[str, Any]				= self.layer_scale_a.unsqueeze(-1).unsqueeze(-1) * pooling_output
							# First residual connection
							__lowerCAmelCase   :						Union[str, Any]				= hidden_states + self.drop_path(_a)
							__lowerCAmelCase   :						List[Any]				= ()

							__lowerCAmelCase   :						List[str]				= self.output(self.after_norm(_a))
							__lowerCAmelCase   :						Optional[Any]				= self.layer_scale_a.unsqueeze(-1).unsqueeze(-1) * layer_output
							# Second residual connection
							__lowerCAmelCase   :						List[Any]				= hidden_states + self.drop_path(_a)

							__lowerCAmelCase   :						Union[str, Any]				= (output,) + outputs
							return outputs

				else:
							__lowerCAmelCase   :						int				= self.drop_path(self.pooling(self.before_norm(_a)))
							# First residual connection
							__lowerCAmelCase   :						int				= pooling_output + hidden_states
							__lowerCAmelCase   :						Dict				= ()

							# Second residual connection inside the PoolFormerOutput block
							__lowerCAmelCase   :						int				= self.drop_path(self.output(self.after_norm(_a)))
							__lowerCAmelCase   :						Union[str, Any]				= hidden_states + layer_output

							__lowerCAmelCase   :						Dict				= (output,) + outputs
							return outputs






class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	List[str] ,    _SCREAMING_SNAKE_CASE:	str)       -> Optional[int]:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						Union[str, Any]				= config
				# stochastic depth decay rule
				__lowerCAmelCase   :						Union[str, Any]				= [x.item() for x in torch.linspace(0 ,    config.drop_path_rate ,    sum(config.depths))]

				# patch embeddings
				__lowerCAmelCase   :						Optional[Any]				= []
				for i in range(config.num_encoder_blocks):
							embeddings.append(
							    PoolFormerEmbeddings(
							        patch_size=config.patch_sizes[i] ,    stride=config.strides[i] ,    padding=config.padding[i] ,    num_channels=config.num_channels if i == 0 else config.hidden_sizes[i - 1] ,    hidden_size=config.hidden_sizes[i] ,    ))
				__lowerCAmelCase   :						Optional[int]				= nn.ModuleList(_a)

				# Transformer blocks
				__lowerCAmelCase   :						Optional[Any]				= []
				__lowerCAmelCase   :						Tuple				= 0
				for i in range(config.num_encoder_blocks):
							# each block consists of layers
							__lowerCAmelCase   :						List[Any]				= []
							if i != 0:
										cur += config.depths[i - 1]
							for j in range(config.depths[i]):
										layers.append(
										    PoolFormerLayer(
										        _a ,    num_channels=config.hidden_sizes[i] ,    pool_size=config.pool_size ,    hidden_size=config.hidden_sizes[i] ,    intermediate_size=int(config.hidden_sizes[i] * config.mlp_ratio) ,    drop_path=dpr[cur + j] ,    ))
							blocks.append(nn.ModuleList(_a))

				__lowerCAmelCase   :						List[Any]				= nn.ModuleList(_a)







	def 							_SCREAMING_SNAKE_CASE     (  self:	int ,    _SCREAMING_SNAKE_CASE:	Optional[Any] ,    _SCREAMING_SNAKE_CASE:	str=False ,    _SCREAMING_SNAKE_CASE:	List[str]=True)       -> Tuple:

				"""simple docstring"""

				__lowerCAmelCase   :						Tuple				= () if output_hidden_states else None

				__lowerCAmelCase   :						List[str]				= pixel_values
				for idx, layers in enumerate(zip(self.patch_embeddings ,    self.block)):
							__lowerCAmelCase   :						Union[str, Any]				= layers
							# Get patch embeddings from hidden_states
							__lowerCAmelCase   :						str				= embedding_layer(_a)
							# Send the embeddings through the blocks
							for _, blk in enumerate(_a):
										__lowerCAmelCase   :						Optional[Any]				= blk(_a)
										__lowerCAmelCase   :						List[str]				= layer_outputs[0]

							if output_hidden_states:
										__lowerCAmelCase   :						Any				= all_hidden_states + (hidden_states,)

				if not return_dict:
							return tuple(v for v in [hidden_states, all_hidden_states] if v is not None)

				return BaseModelOutputWithNoAttention(last_hidden_state=_a ,    hidden_states=_a)






class     A__     (						UpperCamelCase__							):




	'''simple docstring'''





	SCREAMING_SNAKE_CASE					   =			PoolFormerConfig
	SCREAMING_SNAKE_CASE					   =			'poolformer'
	SCREAMING_SNAKE_CASE					   =			'pixel_values'
	SCREAMING_SNAKE_CASE					   =			True

	def 							_SCREAMING_SNAKE_CASE     (  self:	Dict ,    _SCREAMING_SNAKE_CASE:	Any)       -> Tuple:

				"""simple docstring"""

				if isinstance(_a ,    (nn.Linear, nn.Convad)):
							module.weight.data.normal_(mean=0.0 ,    std=self.config.initializer_range)
							if module.bias is not None:
										module.bias.data.zero_()
				elif isinstance(_a ,    nn.LayerNorm):
							module.bias.data.zero_()
							module.weight.data.fill_(1.0)







	def 							_SCREAMING_SNAKE_CASE     (  self:	Dict ,    _SCREAMING_SNAKE_CASE:	List[Any] ,    _SCREAMING_SNAKE_CASE:	Any=False)       -> Union[str, Any]:

				"""simple docstring"""

				if isinstance(_a ,    _a):
							__lowerCAmelCase   :						List[str]				= value






__snake_case							:     int   =					R'\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) sub-class. Use\n    it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and\n    behavior.\n\n    Parameters:\n        config ([`PoolFormerConfig`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n'

__snake_case							:     str   =					R'\n    Args:\n        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See\n            [`PoolFormerImageProcessor.__call__`] for details.\n'



@add_start_docstrings(
    'The bare PoolFormer Model transformer outputting raw hidden-states without any specific head on top.'						,	UpperCamelCase__						,	)
class     A__     (						UpperCamelCase__							):




	'''simple docstring'''





	def __init__(  self:	str ,    _SCREAMING_SNAKE_CASE:	Optional[int])       -> List[str]:

				"""simple docstring"""

				super().__init__(_a)
				__lowerCAmelCase   :						Any				= config

				__lowerCAmelCase   :						Union[str, Any]				= PoolFormerEncoder(_a)

				# Initialize weights and apply final processing
				self.post_init()

	def 							_SCREAMING_SNAKE_CASE     (  self:	Union[str, Any])       -> str:

				"""simple docstring"""

				return self.embeddings.patch_embeddings







	@add_start_docstrings_to_model_forward(_a)
	@add_code_sample_docstrings(
	    checkpoint=_CHECKPOINT_FOR_DOC ,    output_type=_a ,    config_class=_CONFIG_FOR_DOC ,    modality="vision" ,    expected_output=_EXPECTED_OUTPUT_SHAPE ,    )
	def 							_SCREAMING_SNAKE_CASE     (  self:	int ,    _SCREAMING_SNAKE_CASE:	Any = None ,    _SCREAMING_SNAKE_CASE:	Optional[int] = None ,    _SCREAMING_SNAKE_CASE:	str = None ,    )       -> Union[Tuple, BaseModelOutputWithNoAttention]:

				"""simple docstring"""

				__lowerCAmelCase   :						Optional[int]				= (
				    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
				)
				__lowerCAmelCase   :						List[str]				= return_dict if return_dict is not None else self.config.use_return_dict

				if pixel_values is None:
							raise ValueError("You have to specify pixel_values")

				__lowerCAmelCase   :						Any				= self.encoder(
				    _a ,    output_hidden_states=_a ,    return_dict=_a ,    )
				__lowerCAmelCase   :						Optional[int]				= encoder_outputs[0]

				if not return_dict:
							return (sequence_output, None) + encoder_outputs[1:]

				return BaseModelOutputWithNoAttention(
				    last_hidden_state=_a ,    hidden_states=encoder_outputs.hidden_states ,    )






class     A__     (						nn.Module							):




	'''simple docstring'''





	def __init__(  self:	Tuple ,    _SCREAMING_SNAKE_CASE:	Union[str, Any])       -> Tuple:

				"""simple docstring"""

				super().__init__()
				__lowerCAmelCase   :						List[str]				= nn.Linear(config.hidden_size ,    config.hidden_size)







	def 							_SCREAMING_SNAKE_CASE     (  self:	List[str] ,    _SCREAMING_SNAKE_CASE:	Union[str, Any])       -> Tuple:

				"""simple docstring"""

				__lowerCAmelCase   :						str				= self.dense(_a)
				return output






@add_start_docstrings(
    '\n    PoolFormer Model transformer with an image classification head on top\n    '						,	UpperCamelCase__						,	)
class     A__     (						UpperCamelCase__							):




	'''simple docstring'''





	def __init__(  self:	Any ,    _SCREAMING_SNAKE_CASE:	Union[str, Any])       -> List[Any]:

				"""simple docstring"""

				super().__init__(_a)
				__lowerCAmelCase   :						Tuple				= config.num_labels
				__lowerCAmelCase   :						Optional[Any]				= PoolFormerModel(_a)

				# Final norm
				__lowerCAmelCase   :						Optional[int]				= PoolFormerGroupNorm(config.hidden_sizes[-1])
				# Classifier head
				__lowerCAmelCase   :						str				= (
				    nn.Linear(config.hidden_sizes[-1] ,    config.num_labels) if config.num_labels > 0 else nn.Identity()
				)

				# Initialize weights and apply final processing
				self.post_init()







	@add_start_docstrings_to_model_forward(_a)
	@add_code_sample_docstrings(
	    checkpoint=_IMAGE_CLASS_CHECKPOINT ,    output_type=_a ,    config_class=_CONFIG_FOR_DOC ,    expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT ,    )
	def 							_SCREAMING_SNAKE_CASE     (  self:	int ,    _SCREAMING_SNAKE_CASE:	Union[str, Any] = None ,    _SCREAMING_SNAKE_CASE:	List[str] = None ,    _SCREAMING_SNAKE_CASE:	Optional[Any] = None ,    _SCREAMING_SNAKE_CASE:	List[Any] = None ,    )       -> Union[Tuple, ImageClassifierOutputWithNoAttention]:

				"""simple docstring"""

				__lowerCAmelCase   :						Optional[Any]				= return_dict if return_dict is not None else self.config.use_return_dict

				__lowerCAmelCase   :						Optional[Any]				= self.poolformer(
				    _a ,    output_hidden_states=_a ,    return_dict=_a ,    )

				__lowerCAmelCase   :						Tuple				= outputs[0]

				__lowerCAmelCase   :						Any				= self.classifier(self.norm(_a).mean([-2, -1]))

				__lowerCAmelCase   :						Dict				= None
				if labels is not None:
							if self.config.problem_type is None:
										if self.num_labels == 1:
													__lowerCAmelCase   :						Union[str, Any]				= """regression"""
										elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
													__lowerCAmelCase   :						Optional[int]				= """single_label_classification"""
										else:
													__lowerCAmelCase   :						List[Any]				= """multi_label_classification"""

							if self.config.problem_type == "regression":
										__lowerCAmelCase   :						Union[str, Any]				= MSELoss()
										if self.num_labels == 1:
													__lowerCAmelCase   :						str				= loss_fct(logits.squeeze() ,    labels.squeeze())
										else:
													__lowerCAmelCase   :						Optional[Any]				= loss_fct(_a ,    _a)
							elif self.config.problem_type == "single_label_classification":
										__lowerCAmelCase   :						List[str]				= CrossEntropyLoss()
										__lowerCAmelCase   :						Any				= loss_fct(logits.view(-1 ,    self.num_labels) ,    labels.view(-1))
							elif self.config.problem_type == "multi_label_classification":
										__lowerCAmelCase   :						List[Any]				= BCEWithLogitsLoss()
										__lowerCAmelCase   :						Dict				= loss_fct(_a ,    _a)

				if not return_dict:
							__lowerCAmelCase   :						List[str]				= (logits,) + outputs[2:]
							return ((loss,) + output) if loss is not None else output

				return ImageClassifierOutputWithNoAttention(loss=_a ,    logits=_a ,    hidden_states=outputs.hidden_states)




"""simple docstring"""






import pyarrow.parquet as pq
import pytest

from datasets import Audio, Dataset, DatasetDict, Features, NamedSplit, Sequence, Value, config
from datasets.features.image import Image
from datasets.io.parquet import ParquetDatasetReader, ParquetDatasetWriter, get_writer_batch_size

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases


def     _lowercase       (						__snake_case      ,__snake_case			)		->     Dict:
			assert isinstance(__snake_case      ,__snake_case			)
			assert dataset.num_rows == 4
			assert dataset.num_columns == 3
			assert dataset.column_names == ["col_1", "col_2", "col_3"]
			for feature, expected_dtype in expected_features.items():
						assert dataset.features[feature].dtype == expected_dtype


@pytest.mark.parametrize("keep_in_memory"      ,[False, True]			)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     Any:
			__lowerCAmelCase   :						Dict				= tmp_path / "cache"
			__lowerCAmelCase   :						Union[str, Any]				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
						__lowerCAmelCase   :						Optional[Any]				= ParquetDatasetReader(__snake_case      ,cache_dir=__snake_case      ,keep_in_memory=__snake_case			).read()
			_check_parquet_dataset(__snake_case      ,__snake_case			)


@pytest.mark.parametrize(
    "features"      ,[
        None,
        {"col_1": "string", "col_2": "int64", "col_3": "float64"},
        {"col_1": "string", "col_2": "string", "col_3": "string"},
        {"col_1": "int32", "col_2": "int32", "col_3": "int32"},
        {"col_1": "float32", "col_2": "float32", "col_3": "float32"},
    ]      ,)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     str:
			__lowerCAmelCase   :						Tuple				= tmp_path / "cache"
			__lowerCAmelCase   :						Dict				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			__lowerCAmelCase   :						List[Any]				= features.copy() if features else default_expected_features
			__lowerCAmelCase   :						Dict				= (
			    Features({feature: Value(__snake_case			) for feature, dtype in features.items()}			) if features is not None else None
			)
			__lowerCAmelCase   :						Optional[int]				= ParquetDatasetReader(__snake_case      ,features=__snake_case      ,cache_dir=__snake_case			).read()
			_check_parquet_dataset(__snake_case      ,__snake_case			)


@pytest.mark.parametrize("split"      ,[None, NamedSplit("train"			), "train", "test"]			)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     Any:
			__lowerCAmelCase   :						List[str]				= tmp_path / "cache"
			__lowerCAmelCase   :						List[str]				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			__lowerCAmelCase   :						List[Any]				= ParquetDatasetReader(__snake_case      ,cache_dir=__snake_case      ,split=__snake_case			).read()
			_check_parquet_dataset(__snake_case      ,__snake_case			)
			assert dataset.split == split if split else "train"


@pytest.mark.parametrize("path_type"      ,[str, list]			)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     Dict:
			if issubclass(__snake_case      ,__snake_case			):
						__lowerCAmelCase   :						List[Any]				= parquet_path
			elif issubclass(__snake_case      ,__snake_case			):
						__lowerCAmelCase   :						List[Any]				= [parquet_path]
			__lowerCAmelCase   :						str				= tmp_path / "cache"
			__lowerCAmelCase   :						List[str]				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			__lowerCAmelCase   :						Tuple				= ParquetDatasetReader(__snake_case      ,cache_dir=__snake_case			).read()
			_check_parquet_dataset(__snake_case      ,__snake_case			)


def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case=("train",)			)		->     int:
			assert isinstance(__snake_case      ,__snake_case			)
			for split in splits:
						__lowerCAmelCase   :						str				= dataset_dict[split]
						assert dataset.num_rows == 4
						assert dataset.num_columns == 3
						assert dataset.column_names == ["col_1", "col_2", "col_3"]
						for feature, expected_dtype in expected_features.items():
									assert dataset.features[feature].dtype == expected_dtype


@pytest.mark.parametrize("keep_in_memory"      ,[False, True]			)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     str:
			__lowerCAmelCase   :						Any				= tmp_path / "cache"
			__lowerCAmelCase   :						int				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
						__lowerCAmelCase   :						str				= ParquetDatasetReader(
						    {"train": parquet_path}      ,cache_dir=__snake_case      ,keep_in_memory=__snake_case			).read()
			_check_parquet_datasetdict(__snake_case      ,__snake_case			)


@pytest.mark.parametrize(
    "features"      ,[
        None,
        {"col_1": "string", "col_2": "int64", "col_3": "float64"},
        {"col_1": "string", "col_2": "string", "col_3": "string"},
        {"col_1": "int32", "col_2": "int32", "col_3": "int32"},
        {"col_1": "float32", "col_2": "float32", "col_3": "float32"},
    ]      ,)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     Dict:
			__lowerCAmelCase   :						List[Any]				= tmp_path / "cache"
			__lowerCAmelCase   :						Dict				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			__lowerCAmelCase   :						str				= features.copy() if features else default_expected_features
			__lowerCAmelCase   :						str				= (
			    Features({feature: Value(__snake_case			) for feature, dtype in features.items()}			) if features is not None else None
			)
			__lowerCAmelCase   :						str				= ParquetDatasetReader({"train": parquet_path}      ,features=__snake_case      ,cache_dir=__snake_case			).read()
			_check_parquet_datasetdict(__snake_case      ,__snake_case			)


@pytest.mark.parametrize("split"      ,[None, NamedSplit("train"			), "train", "test"]			)
def     _lowercase       (						__snake_case      ,__snake_case      ,__snake_case			)		->     str:
			if split:
						__lowerCAmelCase   :						Optional[int]				= {split: parquet_path}
			else:
						__lowerCAmelCase   :						str				= "train"
						__lowerCAmelCase   :						Optional[Any]				= {"train": parquet_path, "test": parquet_path}
			__lowerCAmelCase   :						Tuple				= tmp_path / "cache"
			__lowerCAmelCase   :						int				= {"col_1": "string", "col_2": "int64", "col_3": "float64"}
			__lowerCAmelCase   :						Union[str, Any]				= ParquetDatasetReader(__snake_case      ,cache_dir=__snake_case			).read()
			_check_parquet_datasetdict(__snake_case      ,__snake_case      ,splits=list(path.keys()			)			)
			assert all(dataset[split].split == split for split in path.keys()			)


def     _lowercase       (						__snake_case      ,__snake_case			)		->     Union[str, Any]:
			__lowerCAmelCase   :						int				= ParquetDatasetWriter(__snake_case      ,tmp_path / "foo.parquet"			)
			assert writer.write() > 0
			__lowerCAmelCase   :						str				= pq.ParquetFile(tmp_path / "foo.parquet"			)
			__lowerCAmelCase   :						int				= pf.read()
			assert dataset.data.table == output_table


def     _lowercase       (						__snake_case      ,__snake_case			)		->     int:
			__lowerCAmelCase   :						int				= str(shared_datadir / "test_image_rgb.jpg"			)
			__lowerCAmelCase   :						Any				= {"image": [image_path]}
			__lowerCAmelCase   :						List[Any]				= Features({"image": Image()}			)
			__lowerCAmelCase   :						Tuple				= Dataset.from_dict(__snake_case      ,features=__snake_case			)
			__lowerCAmelCase   :						Dict				= ParquetDatasetWriter(__snake_case      ,tmp_path / "foo.parquet"			)
			assert writer.write() > 0

			__lowerCAmelCase   :						List[Any]				= Dataset.from_parquet(str(tmp_path / "foo.parquet"			)			)
			assert dataset.features == reloaded_dataset.features

			__lowerCAmelCase   :						List[Any]				= ParquetDatasetReader(str(tmp_path / "foo.parquet"			)      ,streaming=__snake_case			).read()
			assert dataset.features == reloaded_iterable_dataset.features





@pytest.mark.parametrize(
    "feature, expected"      ,[
        (Features({"foo": Value("int32"			)}			), None),
        (Features({"image": Image(), "foo": Value("int32"			)}			), config.PARQUET_ROW_GROUP_SIZE_FOR_IMAGE_DATASETS),
        (Features({"nested": Sequence(Audio()			)}			), config.PARQUET_ROW_GROUP_SIZE_FOR_AUDIO_DATASETS),
    ]      ,)
def     _lowercase       (						__snake_case      ,__snake_case			)		->     List[Any]:
			assert get_writer_batch_size(__snake_case			) == expected