





"""simple docstring"""


from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin



@dataclass
class        lowercase_			(       lowercase_							):
							snake_case__			:							Optional[Any]										=       42



class        lowercase_			(       lowercase_			,lowercase_							):

							@register_to_config
							def __init__(     self				,   __SCREAMING_SNAKE_CASE = 16				,   __SCREAMING_SNAKE_CASE = 88				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = 1				,   __SCREAMING_SNAKE_CASE = 0.0				,   __SCREAMING_SNAKE_CASE = 32				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = False				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = "geglu"				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = False				,   __SCREAMING_SNAKE_CASE = False				,   __SCREAMING_SNAKE_CASE = False				,   __SCREAMING_SNAKE_CASE = "layer_norm"				,   __SCREAMING_SNAKE_CASE = True				,   ):






								'''simple docstring'''

								super().__init__()
								snake_case__      							=			use_linear_projection
								snake_case__      							=			num_attention_heads
								snake_case__      							=			attention_head_dim
								snake_case__      							=			num_attention_heads * attention_head_dim

								# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
								# Define whether input is continuous or discrete depending on configuration
								snake_case__      							=			(in_channels is not None) and (patch_size is None)
								snake_case__      							=			num_vector_embeds is not None
								snake_case__      							=			in_channels is not None and patch_size is not None

								if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
									snake_case__      							=			(
									    F"The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or"
									    " incorrectly set to `'layer_norm'`.Make sure to set `norm_type` to `'ada_norm'` in the config."
									    " Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect"
									    " results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it"
									    " would be very nice if you could open a Pull request for the `transformer/config.json` file"
									)
									deprecate("norm_type!=num_embeds_ada_norm"				,   "1.0.0"				,   __UpperCamelCase				,   standard_warn=__UpperCamelCase       )
									snake_case__      							=			"ada_norm"

								if self.is_input_continuous and self.is_input_vectorized:
									raise ValueError(
									    F"Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make"
									    " sure that either `in_channels` or `num_vector_embeds` is None."       )
								elif self.is_input_vectorized and self.is_input_patches:
									raise ValueError(
									    F"Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make"
									    " sure that either `num_vector_embeds` or `num_patches` is None."       )
								elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
									raise ValueError(
									    F"Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:"
									    F" {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None."       )

								# 2. Define input layers
								if self.is_input_continuous:
									snake_case__      							=			in_channels

									snake_case__      							=			torch.nn.GroupNorm(num_groups=__UpperCamelCase				,   num_channels=__UpperCamelCase				,   eps=1e-6				,   affine=__UpperCamelCase       )
									if use_linear_projection:
										snake_case__      							=			nn.Linear(__UpperCamelCase				,   __UpperCamelCase       )
									else:
										snake_case__      							=			nn.Convad(__UpperCamelCase				,   __UpperCamelCase				,   kernel_size=1				,   stride=1				,   padding=0       )
								elif self.is_input_vectorized:
									assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
									assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

									snake_case__      							=			sample_size
									snake_case__      							=			sample_size
									snake_case__      							=			num_vector_embeds
									snake_case__      							=			self.height * self.width

									snake_case__      							=			ImagePositionalEmbeddings(
									    num_embed=__UpperCamelCase				,   embed_dim=__UpperCamelCase				,   height=self.height				,   width=self.width       )
								elif self.is_input_patches:
									assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

									snake_case__      							=			sample_size
									snake_case__      							=			sample_size

									snake_case__      							=			patch_size
									snake_case__      							=			PatchEmbed(
									    height=__UpperCamelCase				,   width=__UpperCamelCase				,   patch_size=__UpperCamelCase				,   in_channels=__UpperCamelCase				,   embed_dim=__UpperCamelCase				,   )

								# 3. Define transformers blocks
								snake_case__      							=			nn.ModuleList(
								    [
								        BasicTransformerBlock(
								            __UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase				,   dropout=__UpperCamelCase				,   cross_attention_dim=__UpperCamelCase				,   activation_fn=__UpperCamelCase				,   num_embeds_ada_norm=__UpperCamelCase				,   attention_bias=__UpperCamelCase				,   only_cross_attention=__UpperCamelCase				,   upcast_attention=__UpperCamelCase				,   norm_type=__UpperCamelCase				,   norm_elementwise_affine=__UpperCamelCase				,   )
								        for d in range(__UpperCamelCase       )
								    ]       )

								# 4. Define output layers
								snake_case__      							=			in_channels if out_channels is None else out_channels
								if self.is_input_continuous:
									# TODO: should use out_channels for continuous projections
									if use_linear_projection:
										snake_case__      							=			nn.Linear(__UpperCamelCase				,   __UpperCamelCase       )
									else:
										snake_case__      							=			nn.Convad(__UpperCamelCase				,   __UpperCamelCase				,   kernel_size=1				,   stride=1				,   padding=0       )
								elif self.is_input_vectorized:
									snake_case__      							=			nn.LayerNorm(__UpperCamelCase       )
									snake_case__      							=			nn.Linear(__UpperCamelCase				,   self.num_vector_embeds - 1       )
								elif self.is_input_patches:
									snake_case__      							=			nn.LayerNorm(__UpperCamelCase				,   elementwise_affine=__UpperCamelCase				,   eps=1e-6       )
									snake_case__      							=			nn.Linear(__UpperCamelCase				,   2 * inner_dim       )
									snake_case__      							=			nn.Linear(__UpperCamelCase				,   patch_size * patch_size * self.out_channels       )

							def 						_SCREAMING_SNAKE_CASE(     self				,   __SCREAMING_SNAKE_CASE				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = None				,   __SCREAMING_SNAKE_CASE = True				,   ):






								'''simple docstring'''

								if attention_mask is not None and attention_mask.ndim == 2:
									# assume that mask is expressed as:
									#   (1 = keep,      0 = discard)
									# convert mask into a bias that can be added to attention scores:
									#       (keep = +0,     discard = -10000.0)
									snake_case__      							=			(1 - attention_mask.to(hidden_states.dtype       )) * -1_0000.0
									snake_case__      							=			attention_mask.unsqueeze(1       )

								# convert encoder_attention_mask to a bias the same way we do for attention_mask
								if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
									snake_case__      							=			(1 - encoder_attention_mask.to(hidden_states.dtype       )) * -1_0000.0
									snake_case__      							=			encoder_attention_mask.unsqueeze(1       )

								# 1. Input
								if self.is_input_continuous:
									snake_case__							,				snake_case__							,				snake_case__							,				snake_case__      							=			hidden_states.shape
									snake_case__      							=			hidden_states

									snake_case__      							=			self.norm(__UpperCamelCase       )
									if not self.use_linear_projection:
										snake_case__      							=			self.proj_in(__UpperCamelCase       )
										snake_case__      							=			hidden_states.shape[1]
										snake_case__      							=			hidden_states.permute(0				,   2				,   3				,   1       ).reshape(__UpperCamelCase				,   height * width				,   __UpperCamelCase       )
									else:
										snake_case__      							=			hidden_states.shape[1]
										snake_case__      							=			hidden_states.permute(0				,   2				,   3				,   1       ).reshape(__UpperCamelCase				,   height * width				,   __UpperCamelCase       )
										snake_case__      							=			self.proj_in(__UpperCamelCase       )
								elif self.is_input_vectorized:
									snake_case__      							=			self.latent_image_embedding(__UpperCamelCase       )
								elif self.is_input_patches:
									snake_case__      							=			self.pos_embed(__UpperCamelCase       )

								# 2. Blocks
								for block in self.transformer_blocks:
									snake_case__      							=			block(
									    __UpperCamelCase				,   attention_mask=__UpperCamelCase				,   encoder_hidden_states=__UpperCamelCase				,   encoder_attention_mask=__UpperCamelCase				,   timestep=__UpperCamelCase				,   cross_attention_kwargs=__UpperCamelCase				,   class_labels=__UpperCamelCase				,   )

								# 3. Output
								if self.is_input_continuous:
									if not self.use_linear_projection:
										snake_case__      							=			hidden_states.reshape(__UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase       ).permute(0				,   3				,   1				,   2       ).contiguous()
										snake_case__      							=			self.proj_out(__UpperCamelCase       )
									else:
										snake_case__      							=			self.proj_out(__UpperCamelCase       )
										snake_case__      							=			hidden_states.reshape(__UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase       ).permute(0				,   3				,   1				,   2       ).contiguous()

									snake_case__      							=			hidden_states + residual
								elif self.is_input_vectorized:
									snake_case__      							=			self.norm_out(__UpperCamelCase       )
									snake_case__      							=			self.out(__UpperCamelCase       )
									# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
									snake_case__      							=			logits.permute(0				,   2				,   1       )

									# log(p(x_0))
									snake_case__      							=			F.log_softmax(logits.double()				,   dim=1       ).float()
								elif self.is_input_patches:
									# TODO: cleanup!
									snake_case__      							=			self.transformer_blocks[0].norma.emb(
									    __UpperCamelCase				,   __UpperCamelCase				,   hidden_dtype=hidden_states.dtype       )
									snake_case__							,				snake_case__      							=			self.proj_out_a(F.silu(__UpperCamelCase       )       ).chunk(2				,   dim=1       )
									snake_case__      							=			self.norm_out(__UpperCamelCase       ) * (1 + scale[:, None]) + shift[:, None]
									snake_case__      							=			self.proj_out_a(__UpperCamelCase       )

									# unpatchify
									snake_case__      							=			snake_case__      							=			int(hidden_states.shape[1] ** 0.5       )
									snake_case__      							=			hidden_states.reshape(
									    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)       )
									snake_case__      							=			torch.einsum("nhwpqc->nchpwq"				,   __UpperCamelCase       )
									snake_case__      							=			hidden_states.reshape(
									    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size)       )

								if not return_dict:
									return (output,)

								return TransformeraDModelOutput(sample=__UpperCamelCase       )








"""simple docstring"""


import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig

from diffusers import AutoencoderKL, PaintByExamplePipeline, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.paint_by_example import PaintByExampleImageEncoder
from diffusers.utils import floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS, IMAGE_GUIDED_IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



class        lowercase_			(       __lowerCAmelCase			,unittest.TestCase							):
       snake_case__			:							Tuple										=       PaintByExamplePipeline
       snake_case__			:							List[str]										=       IMAGE_GUIDED_IMAGE_INPAINTING_PARAMS
       snake_case__			:							List[str]										=       IMAGE_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
       snake_case__			:							List[str]										=       frozenset([]							)  # TO_DO: update the image_prams once refactored VaeImageProcessor.preprocess

       def 						_SCREAMING_SNAKE_CASE(     self       ):






        '''simple docstring'''

        torch.manual_seed(0       )
        snake_case__      							=			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				,   )
        snake_case__      							=			PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE       )
        torch.manual_seed(0       )
        snake_case__      							=			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       )
        snake_case__      							=			CLIPVisionConfig(
            hidden_size=32				,   projection_dim=32				,   intermediate_size=37				,   layer_norm_eps=1e-05				,   num_attention_heads=4				,   num_hidden_layers=5				,   image_size=32				,   patch_size=4				,   )
        snake_case__      							=			PaintByExampleImageEncoder(__SCREAMING_SNAKE_CASE				,   proj_size=32       )
        snake_case__      							=			CLIPImageProcessor(crop_size=32				,   size=32       )

        snake_case__      							=			{
            "unet": unet,
            "scheduler": scheduler,
            "vae": vae,
            "image_encoder": image_encoder,
            "safety_checker": None,
            "feature_extractor": feature_extractor,
        }
        return components

       def 						_SCREAMING_SNAKE_CASE(     self				,   __SCREAMING_SNAKE_CASE       ):






        '''simple docstring'''

        snake_case__      							=			np.array(image.convert("RGB"       )       )
        snake_case__      							=			image[None].transpose(0				,   3				,   1				,   2       )
        snake_case__      							=			torch.from_numpy(__SCREAMING_SNAKE_CASE       ).to(dtype=torch.floataa       ) / 127.5 - 1.0
        return image

       def 						_SCREAMING_SNAKE_CASE(     self				,   __SCREAMING_SNAKE_CASE="cpu"				,   __SCREAMING_SNAKE_CASE=0       ):






        '''simple docstring'''

        snake_case__      							=			floats_tensor((1, 3, 32, 32)				,   rng=random.Random(__SCREAMING_SNAKE_CASE       )       ).to(__SCREAMING_SNAKE_CASE       )
        snake_case__      							=			image.cpu().permute(0				,   2				,   3				,   1       )[0]
        snake_case__      							=			Image.fromarray(np.uinta(__SCREAMING_SNAKE_CASE       )       ).convert("RGB"       ).resize((64, 64)       )
        snake_case__      							=			Image.fromarray(np.uinta(image + 4       )       ).convert("RGB"       ).resize((64, 64)       )
        snake_case__      							=			Image.fromarray(np.uinta(__SCREAMING_SNAKE_CASE       )       ).convert("RGB"       ).resize((32, 32)       )

        if str(__SCREAMING_SNAKE_CASE       ).startswith("mps"       ):
         snake_case__      							=			torch.manual_seed(__SCREAMING_SNAKE_CASE       )
        else:
         snake_case__      							=			torch.Generator(device=__SCREAMING_SNAKE_CASE       ).manual_seed(__SCREAMING_SNAKE_CASE       )
        snake_case__      							=			{
            "example_image": example_image,
            "image": init_image,
            "mask_image": mask_image,
            "generator": generator,
            "num_inference_steps": 2,
            "guidance_scale": 6.0,
            "output_type": "numpy",
        }
        return inputs

       def 						_SCREAMING_SNAKE_CASE(     self       ):






        '''simple docstring'''

        snake_case__      							=			self.get_dummy_components()

        # make sure here that pndm scheduler skips prk
        snake_case__      							=			PaintByExamplePipeline(**__SCREAMING_SNAKE_CASE       )
        snake_case__      							=			pipe.to("cpu"       )
        pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

        snake_case__      							=			self.get_dummy_inputs()
        snake_case__      							=			pipe(**__SCREAMING_SNAKE_CASE       )
        snake_case__      							=			output.images

        snake_case__      							=			image[0, -3:, -3:, -1]

        assert image.shape == (1, 64, 64, 3)
        snake_case__      							=			np.array([0.4686, 0.5687, 0.4007, 0.5218, 0.5741, 0.4482, 0.4940, 0.4629, 0.4503]       )

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

       def 						_SCREAMING_SNAKE_CASE(     self       ):






        '''simple docstring'''

        snake_case__      							=			"cpu"
        snake_case__      							=			self.get_dummy_inputs()
        inputs.pop("mask_image"       )
        snake_case__      							=			self.convert_to_pt(inputs.pop("image"       )       )
        snake_case__      							=			image.clamp(0				,   1       ) / 2

        # make sure here that pndm scheduler skips prk
        snake_case__      							=			PaintByExamplePipeline(**self.get_dummy_components()       )
        snake_case__      							=			pipe.to(__SCREAMING_SNAKE_CASE       )
        pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

        snake_case__      							=			pipe(image=__SCREAMING_SNAKE_CASE				,   mask_image=mask_image[:, 0]				,   **__SCREAMING_SNAKE_CASE       )
        snake_case__      							=			output.images

        snake_case__      							=			image.cpu().permute(0				,   2				,   3				,   1       )[0]
        snake_case__      							=			mask_image.cpu().permute(0				,   2				,   3				,   1       )[0]

        snake_case__      							=			Image.fromarray(np.uinta(__SCREAMING_SNAKE_CASE       )       ).convert("RGB"       )
        snake_case__      							=			Image.fromarray(np.uinta(__SCREAMING_SNAKE_CASE       )       ).convert("RGB"       )

        snake_case__      							=			pipe(**self.get_dummy_inputs()       )
        snake_case__      							=			output.images

        assert out_a.shape == (1, 64, 64, 3)
        assert np.abs(out_a.flatten() - out_a.flatten()       ).max() < 5e-2

       def 						_SCREAMING_SNAKE_CASE(     self       ):






        '''simple docstring'''

        super().test_inference_batch_single_identical(expected_max_diff=3e-3       )



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

       def 						_SCREAMING_SNAKE_CASE(     self       ):






        '''simple docstring'''

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

       def 						_SCREAMING_SNAKE_CASE(     self       ):






        '''simple docstring'''

        snake_case__      							=			load_image(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
            "/paint_by_example/dog_in_bucket.png"       )
        snake_case__      							=			load_image(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
            "/paint_by_example/mask.png"       )
        snake_case__      							=			load_image(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
            "/paint_by_example/panda.jpg"       )

        snake_case__      							=			PaintByExamplePipeline.from_pretrained("Fantasy-Studio/Paint-by-Example"       )
        snake_case__      							=			pipe.to(__SCREAMING_SNAKE_CASE       )
        pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

        snake_case__      							=			torch.manual_seed(321       )
        snake_case__      							=			pipe(
            image=__SCREAMING_SNAKE_CASE				,   mask_image=__SCREAMING_SNAKE_CASE				,   example_image=__SCREAMING_SNAKE_CASE				,   generator=__SCREAMING_SNAKE_CASE				,   guidance_scale=5.0				,   num_inference_steps=50				,   output_type="np"				,   )

        snake_case__      							=			output.images

        snake_case__      							=			image[0, -3:, -3:, -1]

        assert image.shape == (1, 512, 512, 3)
        snake_case__      							=			np.array([0.4834, 0.4811, 0.4874, 0.5122, 0.5081, 0.5144, 0.5291, 0.5290, 0.5374]       )

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

