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

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin






@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->UnCLIP
class   UpperCAmelCase		(						A_ ):
   A__					:		torch.FloatTensor
   A__					:		Optional[torch.FloatTensor]						     =     None



def 			UpperCamelCase					(    __lowerCamelCase     : Tuple				,					__lowerCamelCase     : Tuple=0.999				,					__lowerCamelCase     : Optional[Any]="cosine"				,					):
       if alpha_transform_type == "cosine":

              def alpha_bar_fn(__lowerCamelCase     : List[Any]    ):
                     return math.cos((t + 0.008) / 1.008 * math.pi / 2    ) ** 2

       elif alpha_transform_type == "exp":

              def alpha_bar_fn(__lowerCamelCase     : Any    ):
                     return math.exp(t * -12.0    )

       else:
              raise ValueError(f"""Unsupported alpha_tranform_type: {alpha_transform_type}"""    )

       snake_case       :   Optional[Any]							=      []
       for i in range(__lowerCamelCase    ):
              snake_case       :   List[str]							=      i / num_diffusion_timesteps
              snake_case       :   List[Any]							=      (i + 1) / num_diffusion_timesteps
              betas.append(min(1 - alpha_bar_fn(__lowerCamelCase    ) / alpha_bar_fn(__lowerCamelCase    )				,					__lowerCamelCase    )    )
       return torch.tensor(__lowerCamelCase				,					dtype=torch.floataa    )






class   UpperCAmelCase		(						A_       ,A_ ):



   @register_to_config
   def __init__(self		:					str    ,						snake_case__		:					int = 10_00    ,						snake_case__		:					str = "fixed_small_log"    ,						snake_case__		:					bool = True    ,						snake_case__		:					Optional[float] = 1.0    ,						snake_case__		:					str = "epsilon"    ,						snake_case__		:					str = "squaredcos_cap_v2"    ,						)		->	List[str]:


          '''simple docstring'''


          if beta_schedule != "squaredcos_cap_v2":
                 raise ValueError("UnCLIPScheduler only supports `beta_schedule`: 'squaredcos_cap_v2'" )

          snake_case       :   Union[str, Any]							=      betas_for_alpha_bar(snake_case__ )

          snake_case       :   str							=      1.0 - self.betas
          snake_case       :   Optional[Any]							=      torch.cumprod(self.alphas    ,						dim=0 )
          snake_case       :   str							=      torch.tensor(1.0 )

          # standard deviation of the initial noise distribution
          snake_case       :   Optional[Any]							=      1.0

          # setable values
          snake_case       :   Optional[Any]							=      None
          snake_case       :   List[str]							=      torch.from_numpy(np.arange(0    ,						snake_case__ )[::-1].copy() )

          snake_case       :   str							=      variance_type



   def 				_SCREAMING_SNAKE_CASE   (self		:					int    ,						snake_case__		:					torch.FloatTensor    ,						snake_case__		:					Optional[int] = None )		->	torch.FloatTensor:


          '''simple docstring'''


          return sample



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int]    ,						snake_case__		:					int    ,						snake_case__		:					Union[str, torch.device] = None )		->	int:


          '''simple docstring'''


          snake_case       :   int							=      num_inference_steps
          snake_case       :   Tuple							=      (self.config.num_train_timesteps - 1) / (self.num_inference_steps - 1)
          snake_case       :   Union[str, Any]							=      (np.arange(0    ,						snake_case__ ) * step_ratio).round()[::-1].copy().astype(np.intaa )
          snake_case       :   List[str]							=      torch.from_numpy(snake_case__ ).to(snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int]    ,						snake_case__		:					Dict    ,						snake_case__		:					Union[str, Any]=None    ,						snake_case__		:					Optional[Any]=None    ,						snake_case__		:					Tuple=None )		->	Union[str, Any]:


          '''simple docstring'''


          if prev_timestep is None:
                 snake_case       :   Optional[int]							=      t - 1

          snake_case       :   Union[str, Any]							=      self.alphas_cumprod[t]
          snake_case       :   int							=      self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
          snake_case       :   Union[str, Any]							=      1 - alpha_prod_t
          snake_case       :   int							=      1 - alpha_prod_t_prev

          if prev_timestep == t - 1:
                 snake_case       :   List[str]							=      self.betas[t]
          else:
                 snake_case       :   Any							=      1 - alpha_prod_t / alpha_prod_t_prev

          # For t > 0, compute predicted variance βt (see formula (6) and (7) from https://arxiv.org/pdf/2006.11239.pdf)
          # and sample from it to get previous sample
          # x_{t-1} ~ N(pred_prev_sample, variance) == add variance to pred_sample
          snake_case       :   str							=      beta_prod_t_prev / beta_prod_t * beta

          if variance_type is None:
                 snake_case       :   int							=      self.config.variance_type

          # hacks - were probably added for training stability
          if variance_type == "fixed_small_log":
                 snake_case       :   Union[str, Any]							=      torch.log(torch.clamp(snake_case__    ,						min=1e-20 ) )
                 snake_case       :   Any							=      torch.exp(0.5 * variance )
          elif variance_type == "learned_range":
                 # NOTE difference with DDPM scheduler
                 snake_case       :   Union[str, Any]							=      variance.log()
                 snake_case       :   Union[str, Any]							=      beta.log()

                 snake_case       :   Optional[Any]							=      (predicted_variance + 1) / 2
                 snake_case       :   Dict							=      frac * max_log + (1 - frac) * min_log

          return variance



   def 				_SCREAMING_SNAKE_CASE   (self		:					List[str]    ,						snake_case__		:					torch.FloatTensor    ,						snake_case__		:					int    ,						snake_case__		:					torch.FloatTensor    ,						snake_case__		:					Optional[int] = None    ,						snake_case__		:					int=None    ,						snake_case__		:					bool = True    ,						)		->	Union[UnCLIPSchedulerOutput, Tuple]:


          '''simple docstring'''


          snake_case       :   Tuple							=      timestep

          if model_output.shape[1] == sample.shape[1] * 2 and self.variance_type == "learned_range":
                 snake_case				,					snake_case       :   int							=      torch.split(snake_case__    ,						sample.shape[1]    ,						dim=1 )
          else:
                 snake_case       :   List[Any]							=      None

          # 1. compute alphas, betas
          if prev_timestep is None:
                 snake_case       :   Any							=      t - 1

          snake_case       :   int							=      self.alphas_cumprod[t]
          snake_case       :   Tuple							=      self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
          snake_case       :   str							=      1 - alpha_prod_t
          snake_case       :   Tuple							=      1 - alpha_prod_t_prev

          if prev_timestep == t - 1:
                 snake_case       :   int							=      self.betas[t]
                 snake_case       :   Union[str, Any]							=      self.alphas[t]
          else:
                 snake_case       :   str							=      1 - alpha_prod_t / alpha_prod_t_prev
                 snake_case       :   List[str]							=      1 - beta

          # 2. compute predicted original sample from predicted noise also called
          # "predicted x_0" of formula (15) from https://arxiv.org/pdf/2006.11239.pdf
          if self.config.prediction_type == "epsilon":
                 snake_case       :   Optional[int]							=      (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
          elif self.config.prediction_type == "sample":
                 snake_case       :   Any							=      model_output
          else:
                 raise ValueError(
                     f"""prediction_type given as {self.config.prediction_type} must be one of `epsilon` or `sample`"""
                     " for the UnCLIPScheduler." )

          # 3. Clip "predicted x_0"
          if self.config.clip_sample:
                 snake_case       :   List[Any]							=      torch.clamp(
                     snake_case__    ,						-self.config.clip_sample_range    ,						self.config.clip_sample_range )

          # 4. Compute coefficients for pred_original_sample x_0 and current sample x_t
          # See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
          snake_case       :   Union[str, Any]							=      (alpha_prod_t_prev ** 0.5 * beta) / beta_prod_t
          snake_case       :   str							=      alpha ** 0.5 * beta_prod_t_prev / beta_prod_t

          # 5. Compute predicted previous sample µ_t
          # See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
          snake_case       :   int							=      pred_original_sample_coeff * pred_original_sample + current_sample_coeff * sample

          # 6. Add noise
          snake_case       :   Tuple							=      0
          if t > 0:
                 snake_case       :   Optional[int]							=      randn_tensor(
                     model_output.shape    ,						dtype=model_output.dtype    ,						generator=snake_case__    ,						device=model_output.device )

                 snake_case       :   str							=      self._get_variance(
                     snake_case__    ,						predicted_variance=snake_case__    ,						prev_timestep=snake_case__    ,						)

                 if self.variance_type == "fixed_small_log":
                        snake_case       :   Optional[Any]							=      variance
                 elif self.variance_type == "learned_range":
                        snake_case       :   Optional[Any]							=      (0.5 * variance).exp()
                 else:
                        raise ValueError(
                            f"""variance_type given as {self.variance_type} must be one of `fixed_small_log` or `learned_range`"""
                            " for the UnCLIPScheduler." )

                 snake_case       :   Any							=      variance * variance_noise

          snake_case       :   Tuple							=      pred_prev_sample + variance

          if not return_dict:
                 return (pred_prev_sample,)

          return UnCLIPSchedulerOutput(prev_sample=snake_case__    ,						pred_original_sample=snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Dict    ,						snake_case__		:					torch.FloatTensor    ,						snake_case__		:					torch.FloatTensor    ,						snake_case__		:					torch.IntTensor    ,						)		->	torch.FloatTensor:


          '''simple docstring'''


          snake_case       :   Optional[Any]							=      self.alphas_cumprod.to(device=original_samples.device    ,						dtype=original_samples.dtype )
          snake_case       :   List[Any]							=      timesteps.to(original_samples.device )

          snake_case       :   Optional[int]							=      alphas_cumprod[timesteps] ** 0.5
          snake_case       :   List[Any]							=      sqrt_alpha_prod.flatten()
          while len(sqrt_alpha_prod.shape ) < len(original_samples.shape ):
                 snake_case       :   Tuple							=      sqrt_alpha_prod.unsqueeze(-1 )

          snake_case       :   List[Any]							=      (1 - alphas_cumprod[timesteps]) ** 0.5
          snake_case       :   str							=      sqrt_one_minus_alpha_prod.flatten()
          while len(sqrt_one_minus_alpha_prod.shape ) < len(original_samples.shape ):
                 snake_case       :   Optional[Any]							=      sqrt_one_minus_alpha_prod.unsqueeze(-1 )

          snake_case       :   List[Any]							=      sqrt_alpha_prod * original_samples + sqrt_one_minus_alpha_prod * noise
          return noisy_samples










'''simple docstring'''





# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

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,
    StableDiffusionControlNetImgaImgPipeline,
    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 (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   unittest.TestCase     ):
  """simple docstring"""




  a_												=StableDiffusionControlNetImgaImgPipeline
  a_												=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
  a_												=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  a_												=IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}     )
  a_												=IMAGE_TO_IMAGE_IMAGE_PARAMS




  def 						_lowercase     ( self      :	Union[str, Any]   )					->					List[str]:
    torch.manual_seed(0   )
    __lowerCamelCase       :       Tuple           = 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   )
    __lowerCamelCase       :       Any           = 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   )
    __lowerCamelCase       :       Any           = DDIMScheduler(
        beta_start=0.00085			,	beta_end=0.012			,	beta_schedule='scaled_linear'			,	clip_sample=_a			,	set_alpha_to_one=_a			,	)
    torch.manual_seed(0   )
    __lowerCamelCase       :       Dict           = 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   )
    __lowerCamelCase       :       Dict           = 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=1000			,	)
    __lowerCamelCase       :       Union[str, Any]           = CLIPTextModel(_a   )
    __lowerCamelCase       :       Any           = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

    __lowerCamelCase       :       int           = {
        'unet': unet,
        'controlnet': controlnet,
        'scheduler': scheduler,
        'vae': vae,
        'text_encoder': text_encoder,
        'tokenizer': tokenizer,
        'safety_checker': None,
        'feature_extractor': None,
    }
    return components




  def 						_lowercase     ( self      :	Union[str, Any]			,	_a      :	Any			,	_a      :	List[str]=0   )					->					List[str]:
    if str(_a   ).startswith('mps'   ):
      __lowerCamelCase       :       Any           = torch.manual_seed(_a   )
    else:
      __lowerCamelCase       :       Any           = torch.Generator(device=_a   ).manual_seed(_a   )

    __lowerCamelCase       :       Union[str, Any]           = 2
    __lowerCamelCase       :       Any           = randn_tensor(
        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)			,	generator=_a			,	device=torch.device(_a   )			,	)
    __lowerCamelCase       :       Optional[Any]           = floats_tensor(control_image.shape			,	rng=random.Random(_a   )   ).to(_a   )
    __lowerCamelCase       :       List[Any]           = image.cpu().permute(0			,	2			,	3			,	1   )[0]
    __lowerCamelCase       :       int           = Image.fromarray(np.uinta(_a   )   ).convert('RGB'   ).resize((64, 64)   )
    __lowerCamelCase       :       int           = {
        'prompt': 'A painting of a squirrel eating a burger',
        'generator': generator,
        'num_inference_steps': 2,
        'guidance_scale': 6.0,
        'output_type': 'numpy',
        'image': image,
        'control_image': control_image,
    }

    return inputs




  def 						_lowercase     ( self      :	List[str]   )					->					Optional[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 						_lowercase     ( self      :	str   )					->					List[str]:
    self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3   )




  def 						_lowercase     ( self      :	Any   )					->					List[str]:
    self._test_inference_batch_single_identical(expected_max_diff=2e-3   )






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   unittest.TestCase     ):
  """simple docstring"""




  a_												=StableDiffusionControlNetImgaImgPipeline
  a_												=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
  a_												=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  a_												=frozenset([]     )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess




  def 						_lowercase     ( self      :	List[Any]   )					->					List[str]:
    torch.manual_seed(0   )
    __lowerCamelCase       :       Any           = 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   )

    def init_weights(_a      :	str   ):
      if isinstance(_a			,	torch.nn.Convad   ):
        torch.nn.init.normal(m.weight   )
        m.bias.data.fill_(1.0   )

    __lowerCamelCase       :       Any           = 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(_a   )

    torch.manual_seed(0   )
    __lowerCamelCase       :       Any           = 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(_a   )

    torch.manual_seed(0   )
    __lowerCamelCase       :       List[Any]           = DDIMScheduler(
        beta_start=0.00085			,	beta_end=0.012			,	beta_schedule='scaled_linear'			,	clip_sample=_a			,	set_alpha_to_one=_a			,	)
    torch.manual_seed(0   )
    __lowerCamelCase       :       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			,	)
    torch.manual_seed(0   )
    __lowerCamelCase       :       Optional[int]           = 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=1000			,	)
    __lowerCamelCase       :       Dict           = CLIPTextModel(_a   )
    __lowerCamelCase       :       Optional[Any]           = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

    __lowerCamelCase       :       List[str]           = MultiControlNetModel([controlneta, controlneta]   )

    __lowerCamelCase       :       str           = {
        'unet': unet,
        'controlnet': controlnet,
        'scheduler': scheduler,
        'vae': vae,
        'text_encoder': text_encoder,
        'tokenizer': tokenizer,
        'safety_checker': None,
        'feature_extractor': None,
    }
    return components




  def 						_lowercase     ( self      :	Dict			,	_a      :	int			,	_a      :	Union[str, Any]=0   )					->					Union[str, Any]:
    if str(_a   ).startswith('mps'   ):
      __lowerCamelCase       :       Optional[Any]           = torch.manual_seed(_a   )
    else:
      __lowerCamelCase       :       Union[str, Any]           = torch.Generator(device=_a   ).manual_seed(_a   )

    __lowerCamelCase       :       Any           = 2

    __lowerCamelCase       :       Any           = [
        randn_tensor(
            (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)			,	generator=_a			,	device=torch.device(_a   )			,	),
        randn_tensor(
            (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)			,	generator=_a			,	device=torch.device(_a   )			,	),
    ]

    __lowerCamelCase       :       List[Any]           = floats_tensor(control_image[0].shape			,	rng=random.Random(_a   )   ).to(_a   )
    __lowerCamelCase       :       Union[str, Any]           = image.cpu().permute(0			,	2			,	3			,	1   )[0]
    __lowerCamelCase       :       Dict           = Image.fromarray(np.uinta(_a   )   ).convert('RGB'   ).resize((64, 64)   )
    __lowerCamelCase       :       Dict           = {
        'prompt': 'A painting of a squirrel eating a burger',
        'generator': generator,
        'num_inference_steps': 2,
        'guidance_scale': 6.0,
        'output_type': 'numpy',
        'image': image,
        'control_image': control_image,
    }

    return inputs




  def 						_lowercase     ( self      :	Tuple   )					->					Union[str, Any]:
    __lowerCamelCase       :       Any           = self.get_dummy_components()
    __lowerCamelCase       :       Any           = self.pipeline_class(**_a   )
    pipe.to(_a   )

    __lowerCamelCase       :       Optional[int]           = 10.0
    __lowerCamelCase       :       Tuple           = 4

    __lowerCamelCase       :       str           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       int           = steps
    __lowerCamelCase       :       List[Any]           = scale
    __lowerCamelCase       :       Optional[Any]           = pipe(**_a   )[0]

    __lowerCamelCase       :       Dict           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       List[str]           = steps
    __lowerCamelCase       :       List[Any]           = scale
    __lowerCamelCase       :       Dict           = pipe(**_a			,	control_guidance_start=0.1			,	control_guidance_end=0.2   )[0]

    __lowerCamelCase       :       Any           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       Union[str, Any]           = steps
    __lowerCamelCase       :       Any           = scale
    __lowerCamelCase       :       Optional[int]           = pipe(**_a			,	control_guidance_start=[0.1, 0.3]			,	control_guidance_end=[0.2, 0.7]   )[0]

    __lowerCamelCase       :       Any           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       Tuple           = steps
    __lowerCamelCase       :       List[Any]           = scale
    __lowerCamelCase       :       List[str]           = pipe(**_a			,	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 						_lowercase     ( self      :	Optional[Any]   )					->					List[str]:
    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 						_lowercase     ( self      :	Dict   )					->					Tuple:
    self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3   )




  def 						_lowercase     ( self      :	str   )					->					int:
    self._test_inference_batch_single_identical(expected_max_diff=2e-3   )




  def 						_lowercase     ( self      :	List[Any]   )					->					str:
    __lowerCamelCase       :       int           = self.get_dummy_components()
    __lowerCamelCase       :       List[Any]           = self.pipeline_class(**_a   )
    pipe.to(_a   )
    pipe.set_progress_bar_config(disable=_a   )
    with tempfile.TemporaryDirectory() as tmpdir:
      try:
        # save_pretrained is not implemented for Multi-ControlNet
        pipe.save_pretrained(_a   )
      except NotImplementedError:
        pass






@slow
@require_torch_gpu
class 					lowerCamelCase_    (				unittest.TestCase     ):
  """simple docstring"""




  def 						_lowercase     ( self      :	List[str]   )					->					List[str]:
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()




  def 						_lowercase     ( self      :	Optional[Any]   )					->					Union[str, Any]:
    __lowerCamelCase       :       Optional[Any]           = ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'   )

    __lowerCamelCase       :       List[str]           = StableDiffusionControlNetImgaImgPipeline.from_pretrained(
        'runwayml/stable-diffusion-v1-5'			,	safety_checker=_a			,	controlnet=_a   )
    pipe.enable_model_cpu_offload()
    pipe.set_progress_bar_config(disable=_a   )

    __lowerCamelCase       :       Optional[int]           = torch.Generator(device='cpu'   ).manual_seed(0   )
    __lowerCamelCase       :       List[Any]           = 'evil space-punk bird'
    __lowerCamelCase       :       List[str]           = load_image(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png'   ).resize((512, 512)   )
    __lowerCamelCase       :       Any           = load_image(
        'https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png'   ).resize((512, 512)   )

    __lowerCamelCase       :       Dict           = pipe(
        _a			,	_a			,	control_image=_a			,	generator=_a			,	output_type='np'			,	num_inference_steps=50			,	strength=0.6			,	)

    __lowerCamelCase       :       Optional[Any]           = output.images[0]

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

    __lowerCamelCase       :       List[str]           = load_numpy(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy'   )

    assert np.abs(expected_image - image   ).max() < 9e-2






