





# 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        __magic_name__	(				lowerCamelCase__     ,					lowerCamelCase__     ,					lowerCamelCase__     ,					unittest.TestCase				):





     """simple docstring"""



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



     def     SCREAMING_SNAKE_CASE		(				self							:Union[str, Any]	):





        '''simple docstring'''




        torch.manual_seed(0	)
        A_       :					List[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	)
        A_       :					List[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	)
        A_       :					List[str]        =      DDIMScheduler(
            beta_start=0.00085      ,   beta_end=0.012      ,   beta_schedule="scaled_linear"      ,   clip_sample=snake_case      ,   set_alpha_to_one=snake_case      ,   )
        torch.manual_seed(0	)
        A_       :					Any        =      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_       :					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=1_000      ,   )
        A_       :					Optional[int]        =      CLIPTextModel(snake_case	)
        A_       :					Optional[int]        =      CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"	)

        A_       :					Any        =      {
            "unet": unet,
            "controlnet": controlnet,
            "scheduler": scheduler,
            "vae": vae,
            "text_encoder": text_encoder,
            "tokenizer": tokenizer,
            "safety_checker": None,
            "feature_extractor": None,
        }
        return components



     def     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]      ,   snake_case							:int      ,   snake_case							:int=0	):





        '''simple docstring'''




        if str(snake_case	).startswith("mps"	):
           A_       :					Optional[Any]        =      torch.manual_seed(snake_case	)
        else:
           A_       :					Union[str, Any]        =      torch.Generator(device=snake_case	).manual_seed(snake_case	)

        A_       :					Union[str, Any]        =      2
        A_       :					Dict        =      randn_tensor(
            (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)      ,   generator=snake_case      ,   device=torch.device(snake_case	)      ,   )
        A_       :					Any        =      floats_tensor(control_image.shape      ,   rng=random.Random(snake_case	)	).to(snake_case	)
        A_       :					Dict        =      image.cpu().permute(0      ,   2      ,   3      ,   1	)[0]
        A_       :					Union[str, Any]        =      Image.fromarray(np.uinta(snake_case	)	).convert("RGB"	).resize((64, 64)	)
        A_       :					Union[str, Any]        =      {
            "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     SCREAMING_SNAKE_CASE		(				self							:Optional[int]	):





        '''simple docstring'''




        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     SCREAMING_SNAKE_CASE		(				self							:str	):





        '''simple docstring'''




        self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3	)





     def     SCREAMING_SNAKE_CASE		(				self							:Dict	):





        '''simple docstring'''




        self._test_inference_batch_single_identical(expected_max_diff=2e-3	)




class        __magic_name__	(				lowerCamelCase__     ,					lowerCamelCase__     ,					unittest.TestCase				):





     """simple docstring"""



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



     def     SCREAMING_SNAKE_CASE		(				self							:int	):





        '''simple docstring'''




        torch.manual_seed(0	)
        A_       :					Optional[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(snake_case							:Optional[int]	):
           if isinstance(snake_case      ,   torch.nn.Convad	):
              torch.nn.init.normal(m.weight	)
              m.bias.data.fill_(1.0	)

        A_       :					str        =      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(snake_case	)

        torch.manual_seed(0	)
        A_       :					Union[str, 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(snake_case	)

        torch.manual_seed(0	)
        A_       :					Optional[int]        =      DDIMScheduler(
            beta_start=0.00085      ,   beta_end=0.012      ,   beta_schedule="scaled_linear"      ,   clip_sample=snake_case      ,   set_alpha_to_one=snake_case      ,   )
        torch.manual_seed(0	)
        A_       :					Optional[int]        =      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_       :					Union[str, Any]        =      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_       :					Dict        =      CLIPTextModel(snake_case	)
        A_       :					Optional[int]        =      CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"	)

        A_       :					Optional[Any]        =      MultiControlNetModel([controlneta, controlneta]	)

        A_       :					Union[str, Any]        =      {
            "unet": unet,
            "controlnet": controlnet,
            "scheduler": scheduler,
            "vae": vae,
            "text_encoder": text_encoder,
            "tokenizer": tokenizer,
            "safety_checker": None,
            "feature_extractor": None,
        }
        return components



     def     SCREAMING_SNAKE_CASE		(				self							:List[Any]      ,   snake_case							:str      ,   snake_case							:Optional[int]=0	):





        '''simple docstring'''




        if str(snake_case	).startswith("mps"	):
           A_       :					List[Any]        =      torch.manual_seed(snake_case	)
        else:
           A_       :					Any        =      torch.Generator(device=snake_case	).manual_seed(snake_case	)

        A_       :					Dict        =      2

        A_       :					Tuple        =      [
            randn_tensor(
                (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)      ,   generator=snake_case      ,   device=torch.device(snake_case	)      ,   ),
            randn_tensor(
                (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)      ,   generator=snake_case      ,   device=torch.device(snake_case	)      ,   ),
        ]

        A_       :					Dict        =      floats_tensor(control_image[0].shape      ,   rng=random.Random(snake_case	)	).to(snake_case	)
        A_       :					List[Any]        =      image.cpu().permute(0      ,   2      ,   3      ,   1	)[0]
        A_       :					List[str]        =      Image.fromarray(np.uinta(snake_case	)	).convert("RGB"	).resize((64, 64)	)
        A_       :					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     SCREAMING_SNAKE_CASE		(				self							:int	):





        '''simple docstring'''




        A_       :					str        =      self.get_dummy_components()
        A_       :					List[str]        =      self.pipeline_class(**snake_case	)
        pipe.to(snake_case	)

        A_       :					str        =      10.0
        A_       :					List[Any]        =      4

        A_       :					Tuple        =      self.get_dummy_inputs(snake_case	)
        A_       :					int        =      steps
        A_       :					Any        =      scale
        A_       :					List[Any]        =      pipe(**snake_case	)[0]

        A_       :					Optional[Any]        =      self.get_dummy_inputs(snake_case	)
        A_       :					Union[str, Any]        =      steps
        A_       :					Union[str, Any]        =      scale
        A_       :					Tuple        =      pipe(**snake_case      ,   control_guidance_start=0.1      ,   control_guidance_end=0.2	)[0]

        A_       :					Any        =      self.get_dummy_inputs(snake_case	)
        A_       :					List[Any]        =      steps
        A_       :					Tuple        =      scale
        A_       :					Any        =      pipe(**snake_case      ,   control_guidance_start=[0.1, 0.3]      ,   control_guidance_end=[0.2, 0.7]	)[0]

        A_       :					Optional[Any]        =      self.get_dummy_inputs(snake_case	)
        A_       :					str        =      steps
        A_       :					Dict        =      scale
        A_       :					str        =      pipe(**snake_case      ,   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     SCREAMING_SNAKE_CASE		(				self							:List[Any]	):





        '''simple docstring'''




        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     SCREAMING_SNAKE_CASE		(				self							:List[Any]	):





        '''simple docstring'''




        self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3	)



     def     SCREAMING_SNAKE_CASE		(				self							:List[str]	):





        '''simple docstring'''




        self._test_inference_batch_single_identical(expected_max_diff=2e-3	)





     def     SCREAMING_SNAKE_CASE		(				self							:Any	):





        '''simple docstring'''




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




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





     """simple docstring"""



     def     SCREAMING_SNAKE_CASE		(				self							:Tuple	):





        '''simple docstring'''




        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()





     def     SCREAMING_SNAKE_CASE		(				self							:int	):





        '''simple docstring'''




        A_       :					Tuple        =      ControlNetModel.from_pretrained("lllyasviel/sd-controlnet-canny"	)

        A_       :					int        =      StableDiffusionControlNetImgaImgPipeline.from_pretrained(
            "runwayml/stable-diffusion-v1-5"      ,   safety_checker=snake_case      ,   controlnet=snake_case	)
        pipe.enable_model_cpu_offload()
        pipe.set_progress_bar_config(disable=snake_case	)

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

        A_       :					Any        =      pipe(
            snake_case      ,   snake_case      ,   control_image=snake_case      ,   generator=snake_case      ,   output_type="np"      ,   num_inference_steps=50      ,   strength=0.6      ,   )

        A_       :					Dict        =      output.images[0]

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

        A_       :					Union[str, Any]        =      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







from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxSeqaSeqConfigWithPast
from ...utils import logging


_lowerCAmelCase				:   Optional[int]						       =						logging.get_logger(__name__)

_lowerCAmelCase				:   List[str]						       =						{
    '''google/umt5-small''': '''https://huggingface.co/google/umt5-small/resolve/main/config.json''',
    # See all umt5 models at https://huggingface.co/models?filter=umt5
}


class        __magic_name__	(				lowerCamelCase__				):





     """simple docstring"""



     __UpperCamelCase            =  '''umt5'''
     __UpperCamelCase            =  ['''past_key_values''']



     def __init__(				self							:int      ,   snake_case							:Optional[Any]=250_112      ,   snake_case							:Optional[int]=512      ,   snake_case							:Any=64      ,   snake_case							:Union[str, Any]=1_024      ,   snake_case							:Tuple=8      ,   snake_case							:Optional[int]=None      ,   snake_case							:Union[str, Any]=6      ,   snake_case							:List[Any]=32      ,   snake_case							:Dict=128      ,   snake_case							:List[str]=0.1      ,   snake_case							:List[Any]=1e-6      ,   snake_case							:Dict=1.0      ,   snake_case							:Union[str, Any]="gated-gelu"      ,   snake_case							:Union[str, Any]=True      ,   snake_case							:Any=True      ,   snake_case							:List[str]="T5Tokenizer"      ,   snake_case							:Union[str, Any]=True      ,   snake_case							:Union[str, Any]=0      ,   snake_case							:List[Any]=1      ,   snake_case							:List[Any]=0      ,   **snake_case							:Any      ,   ):





        '''simple docstring'''




        super().__init__(
            is_encoder_decoder=snake_case      ,   tokenizer_class=snake_case      ,   tie_word_embeddings=snake_case      ,   pad_token_id=snake_case      ,   eos_token_id=snake_case      ,   decoder_start_token_id=snake_case      ,   **snake_case      ,   )
        A_       :					Union[str, Any]        =      vocab_size
        A_       :					Tuple        =      d_model
        A_       :					List[str]        =      d_kv
        A_       :					Union[str, Any]        =      d_ff
        A_       :					Any        =      num_layers
        A_       :					Optional[Any]        =      (
            num_decoder_layers if num_decoder_layers is not None else self.num_layers
        )  # default = symmetry
        A_       :					List[Any]        =      num_heads
        A_       :					List[str]        =      relative_attention_num_buckets
        A_       :					Dict        =      relative_attention_max_distance
        A_       :					Optional[Any]        =      dropout_rate
        A_       :					Any        =      layer_norm_epsilon
        A_       :					List[Any]        =      initializer_factor
        A_       :					Any        =      feed_forward_proj
        A_       :					Optional[Any]        =      use_cache

        A_       :					int        =      self.feed_forward_proj.split("-"	)
        A_       :					Any        =      act_info[-1]
        A_       :					Tuple        =      act_info[0] == "gated"

        if len(snake_case	) > 1 and act_info[0] != "gated" or len(snake_case	) > 2:
           raise ValueError(
               f"`feed_forward_proj`: {feed_forward_proj} is not a valid activation function of the dense layer."
               "Please make sure `feed_forward_proj` is of the format `gated-{ACT_FN}` or `{ACT_FN}`, e.g. "
               "'gated-gelu' or 'relu'"	)

        if feed_forward_proj == "gated-gelu":
           A_       :					Optional[Any]        =      "gelu_new"



     @property
     def     SCREAMING_SNAKE_CASE		(				self							:str	):





        '''simple docstring'''




        return self.d_model



     @property
     def     SCREAMING_SNAKE_CASE		(				self							:str	):





        '''simple docstring'''




        return self.num_heads





     @property
     def     SCREAMING_SNAKE_CASE		(				self							:Tuple	):





        '''simple docstring'''




        return self.num_layers




class        __magic_name__	(				lowerCamelCase__				):





     """simple docstring"""



     @property
     # Copied from transformers.models.t5.configuration_t5.T5OnnxConfig.inputs
     def     SCREAMING_SNAKE_CASE		(				self							:Union[str, Any]	):





        '''simple docstring'''




        A_       :					List[Any]        =      {
            "input_ids": {0: "batch", 1: "encoder_sequence"},
            "attention_mask": {0: "batch", 1: "encoder_sequence"},
        }
        if self.use_past:
           A_       :					Any        =      "past_encoder_sequence + sequence"
           A_       :					Union[str, Any]        =      {0: "batch"}
           A_       :					Optional[int]        =      {0: "batch", 1: "past_decoder_sequence + sequence"}
        else:
           A_       :					Dict        =      {0: "batch", 1: "decoder_sequence"}
           A_       :					str        =      {0: "batch", 1: "decoder_sequence"}

        if self.use_past:
           self.fill_with_past_key_values_(snake_case      ,   direction="inputs"	)

        return common_inputs



     @property
     # Copied from transformers.models.t5.configuration_t5.T5OnnxConfig.default_onnx_opset
     def     SCREAMING_SNAKE_CASE		(				self							:Optional[int]	):





        '''simple docstring'''




        return 13





     @property
     def     SCREAMING_SNAKE_CASE		(				self							:List[str]	):





        '''simple docstring'''




        return 5e-4
