




'''simple docstring'''
from typing import Optional, Tuple, Union

import tensorflow as tf

from ...activations_tf import ACTaFN
from ...modeling_tf_outputs import (
    TFBaseModelOutputWithNoAttention,
    TFBaseModelOutputWithPoolingAndNoAttention,
    TFImageClassifierOutputWithNoAttention,
)
from ...modeling_tf_utils import TFPreTrainedModel, TFSequenceClassificationLoss, keras_serializable, unpack_inputs
from ...tf_utils import shape_list
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_resnet import ResNetConfig


A							:      List[Any] 	=       logging.get_logger(__name__)

# General docstring
A							:      Tuple 	=       """ResNetConfig"""

# Base docstring
A							:      List[Any] 	=       """microsoft/resnet-50"""
A							:      Optional[int] 	=       [1, 2_048, 7, 7]

# Image classification docstring
A							:      List[Any] 	=       """microsoft/resnet-50"""
A							:      Union[str, Any] 	=       """tiger cat"""

A							:      List[Any] 	=       [
    """microsoft/resnet-50""",
    # See all resnet models at https://huggingface.co/models?filter=resnet
]






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__ = 3					,		lowerCamelCase__ = 1					,		lowerCamelCase__ = "relu"					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           __lowercase							=						kernel_size // 2
           __lowercase							=						tf.keras.layers.ConvaD(
               lowerCamelCase__					,		kernel_size=lowerCamelCase__					,		strides=lowerCamelCase__					,		padding="""valid"""					,		use_bias=lowerCamelCase__					,		name="""convolution"""     )
           # Use same default momentum and epsilon as PyTorch equivalent
           __lowercase							=						tf.keras.layers.BatchNormalization(epsilon=1E-5					,		momentum=0.9					,		name="""normalization"""     )
           __lowercase							=						ACTaFN[activation] if activation is not None else tf.keras.layers.Activation("""linear"""     )

       def     snake_case__			(    self					,		lowerCamelCase__     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						(self.pad_value, self.pad_value)
           __lowercase							=						tf.pad(lowerCamelCase__					,		[(0, 0), height_pad, width_pad, (0, 0)]     )
           __lowercase							=						self.conv(lowerCamelCase__     )
           return hidden_state




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						self.convolution(lowerCamelCase__     )
           __lowercase							=						self.normalization(lowerCamelCase__					,		training=lowerCamelCase__     )
           __lowercase							=						self.activation(lowerCamelCase__     )
           return hidden_state






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           __lowercase							=						TFResNetConvLayer(
               config.embedding_size					,		kernel_size=7					,		stride=2					,		activation=config.hidden_act					,		name="""embedder"""					,		)
           __lowercase							=						tf.keras.layers.MaxPoolaD(pool_size=3					,		strides=2					,		padding="""valid"""					,		name="""pooler"""     )
           __lowercase							=						config.num_channels




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						shape_list(lowerCamelCase__     )
           if tf.executing_eagerly() and num_channels != self.num_channels:
               raise ValueError(
                   """Make sure that the channel dimension of the pixel values match with the one set in the configuration."""     )
           __lowercase							=						pixel_values
           __lowercase							=						self.embedder(lowerCamelCase__     )
           __lowercase							=						tf.pad(lowerCamelCase__					,		[[0, 0], [1, 1], [1, 1], [0, 0]]     )
           __lowercase							=						self.pooler(lowerCamelCase__     )
           return hidden_state






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__ = 2					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           __lowercase							=						tf.keras.layers.ConvaD(
               lowerCamelCase__					,		kernel_size=1					,		strides=lowerCamelCase__					,		use_bias=lowerCamelCase__					,		name="""convolution"""     )
           # Use same default momentum and epsilon as PyTorch equivalent
           __lowercase							=						tf.keras.layers.BatchNormalization(epsilon=1E-5					,		momentum=0.9					,		name="""normalization"""     )




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						x
           __lowercase							=						self.convolution(lowerCamelCase__     )
           __lowercase							=						self.normalization(lowerCamelCase__					,		training=lowerCamelCase__     )
           return hidden_state






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__ = 1					,		lowerCamelCase__ = "relu"					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           __lowercase							=						in_channels != out_channels or stride != 1
           __lowercase							=						TFResNetConvLayer(lowerCamelCase__					,		stride=lowerCamelCase__					,		name="""layer.0"""     )
           __lowercase							=						TFResNetConvLayer(lowerCamelCase__					,		activation=lowerCamelCase__					,		name="""layer.1"""     )
           __lowercase							=						(
               TFResNetShortCut(lowerCamelCase__					,		stride=lowerCamelCase__					,		name="""shortcut"""     )
               if should_apply_shortcut
               else tf.keras.layers.Activation("""linear"""					,		name="""shortcut"""     )
           )
           __lowercase							=						ACTaFN[activation]




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						hidden_state
           __lowercase							=						self.conva(lowerCamelCase__					,		training=lowerCamelCase__     )
           __lowercase							=						self.conva(lowerCamelCase__					,		training=lowerCamelCase__     )
           __lowercase							=						self.shortcut(lowerCamelCase__					,		training=lowerCamelCase__     )
           hidden_state += residual
           __lowercase							=						self.activation(lowerCamelCase__     )
           return hidden_state






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__ = 1					,		lowerCamelCase__ = "relu"					,		lowerCamelCase__ = 4					,		**lowerCamelCase__					,		)    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           __lowercase							=						in_channels != out_channels or stride != 1
           __lowercase							=						out_channels // reduction
           __lowercase							=						TFResNetConvLayer(lowerCamelCase__					,		kernel_size=1					,		name="""layer.0"""     )
           __lowercase							=						TFResNetConvLayer(lowerCamelCase__					,		stride=lowerCamelCase__					,		name="""layer.1"""     )
           __lowercase							=						TFResNetConvLayer(lowerCamelCase__					,		kernel_size=1					,		activation=lowerCamelCase__					,		name="""layer.2"""     )
           __lowercase							=						(
               TFResNetShortCut(lowerCamelCase__					,		stride=lowerCamelCase__					,		name="""shortcut"""     )
               if should_apply_shortcut
               else tf.keras.layers.Activation("""linear"""					,		name="""shortcut"""     )
           )
           __lowercase							=						ACTaFN[activation]




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						hidden_state
           __lowercase							=						self.conva(lowerCamelCase__					,		training=lowerCamelCase__     )
           __lowercase							=						self.conva(lowerCamelCase__					,		training=lowerCamelCase__     )
           __lowercase							=						self.conva(lowerCamelCase__					,		training=lowerCamelCase__     )
           __lowercase							=						self.shortcut(lowerCamelCase__					,		training=lowerCamelCase__     )
           hidden_state += residual
           __lowercase							=						self.activation(lowerCamelCase__     )
           return hidden_state






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__ = 2					,		lowerCamelCase__ = 2					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )

           __lowercase							=						TFResNetBottleNeckLayer if config.layer_type == """bottleneck""" else TFResNetBasicLayer

           __lowercase							=						[layer(lowerCamelCase__					,		lowerCamelCase__					,		stride=lowerCamelCase__					,		activation=config.hidden_act					,		name="""layers.0"""     )]
           layers += [
               layer(lowerCamelCase__					,		lowerCamelCase__					,		activation=config.hidden_act					,		name=F'layers.{i + 1}'     )
               for i in range(depth - 1     )
           ]
           __lowercase							=						layers




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False     )    ->					tf.Tensor:

           """simple docstring"""



           for layer in self.stage_layers:
               __lowercase							=						layer(lowerCamelCase__					,		training=lowerCamelCase__     )
           return hidden_state






class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):

       def __init__(    self					,		lowerCamelCase__					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           # based on `downsample_in_first_stage` the first layer of the first stage may or may not downsample the input
           __lowercase							=						[
               TFResNetStage(
                   lowerCamelCase__					,		config.embedding_size					,		config.hidden_sizes[0]					,		stride=2 if config.downsample_in_first_stage else 1					,		depth=config.depths[0]					,		name="""stages.0"""					,		)
           ]
           for i, (in_channels, out_channels, depth) in enumerate(
               zip(config.hidden_sizes					,		config.hidden_sizes[1:]					,		config.depths[1:]     )     ):
               self.stages.append(TFResNetStage(lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__					,		depth=lowerCamelCase__					,		name=F'stages.{i + 1}'     )     )




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = False					,		lowerCamelCase__ = True					,		lowerCamelCase__ = False					,		)    ->					TFBaseModelOutputWithNoAttention:

           """simple docstring"""



           __lowercase							=						() if output_hidden_states else None

           for stage_module in self.stages:
               if output_hidden_states:
                   __lowercase							=						hidden_states + (hidden_state,)

               __lowercase							=						stage_module(lowerCamelCase__					,		training=lowerCamelCase__     )

           if output_hidden_states:
               __lowercase							=						hidden_states + (hidden_state,)

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

           return TFBaseModelOutputWithNoAttention(last_hidden_state=lowerCamelCase__					,		hidden_states=lowerCamelCase__     )






class        SCREAMING_SNAKE_CASE(				a_     ):
       snake_case_ :     Optional[Any]    =					ResNetConfig
       snake_case_ :     List[Any]    =					"resnet"
       snake_case_ :     List[str]    =					"pixel_values"

       @property
       def     snake_case__			(    self     )    ->					Union[str, Any]:

           """simple docstring"""



           return {"pixel_values": tf.TensorSpec(shape=(None, self.config.num_channels, 224, 224)					,		dtype=tf.floataa     )}

A							:      List[str] 	=       r"""
    This model is a TensorFlow
    [tf.keras.layers.Layer](https://www.tensorflow.org/api_docs/python/tf/keras/layers/Layer) sub-class. Use it as a
    regular TensorFlow Module and refer to the TensorFlow documentation for all matter related to general usage and
    behavior.

    Parameters:
        config ([`ResNetConfig`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~TFPreTrainedModel.from_pretrained`] method to load the model weights.
"""


A							:      List[str] 	=       r"""
    Args:
        pixel_values (`tf.Tensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`ConvNextImageProcessor.__call__`] for details.

        output_hidden_states (`bool`, *optional*):
            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for
            more detail.
        return_dict (`bool`, *optional*):
            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
"""






@keras_serializable
class        SCREAMING_SNAKE_CASE(				tf.keras.layers.Layer     ):
       snake_case_ :     int    =					ResNetConfig

       def __init__(    self					,		lowerCamelCase__					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(**lowerCamelCase__     )
           __lowercase							=						config
           __lowercase							=						TFResNetEmbeddings(lowerCamelCase__					,		name="""embedder"""     )
           __lowercase							=						TFResNetEncoder(lowerCamelCase__					,		name="""encoder"""     )
           __lowercase							=						tf.keras.layers.GlobalAveragePoolingaD(keepdims=lowerCamelCase__     )




       @unpack_inputs
       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = None					,		lowerCamelCase__ = None					,		lowerCamelCase__ = False					,		)    ->					Union[Tuple[tf.Tensor], TFBaseModelOutputWithPoolingAndNoAttention]:

           """simple docstring"""



           __lowercase							=						(
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           __lowercase							=						return_dict if return_dict is not None else self.config.use_return_dict

           # TF 2.0 image layers can't use NCHW format when running on CPU.
           # We transpose to NHWC format and then transpose back after the full forward pass.
           # (batch_size, num_channels, height, width) -> (batch_size, height, width, num_channels)
           __lowercase							=						tf.transpose(lowerCamelCase__					,		perm=[0, 2, 3, 1]     )
           __lowercase							=						self.embedder(lowerCamelCase__					,		training=lowerCamelCase__     )

           __lowercase							=						self.encoder(
               lowerCamelCase__					,		output_hidden_states=lowerCamelCase__					,		return_dict=lowerCamelCase__					,		training=lowerCamelCase__     )

           __lowercase							=						encoder_outputs[0]

           __lowercase							=						self.pooler(lowerCamelCase__     )

           # Transpose all the outputs to the NCHW format
           # (batch_size, height, width, num_channels) -> (batch_size, num_channels, height, width)
           __lowercase							=						tf.transpose(lowerCamelCase__					,		(0, 3, 1, 2)     )
           __lowercase							=						tf.transpose(lowerCamelCase__					,		(0, 3, 1, 2)     )
           __lowercase							=						()
           for hidden_state in encoder_outputs[1:]:
               __lowercase							=						hidden_states + tuple(tf.transpose(lowerCamelCase__					,		(0, 3, 1, 2)     ) for h in hidden_state     )

           if not return_dict:
               return (last_hidden_state, pooled_output) + hidden_states

           __lowercase							=						hidden_states if output_hidden_states else None

           return TFBaseModelOutputWithPoolingAndNoAttention(
               last_hidden_state=lowerCamelCase__					,		pooler_output=lowerCamelCase__					,		hidden_states=lowerCamelCase__					,		)






@add_start_docstrings(
    """The bare ResNet model outputting raw features without any specific head on top."""   ,      a_   ,      )
class        SCREAMING_SNAKE_CASE(				a_     ):

       def __init__(    self					,		lowerCamelCase__					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(lowerCamelCase__					,		**lowerCamelCase__     )
           __lowercase							=						TFResNetMainLayer(config=lowerCamelCase__					,		name="""resnet"""     )




       @add_start_docstrings_to_model_forward(lowerCamelCase__     )
       @add_code_sample_docstrings(
           checkpoint=_CHECKPOINT_FOR_DOC					,		output_type=lowerCamelCase__					,		config_class=_CONFIG_FOR_DOC					,		modality="""vision"""					,		expected_output=_EXPECTED_OUTPUT_SHAPE					,		)
       @unpack_inputs
       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__ = None					,		lowerCamelCase__ = None					,		lowerCamelCase__ = False					,		)    ->					Union[Tuple[tf.Tensor], TFBaseModelOutputWithPoolingAndNoAttention]:

           """simple docstring"""



           __lowercase							=						(
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           __lowercase							=						return_dict if return_dict is not None else self.config.use_return_dict

           __lowercase							=						self.resnet(
               pixel_values=lowerCamelCase__					,		output_hidden_states=lowerCamelCase__					,		return_dict=lowerCamelCase__					,		training=lowerCamelCase__					,		)
           return resnet_outputs






@add_start_docstrings(
    """\n    ResNet Model with an image classification head on top (a linear layer on top of the pooled features), e.g. for\n    ImageNet.\n    """   ,      a_   ,      )
class        SCREAMING_SNAKE_CASE(				a_   ,      a_     ):

       def __init__(    self					,		lowerCamelCase__					,		**lowerCamelCase__     )    ->					None:

           """simple docstring"""



           super().__init__(lowerCamelCase__					,		**lowerCamelCase__     )
           __lowercase							=						config.num_labels
           __lowercase							=						TFResNetMainLayer(lowerCamelCase__					,		name="""resnet"""     )
           # classification head
           __lowercase							=						(
               tf.keras.layers.Dense(config.num_labels					,		name="""classifier.1"""     )
               if config.num_labels > 0
               else tf.keras.layers.Activation("""linear"""					,		name="""classifier.1"""     )
           )

       def     snake_case__			(    self					,		lowerCamelCase__     )    ->					tf.Tensor:

           """simple docstring"""



           __lowercase							=						tf.keras.layers.Flatten()(lowerCamelCase__     )
           __lowercase							=						self.classifier_layer(lowerCamelCase__     )
           return logits




       @add_start_docstrings_to_model_forward(lowerCamelCase__     )
       @add_code_sample_docstrings(
           checkpoint=_IMAGE_CLASS_CHECKPOINT					,		output_type=lowerCamelCase__					,		config_class=_CONFIG_FOR_DOC					,		expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT					,		)
       @unpack_inputs
       def     snake_case__			(    self					,		lowerCamelCase__ = None					,		lowerCamelCase__ = None					,		lowerCamelCase__ = None					,		lowerCamelCase__ = None					,		lowerCamelCase__ = False					,		)    ->					Union[Tuple[tf.Tensor], TFImageClassifierOutputWithNoAttention]:

           """simple docstring"""



           __lowercase							=						return_dict if return_dict is not None else self.config.use_return_dict

           __lowercase							=						self.resnet(
               lowerCamelCase__					,		output_hidden_states=lowerCamelCase__					,		return_dict=lowerCamelCase__					,		training=lowerCamelCase__     )

           __lowercase							=						outputs.pooler_output if return_dict else outputs[1]

           __lowercase							=						self.classifier(lowerCamelCase__     )

           __lowercase							=						None if labels is None else self.hf_compute_loss(lowerCamelCase__					,		lowerCamelCase__     )

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

           return TFImageClassifierOutputWithNoAttention(loss=lowerCamelCase__					,		logits=lowerCamelCase__					,		hidden_states=outputs.hidden_states     )







import csv
import os

import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.csv import CsvDatasetReader, CsvDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases





def 	snake_case_      (__A     :						Any					,   __A     :						Union[str, Any]     )			->	int:
   assert isinstance(__A					,   __A     )
   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 	snake_case_      (__A     :						int					,   __A     :						int					,   __A     :						List[str]     )			->	Tuple:
   __lowerCAmelCase  :			Dict					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			int					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader(__A					,   cache_dir=__A					,   keep_in_memory=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





@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 	snake_case_      (__A     :						Dict					,   __A     :						List[Any]					,   __A     :						List[str]     )			->	Union[str, Any]:
   __lowerCAmelCase  :			str					    =						tmp_path / """cache"""
   # CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
   __lowerCAmelCase  :			Tuple					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			int					    =						features.copy() if features else default_expected_features
   __lowerCAmelCase  :			List[Any]					    =						(
       Features({feature: Value(__A     ) for feature, dtype in features.items()}     ) if features is not None else None
   )
   __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader(__A					,   features=__A					,   cache_dir=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





@pytest.mark.parametrize("""split"""					,   [None, NamedSplit("""train"""     ), """train""", """test"""]     )
def 	snake_case_      (__A     :						str					,   __A     :						Optional[int]					,   __A     :						int     )			->	Union[str, Any]:
   __lowerCAmelCase  :			str					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Tuple					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			Optional[Any]					    =						CsvDatasetReader(__A					,   cache_dir=__A					,   split=__A     ).read()
   _check_csv_dataset(__A					,   __A     )
   assert dataset.split == split if split else "train"





@pytest.mark.parametrize("""path_type"""					,   [str, list]     )
def 	snake_case_      (__A     :						Union[str, Any]					,   __A     :						Any					,   __A     :						List[str]     )			->	Optional[int]:
   if issubclass(__A					,   __A     ):
      __lowerCAmelCase  :			Dict					    =						csv_path
   elif issubclass(__A					,   __A     ):
      __lowerCAmelCase  :			Tuple					    =						[csv_path]
   __lowerCAmelCase  :			Dict					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Dict					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader(__A					,   cache_dir=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





def 	snake_case_      (__A     :						int					,   __A     :						Optional[int]					,   __A     :						Any=("train",)     )			->	Optional[int]:
   assert isinstance(__A					,   __A     )
   for split in splits:
      __lowerCAmelCase  :			int					    =						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 	snake_case_      (__A     :						Optional[Any]					,   __A     :						List[str]					,   __A     :						Any     )			->	List[str]:
   __lowerCAmelCase  :			int					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Optional[Any]					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A					,   keep_in_memory=__A     ).read()
   _check_csv_datasetdict(__A					,   __A     )





@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 	snake_case_      (__A     :						List[str]					,   __A     :						List[Any]					,   __A     :						Optional[int]     )			->	Optional[int]:
   __lowerCAmelCase  :			Optional[int]					    =						tmp_path / """cache"""
   # CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
   __lowerCAmelCase  :			Union[str, Any]					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			Optional[int]					    =						features.copy() if features else default_expected_features
   __lowerCAmelCase  :			Optional[Any]					    =						(
       Features({feature: Value(__A     ) for feature, dtype in features.items()}     ) if features is not None else None
   )
   __lowerCAmelCase  :			Union[str, Any]					    =						CsvDatasetReader({"""train""": csv_path}					,   features=__A					,   cache_dir=__A     ).read()
   _check_csv_datasetdict(__A					,   __A     )





@pytest.mark.parametrize("""split"""					,   [None, NamedSplit("""train"""     ), """train""", """test"""]     )
def 	snake_case_      (__A     :						Optional[Any]					,   __A     :						Optional[Any]					,   __A     :						Union[str, Any]     )			->	Optional[int]:
   if split:
      __lowerCAmelCase  :			Any					    =						{split: csv_path}
   else:
      __lowerCAmelCase  :			Dict					    =						{"""train""": csv_path, """test""": csv_path}
   __lowerCAmelCase  :			List[str]					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			int					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			int					    =						CsvDatasetReader(__A					,   cache_dir=__A     ).read()
   _check_csv_datasetdict(__A					,   __A					,   splits=list(path.keys()     )     )
   assert all(dataset[split].split == split for split in path.keys()     )





def 	snake_case_      (__A     :						List[str]     )			->	Optional[Any]:
   with open(__A					,   encoding="""utf-8"""     ) as csvfile:
      yield from csv.reader(__A     )





def 	snake_case_      (__A     :						Optional[int]					,   __A     :						Optional[int]     )			->	Union[str, Any]:
   __lowerCAmelCase  :			Optional[Any]					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Tuple					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=1     ).write()

   __lowerCAmelCase  :			Dict					    =						iter_csv_file(__A     )
   __lowerCAmelCase  :			Union[str, Any]					    =						iter_csv_file(__A     )

   for rowa, rowa in zip(__A					,   __A     ):
      assert rowa == rowa





def 	snake_case_      (__A     :						int					,   __A     :						List[str]     )			->	str:
   __lowerCAmelCase  :			Any					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			List[str]					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			Dict					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=2     ).write()

   __lowerCAmelCase  :			Union[str, Any]					    =						iter_csv_file(__A     )
   __lowerCAmelCase  :			Dict					    =						iter_csv_file(__A     )

   for rowa, rowa in zip(__A					,   __A     ):
      assert rowa == rowa





def 	snake_case_      (__A     :						int					,   __A     :						Optional[Any]     )			->	List[str]:
   __lowerCAmelCase  :			Any					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			List[Any]					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   with pytest.raises(__A     ):
      CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=0     )


