



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






 """simple docstring"""


 a__				: Any		       =							StableDiffusionPanoramaPipeline
 a__				: Any		       =							TEXT_TO_IMAGE_PARAMS
 a__				: Optional[int]		       =							TEXT_TO_IMAGE_BATCH_PARAMS
 a__				: Optional[Any]		       =							TEXT_TO_IMAGE_IMAGE_PARAMS
 a__				: str		       =							TEXT_TO_IMAGE_IMAGE_PARAMS




 def   _SCREAMING_SNAKE_CASE		(     self     :						Any					)      ->			int:
       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-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     :						Any						,  __UpperCAmelCase     :						Optional[Any]						,  __UpperCAmelCase     :						List[str]=0					)      ->			List[str]:
       UpperCAmelCase_=       torch.manual_seed(__UpperCAmelCase					)
       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   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]					)      ->			int:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline(**__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, 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   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Optional[int]:
       super().test_inference_batch_consistent(batch_sizes=[1, 2]					)




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




 def   _SCREAMING_SNAKE_CASE		(     self     :						List[str]					)      ->			List[Any]:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline(**__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, 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   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			str:
       UpperCAmelCase_=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_=       self.get_dummy_components()
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline(**__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe.to(__UpperCAmelCase					)
       sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

       UpperCAmelCase_=       self.get_dummy_inputs(__UpperCAmelCase					)
       UpperCAmelCase_=       sd_pipe(**__UpperCAmelCase						,  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   _SCREAMING_SNAKE_CASE		(     self     :						Any					)      ->			Optional[int]:
       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(**__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, 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   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Optional[int]:
       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=__UpperCAmelCase					)
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline(**__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, 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 		lowercase						(			unittest.TestCase):






 """simple docstring"""


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




 def   _SCREAMING_SNAKE_CASE		(     self     :						Union[str, Any]						,  __UpperCAmelCase     :						int=0					)      ->			Any:
       UpperCAmelCase_=       torch.manual_seed(__UpperCAmelCase					)
       UpperCAmelCase_=       {
           """prompt""": """a photo of the dolomites""",
           """generator""": generator,
           """num_inference_steps""": 3,
           """guidance_scale""": 7.5,
           """output_type""": """numpy""",
       }
       return inputs




 def   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Union[str, Any]:
       UpperCAmelCase_=       """stabilityai/stable-diffusion-2-base"""
       UpperCAmelCase_=       DDIMScheduler.from_pretrained(__UpperCAmelCase						,  subfolder="""scheduler"""					)
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline.from_pretrained(__UpperCAmelCase						,  scheduler=__UpperCAmelCase						,  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, 2_048, 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   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]					)      ->			Dict:
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline.from_pretrained(
           """stabilityai/stable-diffusion-2-base"""						,  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, 2_048, 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   _SCREAMING_SNAKE_CASE		(     self     :						int					)      ->			List[Any]:
       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, 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(__UpperCAmelCase						,  subfolder="""scheduler"""					)
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline.from_pretrained(__UpperCAmelCase						,  scheduler=__UpperCAmelCase						,  safety_checker=__UpperCAmelCase					)
       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     :						Tuple					)      ->			List[str]:
       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(__UpperCAmelCase						,  subfolder="""scheduler"""					)
       UpperCAmelCase_=       StableDiffusionPanoramaPipeline.from_pretrained(__UpperCAmelCase						,  scheduler=__UpperCAmelCase						,  safety_checker=__UpperCAmelCase					)
       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 5.2 GB is allocated
       assert mem_bytes < 5.5 * 10**9









import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available


if is_vision_available():
	from PIL import Image

	from transformers import AutoProcessor, BlipaProcessor, BlipImageProcessor, GPTaTokenizer, PreTrainedTokenizerFast






@require_vision
class 		lowercase						(			unittest.TestCase):






	"""simple docstring"""


	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			List[Any]:
							UpperCAmelCase_=       tempfile.mkdtemp()

							UpperCAmelCase_=       BlipImageProcessor()
							UpperCAmelCase_=       GPTaTokenizer.from_pretrained("""hf-internal-testing/tiny-random-GPT2Model"""					)

							UpperCAmelCase_=       BlipaProcessor(__UpperCAmelCase						,  __UpperCAmelCase					)

							processor.save_pretrained(self.tmpdirname					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]						,  **__UpperCAmelCase     :						Union[str, Any]					)      ->			int:
							return AutoProcessor.from_pretrained(self.tmpdirname						,  **__UpperCAmelCase					).tokenizer




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]						,  **__UpperCAmelCase     :						str					)      ->			Optional[int]:
							return AutoProcessor.from_pretrained(self.tmpdirname						,  **__UpperCAmelCase					).image_processor




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			List[Any]:
							shutil.rmtree(self.tmpdirname					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]					)      ->			Any:
							UpperCAmelCase_=       [np.random.randint(255						,  size=(3, 30, 400)						,  dtype=np.uinta					)]

							UpperCAmelCase_=       [Image.fromarray(np.moveaxis(__UpperCAmelCase						,  0						,  -1					)					) for x in image_inputs]

							return image_inputs




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			Dict:
							UpperCAmelCase_=       BlipaProcessor(tokenizer=self.get_tokenizer()						,  image_processor=self.get_image_processor()					)
							processor.save_pretrained(self.tmpdirname					)

							UpperCAmelCase_=       self.get_tokenizer(bos_token="""(BOS)"""						,  eos_token="""(EOS)"""					)
							UpperCAmelCase_=       self.get_image_processor(do_normalize=__UpperCAmelCase						,  padding_value=1.0					)

							UpperCAmelCase_=       BlipaProcessor.from_pretrained(
							    self.tmpdirname						,  bos_token="""(BOS)"""						,  eos_token="""(EOS)"""						,  do_normalize=__UpperCAmelCase						,  padding_value=1.0					)

							self.assertEqual(processor.tokenizer.get_vocab()						,  tokenizer_add_kwargs.get_vocab()					)
							self.assertIsInstance(processor.tokenizer						,  __UpperCAmelCase					)

							self.assertEqual(processor.image_processor.to_json_string()						,  image_processor_add_kwargs.to_json_string()					)
							self.assertIsInstance(processor.image_processor						,  __UpperCAmelCase					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Dict					)      ->			Optional[int]:
							UpperCAmelCase_=       self.get_image_processor()
							UpperCAmelCase_=       self.get_tokenizer()

							UpperCAmelCase_=       BlipaProcessor(tokenizer=__UpperCAmelCase						,  image_processor=__UpperCAmelCase					)

							UpperCAmelCase_=       self.prepare_image_inputs()

							UpperCAmelCase_=       image_processor(__UpperCAmelCase						,  return_tensors="""np"""					)
							UpperCAmelCase_=       processor(images=__UpperCAmelCase						,  return_tensors="""np"""					)

							for key in input_feat_extract.keys():
													self.assertAlmostEqual(input_feat_extract[key].sum()						,  input_processor[key].sum()						,  delta=1E-2					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			Any:
							UpperCAmelCase_=       self.get_image_processor()
							UpperCAmelCase_=       self.get_tokenizer()

							UpperCAmelCase_=       BlipaProcessor(tokenizer=__UpperCAmelCase						,  image_processor=__UpperCAmelCase					)

							UpperCAmelCase_=       """lower newer"""

							UpperCAmelCase_=       processor(text=__UpperCAmelCase					)

							UpperCAmelCase_=       tokenizer(__UpperCAmelCase						,  return_token_type_ids=__UpperCAmelCase					)

							for key in encoded_tok.keys():
													self.assertListEqual(encoded_tok[key]						,  encoded_processor[key]					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Union[str, Any]					)      ->			Dict:
							UpperCAmelCase_=       self.get_image_processor()
							UpperCAmelCase_=       self.get_tokenizer()

							UpperCAmelCase_=       BlipaProcessor(tokenizer=__UpperCAmelCase						,  image_processor=__UpperCAmelCase					)

							UpperCAmelCase_=       """lower newer"""
							UpperCAmelCase_=       self.prepare_image_inputs()

							UpperCAmelCase_=       processor(text=__UpperCAmelCase						,  images=__UpperCAmelCase					)

							self.assertListEqual(list(inputs.keys()					)						,  ["""pixel_values""", """input_ids""", """attention_mask"""]					)

							# test if it raises when no input is passed
							with pytest.raises(__UpperCAmelCase					):
													processor()




	def   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Any:
							UpperCAmelCase_=       self.get_image_processor()
							UpperCAmelCase_=       self.get_tokenizer()

							UpperCAmelCase_=       BlipaProcessor(tokenizer=__UpperCAmelCase						,  image_processor=__UpperCAmelCase					)

							UpperCAmelCase_=       [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

							UpperCAmelCase_=       processor.batch_decode(__UpperCAmelCase					)
							UpperCAmelCase_=       tokenizer.batch_decode(__UpperCAmelCase					)

							self.assertListEqual(__UpperCAmelCase						,  __UpperCAmelCase					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Dict					)      ->			int:
							UpperCAmelCase_=       self.get_image_processor()
							UpperCAmelCase_=       self.get_tokenizer()

							UpperCAmelCase_=       BlipaProcessor(tokenizer=__UpperCAmelCase						,  image_processor=__UpperCAmelCase					)

							UpperCAmelCase_=       """lower newer"""
							UpperCAmelCase_=       self.prepare_image_inputs()

							UpperCAmelCase_=       processor(text=__UpperCAmelCase						,  images=__UpperCAmelCase					)

							# For now the processor supports only ['pixel_values', 'input_ids', 'attention_mask']
							self.assertListEqual(list(inputs.keys()					)						,  ["""pixel_values""", """input_ids""", """attention_mask"""]					)




