code: 
  
  
  '''simple docstring'''
  
  
  
  
  
  from typing import Callable, List, Optional, Union
  
  import torch
  
  from ...models import UNetaDModel
  from ...schedulers import CMStochasticIterativeScheduler
  from ...utils import (
      is_accelerate_available,
      is_accelerate_version,
      logging,
      randn_tensor,
      replace_example_docstring,
  )
  from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput
  
  
  UpperCAmelCase__      =							logging.get_logger(__name__)  # pylint: disable=invalid-name
  
  
  UpperCAmelCase__      =							'''
      Examples:
          ```py
          >>> import torch
  
          >>> from diffusers import ConsistencyModelPipeline
  
          >>> device = "cuda"
          >>> # Load the cd_imagenet64_l2 checkpoint.
          >>> model_id_or_path = "openai/diffusers-cd_imagenet64_l2"
          >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)
          >>> pipe.to(device)
  
          >>> # Onestep Sampling
          >>> image = pipe(num_inference_steps=1).images[0]
          >>> image.save("cd_imagenet64_l2_onestep_sample.png")
  
          >>> # Onestep sampling, class-conditional image generation
          >>> # ImageNet-64 class label 145 corresponds to king penguins
          >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]
          >>> image.save("cd_imagenet64_l2_onestep_sample_penguin.png")
  
          >>> # Multistep sampling, class-conditional image generation
          >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:
          >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77
          >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]
          >>> image.save("cd_imagenet64_l2_multistep_sample_penguin.png")
          ```
  '''
  class 	a__				(				a_      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			def __init__(     self       : Any ,						lowerCAmelCase_       : UNetaDModel ,						lowerCAmelCase_       : CMStochasticIterativeScheduler  )			->		None:
  						super().__init__()
  
  						self.register_modules(
  						    unet=lowerCAmelCase_ ,						scheduler=lowerCAmelCase_ ,						)
  
  						__A=  None
  			def 				lowerCAmelCase    (     self       : Dict ,						lowerCAmelCase_       : Any=0  )			->		Dict:
  						if is_accelerate_available() and is_accelerate_version('>=' ,						'0.14.0'  ):
  									from accelerate import cpu_offload
  						else:
  									raise ImportError('`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher'  )
  
  						__A=  torch.device(F"""cuda:{gpu_id}"""  )
  
  						if self.device.type != "cpu":
  									self.to('cpu' ,						silence_dtype_warnings=lowerCAmelCase_  )
  									torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)
  
  						for cpu_offloaded_model in [self.unet]:
  									cpu_offload(lowerCAmelCase_ ,						lowerCAmelCase_  )
  
  						if self.safety_checker is not None:
  									cpu_offload(self.safety_checker ,						execution_device=lowerCAmelCase_ ,						offload_buffers=lowerCAmelCase_  )
  			def 				lowerCAmelCase    (     self       : List[str] ,						lowerCAmelCase_       : List[Any]=0  )			->		str:
  						if is_accelerate_available() and is_accelerate_version('>=' ,						'0.17.0.dev0'  ):
  									from accelerate import cpu_offload_with_hook
  						else:
  									raise ImportError('`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'  )
  
  						__A=  torch.device(F"""cuda:{gpu_id}"""  )
  
  						if self.device.type != "cpu":
  									self.to('cpu' ,						silence_dtype_warnings=lowerCAmelCase_  )
  									torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)
  
  						__A=  None
  						for cpu_offloaded_model in [self.unet]:
  									__A,  __A=  cpu_offload_with_hook(lowerCAmelCase_ ,						lowerCAmelCase_ ,						prev_module_hook=lowerCAmelCase_  )
  
  						if self.safety_checker is not None:
  									__A,  __A=  cpu_offload_with_hook(self.safety_checker ,						lowerCAmelCase_ ,						prev_module_hook=lowerCAmelCase_  )
  
  						# We'll offload the last model manually.
  						__A=  hook
  			@property
  			# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  			def 				lowerCAmelCase    (     self       : Optional[Any]  )			->		Any:
  						if not hasattr(self.unet ,						'_hf_hook'  ):
  									return self.device
  						for module in self.unet.modules():
  									if (
  									    hasattr(lowerCAmelCase_ ,						'_hf_hook'  )
  									    and hasattr(module._hf_hook ,						'execution_device'  )
  									    and module._hf_hook.execution_device is not None
  									):
  												return torch.device(module._hf_hook.execution_device  )
  						return self.device
  			def 				lowerCAmelCase    (     self       : Optional[int] ,						lowerCAmelCase_       : Any ,						lowerCAmelCase_       : List[Any] ,						lowerCAmelCase_       : int ,						lowerCAmelCase_       : Optional[Any] ,						lowerCAmelCase_       : Dict ,						lowerCAmelCase_       : str ,						lowerCAmelCase_       : List[Any] ,						lowerCAmelCase_       : Optional[int]=None  )			->		Dict:
  						__A=  (batch_size, num_channels, height, width)
  						if isinstance(lowerCAmelCase_ ,						lowerCAmelCase_  ) and len(lowerCAmelCase_  ) != batch_size:
  									raise ValueError(
  									    F"""You have passed a list of generators of length {len(lowerCAmelCase_  )}, but requested an effective batch"""
  									    F""" size of {batch_size}. Make sure the batch size matches the length of the generators."""  )
  
  						if latents is None:
  									__A=  randn_tensor(lowerCAmelCase_ ,						generator=lowerCAmelCase_ ,						device=lowerCAmelCase_ ,						dtype=lowerCAmelCase_  )
  						else:
  									__A=  latents.to(device=lowerCAmelCase_ ,						dtype=lowerCAmelCase_  )
  
  						# scale the initial noise by the standard deviation required by the scheduler
  						__A=  latents * self.scheduler.init_noise_sigma
  						return latents
  			def 				lowerCAmelCase    (     self       : Any ,						lowerCAmelCase_       : torch.FloatTensor ,						lowerCAmelCase_       : str = "pil"  )			->		Any:
  						if output_type not in ["pt", "np", "pil"]:
  									raise ValueError(
  									    F"""output_type={output_type} is not supported. Make sure to choose one of ['pt', 'np', or 'pil']"""  )
  
  						# Equivalent to diffusers.VaeImageProcessor.denormalize
  						__A=  (sample / 2 + 0.5).clamp(0 ,						1  )
  						if output_type == "pt":
  									return sample
  
  						# Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
  						__A=  sample.cpu().permute(0 ,						2 ,						3 ,						1  ).numpy()
  						if output_type == "np":
  									return sample
  
  						# Output_type must be 'pil'
  						__A=  self.numpy_to_pil(lowerCAmelCase_  )
  						return sample
  			def 				lowerCAmelCase    (     self       : Optional[int] ,						lowerCAmelCase_       : Optional[int] ,						lowerCAmelCase_       : str ,						lowerCAmelCase_       : Tuple=None  )			->		Tuple:
  						if self.unet.config.num_class_embeds is not None:
  									if isinstance(lowerCAmelCase_ ,						lowerCAmelCase_  ):
  												__A=  torch.tensor(lowerCAmelCase_ ,						dtype=torch.int  )
  									elif isinstance(lowerCAmelCase_ ,						lowerCAmelCase_  ):
  												assert batch_size == 1, "Batch size must be 1 if classes is an int"
  												__A=  torch.tensor([class_labels] ,						dtype=torch.int  )
  									elif class_labels is None:
  												# Randomly generate batch_size class labels
  												# TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
  												__A=  torch.randint(0 ,						self.unet.config.num_class_embeds ,						size=(batch_size,)  )
  									__A=  class_labels.to(lowerCAmelCase_  )
  						else:
  									__A=  None
  						return class_labels
  			def 				lowerCAmelCase    (     self       : int ,						lowerCAmelCase_       : List[Any] ,						lowerCAmelCase_       : int ,						lowerCAmelCase_       : int ,						lowerCAmelCase_       : Dict ,						lowerCAmelCase_       : Any ,						lowerCAmelCase_       : Any  )			->		str:
  						if num_inference_steps is None and timesteps is None:
  									raise ValueError('Exactly one of `num_inference_steps` or `timesteps` must be supplied.'  )
  
  						if num_inference_steps is not None and timesteps is not None:
  									logger.warning(
  									    F"""Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;"""
  									    ' `timesteps` will be used over `num_inference_steps`.'  )
  
  						if latents is not None:
  									__A=  (batch_size, 3, img_size, img_size)
  									if latents.shape != expected_shape:
  												raise ValueError(F"""The shape of latents is {latents.shape} but is expected to be {expected_shape}."""  )
  
  						if (callback_steps is None) or (
  						    callback_steps is not None and (not isinstance(lowerCAmelCase_ ,						lowerCAmelCase_  ) or callback_steps <= 0)
  						):
  									raise ValueError(
  									    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
  									    F""" {type(lowerCAmelCase_  )}."""  )
  
  			@torch.no_grad()
  			@replace_example_docstring(lowerCAmelCase_  )
  			def __call__(     self       : Union[str, Any] ,						lowerCAmelCase_       : int = 1 ,						lowerCAmelCase_       : Optional[Union[torch.Tensor, List[int], int]] = None ,						lowerCAmelCase_       : int = 1 ,						lowerCAmelCase_       : List[int] = None ,						lowerCAmelCase_       : Optional[Union[torch.Generator, List[torch.Generator]]] = None ,						lowerCAmelCase_       : Optional[torch.FloatTensor] = None ,						lowerCAmelCase_       : Optional[str] = "pil" ,						lowerCAmelCase_       : bool = True ,						lowerCAmelCase_       : Optional[Callable[[int, int, torch.FloatTensor], None]] = None ,						lowerCAmelCase_       : int = 1 ,						)			->		str:
  						__A=  self.unet.config.sample_size
  						__A=  self._execution_device
  
  						# 1. Check inputs
  						self.check_inputs(lowerCAmelCase_ ,						lowerCAmelCase_ ,						lowerCAmelCase_ ,						lowerCAmelCase_ ,						lowerCAmelCase_ ,						lowerCAmelCase_  )
  
  						# 2. Prepare image latents
  						# Sample image latents x_0 ~ N(0, sigma_0^2 * I)
  						__A=  self.prepare_latents(
  						    batch_size=lowerCAmelCase_ ,						num_channels=self.unet.config.in_channels ,						height=lowerCAmelCase_ ,						width=lowerCAmelCase_ ,						dtype=self.unet.dtype ,						device=lowerCAmelCase_ ,						generator=lowerCAmelCase_ ,						latents=lowerCAmelCase_ ,						)
  
  						# 3. Handle class_labels for class-conditional models
  						__A=  self.prepare_class_labels(lowerCAmelCase_ ,						lowerCAmelCase_ ,						class_labels=lowerCAmelCase_  )
  
  						# 4. Prepare timesteps
  						if timesteps is not None:
  									self.scheduler.set_timesteps(timesteps=lowerCAmelCase_ ,						device=lowerCAmelCase_  )
  									__A=  self.scheduler.timesteps
  									__A=  len(lowerCAmelCase_  )
  						else:
  									self.scheduler.set_timesteps(lowerCAmelCase_  )
  									__A=  self.scheduler.timesteps
  
  						# 5. Denoising loop
  						# Multistep sampling: implements Algorithm 1 in the paper
  						with self.progress_bar(total=lowerCAmelCase_  ) as progress_bar:
  									for i, t in enumerate(lowerCAmelCase_  ):
  												__A=  self.scheduler.scale_model_input(lowerCAmelCase_ ,						lowerCAmelCase_  )
  												__A=  self.unet(lowerCAmelCase_ ,						lowerCAmelCase_ ,						class_labels=lowerCAmelCase_ ,						return_dict=lowerCAmelCase_  )[0]
  
  												__A=  self.scheduler.step(lowerCAmelCase_ ,						lowerCAmelCase_ ,						lowerCAmelCase_ ,						generator=lowerCAmelCase_  )[0]
  
  												# call the callback, if provided
  												progress_bar.update()
  												if callback is not None and i % callback_steps == 0:
  															callback(lowerCAmelCase_ ,						lowerCAmelCase_ ,						lowerCAmelCase_  )
  
          # 6. Post-process image sample
  						__A=  self.postprocess_image(lowerCAmelCase_ ,						output_type=lowerCAmelCase_  )
  
  						# Offload last model to CPU
  						if hasattr(self ,						'final_offload_hook'  ) and self.final_offload_hook is not None:
  									self.final_offload_hook.offload()
  
  						if not return_dict:
  									return (image,)
  
  						return ImagePipelineOutput(images=lowerCAmelCase_  )
  
  
  
  
style_context: 
  
  
  '''simple docstring'''
  
  
  
  
  
  # This model implementation is heavily based on:
  
  import gc
  import random
  import tempfile
  import unittest
  
  import numpy as np
  import torch
  from PIL import Image
  from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer
  
  from diffusers import (
      AutoencoderKL,
      ControlNetModel,
      DDIMScheduler,
      StableDiffusionControlNetInpaintPipeline,
      UNetaDConditionModel,
  )
  from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
  from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
  from diffusers.utils.import_utils import is_xformers_available
  from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu
  
  from ..pipeline_params import (
      TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
      TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
      TEXT_TO_IMAGE_IMAGE_PARAMS,
  )
  from ..test_pipelines_common import (
      PipelineKarrasSchedulerTesterMixin,
      PipelineLatentTesterMixin,
      PipelineTesterMixin,
  )
  
  
  enable_full_determinism()
  class 	a__				(				a_			,     a_			,     a_			,     unittest.TestCase      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			A	:							Optional[int]							= StableDiffusionControlNetInpaintPipeline
  			A	:							List[str]							= TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
  			A	:							List[Any]							= TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  			A	:							Any							= frozenset({'''control_image'''}      )  # skip `image` and `mask` for now, only test for control_image
  			A	:							List[Any]							= TEXT_TO_IMAGE_IMAGE_PARAMS
  			def 				lowerCAmelCase    (     self       : Tuple  )			->		List[str]:
  						torch.manual_seed(0  )
  						__A=  UNetaDConditionModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						sample_size=32 ,						in_channels=9 ,						out_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D') ,						cross_attention_dim=32 ,						)
  						torch.manual_seed(0  )
  						__A=  ControlNetModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						in_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						cross_attention_dim=32 ,						conditioning_embedding_out_channels=(16, 32) ,						)
  						torch.manual_seed(0  )
  						__A=  DDIMScheduler(
  						    beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						beta_schedule='scaled_linear' ,						clip_sample=lowerCAmelCase_ ,						set_alpha_to_one=lowerCAmelCase_ ,						)
  						torch.manual_seed(0  )
  						__A=  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 ,						)
  						torch.manual_seed(0  )
  						__A=  CLIPTextConfig(
  						    bos_token_id=0 ,						eos_token_id=2 ,						hidden_size=32 ,						intermediate_size=37 ,						layer_norm_eps=1E-05 ,						num_attention_heads=4 ,						num_hidden_layers=5 ,						pad_token_id=1 ,						vocab_size=1_000 ,						)
  						__A=  CLIPTextModel(lowerCAmelCase_  )
  						__A=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )
  
  						__A=  {
  						    'unet': unet,
  						    'controlnet': controlnet,
  						    'scheduler': scheduler,
  						    'vae': vae,
  						    'text_encoder': text_encoder,
  						    'tokenizer': tokenizer,
  						    'safety_checker': None,
  						    'feature_extractor': None,
  						}
  						return components
  			def 				lowerCAmelCase    (     self       : List[str] ,						lowerCAmelCase_       : List[Any] ,						lowerCAmelCase_       : int=0  )			->		str:
  						if str(lowerCAmelCase_  ).startswith('mps'  ):
  									__A=  torch.manual_seed(lowerCAmelCase_  )
  						else:
  									__A=  torch.Generator(device=lowerCAmelCase_  ).manual_seed(lowerCAmelCase_  )
  
  						__A=  2
  						__A=  randn_tensor(
  						    (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor) ,						generator=lowerCAmelCase_ ,						device=torch.device(lowerCAmelCase_  ) ,						)
  						__A=  floats_tensor((1, 3, 32, 32) ,						rng=random.Random(lowerCAmelCase_  )  ).to(lowerCAmelCase_  )
  						__A=  init_image.cpu().permute(0 ,						2 ,						3 ,						1  )[0]
  
  						__A=  Image.fromarray(np.uinta(lowerCAmelCase_  )  ).convert('RGB'  ).resize((64, 64)  )
  						__A=  Image.fromarray(np.uinta(init_image + 4  )  ).convert('RGB'  ).resize((64, 64)  )
  
  						__A=  {
  						    'prompt': 'A painting of a squirrel eating a burger',
  						    'generator': generator,
  						    'num_inference_steps': 2,
  						    'guidance_scale': 6.0,
  						    'output_type': 'numpy',
  						    'image': image,
  						    'mask_image': mask_image,
  						    'control_image': control_image,
  						}
  
  						return inputs
  			def 				lowerCAmelCase    (     self       : Tuple  )			->		Optional[int]:
  						return self._test_attention_slicing_forward_pass(expected_max_diff=2E-3  )
  			@unittest.skipIf(
  			    torch_device != 'cuda' or not is_xformers_available() ,						reason='XFormers attention is only available with CUDA and `xformers` installed' ,						)
  			def 				lowerCAmelCase    (     self       : Optional[int]  )			->		str:
  						self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3  )
  
  			def 				lowerCAmelCase    (     self       : List[Any]  )			->		str:
  						self._test_inference_batch_single_identical(expected_max_diff=2E-3  )
  
  
  class 	a__				(				a_      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			A	:							int							= StableDiffusionControlNetInpaintPipeline
  			A	:							List[str]							= TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
  			A	:							Tuple							= TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  			A	:							int							= frozenset([]      )
  			def 				lowerCAmelCase    (     self       : int  )			->		List[Any]:
  						torch.manual_seed(0  )
  						__A=  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 ,						)
  						torch.manual_seed(0  )
  						__A=  ControlNetModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						in_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						cross_attention_dim=32 ,						conditioning_embedding_out_channels=(16, 32) ,						)
  						torch.manual_seed(0  )
  						__A=  DDIMScheduler(
  						    beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						beta_schedule='scaled_linear' ,						clip_sample=lowerCAmelCase_ ,						set_alpha_to_one=lowerCAmelCase_ ,						)
  						torch.manual_seed(0  )
  						__A=  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 ,						)
  						torch.manual_seed(0  )
  						__A=  CLIPTextConfig(
  						    bos_token_id=0 ,						eos_token_id=2 ,						hidden_size=32 ,						intermediate_size=37 ,						layer_norm_eps=1E-05 ,						num_attention_heads=4 ,						num_hidden_layers=5 ,						pad_token_id=1 ,						vocab_size=1_000 ,						)
  						__A=  CLIPTextModel(lowerCAmelCase_  )
  						__A=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )
  
  						__A=  {
  						    'unet': unet,
  						    'controlnet': controlnet,
  						    'scheduler': scheduler,
  						    'vae': vae,
  						    'text_encoder': text_encoder,
  						    'tokenizer': tokenizer,
  						    'safety_checker': None,
  						    'feature_extractor': None,
  						}
  						return components
  
  
  class 	a__				(				a_			,     a_			,     unittest.TestCase      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			A	:							Tuple							= StableDiffusionControlNetInpaintPipeline
  			A	:							int							= TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
  			A	:							Optional[int]							= TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  			def 				lowerCAmelCase    (     self       : List[Any]  )			->		int:
  						torch.manual_seed(0  )
  						__A=  UNetaDConditionModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						sample_size=32 ,						in_channels=9 ,						out_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D') ,						cross_attention_dim=32 ,						)
  						torch.manual_seed(0  )
  
  						def init_weights(lowerCAmelCase_       : Optional[int]  ):
  									if isinstance(lowerCAmelCase_ ,						torch.nn.Convad  ):
  												torch.nn.init.normal(m.weight  )
  												m.bias.data.fill_(1.0  )
  
  						__A=  ControlNetModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						in_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						cross_attention_dim=32 ,						conditioning_embedding_out_channels=(16, 32) ,						)
  						controlneta.controlnet_down_blocks.apply(lowerCAmelCase_  )
  
  						torch.manual_seed(0  )
  						__A=  ControlNetModel(
  						    block_out_channels=(32, 64) ,						layers_per_block=2 ,						in_channels=4 ,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D') ,						cross_attention_dim=32 ,						conditioning_embedding_out_channels=(16, 32) ,						)
  						controlneta.controlnet_down_blocks.apply(lowerCAmelCase_  )
  
  						torch.manual_seed(0  )
  						__A=  DDIMScheduler(
  						    beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						beta_schedule='scaled_linear' ,						clip_sample=lowerCAmelCase_ ,						set_alpha_to_one=lowerCAmelCase_ ,						)
  						torch.manual_seed(0  )
  						__A=  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 ,						)
  						torch.manual_seed(0  )
  						__A=  CLIPTextConfig(
  						    bos_token_id=0 ,						eos_token_id=2 ,						hidden_size=32 ,						intermediate_size=37 ,						layer_norm_eps=1E-05 ,						num_attention_heads=4 ,						num_hidden_layers=5 ,						pad_token_id=1 ,						vocab_size=1_000 ,						)
  						__A=  CLIPTextModel(lowerCAmelCase_  )
  						__A=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )
  
  						__A=  MultiControlNetModel([controlneta, controlneta]  )
  
  						__A=  {
  						    'unet': unet,
  						    'controlnet': controlnet,
  						    'scheduler': scheduler,
  						    'vae': vae,
  						    'text_encoder': text_encoder,
  						    'tokenizer': tokenizer,
  						    'safety_checker': None,
  						    'feature_extractor': None,
  						}
  						return components
  			def 				lowerCAmelCase    (     self       : Union[str, Any] ,						lowerCAmelCase_       : str ,						lowerCAmelCase_       : List[Any]=0  )			->		List[Any]:
  						if str(lowerCAmelCase_  ).startswith('mps'  ):
  									__A=  torch.manual_seed(lowerCAmelCase_  )
  						else:
  									__A=  torch.Generator(device=lowerCAmelCase_  ).manual_seed(lowerCAmelCase_  )
  
  						__A=  2
  
  						__A=  [
  						    randn_tensor(
  						        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor) ,						generator=lowerCAmelCase_ ,						device=torch.device(lowerCAmelCase_  ) ,						),
  						    randn_tensor(
  						        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor) ,						generator=lowerCAmelCase_ ,						device=torch.device(lowerCAmelCase_  ) ,						),
  						]
  						__A=  floats_tensor((1, 3, 32, 32) ,						rng=random.Random(lowerCAmelCase_  )  ).to(lowerCAmelCase_  )
  						__A=  init_image.cpu().permute(0 ,						2 ,						3 ,						1  )[0]
  
  						__A=  Image.fromarray(np.uinta(lowerCAmelCase_  )  ).convert('RGB'  ).resize((64, 64)  )
  						__A=  Image.fromarray(np.uinta(init_image + 4  )  ).convert('RGB'  ).resize((64, 64)  )
  
  						__A=  {
  						    'prompt': 'A painting of a squirrel eating a burger',
  						    'generator': generator,
  						    'num_inference_steps': 2,
  						    'guidance_scale': 6.0,
  						    'output_type': 'numpy',
  						    'image': image,
  						    'mask_image': mask_image,
  						    'control_image': control_image,
  						}
  
  						return inputs
  			def 				lowerCAmelCase    (     self       : Union[str, Any]  )			->		Any:
  						__A=  self.get_dummy_components()
  						__A=  self.pipeline_class(**lowerCAmelCase_  )
  						pipe.to(lowerCAmelCase_  )
  
  						__A=  10.0
  						__A=  4
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_  )[0]
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_ ,						control_guidance_start=0.1 ,						control_guidance_end=0.2  )[0]
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_ ,						control_guidance_start=[0.1, 0.3] ,						control_guidance_end=[0.2, 0.7]  )[0]
  
  						__A=  self.get_dummy_inputs(lowerCAmelCase_  )
  						__A=  steps
  						__A=  scale
  						__A=  pipe(**lowerCAmelCase_ ,						control_guidance_start=0.4 ,						control_guidance_end=[0.5, 0.8]  )[0]
  
  						# make sure that all outputs are different
  						assert np.sum(np.abs(output_a - output_a  )  ) > 1E-3
  						assert np.sum(np.abs(output_a - output_a  )  ) > 1E-3
  						assert np.sum(np.abs(output_a - output_a  )  ) > 1E-3
  			def 				lowerCAmelCase    (     self       : str  )			->		Any:
  						return self._test_attention_slicing_forward_pass(expected_max_diff=2E-3  )
  			@unittest.skipIf(
  			    torch_device != 'cuda' or not is_xformers_available() ,						reason='XFormers attention is only available with CUDA and `xformers` installed' ,						)
  			def 				lowerCAmelCase    (     self       : Any  )			->		int:
  						self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3  )
  			def 				lowerCAmelCase    (     self       : Dict  )			->		Dict:
  						self._test_inference_batch_single_identical(expected_max_diff=2E-3  )
  
  			def 				lowerCAmelCase    (     self       : Union[str, Any]  )			->		Optional[int]:
  						__A=  self.get_dummy_components()
  						__A=  self.pipeline_class(**lowerCAmelCase_  )
  						pipe.to(lowerCAmelCase_  )
  						pipe.set_progress_bar_config(disable=lowerCAmelCase_  )
  						with tempfile.TemporaryDirectory() as tmpdir:
  									try:
  												# save_pretrained is not implemented for Multi-ControlNet
  												pipe.save_pretrained(lowerCAmelCase_  )
  									except NotImplementedError:
  												pass
  
  
  @slow
  @require_torch_gpu
  class 	a__				(				unittest.TestCase      ):
  
  
  
  
  
  			'''simple docstring'''
  
  			def 				lowerCAmelCase    (     self       : Optional[int]  )			->		Tuple:
  						super().tearDown()
  						gc.collect()
  						torch.cuda.empty_cache()
  			def 				lowerCAmelCase    (     self       : Union[str, Any]  )			->		Tuple:
  						__A=  ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'  )
  
  						__A=  StableDiffusionControlNetInpaintPipeline.from_pretrained(
  						    'runwayml/stable-diffusion-inpainting' ,						safety_checker=lowerCAmelCase_ ,						controlnet=lowerCAmelCase_  )
  						pipe.enable_model_cpu_offload()
  						pipe.set_progress_bar_config(disable=lowerCAmelCase_  )
  
  						__A=  torch.Generator(device='cpu'  ).manual_seed(0  )
  						__A=  load_image(
  						    'https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png'  ).resize((512, 512)  )
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
  						    '/stable_diffusion_inpaint/input_bench_mask.png'  ).resize((512, 512)  )
  
  						__A=  'pitch black hole'
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png'  ).resize((512, 512)  )
  
  						__A=  pipe(
  						    lowerCAmelCase_ ,						image=lowerCAmelCase_ ,						mask_image=lowerCAmelCase_ ,						control_image=lowerCAmelCase_ ,						generator=lowerCAmelCase_ ,						output_type='np' ,						num_inference_steps=3 ,						)
  
  						__A=  output.images[0]
  
  						assert image.shape == (512, 512, 3)
  
  						__A=  load_numpy(
  						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/inpaint.npy'  )
  
  						assert np.abs(expected_image - image  ).max() < 9E-2
  
  			def 				lowerCAmelCase    (     self       : str  )			->		Tuple:
  						__A=  ControlNetModel.from_pretrained('lllyasviel/control_v11p_sd15_inpaint'  )
  
  						__A=  StableDiffusionControlNetInpaintPipeline.from_pretrained(
  						    'runwayml/stable-diffusion-v1-5' ,						safety_checker=lowerCAmelCase_ ,						controlnet=lowerCAmelCase_  )
  						__A=  DDIMScheduler.from_config(pipe.scheduler.config  )
  						pipe.enable_model_cpu_offload()
  						pipe.set_progress_bar_config(disable=lowerCAmelCase_  )
  
  						__A=  torch.Generator(device='cpu'  ).manual_seed(33  )
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy.png'  )
  						__A=  init_image.resize((512, 512)  )
  
  						__A=  load_image(
  						    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy_mask.png'  )
  						__A=  mask_image.resize((512, 512)  )
  
  						__A=  'a handsome man with ray-ban sunglasses'
  
  						def make_inpaint_condition(lowerCAmelCase_       : int ,						lowerCAmelCase_       : str  ):
  									__A=  np.array(image.convert('RGB'  )  ).astype(np.floataa  ) / 2_55.0
  									__A=  np.array(image_mask.convert('L'  )  ).astype(np.floataa  ) / 2_55.0
  
  									assert image.shape[0:1] == image_mask.shape[0:1], "image and image_mask must have the same image size"
  									__A=  -1.0  # set as masked pixel
  									__A=  np.expand_dims(lowerCAmelCase_ ,						0  ).transpose(0 ,						3 ,						1 ,						2  )
  									__A=  torch.from_numpy(lowerCAmelCase_  )
  									return image
  
  						__A=  make_inpaint_condition(lowerCAmelCase_ ,						lowerCAmelCase_  )
  
  						__A=  pipe(
  						    lowerCAmelCase_ ,						image=lowerCAmelCase_ ,						mask_image=lowerCAmelCase_ ,						control_image=lowerCAmelCase_ ,						guidance_scale=9.0 ,						eta=1.0 ,						generator=lowerCAmelCase_ ,						num_inference_steps=20 ,						output_type='np' ,						)
  						__A=  output.images[0]
  
  						assert image.shape == (512, 512, 3)
  
  						__A=  load_numpy(
  						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/boy_ray_ban.npy'  )
  
  						assert np.abs(expected_image - image  ).max() < 9E-2
  
  
  
  