



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_		(  a__	,   a__	,   a__	,   unittest.TestCase							):
       UpperCAmelCase__    : int       					=  StableDiffusionInstructPixaPixPipeline
       UpperCAmelCase__    : Tuple       					=  TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width", "cross_attention_kwargs"}
       UpperCAmelCase__    : Tuple       					=  TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
       UpperCAmelCase__    : List[str]       					=  IMAGE_TO_IMAGE_IMAGE_PARAMS
       UpperCAmelCase__    : List[str]       					=  IMAGE_TO_IMAGE_IMAGE_PARAMS
       def  snake_case_	(      self	)					->		Any:
          torch.manual_seed(0	)
          UpperCamelCase   :  List[Any]						     =     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   :  Dict						     =     PNDMScheduler(skip_prk_steps=__A	)
          torch.manual_seed(0	)
          UpperCamelCase   :  Optional[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	)
          UpperCamelCase   :  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=1000,      )
          UpperCamelCase   :  Optional[int]						     =     CLIPTextModel(__A	)
          UpperCamelCase   :  int						     =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)

          UpperCamelCase   :  int						     =     {
              'unet': unet,
              'scheduler': scheduler,
              'vae': vae,
              'text_encoder': text_encoder,
              'tokenizer': tokenizer,
              'safety_checker': None,
              'feature_extractor': None,
          }
          return components
       def  snake_case_	(      self,      SCREAMING_SNAKE_CASE_,      SCREAMING_SNAKE_CASE_=0	)					->		List[str]:
          UpperCamelCase   :  Dict						     =     floats_tensor((1, 3, 32, 32),      rng=random.Random(__A	)	).to(__A	)
          UpperCamelCase   :  Tuple						     =     image.cpu().permute(0,      2,      3,      1	)[0]
          UpperCamelCase   :  Dict						     =     Image.fromarray(np.uinta(__A	)	).convert('RGB'	)
          if str(__A	).startswith('mps'	):
             UpperCamelCase   :  Optional[int]						     =     torch.manual_seed(__A	)
          else:
             UpperCamelCase   :  str						     =     torch.Generator(device=__A	).manual_seed(__A	)
          UpperCamelCase   :  Optional[int]						     =     {
              '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	)					->		Union[str, Any]:
          UpperCamelCase   :  str						     =     'cpu'  # ensure determinism for the device-dependent torch.Generator
          UpperCamelCase   :  Optional[Any]						     =     self.get_dummy_components()
          UpperCamelCase   :  Optional[Any]						     =     StableDiffusionInstructPixaPixPipeline(**__A	)
          UpperCamelCase   :  List[Any]						     =     sd_pipe.to(__A	)
          sd_pipe.set_progress_bar_config(disable=__A	)

          UpperCamelCase   :  Union[str, Any]						     =     self.get_dummy_inputs(__A	)
          UpperCamelCase   :  Optional[Any]						     =     sd_pipe(**__A	).images
          UpperCamelCase   :  Tuple						     =     image[0, -3:, -3:, -1]
          assert image.shape == (1, 32, 32, 3)
          UpperCamelCase   :  Optional[Any]						     =     np.array([0.75_26, 0.37_50, 0.45_47, 0.61_17, 0.58_66, 0.50_16, 0.43_27, 0.56_42, 0.48_15]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		int:
          UpperCamelCase   :  int						     =     'cpu'  # ensure determinism for the device-dependent torch.Generator
          UpperCamelCase   :  int						     =     self.get_dummy_components()
          UpperCamelCase   :  Any						     =     StableDiffusionInstructPixaPixPipeline(**__A	)
          UpperCamelCase   :  str						     =     sd_pipe.to(__A	)
          sd_pipe.set_progress_bar_config(disable=__A	)

          UpperCamelCase   :  int						     =     self.get_dummy_inputs(__A	)
          UpperCamelCase   :  List[str]						     =     'french fries'
          UpperCamelCase   :  Optional[Any]						     =     sd_pipe(**__A,      negative_prompt=__A	)
          UpperCamelCase   :  str						     =     output.images
          UpperCamelCase   :  int						     =     image[0, -3:, -3:, -1]

          assert image.shape == (1, 32, 32, 3)
          UpperCamelCase   :  int						     =     np.array([0.75_11, 0.36_42, 0.45_53, 0.62_36, 0.57_97, 0.50_13, 0.43_43, 0.56_11, 0.48_31]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		Optional[Any]:
          UpperCamelCase   :  List[Any]						     =     'cpu'  # ensure determinism for the device-dependent torch.Generator
          UpperCamelCase   :  Optional[int]						     =     self.get_dummy_components()
          UpperCamelCase   :  Optional[Any]						     =     StableDiffusionInstructPixaPixPipeline(**__A	)
          UpperCamelCase   :  Tuple						     =     sd_pipe.to(__A	)
          sd_pipe.set_progress_bar_config(disable=__A	)

          UpperCamelCase   :  str						     =     self.get_dummy_inputs(__A	)
          UpperCamelCase   :  int						     =     [inputs['prompt']] * 2

          UpperCamelCase   :  Dict						     =     np.array(inputs['image']	).astype(np.floataa	) / 2_55.0
          UpperCamelCase   :  Tuple						     =     torch.from_numpy(__A	).unsqueeze(0	).to(__A	)
          UpperCamelCase   :  str						     =     image / 2 + 0.5
          UpperCamelCase   :  Dict						     =     image.permute(0,      3,      1,      2	)
          UpperCamelCase   :  Optional[Any]						     =     image.repeat(2,      1,      1,      1	)

          UpperCamelCase   :  Dict						     =     sd_pipe(**__A	).images
          UpperCamelCase   :  Tuple						     =     image[-1, -3:, -3:, -1]

          assert image.shape == (2, 32, 32, 3)
          UpperCamelCase   :  Any						     =     np.array([0.58_12, 0.57_48, 0.52_22, 0.59_08, 0.56_95, 0.71_74, 0.68_04, 0.55_23, 0.55_79]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		Any:
          UpperCamelCase   :  Optional[int]						     =     'cpu'  # ensure determinism for the device-dependent torch.Generator
          UpperCamelCase   :  Optional[Any]						     =     self.get_dummy_components()
          UpperCamelCase   :  Optional[int]						     =     EulerAncestralDiscreteScheduler(
              beta_start=0.0_00_85,      beta_end=0.0_12,      beta_schedule='scaled_linear'	)
          UpperCamelCase   :  int						     =     StableDiffusionInstructPixaPixPipeline(**__A	)
          UpperCamelCase   :  Union[str, Any]						     =     sd_pipe.to(__A	)
          sd_pipe.set_progress_bar_config(disable=__A	)

          UpperCamelCase   :  Optional[int]						     =     self.get_dummy_inputs(__A	)
          UpperCamelCase   :  int						     =     sd_pipe(**__A	).images
          UpperCamelCase   :  Union[str, Any]						     =     image[0, -3:, -3:, -1]

          UpperCamelCase   :  List[str]						     =     [round(__A,      4	) for x in image_slice.flatten().tolist()]
          print(','.join([str(__A	) for x in slice]	)	)

          assert image.shape == (1, 32, 32, 3)
          UpperCamelCase   :  List[Any]						     =     np.array([0.74_17, 0.38_42, 0.47_32, 0.57_76, 0.58_91, 0.51_39, 0.40_52, 0.56_73, 0.49_86]	)

          assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		Any:
          super().test_inference_batch_single_identical(expected_max_diff=3e-3	)






       def  snake_case_	(      self	)					->		str:
          UpperCamelCase   :  Dict						     =     self.get_dummy_components()
          UpperCamelCase   :  Dict						     =     StableDiffusionInstructPixaPixPipeline(**__A	)
          UpperCamelCase   :  Tuple						     =     VaeImageProcessor(do_resize=__A,      do_normalize=__A	)
          UpperCamelCase   :  Optional[Any]						     =     pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)

          UpperCamelCase   :  Optional[Any]						     =     pipe(**self.get_dummy_inputs_by_type(__A,      input_image_type='pt'	)	)[0]

          UpperCamelCase   :  List[str]						     =     components['vae']
          UpperCamelCase   :  Union[str, Any]						     =     self.get_dummy_inputs_by_type(__A,      input_image_type='pt'	)

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

          UpperCamelCase   :  Dict						     =     pipe(**__A	)[0]

          UpperCamelCase   :  Union[str, Any]						     =     np.abs(out - out_latents_inputs	).max()
          self.assertLess(__A,      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	)					->		Union[str, Any]:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()
       def  snake_case_	(      self,      SCREAMING_SNAKE_CASE_=0	)					->		Optional[int]:
          UpperCamelCase   :  int						     =     torch.manual_seed(__A	)
          UpperCamelCase   :  int						     =     load_image(
              'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg'	)
          UpperCamelCase   :  Optional[Any]						     =     {
              '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	)					->		str:
          UpperCamelCase   :  Optional[Any]						     =     StableDiffusionInstructPixaPixPipeline.from_pretrained(
              'timbrooks/instruct-pix2pix',      safety_checker=__A	)
          pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)
          pipe.enable_attention_slicing()

          UpperCamelCase   :  List[str]						     =     self.get_inputs()
          UpperCamelCase   :  Optional[Any]						     =     pipe(**__A	).images
          UpperCamelCase   :  Any						     =     image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          UpperCamelCase   :  Dict						     =     np.array([0.59_02, 0.60_15, 0.60_27, 0.59_83, 0.60_92, 0.60_61, 0.57_65, 0.57_85, 0.55_55]	)

          assert np.abs(expected_slice - image_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		Tuple:
          UpperCamelCase   :  List[str]						     =     StableDiffusionInstructPixaPixPipeline.from_pretrained(
              'timbrooks/instruct-pix2pix',      safety_checker=__A	)
          UpperCamelCase   :  Union[str, Any]						     =     LMSDiscreteScheduler.from_config(pipe.scheduler.config	)
          pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)
          pipe.enable_attention_slicing()

          UpperCamelCase   :  Any						     =     self.get_inputs()
          UpperCamelCase   :  Optional[Any]						     =     pipe(**__A	).images
          UpperCamelCase   :  str						     =     image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          UpperCamelCase   :  Optional[int]						     =     np.array([0.65_78, 0.68_17, 0.69_72, 0.67_61, 0.68_56, 0.69_16, 0.64_28, 0.65_16, 0.63_01]	)

          assert np.abs(expected_slice - image_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		str:
          UpperCamelCase   :  Dict						     =     StableDiffusionInstructPixaPixPipeline.from_pretrained(
              'timbrooks/instruct-pix2pix',      safety_checker=__A	)
          UpperCamelCase   :  str						     =     DDIMScheduler.from_config(pipe.scheduler.config	)
          pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)
          pipe.enable_attention_slicing()

          UpperCamelCase   :  Optional[int]						     =     self.get_inputs()
          UpperCamelCase   :  List[Any]						     =     pipe(**__A	).images
          UpperCamelCase   :  List[Any]						     =     image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          UpperCamelCase   :  Tuple						     =     np.array([0.38_28, 0.38_34, 0.38_18, 0.37_92, 0.38_65, 0.37_52, 0.37_92, 0.38_47, 0.37_53]	)

          assert np.abs(expected_slice - image_slice	).max() < 1e-3
       def  snake_case_	(      self	)					->		List[str]:
          UpperCamelCase   :  Tuple						     =     0

          def callback_fn(SCREAMING_SNAKE_CASE_,      SCREAMING_SNAKE_CASE_,      SCREAMING_SNAKE_CASE_	) -> None:
             UpperCamelCase   :  Any						     =     True
             nonlocal number_of_steps
             number_of_steps += 1
             if step == 1:
                UpperCamelCase   :  Dict						     =     latents.detach().cpu().numpy()
                assert latents.shape == (1, 4, 64, 64)
                UpperCamelCase   :  Dict						     =     latents[0, -3:, -3:, -1]
                UpperCamelCase   :  Dict						     =     np.array([-0.24_63, -0.46_44, -0.97_56, 1.51_76, 1.44_14, 0.78_66, 0.98_97, 0.85_21, 0.79_83]	)

                assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5e-2
             elif step == 2:
                UpperCamelCase   :  Dict						     =     latents.detach().cpu().numpy()
                assert latents.shape == (1, 4, 64, 64)
                UpperCamelCase   :  Tuple						     =     latents[0, -3:, -3:, -1]
                UpperCamelCase   :  Union[str, Any]						     =     np.array([-0.26_44, -0.46_26, -0.96_53, 1.51_76, 1.45_51, 0.76_86, 0.98_05, 0.84_52, 0.81_15]	)

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

          UpperCamelCase   :  Optional[Any]						     =     False

          UpperCamelCase   :  Dict						     =     StableDiffusionInstructPixaPixPipeline.from_pretrained(
              'timbrooks/instruct-pix2pix',      safety_checker=__A,      torch_dtype=torch.floataa	)
          UpperCamelCase   :  List[Any]						     =     pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)
          pipe.enable_attention_slicing()

          UpperCamelCase   :  Optional[int]						     =     self.get_inputs()
          pipe(**__A,      callback=__A,      callback_steps=1	)
          assert callback_fn.has_been_called
          assert number_of_steps == 3
       def  snake_case_	(      self	)					->		List[str]:
          torch.cuda.empty_cache()
          torch.cuda.reset_max_memory_allocated()
          torch.cuda.reset_peak_memory_stats()

          UpperCamelCase   :  Optional[int]						     =     StableDiffusionInstructPixaPixPipeline.from_pretrained(
              'timbrooks/instruct-pix2pix',      safety_checker=__A,      torch_dtype=torch.floataa	)
          UpperCamelCase   :  Optional[int]						     =     pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)
          pipe.enable_attention_slicing(1	)
          pipe.enable_sequential_cpu_offload()

          UpperCamelCase   :  Dict						     =     self.get_inputs()
          UpperCamelCase   :  List[str]						     =     pipe(**__A	)

          UpperCamelCase   :  Tuple						     =     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	)					->		Tuple:
          UpperCamelCase   :  str						     =     self.get_inputs()
          # resize to resolution that is divisible by 8 but not 16 or 32
          UpperCamelCase   :  str						     =     inputs['image'].resize((504, 504)	)

          UpperCamelCase   :  Dict						     =     'timbrooks/instruct-pix2pix'
          UpperCamelCase   :  Optional[Any]						     =     StableDiffusionInstructPixaPixPipeline.from_pretrained(
              __A,      safety_checker=__A,      )
          pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)
          pipe.enable_attention_slicing()

          UpperCamelCase   :  Any						     =     pipe(**__A	)
          UpperCamelCase   :  Tuple						     =     output.images[0]

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

          assert image.shape == (504, 504, 3)
          UpperCamelCase   :  str						     =     np.array([0.27_26, 0.25_29, 0.26_64, 0.26_55, 0.26_41, 0.26_42, 0.25_91, 0.26_49, 0.25_90]	)

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




import hashlib
import unittest

from transformers import MODEL_FOR_DEPTH_ESTIMATION_MAPPING, is_torch_available, is_vision_available
from transformers.pipelines import DepthEstimationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_timm,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image
else:



    class     lowerCAmelCase_		:
           @staticmethod
           def  snake_case_	(      *SCREAMING_SNAKE_CASE_,      **SCREAMING_SNAKE_CASE_	)					->		Dict:
              pass
def 		UpperCamelCase     (       snake_case__	:   Image     )     ->					str:
   UpperCamelCase   :  Tuple						     =     hashlib.mda(image.tobytes()     )
   return m.hexdigest()



@is_pipeline_test
@require_vision
@require_timm
@require_torch
class     lowerCAmelCase_		(  unittest.TestCase							):
       UpperCAmelCase__    : str       					=  MODEL_FOR_DEPTH_ESTIMATION_MAPPING
       def  snake_case_	(      self,      SCREAMING_SNAKE_CASE_,      SCREAMING_SNAKE_CASE_,      SCREAMING_SNAKE_CASE_	)					->		Optional[Any]:
          UpperCamelCase   :  List[str]						     =     DepthEstimationPipeline(model=SCREAMING_SNAKE_CASE_,      image_processor=SCREAMING_SNAKE_CASE_	)
          return depth_estimator, [
              "./tests/fixtures/tests_samples/COCO/000000039769.png",
              "./tests/fixtures/tests_samples/COCO/000000039769.png",
          ]
       def  snake_case_	(      self,      SCREAMING_SNAKE_CASE_,      SCREAMING_SNAKE_CASE_	)					->		Union[str, Any]:
          UpperCamelCase   :  Union[str, Any]						     =     depth_estimator('./tests/fixtures/tests_samples/COCO/000000039769.png'	)
          self.assertEqual({'predicted_depth': ANY(torch.Tensor	), 'depth': ANY(Image.Image	)},      SCREAMING_SNAKE_CASE_	)
          import datasets

          UpperCamelCase   :  int						     =     datasets.load_dataset('hf-internal-testing/fixtures_image_utils',      'image',      split='test'	)
          UpperCamelCase   :  Tuple						     =     depth_estimator(
              [
                  Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'	),
                  'http://images.cocodataset.org/val2017/000000039769.jpg',
                  # RGBA
                  dataset[0]['file'],
                  # LA
                  dataset[1]['file'],
                  # L
                  dataset[2]['file'],
              ]	)
          self.assertEqual(
              [
                  {'predicted_depth': ANY(torch.Tensor	), 'depth': ANY(Image.Image	)},
                  {'predicted_depth': ANY(torch.Tensor	), 'depth': ANY(Image.Image	)},
                  {'predicted_depth': ANY(torch.Tensor	), 'depth': ANY(Image.Image	)},
                  {'predicted_depth': ANY(torch.Tensor	), 'depth': ANY(Image.Image	)},
                  {'predicted_depth': ANY(torch.Tensor	), 'depth': ANY(Image.Image	)},
              ],      SCREAMING_SNAKE_CASE_,      )
       @require_tf
       @unittest.skip('Depth estimation is not implemented in TF'	)
       def  snake_case_	(      self	)					->		Optional[int]:
          pass
       @slow
       @require_torch
       def  snake_case_	(      self	)					->		List[Any]:
          UpperCamelCase   :  Optional[Any]						     =     'Intel/dpt-large'
          UpperCamelCase   :  Tuple						     =     pipeline('depth-estimation',      model=SCREAMING_SNAKE_CASE_	)
          UpperCamelCase   :  Any						     =     depth_estimator('http://images.cocodataset.org/val2017/000000039769.jpg'	)
          UpperCamelCase   :  int						     =     hashimage(outputs['depth']	)

          # This seems flaky.
          # self.assertEqual(outputs["depth"], "1a39394e282e9f3b0741a90b9f108977")
          self.assertEqual(nested_simplify(outputs['predicted_depth'].max().item()	),      29.3_04	)
          self.assertEqual(nested_simplify(outputs['predicted_depth'].min().item()	),      2.6_62	)






       @require_torch
       def  snake_case_	(      self	)					->		Optional[int]:
          # This is highly irregular to have no small tests.
          self.skipTest('There is not hf-internal-testing tiny model for either GLPN nor DPT'	)



