






"""simple docstring"""


import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class 	__UpperCamelCase    (    UpperCamelCase      ,    UpperCamelCase      ,    unittest.TestCase	):
					lowercase_						:							List[Any]			=     StableDiffusionParadigmsPipeline
					lowercase_						:							Dict			=     TEXT_TO_IMAGE_PARAMS
					lowercase_						:							Optional[int]			=     TEXT_TO_IMAGE_BATCH_PARAMS
					lowercase_						:							Dict			=     TEXT_TO_IMAGE_IMAGE_PARAMS
					lowercase_						:							Optional[Any]			=     TEXT_TO_IMAGE_IMAGE_PARAMS
					def 					UpperCAmelCase__   (       self       :  Union[str, Any]      )  ->					Any:
											torch.manual_seed(0      )
											lowerCAmelCase      :int				  = UNetaDConditionModel(
											    block_out_channels=(32, 64)    ,   layers_per_block=2    ,   sample_size=32    ,   in_channels=4    ,   out_channels=4    ,   down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')    ,   up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')    ,   cross_attention_dim=32    ,   attention_head_dim=(2, 4)    ,   use_linear_projection=UpperCAmelCase    ,   )
											lowerCAmelCase      :Optional[Any]				  = DDIMParallelScheduler(
											    beta_start=0.0_0_0_8_5    ,   beta_end=0.0_1_2    ,   beta_schedule='scaled_linear'    ,   clip_sample=UpperCAmelCase    ,   set_alpha_to_one=UpperCAmelCase    ,   )
											torch.manual_seed(0      )
											lowerCAmelCase      :List[str]				  = AutoencoderKL(
											    block_out_channels=[32, 64]    ,   in_channels=3    ,   out_channels=3    ,   down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']    ,   up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']    ,   latent_channels=4    ,   sample_size=128    ,   )
											torch.manual_seed(0      )
											lowerCAmelCase      :str				  = CLIPTextConfig(
											    bos_token_id=0    ,   eos_token_id=2    ,   hidden_size=32    ,   intermediate_size=37    ,   layer_norm_eps=1e-0_5    ,   num_attention_heads=4    ,   num_hidden_layers=5    ,   pad_token_id=1    ,   vocab_size=1000    ,   hidden_act='gelu'    ,   projection_dim=512    ,   )
											lowerCAmelCase      :int				  = CLIPTextModel(UpperCAmelCase      )
											lowerCAmelCase      :Union[str, Any]				  = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )

											lowerCAmelCase      :Dict				  = {
											    'unet': unet,
											    'scheduler': scheduler,
											    'vae': vae,
											    'text_encoder': text_encoder,
											    'tokenizer': tokenizer,
											    'safety_checker': None,
											    'feature_extractor': None,
											}
											return components
					def 					UpperCAmelCase__   (       self       :  Union[str, Any]    ,   UpperCAmelCase       :  Optional[Any]    ,   UpperCAmelCase       :  Any=0      )  ->					str:
											if str(UpperCAmelCase      ).startswith('mps'      ):
																	lowerCAmelCase      :Optional[Any]				  = torch.manual_seed(UpperCAmelCase      )
											else:
																	lowerCAmelCase      :Any				  = torch.Generator(device=UpperCAmelCase      ).manual_seed(UpperCAmelCase      )
											lowerCAmelCase      :Optional[int]				  = {
											    'prompt': 'a photograph of an astronaut riding a horse',
											    'generator': generator,
											    'num_inference_steps': 10,
											    'guidance_scale': 6.0,
											    'output_type': 'numpy',
											    'parallel': 3,
											    'debug': True,
											}
											return inputs
					def 					UpperCAmelCase__   (       self       :  Optional[Any]      )  ->					Union[str, Any]:
											lowerCAmelCase      :Any				  = 'cpu'  # ensure determinism for the device-dependent torch.Generator
											lowerCAmelCase      :Dict				  = self.get_dummy_components()
											lowerCAmelCase      :Optional[int]				  = StableDiffusionParadigmsPipeline(**UpperCAmelCase      )
											lowerCAmelCase      :Tuple				  = sd_pipe.to(UpperCAmelCase      )
											sd_pipe.set_progress_bar_config(disable=UpperCAmelCase      )

											lowerCAmelCase      :Any				  = self.get_dummy_inputs(UpperCAmelCase      )
											lowerCAmelCase      :Optional[int]				  = sd_pipe(**UpperCAmelCase      ).images
											lowerCAmelCase      :List[Any]				  = image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											lowerCAmelCase      :Dict				  = np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]      )

											assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2
					def 					UpperCAmelCase__   (       self       :  Any      )  ->					int:
											lowerCAmelCase      :List[Any]				  = 'cpu'  # ensure determinism for the device-dependent torch.Generator
											lowerCAmelCase      :Dict				  = self.get_dummy_components()
											torch.manual_seed(0      )
											lowerCAmelCase      :Any				  = DDPMParallelScheduler()
											torch.manual_seed(0      )
											lowerCAmelCase      :Union[str, Any]				  = StableDiffusionParadigmsPipeline(**UpperCAmelCase      )
											lowerCAmelCase      :List[Any]				  = sd_pipe.to(UpperCAmelCase      )
											sd_pipe.set_progress_bar_config(disable=UpperCAmelCase      )

											lowerCAmelCase      :List[str]				  = self.get_dummy_inputs(UpperCAmelCase      )
											lowerCAmelCase      :int				  = sd_pipe(**UpperCAmelCase      ).images
											lowerCAmelCase      :List[Any]				  = image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											lowerCAmelCase      :List[str]				  = np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]      )

											assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2
					def 					UpperCAmelCase__   (       self       :  Union[str, Any]      )  ->					int:
											super().test_inference_batch_consistent(batch_sizes=[1, 2]      )
					def 					UpperCAmelCase__   (       self       :  List[Any]      )  ->					str:
											super().test_inference_batch_single_identical(batch_size=2    ,   expected_max_diff=3e-3      )
					def 					UpperCAmelCase__   (       self       :  Optional[Any]      )  ->					Any:
											lowerCAmelCase      :int				  = 'cpu'  # ensure determinism for the device-dependent torch.Generator
											lowerCAmelCase      :Dict				  = self.get_dummy_components()
											lowerCAmelCase      :Any				  = StableDiffusionParadigmsPipeline(**UpperCAmelCase      )
											lowerCAmelCase      :List[Any]				  = sd_pipe.to(UpperCAmelCase      )
											sd_pipe.set_progress_bar_config(disable=UpperCAmelCase      )

											lowerCAmelCase      :int				  = self.get_dummy_inputs(UpperCAmelCase      )
											lowerCAmelCase      :Dict				  = 'french fries'
											lowerCAmelCase      :str				  = sd_pipe(**UpperCAmelCase    ,   negative_prompt=UpperCAmelCase      )
											lowerCAmelCase      :int				  = output.images
											lowerCAmelCase      :Any				  = image[0, -3:, -3:, -1]

											assert image.shape == (1, 64, 64, 3)

											lowerCAmelCase      :List[Any]				  = np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]      )

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




@slow
@require_torch_gpu
class 	__UpperCamelCase    (    unittest.TestCase	):
					def 					UpperCAmelCase__   (       self       :  int      )  ->					str:
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()
					def 					UpperCAmelCase__   (       self       :  Tuple    ,   UpperCAmelCase       :  Optional[int]=0      )  ->					List[str]:
											lowerCAmelCase      :Any				  = torch.Generator(device=UpperCAmelCase      ).manual_seed(UpperCAmelCase      )
											lowerCAmelCase      :List[Any]				  = {
											    'prompt': 'a photograph of an astronaut riding a horse',
											    'generator': generator,
											    'num_inference_steps': 10,
											    'guidance_scale': 7.5,
											    'output_type': 'numpy',
											    'parallel': 3,
											    'debug': True,
											}
											return inputs
					def 					UpperCAmelCase__   (       self       :  List[Any]      )  ->					List[Any]:
											lowerCAmelCase      :str				  = 'stabilityai/stable-diffusion-2-base'
											lowerCAmelCase      :Optional[int]				  = DDIMParallelScheduler.from_pretrained(UpperCAmelCase    ,   subfolder='scheduler'      )
											lowerCAmelCase      :Any				  = StableDiffusionParadigmsPipeline.from_pretrained(UpperCAmelCase    ,   scheduler=UpperCAmelCase    ,   safety_checker=UpperCAmelCase      )
											pipe.to(UpperCAmelCase      )
											pipe.set_progress_bar_config(disable=UpperCAmelCase      )
											pipe.enable_attention_slicing()

											lowerCAmelCase      :List[str]				  = self.get_inputs()
											lowerCAmelCase      :Optional[int]				  = pipe(**UpperCAmelCase      ).images
											lowerCAmelCase      :List[Any]				  = image[0, -3:, -3:, -1].flatten()

											assert image.shape == (1, 512, 512, 3)

											lowerCAmelCase      :str				  = np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]      )

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












"""simple docstring"""


from .number_theory.prime_numbers import next_prime




class 	__UpperCamelCase    :
					def __init__(       self       :  List[str]    ,   UpperCAmelCase       :  int    ,   UpperCAmelCase       :  int | None = None    ,   UpperCAmelCase       :  float | None = None    ,   )  ->					None:
											lowerCAmelCase      :Tuple				  = size_table
											lowerCAmelCase      :List[Any]				  = [None] * self.size_table
											lowerCAmelCase      :Tuple				  = 0.7_5 if lim_charge is None else lim_charge
											lowerCAmelCase      :str				  = 1 if charge_factor is None else charge_factor
											lowerCAmelCase      :list				  = []
											lowerCAmelCase      :dict				  = {}
					def 					UpperCAmelCase__   (       self       :  Optional[int]      )  ->					Union[str, Any]:
											return self._keys
					def 					UpperCAmelCase__   (       self       :  Dict      )  ->					Optional[Any]:
											return sum(1 for slot in self.values if slot is not None      ) / (
											    self.size_table * self.charge_factor
											)
					def 					UpperCAmelCase__   (       self       :  Any    ,   UpperCAmelCase       :  List[str]      )  ->					int:
											return key % self.size_table
					def 					UpperCAmelCase__   (       self       :  Dict    ,   UpperCAmelCase       :  Union[str, Any]      )  ->					int:
											print(f"""step {step_ord}"""      )
											print(list(range(len(self.values      )      )      )      )
											print(self.values      )
					def 					UpperCAmelCase__   (       self       :  Any    ,   UpperCAmelCase       :  Dict      )  ->					Any:
											lowerCAmelCase      :List[str]				  = 1
											lowerCAmelCase      :Union[str, Any]				  = values
											for value in values:
																	self.insert_data(UpperCAmelCase      )
																	self._step_by_step(UpperCAmelCase      )
																	i += 1
					def 					UpperCAmelCase__   (       self       :  str    ,   UpperCAmelCase       :  Dict    ,   UpperCAmelCase       :  Union[str, Any]      )  ->					Optional[Any]:
											lowerCAmelCase      :List[Any]				  = data
											lowerCAmelCase      :str				  = data
					def 					UpperCAmelCase__   (       self       :  Optional[int]    ,   UpperCAmelCase       :  str    ,   UpperCAmelCase       :  str=None      )  ->					Tuple:
											lowerCAmelCase      :Optional[Any]				  = self.hash_function(key + 1      )

											while self.values[new_key] is not None and self.values[new_key] != key:
																	if self.values.count(UpperCAmelCase      ) > 0:
																							lowerCAmelCase      :Tuple				  = self.hash_function(new_key + 1      )
																	else:
																							lowerCAmelCase      :Union[str, Any]				  = None
																							break

											return new_key
					def 					UpperCAmelCase__   (       self       :  Any      )  ->					Any:
											lowerCAmelCase      :Optional[Any]				  = [value for value in self.values if value is not None]
											lowerCAmelCase      :List[str]				  = next_prime(self.size_table    ,   factor=2      )
											self._keys.clear()
											lowerCAmelCase      :int				  = [None] * self.size_table  # hell's pointers D: don't DRY ;/
											for value in survivor_values:
																	self.insert_data(UpperCAmelCase      )
					def 					UpperCAmelCase__   (       self       :  Optional[int]    ,   UpperCAmelCase       :  Tuple      )  ->					int:
											lowerCAmelCase      :Union[str, Any]				  = self.hash_function(UpperCAmelCase      )

											if self.values[key] is None:
																	self._set_value(UpperCAmelCase    ,   UpperCAmelCase      )

											elif self.values[key] == data:
																	pass

											else:
																	lowerCAmelCase      :Tuple				  = self._collision_resolution(UpperCAmelCase    ,   UpperCAmelCase      )
																	if collision_resolution is not None:
																							self._set_value(UpperCAmelCase    ,   UpperCAmelCase      )
																	else:
																							self.rehashing()
																							self.insert_data(UpperCAmelCase      )