

import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionModelEditingPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()



@skip_mps
class 						lowercase_					(						__lowercase     ,							__lowercase     ,							__lowercase     ,							unittest.TestCase  ):
      UpperCamelCase_    :	Tuple		   =       StableDiffusionModelEditingPipeline
      UpperCamelCase_    :	int		   =       TEXT_TO_IMAGE_PARAMS
      UpperCamelCase_    :	Optional[int]		   =       TEXT_TO_IMAGE_BATCH_PARAMS
      UpperCamelCase_    :	str		   =       TEXT_TO_IMAGE_IMAGE_PARAMS
      UpperCamelCase_    :	Optional[int]		   =       TEXT_TO_IMAGE_IMAGE_PARAMS



      def 		UpperCamelCase_	(    self   :       Tuple	)					->  List[Any]:
        torch.manual_seed(0	)
        _snake_case		=							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    ,      )
        _snake_case		=							DDIMScheduler()
        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		=							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    ,      )
        _snake_case		=							CLIPTextModel(A__	)
        _snake_case		=							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''	)

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



      def 		UpperCamelCase_	(    self   :       List[Any]    ,      A__   :       int    ,      A__   :       Any=0	)					->  str:
        _snake_case		=							torch.manual_seed(A__	)
        _snake_case		=							{
            '''prompt''': '''A field of roses''',
            '''generator''': generator,
            # Setting height and width to None to prevent OOMs on CPU.
            '''height''': None,
            '''width''': None,
            '''num_inference_steps''': 2,
            '''guidance_scale''': 6.0,
            '''output_type''': '''numpy''',
        }
        return inputs



      def 		UpperCamelCase_	(    self   :       Any	)					->  Union[str, Any]:
        _snake_case		=							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
        _snake_case		=							self.get_dummy_components()
        _snake_case		=							StableDiffusionModelEditingPipeline(**A__	)
        _snake_case		=							sd_pipe.to(A__	)
        sd_pipe.set_progress_bar_config(disable=A__	)

        _snake_case		=							self.get_dummy_inputs(A__	)
        _snake_case		=							sd_pipe(**A__	).images
        _snake_case		=							image[0, -3:, -3:, -1]
        assert image.shape == (1, 64, 64, 3)

        _snake_case		=							np.array([0.4755, 0.5132, 0.4976, 0.3904, 0.3554, 0.4765, 0.5139, 0.5158, 0.4889]	)

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



      def 		UpperCamelCase_	(    self   :       Union[str, Any]	)					->  str:
        _snake_case		=							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
        _snake_case		=							self.get_dummy_components()
        _snake_case		=							StableDiffusionModelEditingPipeline(**A__	)
        _snake_case		=							sd_pipe.to(A__	)
        sd_pipe.set_progress_bar_config(disable=A__	)

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

        assert image.shape == (1, 64, 64, 3)

        _snake_case		=							np.array([0.4992, 0.5101, 0.5004, 0.3949, 0.3604, 0.4735, 0.5216, 0.5204, 0.4913]	)

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



      def 		UpperCamelCase_	(    self   :       List[Any]	)					->  Any:
        _snake_case		=							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
        _snake_case		=							self.get_dummy_components()
        _snake_case		=							EulerAncestralDiscreteScheduler(
            beta_start=0.0_0085    ,      beta_end=0.012    ,      beta_schedule='''scaled_linear'''	)
        _snake_case		=							StableDiffusionModelEditingPipeline(**A__	)
        _snake_case		=							sd_pipe.to(A__	)
        sd_pipe.set_progress_bar_config(disable=A__	)

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

        assert image.shape == (1, 64, 64, 3)

        _snake_case		=							np.array([0.4747, 0.5372, 0.4779, 0.4982, 0.5543, 0.4816, 0.5238, 0.4904, 0.5027]	)

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



      def 		UpperCamelCase_	(    self   :       Dict	)					->  List[Any]:
        _snake_case		=							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
        _snake_case		=							self.get_dummy_components()
        _snake_case		=							PNDMScheduler()
        _snake_case		=							StableDiffusionModelEditingPipeline(**A__	)
        _snake_case		=							sd_pipe.to(A__	)
        sd_pipe.set_progress_bar_config(disable=A__	)

        _snake_case		=							self.get_dummy_inputs(A__	)
        # the pipeline does not expect pndm so test if it raises error.
        with self.assertRaises(A__	):
          _snake_case		=							sd_pipe(**A__	).images



      def 		UpperCamelCase_	(    self   :       List[str]	)					->  str:
        super().test_inference_batch_single_identical(expected_max_diff=5e-3	)







      def 		UpperCamelCase_	(    self   :       Dict	)					->  Optional[Any]:
        super().test_attention_slicing_forward_pass(expected_max_diff=5e-3	)



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



      def 		UpperCamelCase_	(    self   :       Optional[int]	)					->  Optional[Any]:
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()



      def 		UpperCamelCase_	(    self   :       Optional[Any]    ,      A__   :       Dict=0	)					->  str:
        _snake_case		=							torch.manual_seed(A__	)
        _snake_case		=							{
            '''prompt''': '''A field of roses''',
            '''generator''': generator,
            '''num_inference_steps''': 3,
            '''guidance_scale''': 7.5,
            '''output_type''': '''numpy''',
        }
        return inputs



      def 		UpperCamelCase_	(    self   :       Dict	)					->  int:
        _snake_case		=							'''CompVis/stable-diffusion-v1-4'''
        _snake_case		=							StableDiffusionModelEditingPipeline.from_pretrained(A__    ,      safety_checker=A__	)
        pipe.to(A__	)
        pipe.set_progress_bar_config(disable=A__	)
        pipe.enable_attention_slicing()

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

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

        _snake_case		=							np.array(
            [0.674_9496, 0.638_6453, 0.5144_3267, 0.6609_4905, 0.6192_1215, 0.549_1332, 0.574_4417, 0.5807_5106, 0.517_4658]	)

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

        # make sure image changes after editing
        pipe.edit_model('''A pack of roses'''    ,      '''A pack of blue roses'''	)

        _snake_case		=							pipe(**A__	).images
        _snake_case		=							image[0, -3:, -3:, -1].flatten()

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

        assert np.abs(expected_slice - image_slice	).max() > 1e-1







      def 		UpperCamelCase_	(    self   :       Optional[int]	)					->  Union[str, Any]:
        torch.cuda.empty_cache()
        torch.cuda.reset_max_memory_allocated()
        torch.cuda.reset_peak_memory_stats()

        _snake_case		=							'''CompVis/stable-diffusion-v1-4'''
        _snake_case		=							DDIMScheduler.from_pretrained(A__    ,      subfolder='''scheduler'''	)
        _snake_case		=							StableDiffusionModelEditingPipeline.from_pretrained(
            A__    ,      scheduler=A__    ,      safety_checker=A__	)
        _snake_case		=							pipe.to(A__	)
        pipe.set_progress_bar_config(disable=A__	)
        pipe.enable_attention_slicing(1	)
        pipe.enable_sequential_cpu_offload()

        _snake_case		=							self.get_inputs()
        _snake_case		=							pipe(**A__	)

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




from collections import namedtuple

__A    	=			namedtuple('''from_to''', '''from_ to''')

__A    	=			{
    '''atm''': from_to(1, 1),
    '''pascal''': from_to(0.000_0098, 10_13_25),
    '''bar''': from_to(0.98_6923, 1.0_1325),
    '''kilopascal''': from_to(0.0098_6923, 101.325),
    '''megapascal''': from_to(9.8_6923, 0.10_1325),
    '''psi''': from_to(0.06_8046, 14.6959),
    '''inHg''': from_to(0.033_4211, 29.9213),
    '''torr''': from_to(0.0013_1579, 7_60),
}



def    snake_case_(_UpperCamelCase				, _UpperCamelCase				, _UpperCamelCase  )							->			float:



  """simple docstring"""






  if from_type not in PRESSURE_CONVERSION:
    raise ValueError(
        F"""Invalid 'from_type' value: {from_type!r}  Supported values are:\n"""
        + ''', '''.join(_UpperCamelCase  )  )
  if to_type not in PRESSURE_CONVERSION:
    raise ValueError(
        F"""Invalid 'to_type' value: {to_type!r}.  Supported values are:\n"""
        + ''', '''.join(_UpperCamelCase  )  )
  return (
      value * PRESSURE_CONVERSION[from_type].from_ * PRESSURE_CONVERSION[to_type].to
  )


if __name__ == "__main__":
 import doctest

 doctest.testmod()

