"""simple docstring"""


import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


@skip_mps
class 							a					(		lowerCAmelCase_						,				lowerCAmelCase_						,				unittest.TestCase ):
    _snake_case  :	Tuple  				=				StableDiffusionPanoramaPipeline
    _snake_case  :	Any  				=				TEXT_TO_IMAGE_PARAMS
    _snake_case  :	List[str]  				=				TEXT_TO_IMAGE_BATCH_PARAMS
    _snake_case  :	Optional[Any]  				=				TEXT_TO_IMAGE_IMAGE_PARAMS
    _snake_case  :	List[str]  				=				TEXT_TO_IMAGE_IMAGE_PARAMS


    def   lowerCAmelCase_	(       self						:	Optional[Any]       ):
      torch.manual_seed(0       )
      _UpperCAmelCase    			=   UNetaDConditionModel(
          block_out_channels=(32, 64) ,					layers_per_block=1 ,					sample_size=32 ,					in_channels=4 ,					out_channels=4 ,					down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""") ,					up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""") ,					cross_attention_dim=32 ,					)
      _UpperCAmelCase    			=   DDIMScheduler()
      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-0_5 ,					num_attention_heads=4 ,					num_hidden_layers=5 ,					pad_token_id=1 ,					vocab_size=1000 ,					)
      _UpperCAmelCase    			=   CLIPTextModel(__lowerCAmelCase       )
      _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   lowerCAmelCase_	(       self						:	Optional[int] ,					__lowerCAmelCase						:	Dict ,					__lowerCAmelCase						:	int=0       ):
      _UpperCAmelCase    			=   torch.manual_seed(__lowerCAmelCase       )
      _UpperCAmelCase    			=   {
          """prompt""": """a photo of the dolomites""",
          """generator""": generator,
          # Setting height and width to None to prevent OOMs on CPU.
          """height""": None,
          """width""": None,
          """num_inference_steps""": 1,
          """guidance_scale""": 6.0,
          """output_type""": """numpy""",
      }
      return inputs


    def   lowerCAmelCase_	(       self						:	Any       ):
      _UpperCAmelCase    			=   """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _UpperCAmelCase    			=   self.get_dummy_components()
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline(**__lowerCAmelCase       )
      _UpperCAmelCase    			=   sd_pipe.to(__lowerCAmelCase       )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase       )

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

      _UpperCAmelCase    			=   np.array([0.6_186, 0.5_374, 0.4_915, 0.4_135, 0.4_114, 0.4_563, 0.5_128, 0.4_977, 0.4_757]       )

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


    def   lowerCAmelCase_	(       self						:	Optional[int]       ):
      super().test_inference_batch_consistent(batch_sizes=[1, 2]       )


    def   lowerCAmelCase_	(       self						:	Any       ):
      super().test_inference_batch_single_identical(batch_size=2 ,					expected_max_diff=3.2_5e-3       )


    def   lowerCAmelCase_	(       self						:	Tuple       ):
      _UpperCAmelCase    			=   """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _UpperCAmelCase    			=   self.get_dummy_components()
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline(**__lowerCAmelCase       )
      _UpperCAmelCase    			=   sd_pipe.to(__lowerCAmelCase       )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase       )

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

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

      _UpperCAmelCase    			=   np.array([0.6_187, 0.5_375, 0.4_915, 0.4_136, 0.4_114, 0.4_563, 0.5_128, 0.4_976, 0.4_757]       )

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


    def   lowerCAmelCase_	(       self						:	Optional[Any]       ):
      _UpperCAmelCase    			=   """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _UpperCAmelCase    			=   self.get_dummy_components()
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline(**__lowerCAmelCase       )
      _UpperCAmelCase    			=   sd_pipe.to(__lowerCAmelCase       )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase       )

      _UpperCAmelCase    			=   self.get_dummy_inputs(__lowerCAmelCase       )
      _UpperCAmelCase    			=   sd_pipe(**__lowerCAmelCase ,					view_batch_size=2       )
      _UpperCAmelCase    			=   output.images
      _UpperCAmelCase    			=   image[0, -3:, -3:, -1]

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

      _UpperCAmelCase    			=   np.array([0.6_187, 0.5_375, 0.4_915, 0.4_136, 0.4_114, 0.4_563, 0.5_128, 0.4_976, 0.4_757]       )

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


    def   lowerCAmelCase_	(       self						:	List[str]       ):
      _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    			=   StableDiffusionPanoramaPipeline(**__lowerCAmelCase       )
      _UpperCAmelCase    			=   sd_pipe.to(__lowerCAmelCase       )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase       )

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

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

      _UpperCAmelCase    			=   np.array([0.4_024, 0.6_510, 0.4_901, 0.5_378, 0.5_813, 0.5_622, 0.4_795, 0.4_467, 0.4_952]       )

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


    def   lowerCAmelCase_	(       self						:	List[Any]       ):
      _UpperCAmelCase    			=   """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _UpperCAmelCase    			=   self.get_dummy_components()
      _UpperCAmelCase    			=   PNDMScheduler(
          beta_start=0.00_085 ,					beta_end=0.012 ,					beta_schedule="""scaled_linear""" ,					skip_prk_steps=__lowerCAmelCase       )
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline(**__lowerCAmelCase       )
      _UpperCAmelCase    			=   sd_pipe.to(__lowerCAmelCase       )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase       )

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

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

      _UpperCAmelCase    			=   np.array([0.6_391, 0.6_291, 0.4_861, 0.5_134, 0.5_552, 0.4_578, 0.5_032, 0.5_023, 0.4_539]       )

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



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


    def   lowerCAmelCase_	(       self						:	str       ):
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()


    def   lowerCAmelCase_	(       self						:	str ,					__lowerCAmelCase						:	List[str]=0       ):
      _UpperCAmelCase    			=   torch.manual_seed(__lowerCAmelCase       )
      _UpperCAmelCase    			=   {
          """prompt""": """a photo of the dolomites""",
          """generator""": generator,
          """num_inference_steps""": 3,
          """guidance_scale""": 7.5,
          """output_type""": """numpy""",
      }
      return inputs


    def   lowerCAmelCase_	(       self						:	List[str]       ):
      _UpperCAmelCase    			=   """stabilityai/stable-diffusion-2-base"""
      _UpperCAmelCase    			=   DDIMScheduler.from_pretrained(__lowerCAmelCase ,					subfolder="""scheduler"""       )
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline.from_pretrained(__lowerCAmelCase ,					scheduler=__lowerCAmelCase ,					safety_checker=__lowerCAmelCase       )
      pipe.to(__lowerCAmelCase       )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase       )
      pipe.enable_attention_slicing()

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

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

      _UpperCAmelCase    			=   np.array(
          [
              0.36_968_392,
              0.27_025_372,
              0.32_446_766,
              0.28_379_387,
              0.36_363_274,
              0.30_733_347,
              0.27_100_027,
              0.27_054_125,
              0.25_536_096,
          ]       )

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


    def   lowerCAmelCase_	(       self						:	Dict       ):
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline.from_pretrained(
          """stabilityai/stable-diffusion-2-base""" ,					safety_checker=__lowerCAmelCase       )
      _UpperCAmelCase    			=   LMSDiscreteScheduler.from_config(pipe.scheduler.config       )
      pipe.to(__lowerCAmelCase       )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase       )
      pipe.enable_attention_slicing()

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

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

      _UpperCAmelCase    			=   np.array(
          [
              [
                  0.0,
                  0.0,
                  0.0,
                  0.0,
                  0.0,
                  0.0,
                  0.0,
                  0.0,
                  0.0,
              ]
          ]       )

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


    def   lowerCAmelCase_	(       self						:	Any       ):
      _UpperCAmelCase    			=   0

      def callback_fn(__lowerCAmelCase						:	int ,					__lowerCAmelCase						:	int ,					__lowerCAmelCase						:	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, 256)
          _UpperCAmelCase    			=   latents[0, -3:, -3:, -1]

          _UpperCAmelCase    			=   np.array(
              [
                  0.18_681_869,
                  0.33_907_816,
                  0.5_361_276,
                  0.14_432_865,
                  -0.02_856_611,
                  -0.73_941_123,
                  0.23_397_987,
                  0.47_322_682,
                  -0.37_823_164,
              ]       )
          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, 256)
          _UpperCAmelCase    			=   latents[0, -3:, -3:, -1]

          _UpperCAmelCase    			=   np.array(
              [
                  0.18_539_645,
                  0.33_987_248,
                  0.5_378_559,
                  0.14_437_142,
                  -0.02_455_261,
                  -0.7_338_317,
                  0.23_990_755,
                  0.47_356_272,
                  -0.3_786_505,
              ]       )

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

      _UpperCAmelCase    			=   False

      _UpperCAmelCase    			=   """stabilityai/stable-diffusion-2-base"""
      _UpperCAmelCase    			=   DDIMScheduler.from_pretrained(__lowerCAmelCase ,					subfolder="""scheduler"""       )
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline.from_pretrained(__lowerCAmelCase ,					scheduler=__lowerCAmelCase ,					safety_checker=__lowerCAmelCase       )
      _UpperCAmelCase    			=   pipe.to(__lowerCAmelCase       )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase       )
      pipe.enable_attention_slicing()

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


    def   lowerCAmelCase_	(       self						:	Dict       ):
      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

      _UpperCAmelCase    			=   """stabilityai/stable-diffusion-2-base"""
      _UpperCAmelCase    			=   DDIMScheduler.from_pretrained(__lowerCAmelCase ,					subfolder="""scheduler"""       )
      _UpperCAmelCase    			=   StableDiffusionPanoramaPipeline.from_pretrained(__lowerCAmelCase ,					scheduler=__lowerCAmelCase ,					safety_checker=__lowerCAmelCase       )
      _UpperCAmelCase    			=   pipe.to(__lowerCAmelCase       )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase       )
      pipe.enable_attention_slicing(1       )
      pipe.enable_sequential_cpu_offload()

      _UpperCAmelCase    			=   self.get_inputs()
      _UpperCAmelCase    			=   pipe(**__lowerCAmelCase       )

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

"""simple docstring"""


import copy
import json
import os
import tempfile

from transformers import is_torch_available

from .test_configuration_utils import config_common_kwargs


class 							a					(		lowerCAmelCase_ ):


    def __init__(       self						:	Union[str, Any] ,					__lowerCAmelCase						:	Union[str, Any] ,					__lowerCAmelCase						:	str=None ,					__lowerCAmelCase						:	List[Any]=True ,					__lowerCAmelCase						:	Union[str, Any]=None ,					**__lowerCAmelCase						:	Dict       ):
      _UpperCAmelCase    			=   parent
      _UpperCAmelCase    			=   config_class
      _UpperCAmelCase    			=   has_text_modality
      _UpperCAmelCase    			=   kwargs
      _UpperCAmelCase    			=   common_properties


    def   lowerCAmelCase_	(       self						:	Dict       ):
      _UpperCAmelCase    			=   self.config_class(**self.inputs_dict       )
      _UpperCAmelCase    			=   (
          ["""hidden_size""", """num_attention_heads""", """num_hidden_layers"""]
          if self.common_properties is None
          else self.common_properties
      )

      # Add common fields for text models
      if self.has_text_modality:
        common_properties.extend(["""vocab_size"""]       )

      # Test that config has the common properties as getters
      for prop in common_properties:
        self.parent.assertTrue(hasattr(__lowerCAmelCase ,					__lowerCAmelCase       ) ,					msg=f'''`{prop}` does not exist'''       )

      # Test that config has the common properties as setter
      for idx, name in enumerate(__lowerCAmelCase       ):
        try:
          setattr(__lowerCAmelCase ,					__lowerCAmelCase ,					__lowerCAmelCase       )
          self.parent.assertEqual(
              getattr(__lowerCAmelCase ,					__lowerCAmelCase       ) ,					__lowerCAmelCase ,					msg=f'''`{name} value {idx} expected, but was {getattr(__lowerCAmelCase ,					__lowerCAmelCase       )}'''       )
        except NotImplementedError:
          # Some models might not be able to implement setters for common_properties
          # In that case, a NotImplementedError is raised
          pass

        # Test if config class can be called with Config(prop_name=..)
      for idx, name in enumerate(__lowerCAmelCase       ):
        try:
          _UpperCAmelCase    			=   self.config_class(**{name: idx}       )
          self.parent.assertEqual(
              getattr(__lowerCAmelCase ,					__lowerCAmelCase       ) ,					__lowerCAmelCase ,					msg=f'''`{name} value {idx} expected, but was {getattr(__lowerCAmelCase ,					__lowerCAmelCase       )}'''       )
        except NotImplementedError:
          # Some models might not be able to implement setters for common_properties
          # In that case, a NotImplementedError is raised
          pass


    def   lowerCAmelCase_	(       self						:	Optional[Any]       ):
      _UpperCAmelCase    			=   self.config_class(**self.inputs_dict       )
      _UpperCAmelCase    			=   json.loads(config.to_json_string()       )
      for key, value in self.inputs_dict.items():
        self.parent.assertEqual(obj[key] ,					__lowerCAmelCase       )


    def   lowerCAmelCase_	(       self						:	Union[str, Any]       ):
      _UpperCAmelCase    			=   self.config_class(**self.inputs_dict       )

      with tempfile.TemporaryDirectory() as tmpdirname:
        _UpperCAmelCase    			=   os.path.join(__lowerCAmelCase ,					"""config.json"""       )
        config_first.to_json_file(__lowerCAmelCase       )
        _UpperCAmelCase    			=   self.config_class.from_json_file(__lowerCAmelCase       )

      self.parent.assertEqual(config_second.to_dict() ,					config_first.to_dict()       )


    def   lowerCAmelCase_	(       self						:	List[str]       ):
      _UpperCAmelCase    			=   self.config_class(**self.inputs_dict       )

      with tempfile.TemporaryDirectory() as tmpdirname:
        config_first.save_pretrained(__lowerCAmelCase       )
        _UpperCAmelCase    			=   self.config_class.from_pretrained(__lowerCAmelCase       )

      self.parent.assertEqual(config_second.to_dict() ,					config_first.to_dict()       )


    def   lowerCAmelCase_	(       self						:	Optional[Any]       ):
      _UpperCAmelCase    			=   self.config_class(**self.inputs_dict       )

      _UpperCAmelCase    			=   """test"""
      with tempfile.TemporaryDirectory() as tmpdirname:
        _UpperCAmelCase    			=   os.path.join(__lowerCAmelCase ,					__lowerCAmelCase       )
        config_first.save_pretrained(__lowerCAmelCase       )
        _UpperCAmelCase    			=   self.config_class.from_pretrained(__lowerCAmelCase ,					subfolder=__lowerCAmelCase       )

      self.parent.assertEqual(config_second.to_dict() ,					config_first.to_dict()       )


    def   lowerCAmelCase_	(       self						:	List[str]       ):
      _UpperCAmelCase    			=   self.config_class(**self.inputs_dict ,					num_labels=5       )
      self.parent.assertEqual(len(config.idalabel       ) ,					5       )
      self.parent.assertEqual(len(config.labelaid       ) ,					5       )

      _UpperCAmelCase    			=   3
      self.parent.assertEqual(len(config.idalabel       ) ,					3       )
      self.parent.assertEqual(len(config.labelaid       ) ,					3       )


    def   lowerCAmelCase_	(       self						:	List[str]       ):
      if self.config_class.is_composition:
        return
      _UpperCAmelCase    			=   self.config_class()
      self.parent.assertIsNotNone(__lowerCAmelCase       )


    def   lowerCAmelCase_	(       self						:	Any       ):
      _UpperCAmelCase    			=   copy.deepcopy(__lowerCAmelCase       )
      _UpperCAmelCase    			=   self.config_class(**__lowerCAmelCase       )
      _UpperCAmelCase    			=   []
      for key, value in config_common_kwargs.items():
        if key == "torch_dtype":
          if not is_torch_available():
            continue
          else:
            import torch

            if config.torch_dtype != torch.floataa:
              wrong_values.append(("""torch_dtype""", config.torch_dtype, torch.floataa)       )
        elif getattr(__lowerCAmelCase ,					__lowerCAmelCase       ) != value:
          wrong_values.append((key, getattr(__lowerCAmelCase ,					__lowerCAmelCase       ), value)       )

      if len(__lowerCAmelCase       ) > 0:
        _UpperCAmelCase    			=   """\n""".join([f'''- {v[0]}: got {v[1]} instead of {v[2]}''' for v in wrong_values]       )
        raise ValueError(f'''The following keys were not properly set in the config:\n{errors}'''       )


    def   lowerCAmelCase_	(       self						:	Optional[Any]       ):
      self.create_and_test_config_common_properties()
      self.create_and_test_config_to_json_string()
      self.create_and_test_config_to_json_file()
      self.create_and_test_config_from_and_save_pretrained()
      self.create_and_test_config_from_and_save_pretrained_subfolder()
      self.create_and_test_config_with_num_labels()
      self.check_config_can_be_init_without_params()
      self.check_config_arguments_init()
