


import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
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_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()



class 	__lowerCamelCase			(__UpperCamelCase				,      __UpperCamelCase				,      __UpperCamelCase				,      unittest.TestCase						):
       _lowercase											=     StableDiffusionInstructPixaPixPipeline
       _lowercase											=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width", "cross_attention_kwargs"}
       _lowercase											=     TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
       _lowercase											=     IMAGE_TO_IMAGE_IMAGE_PARAMS
       _lowercase											=     IMAGE_TO_IMAGE_IMAGE_PARAMS







       def 		snake_case_	(			self: Optional[int]					):






            '''simple docstring'''




            torch.manual_seed(0					)
            __UpperCamelCase =			UNetaDConditionModel(
                block_out_channels=(32, 64),layers_per_block=2,sample_size=32,in_channels=8,out_channels=4,down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D'),up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D'),cross_attention_dim=32,)
            __UpperCamelCase =			PNDMScheduler(skip_prk_steps=snake_case__					)
            torch.manual_seed(0					)
            __UpperCamelCase =			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					)
            __UpperCamelCase =			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,)
            __UpperCamelCase =			CLIPTextModel(snake_case__					)
            __UpperCamelCase =			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

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







       def 		snake_case_	(			self: Optional[int],A_: Tuple,A_: Union[str, Any]=0					):






            '''simple docstring'''




            __UpperCamelCase =			floats_tensor((1, 3, 32, 32),rng=random.Random(snake_case__					)					).to(snake_case__					)
            __UpperCamelCase =			image.cpu().permute(0,2,3,1					)[0]
            __UpperCamelCase =			Image.fromarray(np.uinta(snake_case__					)					).convert('RGB'					)
            if str(snake_case__					).startswith('mps'					):
                 __UpperCamelCase =			torch.manual_seed(snake_case__					)
            else:
                 __UpperCamelCase =			torch.Generator(device=snake_case__					).manual_seed(snake_case__					)
            __UpperCamelCase =			{
                'prompt': 'A painting of a squirrel eating a burger',
                'image': image,
                'generator': generator,
                'num_inference_steps': 2,
                'guidance_scale': 6.0,
                'image_guidance_scale': 1,
                'output_type': 'numpy',
            }
            return inputs







       def 		snake_case_	(			self: str					):






            '''simple docstring'''




            __UpperCamelCase =			'cpu'  # ensure determinism for the device-dependent torch.Generator
            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline(**snake_case__					)
            __UpperCamelCase =			sd_pipe.to(snake_case__					)
            sd_pipe.set_progress_bar_config(disable=snake_case__					)

            __UpperCamelCase =			self.get_dummy_inputs(snake_case__					)
            __UpperCamelCase =			sd_pipe(**snake_case__					).images
            __UpperCamelCase =			image[0, -3:, -3:, -1]
            assert image.shape == (1, 32, 32, 3)
            __UpperCamelCase =			np.array([0.7_5_2_6, 0.3_7_5_0, 0.4_5_4_7, 0.6_1_1_7, 0.5_8_6_6, 0.5_0_1_6, 0.4_3_2_7, 0.5_6_4_2, 0.4_8_1_5]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-3







       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            __UpperCamelCase =			'cpu'  # ensure determinism for the device-dependent torch.Generator
            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline(**snake_case__					)
            __UpperCamelCase =			sd_pipe.to(snake_case__					)
            sd_pipe.set_progress_bar_config(disable=snake_case__					)

            __UpperCamelCase =			self.get_dummy_inputs(snake_case__					)
            __UpperCamelCase =			'french fries'
            __UpperCamelCase =			sd_pipe(**snake_case__,negative_prompt=snake_case__					)
            __UpperCamelCase =			output.images
            __UpperCamelCase =			image[0, -3:, -3:, -1]

            assert image.shape == (1, 32, 32, 3)
            __UpperCamelCase =			np.array([0.7_5_1_1, 0.3_6_4_2, 0.4_5_5_3, 0.6_2_3_6, 0.5_7_9_7, 0.5_0_1_3, 0.4_3_4_3, 0.5_6_1_1, 0.4_8_3_1]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-3







       def 		snake_case_	(			self: str					):






            '''simple docstring'''




            __UpperCamelCase =			'cpu'  # ensure determinism for the device-dependent torch.Generator
            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline(**snake_case__					)
            __UpperCamelCase =			sd_pipe.to(snake_case__					)
            sd_pipe.set_progress_bar_config(disable=snake_case__					)

            __UpperCamelCase =			self.get_dummy_inputs(snake_case__					)
            __UpperCamelCase =			[inputs['prompt']] * 2

            __UpperCamelCase =			np.array(inputs['image']					).astype(np.floataa					) / 2_5_5.0
            __UpperCamelCase =			torch.from_numpy(snake_case__					).unsqueeze(0					).to(snake_case__					)
            __UpperCamelCase =			image / 2 + 0.5
            __UpperCamelCase =			image.permute(0,3,1,2					)
            __UpperCamelCase =			image.repeat(2,1,1,1					)

            __UpperCamelCase =			sd_pipe(**snake_case__					).images
            __UpperCamelCase =			image[-1, -3:, -3:, -1]

            assert image.shape == (2, 32, 32, 3)
            __UpperCamelCase =			np.array([0.5_8_1_2, 0.5_7_4_8, 0.5_2_2_2, 0.5_9_0_8, 0.5_6_9_5, 0.7_1_7_4, 0.6_8_0_4, 0.5_5_2_3, 0.5_5_7_9]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-3







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






            '''simple docstring'''




            __UpperCamelCase =			'cpu'  # ensure determinism for the device-dependent torch.Generator
            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			EulerAncestralDiscreteScheduler(
                beta_start=0.0_0_0_8_5,beta_end=0.0_1_2,beta_schedule='scaled_linear'					)
            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline(**snake_case__					)
            __UpperCamelCase =			sd_pipe.to(snake_case__					)
            sd_pipe.set_progress_bar_config(disable=snake_case__					)

            __UpperCamelCase =			self.get_dummy_inputs(snake_case__					)
            __UpperCamelCase =			sd_pipe(**snake_case__					).images
            __UpperCamelCase =			image[0, -3:, -3:, -1]

            __UpperCamelCase =			[round(snake_case__,4					) for x in image_slice.flatten().tolist()]
            print(','.join([str(snake_case__					) for x in slice]					)					)

            assert image.shape == (1, 32, 32, 3)
            __UpperCamelCase =			np.array([0.7_4_1_7, 0.3_8_4_2, 0.4_7_3_2, 0.5_7_7_6, 0.5_8_9_1, 0.5_1_3_9, 0.4_0_5_2, 0.5_6_7_3, 0.4_9_8_6]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-3







       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




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







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline(**snake_case__					)
            __UpperCamelCase =			VaeImageProcessor(do_resize=snake_case__,do_normalize=snake_case__					)
            __UpperCamelCase =			pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)

            __UpperCamelCase =			pipe(**self.get_dummy_inputs_by_type(snake_case__,input_image_type='pt'					)					)[0]

            __UpperCamelCase =			components['vae']
            __UpperCamelCase =			self.get_dummy_inputs_by_type(snake_case__,input_image_type='pt'					)

            for image_param in self.image_latents_params:
                 if image_param in inputs.keys():
                      __UpperCamelCase =			vae.encode(inputs[image_param]					).latent_dist.mode()

            __UpperCamelCase =			pipe(**snake_case__					)[0]

            __UpperCamelCase =			np.abs(out - out_latents_inputs					).max()
            self.assertLess(snake_case__,1E-4,'passing latents as image input generate different result from passing image'					)



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







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




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







       def 		snake_case_	(			self: List[Any],A_: int=0					):






            '''simple docstring'''




            __UpperCamelCase =			torch.manual_seed(snake_case__					)
            __UpperCamelCase =			load_image(
                'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg'					)
            __UpperCamelCase =			{
                'prompt': 'turn him into a cyborg',
                'image': image,
                'generator': generator,
                'num_inference_steps': 3,
                'guidance_scale': 7.5,
                'image_guidance_scale': 1.0,
                'output_type': 'numpy',
            }
            return inputs







       def 		snake_case_	(			self: Optional[Any]					):






            '''simple docstring'''




            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
                'timbrooks/instruct-pix2pix',safety_checker=snake_case__					)
            pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)
            pipe.enable_attention_slicing()

            __UpperCamelCase =			self.get_inputs()
            __UpperCamelCase =			pipe(**snake_case__					).images
            __UpperCamelCase =			image[0, -3:, -3:, -1].flatten()

            assert image.shape == (1, 512, 512, 3)
            __UpperCamelCase =			np.array([0.5_9_0_2, 0.6_0_1_5, 0.6_0_2_7, 0.5_9_8_3, 0.6_0_9_2, 0.6_0_6_1, 0.5_7_6_5, 0.5_7_8_5, 0.5_5_5_5]					)

            assert np.abs(expected_slice - image_slice					).max() < 1E-3







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
                'timbrooks/instruct-pix2pix',safety_checker=snake_case__					)
            __UpperCamelCase =			LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
            pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)
            pipe.enable_attention_slicing()

            __UpperCamelCase =			self.get_inputs()
            __UpperCamelCase =			pipe(**snake_case__					).images
            __UpperCamelCase =			image[0, -3:, -3:, -1].flatten()

            assert image.shape == (1, 512, 512, 3)
            __UpperCamelCase =			np.array([0.6_5_7_8, 0.6_8_1_7, 0.6_9_7_2, 0.6_7_6_1, 0.6_8_5_6, 0.6_9_1_6, 0.6_4_2_8, 0.6_5_1_6, 0.6_3_0_1]					)

            assert np.abs(expected_slice - image_slice					).max() < 1E-3







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
                'timbrooks/instruct-pix2pix',safety_checker=snake_case__					)
            __UpperCamelCase =			DDIMScheduler.from_config(pipe.scheduler.config					)
            pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)
            pipe.enable_attention_slicing()

            __UpperCamelCase =			self.get_inputs()
            __UpperCamelCase =			pipe(**snake_case__					).images
            __UpperCamelCase =			image[0, -3:, -3:, -1].flatten()

            assert image.shape == (1, 512, 512, 3)
            __UpperCamelCase =			np.array([0.3_8_2_8, 0.3_8_3_4, 0.3_8_1_8, 0.3_7_9_2, 0.3_8_6_5, 0.3_7_5_2, 0.3_7_9_2, 0.3_8_4_7, 0.3_7_5_3]					)

            assert np.abs(expected_slice - image_slice					).max() < 1E-3







       def 		snake_case_	(			self: str					):






            '''simple docstring'''




            __UpperCamelCase =			0

            def callback_fn(A_: int,A_: int,A_: torch.FloatTensor					) -> None:
                 __UpperCamelCase =			True
                 nonlocal number_of_steps
                 number_of_steps += 1
                 if step == 1:
                      __UpperCamelCase =			latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      __UpperCamelCase =			latents[0, -3:, -3:, -1]
                      __UpperCamelCase =			np.array([-0.2_4_6_3, -0.4_6_4_4, -0.9_7_5_6, 1.5_1_7_6, 1.4_4_1_4, 0.7_8_6_6, 0.9_8_9_7, 0.8_5_2_1, 0.7_9_8_3]					)

                      assert np.abs(latents_slice.flatten() - expected_slice					).max() < 5E-2
                 elif step == 2:
                      __UpperCamelCase =			latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      __UpperCamelCase =			latents[0, -3:, -3:, -1]
                      __UpperCamelCase =			np.array([-0.2_6_4_4, -0.4_6_2_6, -0.9_6_5_3, 1.5_1_7_6, 1.4_5_5_1, 0.7_6_8_6, 0.9_8_0_5, 0.8_4_5_2, 0.8_1_1_5]					)

                      assert np.abs(latents_slice.flatten() - expected_slice					).max() < 5E-2

            __UpperCamelCase =			False

            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
                'timbrooks/instruct-pix2pix',safety_checker=snake_case__,torch_dtype=torch.floataa					)
            __UpperCamelCase =			pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)
            pipe.enable_attention_slicing()

            __UpperCamelCase =			self.get_inputs()
            pipe(**snake_case__,callback=snake_case__,callback_steps=1					)
            assert callback_fn.has_been_called
            assert number_of_steps == 3







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            torch.cuda.empty_cache()
            torch.cuda.reset_max_memory_allocated()
            torch.cuda.reset_peak_memory_stats()

            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
                'timbrooks/instruct-pix2pix',safety_checker=snake_case__,torch_dtype=torch.floataa					)
            __UpperCamelCase =			pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)
            pipe.enable_attention_slicing(1					)
            pipe.enable_sequential_cpu_offload()

            __UpperCamelCase =			self.get_inputs()
            __UpperCamelCase =			pipe(**snake_case__					)

            __UpperCamelCase =			torch.cuda.max_memory_allocated()
            # make sure that less than 2.2 GB is allocated
            assert mem_bytes < 2.2 * 10**9







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            __UpperCamelCase =			self.get_inputs()
            # resize to resolution that is divisible by 8 but not 16 or 32
            __UpperCamelCase =			inputs['image'].resize((504, 504)					)

            __UpperCamelCase =			'timbrooks/instruct-pix2pix'
            __UpperCamelCase =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
                snake_case__,safety_checker=snake_case__,)
            pipe.to(snake_case__					)
            pipe.set_progress_bar_config(disable=snake_case__					)
            pipe.enable_attention_slicing()

            __UpperCamelCase =			pipe(**snake_case__					)
            __UpperCamelCase =			output.images[0]

            __UpperCamelCase =			image[255:258, 383:386, -1]

            assert image.shape == (504, 504, 3)
            __UpperCamelCase =			np.array([0.2_7_2_6, 0.2_5_2_9, 0.2_6_6_4, 0.2_6_5_5, 0.2_6_4_1, 0.2_6_4_2, 0.2_5_9_1, 0.2_6_4_9, 0.2_5_9_0]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 5E-3



"""simple docstring"""




from typing import Optional, Tuple

import jax
import jax.numpy as jnp
from flax import linen as nn
from flax.core.frozen_dict import FrozenDict
from transformers import CLIPConfig, FlaxPreTrainedModel
from transformers.models.clip.modeling_flax_clip import FlaxCLIPVisionModule







def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__=1e-12  ):





		"""simple docstring"""




		lowerCAmelCase__   = jnp.divide(emb_a.T      ,	jnp.clip(jnp.linalg.norm(lowerCamelCase__      ,	axis=1  )      ,	a_min=lowerCamelCase__  )  ).T
		lowerCAmelCase__   = jnp.divide(emb_a.T      ,	jnp.clip(jnp.linalg.norm(lowerCamelCase__      ,	axis=1  )      ,	a_min=lowerCamelCase__  )  ).T
		return jnp.matmul(lowerCamelCase__      ,	norm_emb_a.T  )

class    a_   (  nn.Module      ):
					UpperCamelCase_			:						CLIPConfig
					UpperCamelCase_			:						jnp.dtype      =     jnp.floataa




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = FlaxCLIPVisionModule(self.config.vision_config  )
							lowerCAmelCase__   = nn.Dense(self.config.projection_dim	,    use_bias=snake_case__	,    dtype=self.dtype  )

							lowerCAmelCase__   = self.param("""concept_embeds"""	,    jax.nn.initializers.ones	,    (17, self.config.projection_dim)  )
							lowerCAmelCase__   = self.param(
							    """special_care_embeds"""	,    jax.nn.initializers.ones	,    (3, self.config.projection_dim)  )

							lowerCAmelCase__   = self.param("""concept_embeds_weights"""	,    jax.nn.initializers.ones	,    (17,)  )
							lowerCAmelCase__   = self.param("""special_care_embeds_weights"""	,    jax.nn.initializers.ones	,    (3,)  )




					def __call__(						self    :  Optional[int]	,    snake_case__    :  int  ):
							lowerCAmelCase__   = self.vision_model(snake_case__  )[1]
							lowerCAmelCase__   = self.visual_projection(snake_case__  )

							lowerCAmelCase__   = jax_cosine_distance(snake_case__	,    self.special_care_embeds  )
							lowerCAmelCase__   = jax_cosine_distance(snake_case__	,    self.concept_embeds  )

							# increase this value to create a stronger `nfsw` filter
							# at the cost of increasing the possibility of filtering benign image inputs
							lowerCAmelCase__   = 0.0

							lowerCAmelCase__   = special_cos_dist - self.special_care_embeds_weights[None, :] + adjustment
							lowerCAmelCase__   = jnp.round(snake_case__	,    3  )
							lowerCAmelCase__   = jnp.any(special_scores > 0	,    axis=1	,    keepdims=snake_case__  )
							# Use a lower threshold if an image has any special care concept
							lowerCAmelCase__   = is_special_care * 0.01

							lowerCAmelCase__   = cos_dist - self.concept_embeds_weights[None, :] + special_adjustment
							lowerCAmelCase__   = jnp.round(snake_case__	,    3  )
							lowerCAmelCase__   = jnp.any(concept_scores > 0	,    axis=1  )

							return has_nsfw_concepts

class    a_   (  __UpperCamelCase      ):
					UpperCamelCase_			:						Any      =     CLIPConfig
					UpperCamelCase_			:						str      =     "clip_input"
					UpperCamelCase_			:						str      =     FlaxStableDiffusionSafetyCheckerModule




					def __init__(						self    :  Optional[int]	,    snake_case__    :  CLIPConfig	,    snake_case__    :  Optional[Tuple] = None	,    snake_case__    :  int = 0	,    snake_case__    :  jnp.dtype = jnp.floataa	,    snake_case__    :  bool = True	,    **snake_case__    :  str	,    ):
							if input_shape is None:
									lowerCAmelCase__   = (1, 224, 224, 3)
							lowerCAmelCase__   = self.module_class(config=snake_case__	,    dtype=snake_case__	,    **snake_case__  )
							super().__init__(snake_case__	,    snake_case__	,    input_shape=snake_case__	,    seed=snake_case__	,    dtype=snake_case__	,    _do_init=_do_init  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple	,    snake_case__    :  jax.random.KeyArray	,    snake_case__    :  Tuple	,    snake_case__    :  FrozenDict = None  ):
							# init input tensor
							lowerCAmelCase__   = jax.random.normal(snake_case__	,    snake_case__  )

							lowerCAmelCase__ ,       lowerCAmelCase__   = jax.random.split(snake_case__  )
							lowerCAmelCase__   = {"""params""": params_rng, """dropout""": dropout_rng}

							lowerCAmelCase__   = self.module.init(snake_case__	,    snake_case__  )["""params"""]

							return random_params




					def __call__(						self    :  Union[str, Any]	,    snake_case__    :  Union[str, Any]	,    snake_case__    :  dict = None	,    ):
							lowerCAmelCase__   = jnp.transpose(snake_case__	,    (0, 2, 3, 1)  )

							return self.module.apply(
							    {"""params""": params or self.params}	,    jnp.array(snake_case__	,    dtype=jnp.floataa  )	,    rngs={}	,    )




