


from typing import Optional, Tuple, Union

import tensorflow as tf

from ...activations_tf import ACTaFN
from ...file_utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward
from ...modeling_tf_outputs import (
    TFBaseModelOutputWithNoAttention,
    TFBaseModelOutputWithPoolingAndNoAttention,
    TFSequenceClassifierOutput,
)
from ...modeling_tf_utils import TFPreTrainedModel, TFSequenceClassificationLoss, keras_serializable, unpack_inputs
from ...tf_utils import shape_list
from ...utils import logging
from .configuration_regnet import RegNetConfig


snake_case							:       Optional[Any]										=	logging.get_logger(__name__)

# General docstring
snake_case							:       str										=	'''RegNetConfig'''

# Base docstring
snake_case							:       Union[str, Any]										=	'''facebook/regnet-y-040'''
snake_case							:       List[Any]										=	[1, 10_88, 7, 7]

# Image classification docstring
snake_case							:       Tuple										=	'''facebook/regnet-y-040'''
snake_case							:       Tuple										=	'''tabby, tabby cat'''

snake_case							:       Union[str, Any]										=	[
    '''facebook/regnet-y-040''',
    # See all regnet models at https://huggingface.co/models?filter=regnet
]







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




 def __init__(						self   ,    _lowerCamelCase   ,    _lowerCamelCase = 3   ,    _lowerCamelCase = 1   ,    _lowerCamelCase = 1   ,    _lowerCamelCase = "relu"   ,    **_lowerCamelCase   ,    ):
       super().__init__(**_lowerCamelCase       )
       # The padding and conv has been verified in
       # https://colab.research.google.com/gist/sayakpaul/854bc10eeaf21c9ee2119e0b9f3841a7/scratchpad.ipynb
       a		:List[Any]									=			tf.keras.layers.ZeroPaddingaD(padding=kernel_size // 2       )
       a		:List[Any]									=			tf.keras.layers.ConvaD(
           filters=_lowerCamelCase   ,    kernel_size=_lowerCamelCase   ,    strides=_lowerCamelCase   ,    padding='''VALID'''   ,    groups=_lowerCamelCase   ,    use_bias=_lowerCamelCase   ,    name='''convolution'''   ,    )
       a		:Tuple									=			tf.keras.layers.BatchNormalization(epsilon=1e-5   ,    momentum=0.9   ,    name='''normalization'''       )
       a		:Optional[Any]									=			ACTaFN[activation] if activation is not None else tf.identity




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase       ):
       a		:Union[str, Any]									=			self.convolution(self.padding(_lowerCamelCase       )       )
       a		:str									=			self.normalization(_lowerCamelCase       )
       a		:List[Any]									=			self.activation(_lowerCamelCase       )
       return hidden_state







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




 def __init__(						self   ,    _lowerCamelCase   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:str									=			config.num_channels
       a		:str									=			TFRegNetConvLayer(
           out_channels=config.embedding_size   ,    kernel_size=3   ,    stride=2   ,    activation=config.hidden_act   ,    name='''embedder'''   ,    )




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase       ):
       a		:Dict									=			shape_list(_lowerCamelCase       )[1]
       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.'''       )

       # When running on CPU, `tf.keras.layers.Conv2D` doesn't support `NCHW` format.
       # So change the input format from `NCHW` to `NHWC`.
       # shape = (batch_size, in_height, in_width, in_channels=num_channels)
       a		:Dict									=			tf.transpose(_lowerCamelCase   ,    perm=(0, 2, 3, 1)       )
       a		:Dict									=			self.embedder(_lowerCamelCase       )
       return hidden_state







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




 def __init__(						self   ,    _lowerCamelCase   ,    _lowerCamelCase = 2   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:Optional[int]									=			tf.keras.layers.ConvaD(
           filters=_lowerCamelCase   ,    kernel_size=1   ,    strides=_lowerCamelCase   ,    use_bias=_lowerCamelCase   ,    name='''convolution'''       )
       a		:Union[str, Any]									=			tf.keras.layers.BatchNormalization(epsilon=1e-5   ,    momentum=0.9   ,    name='''normalization'''       )




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase = False       ):
       return self.normalization(self.convolution(_lowerCamelCase       )   ,    training=_lowerCamelCase       )







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




 def __init__(						self   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:Optional[int]									=			tf.keras.layers.GlobalAveragePoolingaD(keepdims=_lowerCamelCase   ,    name='''pooler'''       )
       a		:List[str]									=			[
           tf.keras.layers.ConvaD(filters=_lowerCamelCase   ,    kernel_size=1   ,    activation='''relu'''   ,    name='''attention.0'''       ),
           tf.keras.layers.ConvaD(filters=_lowerCamelCase   ,    kernel_size=1   ,    activation='''sigmoid'''   ,    name='''attention.2'''       ),
       ]




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase       ):
       # [batch_size, h, w, num_channels] -> [batch_size, 1, 1, num_channels]
       a		:str									=			self.pooler(_lowerCamelCase       )
       for layer_module in self.attention:
             a		:Tuple									=			layer_module(_lowerCamelCase       )
       a		:int									=			hidden_state * pooled
       return hidden_state







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




 def __init__(						self   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase = 1   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:Any									=			in_channels != out_channels or stride != 1
       a		:Any									=			max(1   ,    out_channels // config.groups_width       )
       a		:int									=			(
           TFRegNetShortCut(_lowerCamelCase   ,    stride=_lowerCamelCase   ,    name='''shortcut'''       )
           if should_apply_shortcut
           else tf.keras.layers.Activation('''linear'''   ,    name='''shortcut'''       )
       )
       # `self.layers` instead of `self.layer` because that is a reserved argument.
       a		:str									=			[
           TFRegNetConvLayer(_lowerCamelCase   ,    kernel_size=1   ,    activation=config.hidden_act   ,    name='''layer.0'''       ),
           TFRegNetConvLayer(
               _lowerCamelCase   ,    stride=_lowerCamelCase   ,    groups=_lowerCamelCase   ,    activation=config.hidden_act   ,    name='''layer.1'''       ),
           TFRegNetConvLayer(_lowerCamelCase   ,    kernel_size=1   ,    activation=_lowerCamelCase   ,    name='''layer.2'''       ),
       ]
       a		:str									=			ACTaFN[config.hidden_act]




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase       ):
       a		:Any									=			hidden_state
       for layer_module in self.layers:
             a		:Dict									=			layer_module(_lowerCamelCase       )
       a		:List[str]									=			self.shortcut(_lowerCamelCase       )
       hidden_state += residual
       a		:List[Any]									=			self.activation(_lowerCamelCase       )
       return hidden_state







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




 def __init__(						self   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase = 1   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:Dict									=			in_channels != out_channels or stride != 1
       a		:List[str]									=			max(1   ,    out_channels // config.groups_width       )
       a		:Any									=			(
           TFRegNetShortCut(_lowerCamelCase   ,    stride=_lowerCamelCase   ,    name='''shortcut'''       )
           if should_apply_shortcut
           else tf.keras.layers.Activation('''linear'''   ,    name='''shortcut'''       )
       )
       a		:Tuple									=			[
           TFRegNetConvLayer(_lowerCamelCase   ,    kernel_size=1   ,    activation=config.hidden_act   ,    name='''layer.0'''       ),
           TFRegNetConvLayer(
               _lowerCamelCase   ,    stride=_lowerCamelCase   ,    groups=_lowerCamelCase   ,    activation=config.hidden_act   ,    name='''layer.1'''       ),
           TFRegNetSELayer(_lowerCamelCase   ,    reduced_channels=int(round(in_channels / 4       )       )   ,    name='''layer.2'''       ),
           TFRegNetConvLayer(_lowerCamelCase   ,    kernel_size=1   ,    activation=_lowerCamelCase   ,    name='''layer.3'''       ),
       ]
       a		:Any									=			ACTaFN[config.hidden_act]




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase       ):
       a		:List[str]									=			hidden_state
       for layer_module in self.layers:
             a		:Tuple									=			layer_module(_lowerCamelCase       )
       a		:Any									=			self.shortcut(_lowerCamelCase       )
       hidden_state += residual
       a		:int									=			self.activation(_lowerCamelCase       )
       return hidden_state







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




 def __init__(						self   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase = 2   ,    _lowerCamelCase = 2   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )

       a		:List[str]									=			TFRegNetXLayer if config.layer_type == '''x''' else TFRegNetYLayer
       a		:Optional[Any]									=			[
           # downsampling is done in the first layer with stride of 2
           layer(_lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    stride=_lowerCamelCase   ,    name='''layers.0'''       ),
           *[layer(_lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    name=F'''layers.{i+1}'''       ) for i in range(depth - 1       )],
       ]




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase       ):
       for layer_module in self.layers:
             a		:Tuple									=			layer_module(_lowerCamelCase       )
       return hidden_state







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




 def __init__(						self   ,    _lowerCamelCase   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:int									=			[]
       # based on `downsample_in_first_stage`, the first layer of the first stage may or may not downsample the input
       self.stages.append(
           TFRegNetStage(
               _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'''   ,    )       )
       a		:str									=			zip(config.hidden_sizes   ,    config.hidden_sizes[1:]       )
       for i, ((in_channels, out_channels), depth) in enumerate(zip(_lowerCamelCase   ,    config.depths[1:]       )       ):
             self.stages.append(TFRegNetStage(_lowerCamelCase   ,    _lowerCamelCase   ,    _lowerCamelCase   ,    depth=_lowerCamelCase   ,    name=F'''stages.{i+1}'''       )       )




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase = False   ,    _lowerCamelCase = True       ):
       a		:str									=			() if output_hidden_states else None

       for stage_module in self.stages:
             if output_hidden_states:
                   a		:List[Any]									=			hidden_states + (hidden_state,)

             a		:Optional[Any]									=			stage_module(_lowerCamelCase       )

       if output_hidden_states:
             a		:List[Any]									=			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       )







@keras_serializable
class 						_snake_case    ( tf.keras.layers.Layer				):
 SCREAMING_SNAKE_CASE__ =			RegNetConfig




 def __init__(						self   ,    _lowerCamelCase   ,    **_lowerCamelCase       ):
       super().__init__(**_lowerCamelCase       )
       a		:Tuple									=			config
       a		:List[Any]									=			TFRegNetEmbeddings(_lowerCamelCase   ,    name='''embedder'''       )
       a		:Union[str, Any]									=			TFRegNetEncoder(_lowerCamelCase   ,    name='''encoder'''       )
       a		:Optional[int]									=			tf.keras.layers.GlobalAveragePoolingaD(keepdims=_lowerCamelCase   ,    name='''pooler'''       )




 @unpack_inputs
 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase = None   ,    _lowerCamelCase = None   ,    _lowerCamelCase = False   ,    ):
       a		:Dict									=			(
           output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
       )
       a		:Union[str, Any]									=			return_dict if return_dict is not None else self.config.use_return_dict

       a		:str									=			self.embedder(_lowerCamelCase   ,    training=_lowerCamelCase       )

       a		:Optional[int]									=			self.encoder(
           _lowerCamelCase   ,    output_hidden_states=_lowerCamelCase   ,    return_dict=_lowerCamelCase   ,    training=_lowerCamelCase       )

       a		:Union[str, Any]									=			encoder_outputs[0]
       a		:Optional[int]									=			self.pooler(_lowerCamelCase       )

       # Change to NCHW output format have uniformity in the modules
       a		:Optional[Any]									=			tf.transpose(_lowerCamelCase   ,    perm=(0, 3, 1, 2)       )
       a		:Any									=			tf.transpose(_lowerCamelCase   ,    perm=(0, 3, 1, 2)       )

       # Change the other hidden state outputs to NCHW as well
       if output_hidden_states:
             a		:Optional[int]									=			tuple([tf.transpose(_lowerCamelCase   ,    perm=(0, 3, 1, 2)       ) for h in encoder_outputs[1]]       )

       if not return_dict:
             return (last_hidden_state, pooled_output) + encoder_outputs[1:]

       return TFBaseModelOutputWithPoolingAndNoAttention(
           last_hidden_state=_lowerCamelCase   ,    pooler_output=_lowerCamelCase   ,    hidden_states=hidden_states if output_hidden_states else encoder_outputs.hidden_states   ,    )







class 						_snake_case    ( _snake_case				):
 SCREAMING_SNAKE_CASE__ =			RegNetConfig
 SCREAMING_SNAKE_CASE__ =			'regnet'
 SCREAMING_SNAKE_CASE__ =			'pixel_values'




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







snake_case							:       Union[str, Any]										=	R'''
    Parameters:
    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.
        config ([`RegNetConfig`]): 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.
'''

snake_case							:       Union[str, Any]										=	R'''
    Args:
        pixel_values (`tf.Tensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`ConveNextImageProcessor.__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.
'''







@add_start_docstrings(
    'The bare RegNet model outputting raw features without any specific head on top.'		,       _snake_case		,       )
class 						_snake_case    ( _snake_case				):




 def __init__(						self   ,    _lowerCamelCase   ,    *_lowerCamelCase   ,    **_lowerCamelCase       ):
       super().__init__(_lowerCamelCase   ,    *_lowerCamelCase   ,    **_lowerCamelCase       )
       a		:List[str]									=			TFRegNetMainLayer(_lowerCamelCase   ,    name='''regnet'''       )




 @unpack_inputs
 @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   ,    )
 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase = None   ,    _lowerCamelCase = None   ,    _lowerCamelCase=False   ,    ):
       a		:Optional[Any]									=			(
           output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
       )
       a		:str									=			return_dict if return_dict is not None else self.config.use_return_dict

       a		:Tuple									=			self.regnet(
           pixel_values=_lowerCamelCase   ,    output_hidden_states=_lowerCamelCase   ,    return_dict=_lowerCamelCase   ,    training=_lowerCamelCase   ,    )
       if not return_dict:
             return (outputs[0],) + outputs[1:]

       return TFBaseModelOutputWithPoolingAndNoAttention(
           last_hidden_state=outputs.last_hidden_state   ,    pooler_output=outputs.pooler_output   ,    hidden_states=outputs.hidden_states   ,    )







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




 def __init__(						self   ,    _lowerCamelCase   ,    *_lowerCamelCase   ,    **_lowerCamelCase       ):
       super().__init__(_lowerCamelCase   ,    *_lowerCamelCase   ,    **_lowerCamelCase       )
       a		:Dict									=			config.num_labels
       a		:Union[str, Any]									=			TFRegNetMainLayer(_lowerCamelCase   ,    name='''regnet'''       )
       # classification head
       a		:Optional[Any]									=			[
           tf.keras.layers.Flatten(),
           tf.keras.layers.Dense(config.num_labels   ,    name='''classifier.1'''       ) if config.num_labels > 0 else tf.identity,
       ]




 @unpack_inputs
 @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   ,    )
 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase = None   ,    _lowerCamelCase = None   ,    _lowerCamelCase = None   ,    _lowerCamelCase = None   ,    _lowerCamelCase=False   ,    ):
       a		:List[Any]									=			(
           output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
       )
       a		:Union[str, Any]									=			return_dict if return_dict is not None else self.config.use_return_dict

       a		:Tuple									=			self.regnet(
           _lowerCamelCase   ,    output_hidden_states=_lowerCamelCase   ,    return_dict=_lowerCamelCase   ,    training=_lowerCamelCase       )

       a		:Union[str, Any]									=			outputs.pooler_output if return_dict else outputs[1]

       a		:Any									=			self.classifier[0](_lowerCamelCase       )
       a		:Optional[int]									=			self.classifier[1](_lowerCamelCase       )

       a		:Any									=			None if labels is None else self.hf_compute_loss(labels=_lowerCamelCase   ,    logits=_lowerCamelCase       )

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

       return TFSequenceClassifierOutput(loss=_lowerCamelCase   ,    logits=_lowerCamelCase   ,    hidden_states=outputs.hidden_states       )









import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam







class 						_snake_case    ( datasets.BeamBasedBuilder				):




 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       return datasets.DatasetInfo(
           features=datasets.Features({'''content''': datasets.Value('''string'''       )}       )   ,    supervised_keys=_lowerCamelCase   ,    )




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       return [datasets.SplitGenerator(name=datasets.Split.TRAIN   ,    gen_kwargs={'''examples''': get_test_dummy_examples()}       )]




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       import apache_beam as beam

       return pipeline | "Load Examples" >> beam.Create(_lowerCamelCase       )







class 						_snake_case    ( datasets.BeamBasedBuilder				):




 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       return datasets.DatasetInfo(
           features=datasets.Features({'''a''': datasets.Sequence({'''b''': datasets.Value('''string'''       )}       )}       )   ,    supervised_keys=_lowerCamelCase   ,    )




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       return [
           datasets.SplitGenerator(name=datasets.Split.TRAIN   ,    gen_kwargs={'''examples''': get_test_nested_examples()}       )
       ]




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       import apache_beam as beam

       return pipeline | "Load Examples" >> beam.Create(_lowerCamelCase       )







def 							__lowerCamelCase		(					):




      """simple docstring"""





      return [(i, {"content": content}) for i, content in enumerate(['''foo''', '''bar''', '''foobar''']		)]
def 							__lowerCamelCase		(					):




      """simple docstring"""





      return [(i, {"a": {"b": [content]}}) for i, content in enumerate(['''foo''', '''bar''', '''foobar''']		)]







class 						_snake_case    ( _snake_case				):




 @require_beam
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       a		:Dict									=			len(get_test_dummy_examples()       )
       with tempfile.TemporaryDirectory() as tmp_cache_dir:
             a		:Optional[Any]									=			DummyBeamDataset(cache_dir=_lowerCamelCase   ,    beam_runner='''DirectRunner'''       )
             builder.download_and_prepare()
             self.assertTrue(
                 os.path.exists(
                     os.path.join(_lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    F'''{builder.name}-train.arrow'''       )       )       )
             self.assertDictEqual(builder.info.features   ,    datasets.Features({'''content''': datasets.Value('''string'''       )}       )       )
             a		:str									=			builder.as_dataset()
             self.assertEqual(dset['''train'''].num_rows   ,    _lowerCamelCase       )
             self.assertEqual(dset['''train'''].info.splits['''train'''].num_examples   ,    _lowerCamelCase       )
             self.assertDictEqual(dset['''train'''][0]   ,    get_test_dummy_examples()[0][1]       )
             self.assertDictEqual(
                 dset['''train'''][expected_num_examples - 1]   ,    get_test_dummy_examples()[expected_num_examples - 1][1]       )
             self.assertTrue(
                 os.path.exists(os.path.join(_lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    '''dataset_info.json'''       )       )       )
             del dset




 @require_beam
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       import apache_beam as beam

       a		:Any									=			beam.io.parquetio.WriteToParquet

       a		:Any									=			len(get_test_dummy_examples()       )
       with tempfile.TemporaryDirectory() as tmp_cache_dir:
             a		:Union[str, Any]									=			DummyBeamDataset(cache_dir=_lowerCamelCase   ,    beam_runner='''DirectRunner'''       )
             with patch('''apache_beam.io.parquetio.WriteToParquet'''       ) as write_parquet_mock:
                   a		:str									=			partial(_lowerCamelCase   ,    num_shards=2       )
                   builder.download_and_prepare()
             self.assertTrue(
                 os.path.exists(
                     os.path.join(
                         _lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    F'''{builder.name}-train-00000-of-00002.arrow'''       )       )       )
             self.assertTrue(
                 os.path.exists(
                     os.path.join(
                         _lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    F'''{builder.name}-train-00000-of-00002.arrow'''       )       )       )
             self.assertDictEqual(builder.info.features   ,    datasets.Features({'''content''': datasets.Value('''string'''       )}       )       )
             a		:List[str]									=			builder.as_dataset()
             self.assertEqual(dset['''train'''].num_rows   ,    _lowerCamelCase       )
             self.assertEqual(dset['''train'''].info.splits['''train'''].num_examples   ,    _lowerCamelCase       )
             # Order is not preserved when sharding, so we just check that all the elements are there
             self.assertListEqual(sorted(dset['''train''']['''content''']       )   ,    sorted(['''foo''', '''bar''', '''foobar''']       )       )
             self.assertTrue(
                 os.path.exists(os.path.join(_lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    '''dataset_info.json'''       )       )       )
             del dset




 @require_beam
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       with tempfile.TemporaryDirectory() as tmp_cache_dir:
             a		:Dict									=			DummyBeamDataset(cache_dir=_lowerCamelCase       )
             self.assertRaises(datasets.builder.MissingBeamOptions   ,    builder.download_and_prepare       )




 @require_beam
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       a		:str									=			len(get_test_nested_examples()       )
       with tempfile.TemporaryDirectory() as tmp_cache_dir:
             a		:List[Any]									=			NestedBeamDataset(cache_dir=_lowerCamelCase   ,    beam_runner='''DirectRunner'''       )
             builder.download_and_prepare()
             self.assertTrue(
                 os.path.exists(
                     os.path.join(_lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    F'''{builder.name}-train.arrow'''       )       )       )
             self.assertDictEqual(
                 builder.info.features   ,    datasets.Features({'''a''': datasets.Sequence({'''b''': datasets.Value('''string'''       )}       )}       )       )
             a		:Any									=			builder.as_dataset()
             self.assertEqual(dset['''train'''].num_rows   ,    _lowerCamelCase       )
             self.assertEqual(dset['''train'''].info.splits['''train'''].num_examples   ,    _lowerCamelCase       )
             self.assertDictEqual(dset['''train'''][0]   ,    get_test_nested_examples()[0][1]       )
             self.assertDictEqual(
                 dset['''train'''][expected_num_examples - 1]   ,    get_test_nested_examples()[expected_num_examples - 1][1]       )
             self.assertTrue(
                 os.path.exists(os.path.join(_lowerCamelCase   ,    builder.name   ,    '''default'''   ,    '''0.0.0'''   ,    '''dataset_info.json'''       )       )       )
             del dset





