




import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference







class 						lowercase							(   __lowerCAmelCase,unittest.TestCase		):
     _a							 =						ShapEPipeline
     _a							 =						['''prompt''']
     _a							 =						['''prompt''']
     _a							 =						[
         '''num_images_per_prompt''',
         '''num_inference_steps''',
         '''generator''',
         '''latents''',
         '''guidance_scale''',
         '''frame_size''',
         '''output_type''',
         '''return_dict''',
     ]
     _a							 =						False

     @property
     def 					a__	(  self				)						->							str:
           return 32

     @property
     def 					a__	(  self				)						->							Union[str, Any]:
           return 32

     @property
     def 					a__	(  self				)						->							str:
           return self.time_input_dim * 4

     @property
     def 					a__	(  self				)						->							Optional[Any]:
           return 8

     @property
     def 					a__	(  self				)						->							Dict:
           _A							:     List[Any]			=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""				)
           return tokenizer

     @property
     def 					a__	(  self				)						->							Dict:
           torch.manual_seed(0				)
           _A							:     Optional[int]			=						CLIPTextConfig(
               bos_token_id=0		,     eos_token_id=2		,     hidden_size=self.text_embedder_hidden_size		,     projection_dim=self.text_embedder_hidden_size		,     intermediate_size=37		,     layer_norm_eps=1e-05		,     num_attention_heads=4		,     num_hidden_layers=5		,     pad_token_id=1		,     vocab_size=1000		,     )
           return CLIPTextModelWithProjection(lowerCAmelCase_				)

     @property
     def 					a__	(  self				)						->							Tuple:
           torch.manual_seed(0				)

           _A							:     List[str]			=						{
               """num_attention_heads""": 2,
               """attention_head_dim""": 16,
               """embedding_dim""": self.time_input_dim,
               """num_embeddings""": 32,
               """embedding_proj_dim""": self.text_embedder_hidden_size,
               """time_embed_dim""": self.time_embed_dim,
               """num_layers""": 1,
               """clip_embed_dim""": self.time_input_dim * 2,
               """additional_embeddings""": 0,
               """time_embed_act_fn""": """gelu""",
               """norm_in_type""": """layer""",
               """encoder_hid_proj_type""": None,
               """added_emb_type""": None,
           }

           _A							:     str			=						PriorTransformer(**lowerCAmelCase_				)
           return model

     @property
     def 					a__	(  self				)						->							str:
           torch.manual_seed(0				)

           _A							:     List[Any]			=						{
               """param_shapes""": (
                   (self.renderer_dim, 93),
                   (self.renderer_dim, 8),
                   (self.renderer_dim, 8),
                   (self.renderer_dim, 8),
               ),
               """d_latent""": self.time_input_dim,
               """d_hidden""": self.renderer_dim,
               """n_output""": 12,
               """background""": (
                   0.1,
                   0.1,
                   0.1,
               ),
           }
           _A							:     str			=						ShapERenderer(**lowerCAmelCase_				)
           return model

     def 					a__	(  self				)						->							Union[str, Any]:
           _A							:     Union[str, Any]			=						self.dummy_prior
           _A							:     int			=						self.dummy_text_encoder
           _A							:     Any			=						self.dummy_tokenizer
           _A							:     Any			=						self.dummy_renderer

           _A							:     Union[str, Any]			=						HeunDiscreteScheduler(
               beta_schedule="""exp"""		,     num_train_timesteps=1024		,     prediction_type="""sample"""		,     use_karras_sigmas=lowerCAmelCase_		,     clip_sample=lowerCAmelCase_		,     clip_sample_range=1.0		,     )
           _A							:     Union[str, Any]			=						{
               """prior""": prior,
               """text_encoder""": text_encoder,
               """tokenizer""": tokenizer,
               """renderer""": renderer,
               """scheduler""": scheduler,
           }

           return components

     def 					a__	(  self		,     _a		,     _a=0				)						->							str:
           if str(lowerCAmelCase_				).startswith("""mps"""				):
                 _A							:     Dict			=						torch.manual_seed(lowerCAmelCase_				)
           else:
                 _A							:     List[str]			=						torch.Generator(device=lowerCAmelCase_				).manual_seed(lowerCAmelCase_				)
           _A							:     Optional[Any]			=						{
               """prompt""": """horse""",
               """generator""": generator,
               """num_inference_steps""": 1,
               """frame_size""": 32,
               """output_type""": """np""",
           }
           return inputs

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Optional[Any]			=						"""cpu"""

           _A							:     int			=						self.get_dummy_components()

           _A							:     List[str]			=						self.pipeline_class(**lowerCAmelCase_				)
           _A							:     int			=						pipe.to(lowerCAmelCase_				)

           pipe.set_progress_bar_config(disable=lowerCAmelCase_				)

           _A							:     Any			=						pipe(**self.get_dummy_inputs(lowerCAmelCase_				)				)
           _A							:     Any			=						output.images[0]
           _A							:     Optional[int]			=						image[0, -3:, -3:, -1]

           assert image.shape == (20, 32, 32, 3)

           _A							:     Optional[int]			=						np.array(
               [
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
               ]				)

           assert np.abs(image_slice.flatten() - expected_slice				).max() < 1e-2

     def 					a__	(  self				)						->							str:
           # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
           self._test_inference_batch_consistent(batch_sizes=[1, 2]				)

     def 					a__	(  self				)						->							int:
           _A							:     str			=						torch_device == """cpu"""
           _A							:     Any			=						True

           self._test_inference_batch_single_identical(
               batch_size=2		,     test_max_difference=lowerCAmelCase_		,     relax_max_difference=lowerCAmelCase_		,     )



     def 					a__	(  self				)						->							int:
           _A							:     Optional[Any]			=						self.get_dummy_components()
           _A							:     Dict			=						self.pipeline_class(**lowerCAmelCase_				)
           _A							:     Dict			=						pipe.to(lowerCAmelCase_				)
           pipe.set_progress_bar_config(disable=lowerCAmelCase_				)

           _A							:     List[Any]			=						1
           _A							:     List[Any]			=						2

           _A							:     Optional[Any]			=						self.get_dummy_inputs(lowerCAmelCase_				)

           for key in inputs.keys():
                 if key in self.batch_params:
                       _A							:     List[Any]			=						batch_size * [inputs[key]]

           _A							:     Dict			=						pipe(**lowerCAmelCase_		,     num_images_per_prompt=lowerCAmelCase_				)[0]

           assert images.shape[0] == batch_size * num_images_per_prompt







@slow
@require_torch_gpu
class 						lowercase							(   unittest.TestCase		):

     def 					a__	(  self				)						->							Optional[int]:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()



     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     List[Any]			=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
               """/shap_e/test_shap_e_np_out.npy"""				)
           _A							:     str			=						ShapEPipeline.from_pretrained("""openai/shap-e"""				)
           _A							:     List[Any]			=						pipe.to(lowerCAmelCase_				)
           pipe.set_progress_bar_config(disable=lowerCAmelCase_				)

           _A							:     Tuple			=						torch.Generator(device=lowerCAmelCase_				).manual_seed(0				)

           _A							:     Tuple			=						pipe(
               """a shark"""		,     generator=lowerCAmelCase_		,     guidance_scale=15.0		,     num_inference_steps=64		,     frame_size=64		,     output_type="""np"""		,     ).images[0]

           assert images.shape == (20, 64, 64, 3)

           assert_mean_pixel_difference(lowerCAmelCase_		,     lowerCAmelCase_				)


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  snake_case  (       snake_case__			:Union[str, Any]   ,	snake_case__			:Dict)						->   Any:
       _A        =  []
       for part_id in partition_order:
              _A        =  df.where(F'''SPARK_PARTITION_ID() = {part_id}''').collect()
              for row_idx, row in enumerate(snake_case__):
                     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  snake_case  (       )						->   Optional[Any]:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(100).repartition(1)
       _A        =  Spark(snake_case__)
       # 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  snake_case  (       )						->   Union[str, Any]:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(10).repartition(2)
       _A        =  [1, 0]
       _A        =  _generate_iterable_examples(snake_case__   ,	snake_case__)  # Reverse the partitions.
       _A        =  _get_expected_row_ids_and_row_dicts_for_partition_order(snake_case__   ,	snake_case__)

       for i, (row_id, row_dict) in enumerate(generate_fn()):
              _A   ,  _A        =  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  snake_case  (       )						->   int:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(10).repartition(1)
       _A        =  SparkExamplesIterable(snake_case__)
       assert it.n_shards == 1
       for i, (row_id, row_dict) in enumerate(snake_case__):
              assert row_id == F'''0_{i}'''
              assert row_dict == {"id": i}






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

              _A        =  SparkExamplesIterable(snake_case__).shuffle_data_sources(snake_case__)
              assert shuffled_it.n_shards == 3
              for i, (row_id, row_dict) in enumerate(snake_case__):
                     _A   ,  _A        =  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  snake_case  (       )						->   List[str]:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(20).repartition(4)

       # Partitions 0 and 2
       _A        =  SparkExamplesIterable(snake_case__).shard_data_sources(worker_id=0   ,	num_workers=2)
       assert shard_it_a.n_shards == 2
       _A        =  _get_expected_row_ids_and_row_dicts_for_partition_order(snake_case__   ,	[0, 2])
       for i, (row_id, row_dict) in enumerate(snake_case__):
              _A   ,  _A        =  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        =  SparkExamplesIterable(snake_case__).shard_data_sources(worker_id=1   ,	num_workers=2)
       assert shard_it_a.n_shards == 2
       _A        =  _get_expected_row_ids_and_row_dicts_for_partition_order(snake_case__   ,	[1, 3])
       for i, (row_id, row_dict) in enumerate(snake_case__):
              _A   ,  _A        =  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  snake_case  (       )						->   Tuple:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(100).repartition(1)
       _A        =  Spark(snake_case__)
       # 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




