


'''simple docstring'''







from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[Any]    ,_UpperCAmelCase	:	Optional[int]  )  ->			Any:
       _a    : Any     					=[]
       for part_id in partition_order:
              _a    : List[Any]     					=df.where(F"SPARK_PARTITION_ID() = {part_id}"  ).collect()
              for row_idx, row in enumerate(_UpperCAmelCase  ):
                     expected_row_ids_and_row_dicts.append((F"{part_id}_{row_idx}", row.asDict())  )
       return expected_row_ids_and_row_dicts


@require_not_windows
@require_dill_gt_0_3_2
def 	SCREAMING_SNAKE_CASE_					(  )  ->			Union[str, Any]:
       _a    : Optional[Any]     					=pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
       _a    : Any     					=spark.range(100  ).repartition(1  )
       _a    : Dict     					=Spark(_UpperCAmelCase  )
       # The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
       # that each partition can hold 2 rows.
       spark_builder._repartition_df_if_needed(max_shard_size=16  )
       # Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
       assert spark_builder.df.rdd.getNumPartitions() == 50


@require_not_windows
@require_dill_gt_0_3_2
def 	SCREAMING_SNAKE_CASE_					(  )  ->			str:
       _a    : Dict     					=pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
       _a    : List[Any]     					=spark.range(10  ).repartition(2  )
       _a    : Optional[Any]     					=[1, 0]
       _a    : int     					=_generate_iterable_examples(_UpperCAmelCase    ,_UpperCAmelCase  )  # Reverse the partitions.
       _a    : Union[str, Any]     					=_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCAmelCase    ,_UpperCAmelCase  )

       for i, (row_id, row_dict) in enumerate(generate_fn()  ):
              _a					,    _a    : int     					=expected_row_ids_and_row_dicts[i]
              assert row_id == expected_row_id
              assert row_dict == expected_row_dict


@require_not_windows
@require_dill_gt_0_3_2
def 	SCREAMING_SNAKE_CASE_					(  )  ->			int:
       _a    : str     					=pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
       _a    : str     					=spark.range(10  ).repartition(1  )
       _a    : int     					=SparkExamplesIterable(_UpperCAmelCase  )
       assert it.n_shards == 1
       for i, (row_id, row_dict) in enumerate(_UpperCAmelCase  ):
              assert row_id == F"0_{i}"
              assert row_dict == {"id": i}


@require_not_windows
@require_dill_gt_0_3_2
def 	SCREAMING_SNAKE_CASE_					(  )  ->			Tuple:
       _a    : List[str]     					=pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
       _a    : List[Any]     					=spark.range(30  ).repartition(3  )
       # Mock the generator so that shuffle reverses the partition indices.
       with patch("""numpy.random.Generator"""  ) as generator_mock:
              _a    : Optional[int]     					=lambda _UpperCAmelCase  : x.reverse()
              _a    : Optional[int]     					=_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCAmelCase    ,[2, 1, 0]  )

              _a    : Tuple     					=SparkExamplesIterable(_UpperCAmelCase  ).shuffle_data_sources(_UpperCAmelCase  )
              assert shuffled_it.n_shards == 3
              for i, (row_id, row_dict) in enumerate(_UpperCAmelCase  ):
                     _a					,    _a    : Optional[Any]     					=expected_row_ids_and_row_dicts[i]
                     assert row_id == expected_row_id
                     assert row_dict == expected_row_dict


@require_not_windows
@require_dill_gt_0_3_2
def 	SCREAMING_SNAKE_CASE_					(  )  ->			List[str]:
       _a    : int     					=pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
       _a    : int     					=spark.range(20  ).repartition(4  )

       # Partitions 0 and 2
       _a    : str     					=SparkExamplesIterable(_UpperCAmelCase  ).shard_data_sources(worker_id=0    ,num_workers=2  )
       assert shard_it_a.n_shards == 2
       _a    : Optional[int]     					=_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCAmelCase    ,[0, 2]  )
       for i, (row_id, row_dict) in enumerate(_UpperCAmelCase  ):
              _a					,    _a    : Any     					=expected_row_ids_and_row_dicts_a[i]
              assert row_id == expected_row_id
              assert row_dict == expected_row_dict

       # Partitions 1 and 3
       _a    : int     					=SparkExamplesIterable(_UpperCAmelCase  ).shard_data_sources(worker_id=1    ,num_workers=2  )
       assert shard_it_a.n_shards == 2
       _a    : Dict     					=_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCAmelCase    ,[1, 3]  )
       for i, (row_id, row_dict) in enumerate(_UpperCAmelCase  ):
              _a					,    _a    : Dict     					=expected_row_ids_and_row_dicts_a[i]
              assert row_id == expected_row_id
              assert row_dict == expected_row_dict


@require_not_windows
@require_dill_gt_0_3_2
def 	SCREAMING_SNAKE_CASE_					(  )  ->			Tuple:
       _a    : Optional[int]     					=pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
       _a    : int     					=spark.range(100  ).repartition(1  )
       _a    : List[Any]     					=Spark(_UpperCAmelCase  )
       # Choose a small max_shard_size for maximum partitioning.
       spark_builder._repartition_df_if_needed(max_shard_size=1  )
       # The new number of partitions should not be greater than the number of rows.
       assert spark_builder.df.rdd.getNumPartitions() == 100






'''simple docstring'''







from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer







@dataclass
class A__       (						UpperCAmelCase__      ):
      __UpperCamelCase       :  torch.FloatTensor







class A__       (						UpperCAmelCase__ ,     UpperCAmelCase__      ):






      @register_to_config
      def __init__( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:int = 3     ,						SCREAMING_SNAKE_CASE			:int = 3     ,						SCREAMING_SNAKE_CASE			:Tuple[str] = ("DownEncoderBlock2D",)     ,						SCREAMING_SNAKE_CASE			:Tuple[str] = ("UpDecoderBlock2D",)     ,						SCREAMING_SNAKE_CASE			:Tuple[int] = (6_4,)     ,						SCREAMING_SNAKE_CASE			:int = 1     ,						SCREAMING_SNAKE_CASE			:str = "silu"     ,						SCREAMING_SNAKE_CASE			:int = 3     ,						SCREAMING_SNAKE_CASE			:int = 3_2     ,						SCREAMING_SNAKE_CASE			:int = 2_5_6     ,						SCREAMING_SNAKE_CASE			:int = 3_2     ,						SCREAMING_SNAKE_CASE			:Optional[int] = None     ,						SCREAMING_SNAKE_CASE			:float = 0.18_215     ,						SCREAMING_SNAKE_CASE			:str = "group"     ,						)       ->					Optional[int]:




             '''simple docstring'''
             super().__init__()

             # pass init params to Encoder
             _a    : Union[str, Any]     					=Encoder(
                 in_channels=SCREAMING_SNAKE_CASE     ,						out_channels=SCREAMING_SNAKE_CASE     ,						down_block_types=SCREAMING_SNAKE_CASE     ,						block_out_channels=SCREAMING_SNAKE_CASE     ,						layers_per_block=SCREAMING_SNAKE_CASE     ,						act_fn=SCREAMING_SNAKE_CASE     ,						norm_num_groups=SCREAMING_SNAKE_CASE     ,						double_z=SCREAMING_SNAKE_CASE     ,						)

             _a    : Optional[int]     					=vq_embed_dim if vq_embed_dim is not None else latent_channels

             _a    : Optional[int]     					=nn.Convad(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE     ,						1	)
             _a    : str     					=VectorQuantizer(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE     ,						beta=0.25     ,						remap=SCREAMING_SNAKE_CASE     ,						sane_index_shape=SCREAMING_SNAKE_CASE	)
             _a    : List[str]     					=nn.Convad(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE     ,						1	)

             # pass init params to Decoder
             _a    : List[str]     					=Decoder(
                 in_channels=SCREAMING_SNAKE_CASE     ,						out_channels=SCREAMING_SNAKE_CASE     ,						up_block_types=SCREAMING_SNAKE_CASE     ,						block_out_channels=SCREAMING_SNAKE_CASE     ,						layers_per_block=SCREAMING_SNAKE_CASE     ,						act_fn=SCREAMING_SNAKE_CASE     ,						norm_num_groups=SCREAMING_SNAKE_CASE     ,						norm_type=SCREAMING_SNAKE_CASE     ,						)






      @apply_forward_hook
      def __UpperCAmelCase       ( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:torch.FloatTensor     ,						SCREAMING_SNAKE_CASE			:bool = True	)       ->					VQEncoderOutput:




             '''simple docstring'''
             _a    : Optional[int]     					=self.encoder(SCREAMING_SNAKE_CASE	)
             _a    : int     					=self.quant_conv(SCREAMING_SNAKE_CASE	)

             if not return_dict:
                    return (h,)

             return VQEncoderOutput(latents=SCREAMING_SNAKE_CASE	)






      @apply_forward_hook
      def __UpperCAmelCase       ( self			:List[Any]     ,						SCREAMING_SNAKE_CASE			:torch.FloatTensor     ,						SCREAMING_SNAKE_CASE			:bool = False     ,						SCREAMING_SNAKE_CASE			:bool = True	)       ->					Union[DecoderOutput, torch.FloatTensor]:




             '''simple docstring'''
             # also go through quantization layer
             if not force_not_quantize:
                    _a					,    _a					,    _a    : Tuple     					=self.quantize(SCREAMING_SNAKE_CASE	)
             else:
                    _a    : str     					=h
             _a    : Dict     					=self.post_quant_conv(SCREAMING_SNAKE_CASE	)
             _a    : Union[str, Any]     					=self.decoder(SCREAMING_SNAKE_CASE     ,						quant if self.config.norm_type == """spatial""" else None	)

             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=SCREAMING_SNAKE_CASE	)






      def __UpperCAmelCase       ( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:torch.FloatTensor     ,						SCREAMING_SNAKE_CASE			:bool = True	)       ->					Union[DecoderOutput, torch.FloatTensor]:




             '''simple docstring'''
             _a    : Tuple     					=sample
             _a    : int     					=self.encode(SCREAMING_SNAKE_CASE	).latents
             _a    : List[Any]     					=self.decode(SCREAMING_SNAKE_CASE	).sample

             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=SCREAMING_SNAKE_CASE	)


