
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							(	lowerCAmelCase_						,		lowerCAmelCase_						,		unittest.TestCase       ):
					__a											=     StableDiffusionPanoramaPipeline
					__a											=     TEXT_TO_IMAGE_PARAMS
					__a											=     TEXT_TO_IMAGE_BATCH_PARAMS
					__a											=     TEXT_TO_IMAGE_IMAGE_PARAMS
					__a											=     TEXT_TO_IMAGE_IMAGE_PARAMS
					def lowercase_      (		self     ):
								"""simple docstring"""







								torch.manual_seed(0     )
								lowerCAmelCase__			:     str					   =      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			,						)
								lowerCAmelCase__			:     Tuple					   =      DDIMScheduler()
								torch.manual_seed(0     )
								lowerCAmelCase__			:     List[str]					   =      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     )
								lowerCAmelCase__			:     Optional[int]					   =      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			,						)
								lowerCAmelCase__			:     Dict					   =      CLIPTextModel(__lowerCAmelCase     )
								lowerCAmelCase__			:     List[Any]					   =      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''     )

								lowerCAmelCase__			:     Any					   =      {
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								    '''vae''': vae,
								    '''text_encoder''': text_encoder,
								    '''tokenizer''': tokenizer,
								    '''safety_checker''': None,
								    '''feature_extractor''': None,
								}
								return components
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__=0     ):
								"""simple docstring"""







								lowerCAmelCase__			:     int					   =      torch.manual_seed(__lowerCAmelCase     )
								lowerCAmelCase__			:     List[Any]					   =      {
								    '''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 lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Union[str, Any]					   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowerCAmelCase__			:     Union[str, Any]					   =      self.get_dummy_components()
								lowerCAmelCase__			:     List[Any]					   =      StableDiffusionPanoramaPipeline(**__lowerCAmelCase     )
								lowerCAmelCase__			:     List[Any]					   =      sd_pipe.to(__lowerCAmelCase     )
								sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase     )

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

								lowerCAmelCase__			:     Dict					   =      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 lowercase_      (		self     ):
								"""simple docstring"""







								super().test_inference_batch_consistent(batch_sizes=[1, 2]     )
					def lowercase_      (		self     ):
								"""simple docstring"""







								super().test_inference_batch_single_identical(batch_size=2			,						expected_max_diff=3.25E-3     )
					def lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Any					   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowerCAmelCase__			:     List[Any]					   =      self.get_dummy_components()
								lowerCAmelCase__			:     Tuple					   =      StableDiffusionPanoramaPipeline(**__lowerCAmelCase     )
								lowerCAmelCase__			:     List[str]					   =      sd_pipe.to(__lowerCAmelCase     )
								sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase     )

								lowerCAmelCase__			:     Tuple					   =      self.get_dummy_inputs(__lowerCAmelCase     )
								lowerCAmelCase__			:     List[str]					   =      '''french fries'''
								lowerCAmelCase__			:     List[str]					   =      sd_pipe(**__lowerCAmelCase			,						negative_prompt=__lowerCAmelCase     )
								lowerCAmelCase__			:     Union[str, Any]					   =      output.images
								lowerCAmelCase__			:     str					   =      image[0, -3:, -3:, -1]

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

								lowerCAmelCase__			:     str					   =      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 lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     List[str]					   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowerCAmelCase__			:     Any					   =      self.get_dummy_components()
								lowerCAmelCase__			:     Dict					   =      StableDiffusionPanoramaPipeline(**__lowerCAmelCase     )
								lowerCAmelCase__			:     Optional[Any]					   =      sd_pipe.to(__lowerCAmelCase     )
								sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase     )

								lowerCAmelCase__			:     Any					   =      self.get_dummy_inputs(__lowerCAmelCase     )
								lowerCAmelCase__			:     List[Any]					   =      sd_pipe(**__lowerCAmelCase			,						view_batch_size=2     )
								lowerCAmelCase__			:     Union[str, Any]					   =      output.images
								lowerCAmelCase__			:     Optional[int]					   =      image[0, -3:, -3:, -1]

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

								lowerCAmelCase__			:     List[Any]					   =      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 lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     str					   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowerCAmelCase__			:     int					   =      self.get_dummy_components()
								lowerCAmelCase__			:     Any					   =      EulerAncestralDiscreteScheduler(
								    beta_start=0.00_085			,						beta_end=0.012			,						beta_schedule='''scaled_linear'''     )
								lowerCAmelCase__			:     Dict					   =      StableDiffusionPanoramaPipeline(**__lowerCAmelCase     )
								lowerCAmelCase__			:     Tuple					   =      sd_pipe.to(__lowerCAmelCase     )
								sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase     )

								lowerCAmelCase__			:     str					   =      self.get_dummy_inputs(__lowerCAmelCase     )
								lowerCAmelCase__			:     List[Any]					   =      sd_pipe(**__lowerCAmelCase     ).images
								lowerCAmelCase__			:     List[str]					   =      image[0, -3:, -3:, -1]

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

								lowerCAmelCase__			:     Optional[int]					   =      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 lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Optional[int]					   =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowerCAmelCase__			:     str					   =      self.get_dummy_components()
								lowerCAmelCase__			:     int					   =      PNDMScheduler(
								    beta_start=0.00_085			,						beta_end=0.012			,						beta_schedule='''scaled_linear'''			,						skip_prk_steps=__lowerCAmelCase     )
								lowerCAmelCase__			:     Dict					   =      StableDiffusionPanoramaPipeline(**__lowerCAmelCase     )
								lowerCAmelCase__			:     str					   =      sd_pipe.to(__lowerCAmelCase     )
								sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase     )

								lowerCAmelCase__			:     List[Any]					   =      self.get_dummy_inputs(__lowerCAmelCase     )
								lowerCAmelCase__			:     str					   =      sd_pipe(**__lowerCAmelCase     ).images
								lowerCAmelCase__			:     Tuple					   =      image[0, -3:, -3:, -1]

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

								lowerCAmelCase__			:     Dict					   =      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       ):
					def lowercase_      (		self     ):
								"""simple docstring"""







								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__=0     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Union[str, Any]					   =      torch.manual_seed(__lowerCAmelCase     )
								lowerCAmelCase__			:     Optional[Any]					   =      {
								    '''prompt''': '''a photo of the dolomites''',
								    '''generator''': generator,
								    '''num_inference_steps''': 3,
								    '''guidance_scale''': 7.5,
								    '''output_type''': '''numpy''',
								}
								return inputs
					def lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     List[Any]					   =      '''stabilityai/stable-diffusion-2-base'''
								lowerCAmelCase__			:     Union[str, Any]					   =      DDIMScheduler.from_pretrained(__lowerCAmelCase			,						subfolder='''scheduler'''     )
								lowerCAmelCase__			:     Dict					   =      StableDiffusionPanoramaPipeline.from_pretrained(__lowerCAmelCase			,						scheduler=__lowerCAmelCase			,						safety_checker=__lowerCAmelCase     )
								pipe.to(__lowerCAmelCase     )
								pipe.set_progress_bar_config(disable=__lowerCAmelCase     )
								pipe.enable_attention_slicing()

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

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

								lowerCAmelCase__			:     int					   =      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 lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Any					   =      StableDiffusionPanoramaPipeline.from_pretrained(
								    '''stabilityai/stable-diffusion-2-base'''			,						safety_checker=__lowerCAmelCase     )
								lowerCAmelCase__			:     int					   =      LMSDiscreteScheduler.from_config(pipe.scheduler.config     )
								pipe.to(__lowerCAmelCase     )
								pipe.set_progress_bar_config(disable=__lowerCAmelCase     )
								pipe.enable_attention_slicing()

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

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

								lowerCAmelCase__			:     str					   =      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 lowercase_      (		self     ):
								"""simple docstring"""







								lowerCAmelCase__			:     List[Any]					   =      0

								def callback_fn(SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__     ) -> None:
											lowerCAmelCase__			:     Optional[int]					   =      True
											nonlocal number_of_steps
											number_of_steps += 1
											if step == 1:
														lowerCAmelCase__			:     Optional[Any]					   =      latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 256)
														lowerCAmelCase__			:     Optional[Any]					   =      latents[0, -3:, -3:, -1]

														lowerCAmelCase__			:     int					   =      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:
														lowerCAmelCase__			:     List[str]					   =      latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 256)
														lowerCAmelCase__			:     List[Any]					   =      latents[0, -3:, -3:, -1]

														lowerCAmelCase__			:     str					   =      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

								lowerCAmelCase__			:     Tuple					   =      False

								lowerCAmelCase__			:     Optional[Any]					   =      '''stabilityai/stable-diffusion-2-base'''
								lowerCAmelCase__			:     Optional[Any]					   =      DDIMScheduler.from_pretrained(__lowerCAmelCase			,						subfolder='''scheduler'''     )
								lowerCAmelCase__			:     Tuple					   =      StableDiffusionPanoramaPipeline.from_pretrained(__lowerCAmelCase			,						scheduler=__lowerCAmelCase			,						safety_checker=__lowerCAmelCase     )
								lowerCAmelCase__			:     Optional[Any]					   =      pipe.to(__lowerCAmelCase     )
								pipe.set_progress_bar_config(disable=__lowerCAmelCase     )
								pipe.enable_attention_slicing()

								lowerCAmelCase__			:     List[Any]					   =      self.get_inputs()
								pipe(**__lowerCAmelCase			,						callback=__lowerCAmelCase			,						callback_steps=1     )
								assert callback_fn.has_been_called
								assert number_of_steps == 3
					def lowercase_      (		self     ):
								"""simple docstring"""







								torch.cuda.empty_cache()
								torch.cuda.reset_max_memory_allocated()
								torch.cuda.reset_peak_memory_stats()

								lowerCAmelCase__			:     Union[str, Any]					   =      '''stabilityai/stable-diffusion-2-base'''
								lowerCAmelCase__			:     str					   =      DDIMScheduler.from_pretrained(__lowerCAmelCase			,						subfolder='''scheduler'''     )
								lowerCAmelCase__			:     int					   =      StableDiffusionPanoramaPipeline.from_pretrained(__lowerCAmelCase			,						scheduler=__lowerCAmelCase			,						safety_checker=__lowerCAmelCase     )
								lowerCAmelCase__			:     Dict					   =      pipe.to(__lowerCAmelCase     )
								pipe.set_progress_bar_config(disable=__lowerCAmelCase     )
								pipe.enable_attention_slicing(1     )
								pipe.enable_sequential_cpu_offload()

								lowerCAmelCase__			:     Optional[Any]					   =      self.get_inputs()
								lowerCAmelCase__			:     Dict					   =      pipe(**__lowerCAmelCase     )

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




"""simple docstring"""


def      __UpperCAmelCase       (      lowercase		):







  """simple docstring"""
  _UpperCAmelCase    			=   0
  _UpperCAmelCase    			=   len(lowercase		)

  for i in range(n - 1		):
    for j in range(i + 1 ,lowercase		):
      if arr[i] > arr[j]:
        num_inversions += 1

  return num_inversions






def      __UpperCAmelCase       (      lowercase		):







  """simple docstring"""
  if len(lowercase		) <= 1:
    return arr, 0
  _UpperCAmelCase    			=   len(lowercase		) // 2
  _UpperCAmelCase    			=   arr[0:mid]
  _UpperCAmelCase    			=   arr[mid:]

  _UpperCAmelCase			,					_UpperCAmelCase    			=   count_inversions_recursive(lowercase		)
  _UpperCAmelCase			,					_UpperCAmelCase    			=   count_inversions_recursive(lowercase		)
  _UpperCAmelCase			,					_UpperCAmelCase    			=   _count_cross_inversions(lowercase ,lowercase		)

  _UpperCAmelCase    			=   inversion_p + inversions_q + cross_inversions
  return c, num_inversions






def      __UpperCAmelCase       (      lowercase ,lowercase		):







  """simple docstring"""
  _UpperCAmelCase    			=   []
  _UpperCAmelCase    			=   _UpperCAmelCase    			=   _UpperCAmelCase    			=   0
  while i < len(lowercase		) and j < len(lowercase		):
    if p[i] > q[j]:
      # if P[1] > Q[j], then P[k] > Q[k] for all  i < k <= len(P)
      # These are all inversions. The claim emerges from the
      # property that P is sorted.
      num_inversion += len(lowercase		) - i
      r.append(q[j]		)
      j += 1
    else:
      r.append(p[i]		)
      i += 1

  if i < len(lowercase		):
    r.extend(p[i:]		)
  else:
    r.extend(q[j:]		)

  return r, num_inversion






def      __UpperCAmelCase       (      ):







  """simple docstring"""
  _UpperCAmelCase    			=   [10, 2, 1, 5, 5, 2, 11]

  # this arr has 8 inversions:
  # (10, 2), (10, 1), (10, 5), (10, 5), (10, 2), (2, 1), (5, 2), (5, 2)

  _UpperCAmelCase    			=   count_inversions_bf(lowercase		)
  _UpperCAmelCase			,					_UpperCAmelCase    			=   count_inversions_recursive(lowercase		)

  assert num_inversions_bf == num_inversions_recursive == 8

  print("""number of inversions = """ ,lowercase		)

  # testing an array with zero inversion (a sorted arr_1)

  arr_a.sort()
  _UpperCAmelCase    			=   count_inversions_bf(lowercase		)
  _UpperCAmelCase			,					_UpperCAmelCase    			=   count_inversions_recursive(lowercase		)

  assert num_inversions_bf == num_inversions_recursive == 0
  print("""number of inversions = """ ,lowercase		)

  # an empty list should also have zero inversions
  _UpperCAmelCase    			=   []
  _UpperCAmelCase    			=   count_inversions_bf(lowercase		)
  _UpperCAmelCase			,					_UpperCAmelCase    			=   count_inversions_recursive(lowercase		)

  assert num_inversions_bf == num_inversions_recursive == 0
  print("""number of inversions = """ ,lowercase		)


if __name__ == "__main__":
    main()
