



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 		lowercase						(			snake_case__			,   snake_case__			,   snake_case__			,   unittest.TestCase):






 """simple docstring"""


 a__				: Tuple		       =							StableDiffusionInstructPixaPixPipeline
 a__				: Any		       =							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width", "cross_attention_kwargs"}
 a__				: List[str]		       =							TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
 a__				: Optional[Any]		       =							IMAGE_TO_IMAGE_IMAGE_PARAMS
 a__				: Dict		       =							IMAGE_TO_IMAGE_IMAGE_PARAMS




 def   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]					)      ->			Union[str, Any]:
       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=__UpperCAmelCase					)
       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=1_000						,  )
       UpperCAmelCase_=       CLIPTextModel(__UpperCAmelCase					)
       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   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]						,  __UpperCAmelCase     :						Optional[Any]						,  __UpperCAmelCase     :						List[str]=0					)      ->			Tuple:
       UpperCAmelCase_=       floats_tensor((1, 3, 32, 32)						,  rng=random.Random(__UpperCAmelCase					)					).to(__UpperCAmelCase					)
       UpperCAmelCase_=       image.cpu().permute(0						,  2						,  3						,  1					)[0]
       UpperCAmelCase_=       Image.fromarray(np.uinta(__UpperCAmelCase					)					).convert("""RGB"""					)
       if str(__UpperCAmelCase					).startswith("""mps"""					):
             UpperCAmelCase_=       torch.manual_seed(__UpperCAmelCase					)
       else:
             UpperCAmelCase_=       torch.Generator(device=__UpperCAmelCase					).manual_seed(__UpperCAmelCase					)
       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   _SCREAMING_SNAKE_CASE		(     self     :						List[str]					)      ->			int:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe.to(__UpperCAmelCase					)
       sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

       UpperCAmelCase_=       self.get_dummy_inputs(__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe(**__UpperCAmelCase					).images
       UpperCAmelCase_=       image[0, -3:, -3:, -1]
       assert image.shape == (1, 32, 32, 3)
       UpperCAmelCase_=       np.array([0.7_526, 0.3_750, 0.4_547, 0.6_117, 0.5_866, 0.5_016, 0.4_327, 0.5_642, 0.4_815]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			Union[str, Any]:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe.to(__UpperCAmelCase					)
       sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

       UpperCAmelCase_=       self.get_dummy_inputs(__UpperCAmelCase					)
       UpperCAmelCase_=       """french fries"""
       UpperCAmelCase_=       sd_pipe(**__UpperCAmelCase						,  negative_prompt=__UpperCAmelCase					)
       UpperCAmelCase_=       output.images
       UpperCAmelCase_=       image[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       UpperCAmelCase_=       np.array([0.7_511, 0.3_642, 0.4_553, 0.6_236, 0.5_797, 0.5_013, 0.4_343, 0.5_611, 0.4_831]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						List[str]					)      ->			Dict:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe.to(__UpperCAmelCase					)
       sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

       UpperCAmelCase_=       self.get_dummy_inputs(__UpperCAmelCase					)
       UpperCAmelCase_=       [inputs["""prompt"""]] * 2

       UpperCAmelCase_=       np.array(inputs["""image"""]					).astype(np.floataa					) / 255.0
       UpperCAmelCase_=       torch.from_numpy(__UpperCAmelCase					).unsqueeze(0					).to(__UpperCAmelCase					)
       UpperCAmelCase_=       image / 2 + 0.5
       UpperCAmelCase_=       image.permute(0						,  3						,  1						,  2					)
       UpperCAmelCase_=       image.repeat(2						,  1						,  1						,  1					)

       UpperCAmelCase_=       sd_pipe(**__UpperCAmelCase					).images
       UpperCAmelCase_=       image[-1, -3:, -3:, -1]

       assert image.shape == (2, 32, 32, 3)
       UpperCAmelCase_=       np.array([0.5_812, 0.5_748, 0.5_222, 0.5_908, 0.5_695, 0.7_174, 0.6_804, 0.5_523, 0.5_579]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			Tuple:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       EulerAncestralDiscreteScheduler(
           beta_start=0.00_085						,  beta_end=0.012						,  beta_schedule="""scaled_linear"""					)
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe.to(__UpperCAmelCase					)
       sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

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

       UpperCAmelCase_=       [round(__UpperCAmelCase						,  4					) for x in image_slice.flatten().tolist()]
       print(""",""".join([str(__UpperCAmelCase					) for x in slice]					)					)

       assert image.shape == (1, 32, 32, 3)
       UpperCAmelCase_=       np.array([0.7_417, 0.3_842, 0.4_732, 0.5_776, 0.5_891, 0.5_139, 0.4_052, 0.5_673, 0.4_986]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			List[Any]:
       super().test_inference_batch_single_identical(expected_max_diff=3E-3					)




 def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			Dict:
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase					)
       UpperCAmelCase_=       VaeImageProcessor(do_resize=__UpperCAmelCase						,  do_normalize=__UpperCAmelCase					)
       UpperCAmelCase_=       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

       UpperCAmelCase_=       pipe(**self.get_dummy_inputs_by_type(__UpperCAmelCase						,  input_image_type="""pt"""					)					)[0]

       UpperCAmelCase_=       components["""vae"""]
       UpperCAmelCase_=       self.get_dummy_inputs_by_type(__UpperCAmelCase						,  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(**__UpperCAmelCase					)[0]

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






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






 """simple docstring"""


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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Union[str, Any]						,  __UpperCAmelCase     :						str=0					)      ->			Tuple:
       UpperCAmelCase_=       torch.manual_seed(__UpperCAmelCase					)
       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   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]					)      ->			str:
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline.from_pretrained(
           """timbrooks/instruct-pix2pix"""						,  safety_checker=__UpperCAmelCase					)
       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
       pipe.enable_attention_slicing()

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

       assert image.shape == (1, 512, 512, 3)
       UpperCAmelCase_=       np.array([0.5_902, 0.6_015, 0.6_027, 0.5_983, 0.6_092, 0.6_061, 0.5_765, 0.5_785, 0.5_555]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Union[str, Any]					)      ->			str:
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline.from_pretrained(
           """timbrooks/instruct-pix2pix"""						,  safety_checker=__UpperCAmelCase					)
       UpperCAmelCase_=       LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
       pipe.enable_attention_slicing()

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

       assert image.shape == (1, 512, 512, 3)
       UpperCAmelCase_=       np.array([0.6_578, 0.6_817, 0.6_972, 0.6_761, 0.6_856, 0.6_916, 0.6_428, 0.6_516, 0.6_301]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			Optional[Any]:
       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline.from_pretrained(
           """timbrooks/instruct-pix2pix"""						,  safety_checker=__UpperCAmelCase					)
       UpperCAmelCase_=       DDIMScheduler.from_config(pipe.scheduler.config					)
       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
       pipe.enable_attention_slicing()

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

       assert image.shape == (1, 512, 512, 3)
       UpperCAmelCase_=       np.array([0.3_828, 0.3_834, 0.3_818, 0.3_792, 0.3_865, 0.3_752, 0.3_792, 0.3_847, 0.3_753]					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						List[str]					)      ->			str:
       UpperCAmelCase_=       0

       def callback_fn(__UpperCAmelCase     :						int						,  __UpperCAmelCase     :						int						,  __UpperCAmelCase     :						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_463, -0.4_644, -0.9_756, 1.5_176, 1.4_414, 0.7_866, 0.9_897, 0.8_521, 0.7_983]					)

                   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_644, -0.4_626, -0.9_653, 1.5_176, 1.4_551, 0.7_686, 0.9_805, 0.8_452, 0.8_115]					)

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

       UpperCAmelCase_=       False

       UpperCAmelCase_=       StableDiffusionInstructPixaPixPipeline.from_pretrained(
           """timbrooks/instruct-pix2pix"""						,  safety_checker=__UpperCAmelCase						,  torch_dtype=torch.floataa					)
       UpperCAmelCase_=       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
       pipe.enable_attention_slicing()

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Tuple:
       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=__UpperCAmelCase						,  torch_dtype=torch.floataa					)
       UpperCAmelCase_=       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
       pipe.enable_attention_slicing(1					)
       pipe.enable_sequential_cpu_offload()

       UpperCAmelCase_=       self.get_inputs()
       UpperCAmelCase_=       pipe(**__UpperCAmelCase					)

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




 def   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Union[str, Any]:
       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(
           __UpperCAmelCase						,  safety_checker=__UpperCAmelCase						,  )
       pipe.to(__UpperCAmelCase					)
       pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
       pipe.enable_attention_slicing()

       UpperCAmelCase_=       pipe(**__UpperCAmelCase					)
       UpperCAmelCase_=       output.images[0]

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

       assert image.shape == (504, 504, 3)
       UpperCAmelCase_=       np.array([0.2_726, 0.2_529, 0.2_664, 0.2_655, 0.2_641, 0.2_642, 0.2_591, 0.2_649, 0.2_590]					)

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









from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader






class 		lowercase						(			snake_case__):






	"""simple docstring"""


	def __init__(     self     :						int						,  __UpperCAmelCase     :						pyspark.sql.DataFrame						,  __UpperCAmelCase     :						Optional[NamedSplit] = None						,  __UpperCAmelCase     :						Optional[Features] = None						,  __UpperCAmelCase     :						bool = True						,  __UpperCAmelCase     :						str = None						,  __UpperCAmelCase     :						bool = False						,  __UpperCAmelCase     :						str = None						,  __UpperCAmelCase     :						bool = True						,  __UpperCAmelCase     :						str = "arrow"						,  **__UpperCAmelCase     :						str						,  )      ->			Dict:
							super().__init__(
							    split=__UpperCAmelCase						,  features=__UpperCAmelCase						,  cache_dir=__UpperCAmelCase						,  keep_in_memory=__UpperCAmelCase						,  streaming=__UpperCAmelCase						,  **__UpperCAmelCase						,  )
							UpperCAmelCase_=       load_from_cache_file
							UpperCAmelCase_=       file_format
							UpperCAmelCase_=       Spark(
							    df=__UpperCAmelCase						,  features=__UpperCAmelCase						,  cache_dir=__UpperCAmelCase						,  working_dir=__UpperCAmelCase						,  **__UpperCAmelCase						,  )




	def   _SCREAMING_SNAKE_CASE		(     self     :						Any					)      ->			Dict:
							if self.streaming:
													return self.builder.as_streaming_dataset(split=self.split					)
							UpperCAmelCase_=       None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
							self.builder.download_and_prepare(
							    download_mode=__UpperCAmelCase						,  file_format=self._file_format						,  )
							return self.builder.as_dataset(split=self.split					)




