


'''simple docstring'''






import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

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


enable_full_determinism()
class   __UpperCamelCase		(  a_  ,     a_  ,     unittest.TestCase							):
							lowercase	:						List[Any]      						=     StableDiffusionParadigmsPipeline
							lowercase	:						Tuple      						=     TEXT_TO_IMAGE_PARAMS
							lowercase	:						List[Any]      						=     TEXT_TO_IMAGE_BATCH_PARAMS
							lowercase	:						Union[str, Any]      						=     TEXT_TO_IMAGE_IMAGE_PARAMS
							lowercase	:						Optional[Any]      						=     TEXT_TO_IMAGE_IMAGE_PARAMS





							def 						a__      (     self						:Any	):
									torch.manual_seed(0	)
									snake_case_						:							Optional[int]					=  UNetaDConditionModel(
									    block_out_channels=(3_2, 6_4)     ,layers_per_block=2     ,sample_size=3_2     ,in_channels=4     ,out_channels=4     ,down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")     ,up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")     ,cross_attention_dim=3_2     ,attention_head_dim=(2, 4)     ,use_linear_projection=_UpperCamelCase     ,)
									snake_case_						:							Union[str, Any]					=  DDIMParallelScheduler(
									    beta_start=0.0_00_85     ,beta_end=0.0_12     ,beta_schedule="""scaled_linear"""     ,clip_sample=_UpperCamelCase     ,set_alpha_to_one=_UpperCamelCase     ,)
									torch.manual_seed(0	)
									snake_case_						:							Optional[int]					=  AutoencoderKL(
									    block_out_channels=[3_2, 6_4]     ,in_channels=3     ,out_channels=3     ,down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]     ,up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]     ,latent_channels=4     ,sample_size=1_2_8     ,)
									torch.manual_seed(0	)
									snake_case_						:							List[Any]					=  CLIPTextConfig(
									    bos_token_id=0     ,eos_token_id=2     ,hidden_size=3_2     ,intermediate_size=3_7     ,layer_norm_eps=1E-0_5     ,num_attention_heads=4     ,num_hidden_layers=5     ,pad_token_id=1     ,vocab_size=1_0_0_0     ,hidden_act="""gelu"""     ,projection_dim=5_1_2     ,)
									snake_case_						:							Dict					=  CLIPTextModel(_UpperCamelCase	)
									snake_case_						:							Optional[int]					=  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""	)

									snake_case_						:							List[Any]					=  {
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									    """text_encoder""": text_encoder,
									    """tokenizer""": tokenizer,
									    """safety_checker""": None,
									    """feature_extractor""": None,
									}
									return components





							def 						a__      (     self						:List[str]     ,_UpperCamelCase						:List[str]     ,_UpperCamelCase						:str=0	):







									if str(_UpperCamelCase	).startswith("""mps"""	):
											snake_case_						:							List[Any]					=  torch.manual_seed(_UpperCamelCase	)
									else:
											snake_case_						:							Optional[int]					=  torch.Generator(device=_UpperCamelCase	).manual_seed(_UpperCamelCase	)
									snake_case_						:							Any					=  {
									    """prompt""": """a photograph of an astronaut riding a horse""",
									    """generator""": generator,
									    """num_inference_steps""": 1_0,
									    """guidance_scale""": 6.0,
									    """output_type""": """numpy""",
									    """parallel""": 3,
									    """debug""": True,
									}
									return inputs





							def 						a__      (     self						:List[str]	):
									snake_case_						:							Union[str, Any]					=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
									snake_case_						:							Union[str, Any]					=  self.get_dummy_components()
									snake_case_						:							List[str]					=  StableDiffusionParadigmsPipeline(**_UpperCamelCase	)
									snake_case_						:							Optional[Any]					=  sd_pipe.to(_UpperCamelCase	)
									sd_pipe.set_progress_bar_config(disable=_UpperCamelCase	)

									snake_case_						:							Optional[Any]					=  self.get_dummy_inputs(_UpperCamelCase	)
									snake_case_						:							Union[str, Any]					=  sd_pipe(**_UpperCamelCase	).images
									snake_case_						:							Union[str, Any]					=  image[0, -3:, -3:, -1]
									assert image.shape == (1, 6_4, 6_4, 3)

									snake_case_						:							int					=  np.array([0.47_73, 0.54_17, 0.47_23, 0.49_25, 0.56_31, 0.47_52, 0.52_40, 0.49_35, 0.50_23]	)

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





							def 						a__      (     self						:Optional[Any]	):
									snake_case_						:							Optional[Any]					=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
									snake_case_						:							str					=  self.get_dummy_components()
									torch.manual_seed(0	)
									snake_case_						:							List[Any]					=  DDPMParallelScheduler()
									torch.manual_seed(0	)
									snake_case_						:							Any					=  StableDiffusionParadigmsPipeline(**_UpperCamelCase	)
									snake_case_						:							Any					=  sd_pipe.to(_UpperCamelCase	)
									sd_pipe.set_progress_bar_config(disable=_UpperCamelCase	)

									snake_case_						:							Dict					=  self.get_dummy_inputs(_UpperCamelCase	)
									snake_case_						:							Dict					=  sd_pipe(**_UpperCamelCase	).images
									snake_case_						:							str					=  image[0, -3:, -3:, -1]
									assert image.shape == (1, 6_4, 6_4, 3)

									snake_case_						:							List[str]					=  np.array([0.35_73, 0.44_20, 0.49_60, 0.47_99, 0.37_96, 0.38_79, 0.48_19, 0.43_65, 0.44_68]	)

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





							def 						a__      (     self						:List[Any]	):
									super().test_inference_batch_consistent(batch_sizes=[1, 2]	)





							def 						a__      (     self						:Optional[Any]	):
									super().test_inference_batch_single_identical(batch_size=2     ,expected_max_diff=3E-3	)





							def 						a__      (     self						:Dict	):
									snake_case_						:							List[str]					=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
									snake_case_						:							Union[str, Any]					=  self.get_dummy_components()
									snake_case_						:							str					=  StableDiffusionParadigmsPipeline(**_UpperCamelCase	)
									snake_case_						:							str					=  sd_pipe.to(_UpperCamelCase	)
									sd_pipe.set_progress_bar_config(disable=_UpperCamelCase	)

									snake_case_						:							List[Any]					=  self.get_dummy_inputs(_UpperCamelCase	)
									snake_case_						:							Optional[int]					=  """french fries"""
									snake_case_						:							Union[str, Any]					=  sd_pipe(**_UpperCamelCase     ,negative_prompt=_UpperCamelCase	)
									snake_case_						:							Tuple					=  output.images
									snake_case_						:							int					=  image[0, -3:, -3:, -1]

									assert image.shape == (1, 6_4, 6_4, 3)

									snake_case_						:							List[Any]					=  np.array([0.47_71, 0.54_20, 0.46_83, 0.49_18, 0.56_36, 0.47_25, 0.52_30, 0.49_23, 0.50_15]	)

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





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





							def 						a__      (     self						:Dict	):
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()





							def 						a__      (     self						:Optional[int]     ,_UpperCamelCase						:List[Any]=0	):
									snake_case_						:							Optional[int]					=  torch.Generator(device=_UpperCamelCase	).manual_seed(_UpperCamelCase	)
									snake_case_						:							Any					=  {
									    """prompt""": """a photograph of an astronaut riding a horse""",
									    """generator""": generator,
									    """num_inference_steps""": 1_0,
									    """guidance_scale""": 7.5,
									    """output_type""": """numpy""",
									    """parallel""": 3,
									    """debug""": True,
									}
									return inputs





							def 						a__      (     self						:Tuple	):
									snake_case_						:							Optional[int]					=  """stabilityai/stable-diffusion-2-base"""
									snake_case_						:							Tuple					=  DDIMParallelScheduler.from_pretrained(_UpperCamelCase     ,subfolder="""scheduler"""	)
									snake_case_						:							Union[str, Any]					=  StableDiffusionParadigmsPipeline.from_pretrained(_UpperCamelCase     ,scheduler=_UpperCamelCase     ,safety_checker=_UpperCamelCase	)
									pipe.to(_UpperCamelCase	)
									pipe.set_progress_bar_config(disable=_UpperCamelCase	)
									pipe.enable_attention_slicing()

									snake_case_						:							Union[str, Any]					=  self.get_inputs()
									snake_case_						:							List[Any]					=  pipe(**_UpperCamelCase	).images
									snake_case_						:							List[str]					=  image[0, -3:, -3:, -1].flatten()

									assert image.shape == (1, 5_1_2, 5_1_2, 3)

									snake_case_						:							List[str]					=  np.array([0.96_22, 0.96_02, 0.97_48, 0.95_91, 0.96_30, 0.96_91, 0.96_61, 0.96_31, 0.97_41]	)

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


import copy
import inspect

from transformers.testing_utils import require_torch, torch_device
from transformers.utils.backbone_utils import BackboneType



@require_torch
class SCREAMING_SNAKE_CASE	:
  """simple docstring"""






  lowerCamelCase    :			List[Any]						      =()
  lowerCamelCase    :			Union[str, Any]						      =True







  def        SCREAMING_SNAKE_CASE    (			self :    Union[str, Any]    )							->	Union[str, Any]:






     """simple docstring"""







     __lowerCAmelCase  :			List[Any]					    =						self.config_class

     # test default config
     __lowerCAmelCase  :			Union[str, Any]					    =						config_class()
     self.assertIsNotNone(lowerCAmelCase    )
     __lowerCAmelCase  :			Optional[int]					    =						["""stem"""] + [f'''stage{idx}''' for idx in range(1       ,							len(config.depths    ) + 1    )]
     self.assertEqual(config.stage_names       ,							lowerCAmelCase    )
     self.assertTrue(set(config.out_features    ).issubset(set(config.stage_names    )    )    )

     # Test out_features and out_indices are correctly set
     # out_features and out_indices both None
     __lowerCAmelCase  :			str					    =						config_class(out_features=lowerCAmelCase       ,							out_indices=lowerCAmelCase    )
     self.assertEqual(config.out_features       ,							[config.stage_names[-1]]    )
     self.assertEqual(config.out_indices       ,							[len(config.stage_names    ) - 1]    )

     # out_features and out_indices both set
     __lowerCAmelCase  :			str					    =						config_class(out_features=["""stem""", """stage1"""]       ,							out_indices=[0, 1]    )
     self.assertEqual(config.out_features       ,							["""stem""", """stage1"""]    )
     self.assertEqual(config.out_indices       ,							[0, 1]    )

     # Only out_features set
     __lowerCAmelCase  :			Any					    =						config_class(out_features=["""stage1""", """stage3"""]    )
     self.assertEqual(config.out_features       ,							["""stage1""", """stage3"""]    )
     self.assertEqual(config.out_indices       ,							[1, 3]    )

     # Only out_indices set
     __lowerCAmelCase  :			str					    =						config_class(out_indices=[0, 2]    )
     self.assertEqual(config.out_features       ,							[config.stage_names[0], config.stage_names[2]]    )
     self.assertEqual(config.out_indices       ,							[0, 2]    )

     # Error raised when out_indices do not correspond to out_features
     with self.assertRaises(lowerCAmelCase    ):
        __lowerCAmelCase  :			Dict					    =						config_class(out_features=["""stage1""", """stage2"""]       ,							out_indices=[0, 2]    )







  def        SCREAMING_SNAKE_CASE    (			self :    str    )							->	Optional[int]:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			Union[str, Any]					    =						self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        __lowerCAmelCase  :			List[Any]					    =						model_class(lowerCAmelCase    )
        __lowerCAmelCase  :			str					    =						inspect.signature(model.forward    )
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        __lowerCAmelCase  :			List[Any]					    =						[*signature.parameters.keys()]
        __lowerCAmelCase  :			Union[str, Any]					    =						["""pixel_values"""]
        self.assertListEqual(arg_names[:1]       ,							lowerCAmelCase    )







  def        SCREAMING_SNAKE_CASE    (			self :    List[Any]    )							->	Tuple:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			Dict					    =						self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        __lowerCAmelCase  :			List[Any]					    =						model_class(lowerCAmelCase    )
        self.assertEqual(len(model.channels    )       ,							len(config.out_features    )    )
        __lowerCAmelCase  :			List[Any]					    =						model.num_features
        __lowerCAmelCase  :			List[str]					    =						[config.stage_names.index(lowerCAmelCase    ) for feat in config.out_features]
        __lowerCAmelCase  :			Union[str, Any]					    =						[num_features[idx] for idx in out_indices]
        self.assertListEqual(model.channels       ,							lowerCAmelCase    )

        __lowerCAmelCase  :			int					    =						copy.deepcopy(lowerCAmelCase    )
        __lowerCAmelCase  :			str					    =						None
        __lowerCAmelCase  :			List[Any]					    =						model_class(lowerCAmelCase    )
        self.assertEqual(len(model.channels    )       ,							1    )
        self.assertListEqual(model.channels       ,							[num_features[-1]]    )

        __lowerCAmelCase  :			Optional[Any]					    =						copy.deepcopy(lowerCAmelCase    )
        __lowerCAmelCase  :			List[Any]					    =						None
        __lowerCAmelCase  :			Any					    =						model_class(lowerCAmelCase    )
        self.assertEqual(len(model.channels    )       ,							1    )
        self.assertListEqual(model.channels       ,							[num_features[-1]]    )







  def        SCREAMING_SNAKE_CASE    (			self :    Union[str, Any]    )							->	Optional[int]:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			str					    =						self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        __lowerCAmelCase  :			List[Any]					    =						model_class(lowerCAmelCase    )
        model.to(lowerCAmelCase    )
        model.eval()
        __lowerCAmelCase  :			str					    =						model(**lowerCAmelCase    )

        self.assertEqual(len(result.feature_maps    )       ,							len(config.out_features    )    )
        self.assertEqual(len(model.channels    )       ,							len(config.out_features    )    )
        self.assertEqual(len(result.feature_maps    )       ,							len(config.out_indices    )    )
        self.assertEqual(len(model.channels    )       ,							len(config.out_indices    )    )

        # Check output of last stage is taken if out_features=None, out_indices=None
        __lowerCAmelCase  :			Dict					    =						copy.deepcopy(lowerCAmelCase    )
        __lowerCAmelCase  :			int					    =						None
        __lowerCAmelCase  :			Any					    =						model_class(lowerCAmelCase    )
        model.to(lowerCAmelCase    )
        model.eval()
        __lowerCAmelCase  :			Optional[int]					    =						model(**lowerCAmelCase    )

        self.assertEqual(len(result.feature_maps    )       ,							1    )
        self.assertEqual(len(model.channels    )       ,							1    )

        __lowerCAmelCase  :			List[Any]					    =						copy.deepcopy(lowerCAmelCase    )
        __lowerCAmelCase  :			Optional[Any]					    =						None
        __lowerCAmelCase  :			Union[str, Any]					    =						model_class(lowerCAmelCase    )
        model.to(lowerCAmelCase    )
        model.eval()
        __lowerCAmelCase  :			Dict					    =						model(**lowerCAmelCase    )

        self.assertEqual(len(result.feature_maps    )       ,							1    )
        self.assertEqual(len(model.channels    )       ,							1    )

        # Check backbone can be initialized with fresh weights
        __lowerCAmelCase  :			Union[str, Any]					    =						copy.deepcopy(lowerCAmelCase    )
        __lowerCAmelCase  :			int					    =						False
        __lowerCAmelCase  :			Tuple					    =						model_class(lowerCAmelCase    )
        model.to(lowerCAmelCase    )
        model.eval()
        __lowerCAmelCase  :			Any					    =						model(**lowerCAmelCase    )







  def        SCREAMING_SNAKE_CASE    (			self :    List[str]    )							->	str:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			Any					    =						self.model_tester.prepare_config_and_inputs_for_common()

     for backbone_class in self.all_model_classes:
        __lowerCAmelCase  :			List[Any]					    =						backbone_class(lowerCAmelCase    )

        self.assertTrue(hasattr(lowerCAmelCase       ,							"""backbone_type"""    )    )
        self.assertTrue(hasattr(lowerCAmelCase       ,							"""stage_names"""    )    )
        self.assertTrue(hasattr(lowerCAmelCase       ,							"""num_features"""    )    )
        self.assertTrue(hasattr(lowerCAmelCase       ,							"""out_indices"""    )    )
        self.assertTrue(hasattr(lowerCAmelCase       ,							"""out_features"""    )    )
        self.assertTrue(hasattr(lowerCAmelCase       ,							"""out_feature_channels"""    )    )
        self.assertTrue(hasattr(lowerCAmelCase       ,							"""channels"""    )    )

        self.assertIsInstance(backbone.backbone_type       ,							lowerCAmelCase    )
        # Verify num_features has been initialized in the backbone init
        self.assertIsNotNone(backbone.num_features    )
        self.assertTrue(len(backbone.channels    ) == len(backbone.out_indices    )    )
        self.assertTrue(len(backbone.stage_names    ) == len(backbone.num_features    )    )
        self.assertTrue(len(backbone.channels    ) <= len(backbone.num_features    )    )
        self.assertTrue(len(backbone.out_feature_channels    ) == len(backbone.stage_names    )    )







  def        SCREAMING_SNAKE_CASE    (			self :    List[str]    )							->	List[Any]:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			int					    =						self.model_tester.prepare_config_and_inputs_for_common()
     __lowerCAmelCase  :			Union[str, Any]					    =						inputs_dict["""pixel_values"""].shape[0]

     for backbone_class in self.all_model_classes:
        __lowerCAmelCase  :			Union[str, Any]					    =						backbone_class(lowerCAmelCase    )
        backbone.to(lowerCAmelCase    )
        backbone.eval()

        __lowerCAmelCase  :			Tuple					    =						backbone(**lowerCAmelCase    )

        # Test default outputs and verify feature maps
        self.assertIsInstance(outputs.feature_maps       ,							lowerCAmelCase    )
        self.assertTrue(len(outputs.feature_maps    ) == len(backbone.channels    )    )
        for feature_map, n_channels in zip(outputs.feature_maps       ,							backbone.channels    ):
           self.assertTrue(feature_map.shape[:2]       ,							(batch_size, n_channels)    )
        self.assertIsNone(outputs.hidden_states    )
        self.assertIsNone(outputs.attentions    )

        # Test output_hidden_states=True
        __lowerCAmelCase  :			Optional[Any]					    =						backbone(**lowerCAmelCase       ,							output_hidden_states=lowerCAmelCase    )
        self.assertIsNotNone(outputs.hidden_states    )
        self.assertTrue(len(outputs.hidden_states    )       ,							len(backbone.stage_names    )    )
        for hidden_state, n_channels in zip(outputs.hidden_states       ,							backbone.channels    ):
           self.assertTrue(hidden_state.shape[:2]       ,							(batch_size, n_channels)    )

        # Test output_attentions=True
        if self.has_attentions:
           __lowerCAmelCase  :			Optional[Any]					    =						backbone(**lowerCAmelCase       ,							output_attentions=lowerCAmelCase    )
           self.assertIsNotNone(outputs.attentions    )


