


'''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  lowerCAmelCase_						(       _UpperCamelCase		,    _UpperCamelCase		,    unittest.TestCase      ):





  """simple docstring"""



  a_       :Union[str, Any]												=StableDiffusionPanoramaPipeline
  a_       :Any												=TEXT_TO_IMAGE_PARAMS
  a_       :List[str]												=TEXT_TO_IMAGE_BATCH_PARAMS
  a_       :List[str]												=TEXT_TO_IMAGE_IMAGE_PARAMS
  a_       :Optional[int]												=TEXT_TO_IMAGE_IMAGE_PARAMS

  def __a						(					self     :							Tuple						):





       '''simple docstring'''

       torch.manual_seed(0						)
       __a       =   UNetaDConditionModel(
           block_out_channels=(3_2, 6_4)				,							layers_per_block=1				,							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				,							)
       __a       =   DDIMScheduler()
       torch.manual_seed(0						)
       __a       =   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				,							)
       torch.manual_seed(0						)
       __a       =   CLIPTextConfig(
           bos_token_id=0				,							eos_token_id=2				,							hidden_size=3_2				,							intermediate_size=3_7				,							layer_norm_eps=1E-05				,							num_attention_heads=4				,							num_hidden_layers=5				,							pad_token_id=1				,							vocab_size=1_0_0_0				,							)
       __a       =   CLIPTextModel(_UpperCAmelCase						)
       __a       =   CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

       __a       =   {
           '''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]				,							SCREAMING_SNAKE_CASE__     :							Optional[int]				,							SCREAMING_SNAKE_CASE__     :							List[Any]=0						):





       '''simple docstring'''

       __a       =   torch.manual_seed(_UpperCAmelCase						)
       __a       =   {
           '''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 __a						(					self     :							int						):





       '''simple docstring'''

       __a       =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       __a       =   self.get_dummy_components()
       __a       =   StableDiffusionPanoramaPipeline(**_UpperCAmelCase						)
       __a       =   sd_pipe.to(_UpperCAmelCase						)
       sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase						)

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

       __a       =   np.array([0.6_1_8_6, 0.5_3_7_4, 0.4_9_1_5, 0.4_1_3_5, 0.4_1_1_4, 0.4_5_6_3, 0.5_1_2_8, 0.4_9_7_7, 0.4_7_5_7]						)

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

  def __a						(					self     :							Optional[int]						):





       '''simple docstring'''

       super().test_inference_batch_consistent(batch_sizes=[1, 2]						)

  def __a						(					self     :							Any						):





       '''simple docstring'''

       super().test_inference_batch_single_identical(batch_size=2				,							expected_max_diff=3.25E-3						)

  def __a						(					self     :							int						):





       '''simple docstring'''

       __a       =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       __a       =   self.get_dummy_components()
       __a       =   StableDiffusionPanoramaPipeline(**_UpperCAmelCase						)
       __a       =   sd_pipe.to(_UpperCAmelCase						)
       sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase						)

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

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

       __a       =   np.array([0.6_1_8_7, 0.5_3_7_5, 0.4_9_1_5, 0.4_1_3_6, 0.4_1_1_4, 0.4_5_6_3, 0.5_1_2_8, 0.4_9_7_6, 0.4_7_5_7]						)

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

  def __a						(					self     :							Optional[int]						):





       '''simple docstring'''

       __a       =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       __a       =   self.get_dummy_components()
       __a       =   StableDiffusionPanoramaPipeline(**_UpperCAmelCase						)
       __a       =   sd_pipe.to(_UpperCAmelCase						)
       sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase						)

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

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

       __a       =   np.array([0.6_1_8_7, 0.5_3_7_5, 0.4_9_1_5, 0.4_1_3_6, 0.4_1_1_4, 0.4_5_6_3, 0.5_1_2_8, 0.4_9_7_6, 0.4_7_5_7]						)

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

  def __a						(					self     :							List[Any]						):





       '''simple docstring'''

       __a       =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       __a       =   self.get_dummy_components()
       __a       =   EulerAncestralDiscreteScheduler(
           beta_start=0.0_0_0_8_5				,							beta_end=0.0_1_2				,							beta_schedule="""scaled_linear"""						)
       __a       =   StableDiffusionPanoramaPipeline(**_UpperCAmelCase						)
       __a       =   sd_pipe.to(_UpperCAmelCase						)
       sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase						)

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

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

       __a       =   np.array([0.4_0_2_4, 0.6_5_1_0, 0.4_9_0_1, 0.5_3_7_8, 0.5_8_1_3, 0.5_6_2_2, 0.4_7_9_5, 0.4_4_6_7, 0.4_9_5_2]						)

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


  def __a						(					self     :							Any						):





       '''simple docstring'''

       __a       =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       __a       =   self.get_dummy_components()
       __a       =   PNDMScheduler(
           beta_start=0.0_0_0_8_5				,							beta_end=0.0_1_2				,							beta_schedule="""scaled_linear"""				,							skip_prk_steps=_UpperCAmelCase						)
       __a       =   StableDiffusionPanoramaPipeline(**_UpperCAmelCase						)
       __a       =   sd_pipe.to(_UpperCAmelCase						)
       sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase						)

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

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

       __a       =   np.array([0.6_3_9_1, 0.6_2_9_1, 0.4_8_6_1, 0.5_1_3_4, 0.5_5_5_2, 0.4_5_7_8, 0.5_0_3_2, 0.5_0_2_3, 0.4_5_3_9]						)

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


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





  """simple docstring"""



  def __a						(					self     :							Optional[Any]						):





       '''simple docstring'''

       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()

  def __a						(					self     :							Any				,							SCREAMING_SNAKE_CASE__     :							Optional[int]=0						):





       '''simple docstring'''

       __a       =   torch.manual_seed(_UpperCAmelCase						)
       __a       =   {
           '''prompt''': '''a photo of the dolomites''',
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs

  def __a						(					self     :							Union[str, Any]						):





       '''simple docstring'''

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

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

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

       __a       =   np.array(
           [
               0.3_6_9_6_8_3_9_2,
               0.2_7_0_2_5_3_7_2,
               0.3_2_4_4_6_7_6_6,
               0.2_8_3_7_9_3_8_7,
               0.3_6_3_6_3_2_7_4,
               0.3_0_7_3_3_3_4_7,
               0.2_7_1_0_0_0_2_7,
               0.2_7_0_5_4_1_2_5,
               0.2_5_5_3_6_0_9_6,
           ]						)

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

  def __a						(					self     :							List[str]						):





       '''simple docstring'''

       __a       =   StableDiffusionPanoramaPipeline.from_pretrained(
           """stabilityai/stable-diffusion-2-base"""				,							safety_checker=_UpperCAmelCase						)
       __a       =   LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
       pipe.to(_UpperCAmelCase						)
       pipe.set_progress_bar_config(disable=_UpperCAmelCase						)
       pipe.enable_attention_slicing()

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

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

       __a       =   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 __a						(					self     :							Optional[Any]						):





       '''simple docstring'''

       __a       =   0

       def callback_fn(SCREAMING_SNAKE_CASE__     :							int				,							SCREAMING_SNAKE_CASE__     :							Any				,							SCREAMING_SNAKE_CASE__     :							List[Any]						) -> None:
            __a       =   True
            nonlocal number_of_steps
            number_of_steps += 1
            if step == 1:
                 __a       =   latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 6_4, 2_5_6)
                 __a       =   latents[0, -3:, -3:, -1]

                 __a       =   np.array(
                     [
                         0.1_8_6_8_1_8_6_9,
                         0.3_3_9_0_7_8_1_6,
                         0.5_3_6_1_2_7_6,
                         0.1_4_4_3_2_8_6_5,
                         -0.0_2_8_5_6_6_1_1,
                         -0.7_3_9_4_1_1_2_3,
                         0.2_3_3_9_7_9_8_7,
                         0.4_7_3_2_2_6_8_2,
                         -0.3_7_8_2_3_1_6_4,
                     ]						)
                 assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5E-2
            elif step == 2:
                 __a       =   latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 6_4, 2_5_6)
                 __a       =   latents[0, -3:, -3:, -1]

                 __a       =   np.array(
                     [
                         0.1_8_5_3_9_6_4_5,
                         0.3_3_9_8_7_2_4_8,
                         0.5_3_7_8_5_5_9,
                         0.1_4_4_3_7_1_4_2,
                         -0.0_2_4_5_5_2_6_1,
                         -0.7_3_3_8_3_1_7,
                         0.2_3_9_9_0_7_5_5,
                         0.4_7_3_5_6_2_7_2,
                         -0.3_7_8_6_5_0_5,
                     ]						)

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

       __a       =   False

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

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


  def __a						(					self     :							List[Any]						):





       '''simple docstring'''

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

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

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

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










'''simple docstring'''







import numpy

# List of input, output pairs
SCREAMING_SNAKE_CASE_      =			(
    ((5, 2, 3), 15),
    ((6, 5, 9), 25),
    ((11, 12, 13), 41),
    ((1, 1, 1), 8),
    ((11, 12, 13), 41),
)
SCREAMING_SNAKE_CASE_      =			(((5_15, 22, 13), 5_55), ((61, 35, 49), 1_50))
SCREAMING_SNAKE_CASE_      =			[2, 4, 1, 5]
SCREAMING_SNAKE_CASE_      =			len(train_data)
SCREAMING_SNAKE_CASE_      =			0.0_0_9




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE="train"		)				->					Optional[int]:
     """simple docstring"""




     return calculate_hypothesis_value(__SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE		) - output(
         __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE		)




def        __lowercase						(      __SCREAMING_SNAKE_CASE		)				->					int:
     """simple docstring"""




     __a       =   0
     for i in range(len(__SCREAMING_SNAKE_CASE		) - 1		):
          hyp_val += data_input_tuple[i] * parameter_vector[i + 1]
     hyp_val += parameter_vector[0]
     return hyp_val




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE		)				->					Optional[int]:
     """simple docstring"""




     if data_set == "train":
          return train_data[example_no][1]
     elif data_set == "test":
          return test_data[example_no][1]
     return None




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE		)				->					Tuple:
     """simple docstring"""




     if data_set == "train":
          return _hypothesis_value(train_data[example_no][0]		)
     elif data_set == "test":
          return _hypothesis_value(test_data[example_no][0]		)
     return None




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE=m		)				->					Optional[Any]:
     """simple docstring"""




     __a       =   0
     for i in range(__SCREAMING_SNAKE_CASE		):
          if index == -1:
               summation_value += _error(__SCREAMING_SNAKE_CASE		)
          else:
               summation_value += _error(__SCREAMING_SNAKE_CASE		) * train_data[i][0][index]
     return summation_value




def        __lowercase						(      __SCREAMING_SNAKE_CASE		)				->					List[str]:
     """simple docstring"""




     __a       =   summation_of_cost_derivative(__SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE		) / m
     return cost_derivative_value




def        __lowercase						(      )				->					str:
     """simple docstring"""




     global parameter_vector
     # Tune these values to set a tolerance value for predicted output
     __a       =   0.000_002
     __a       =   0
     __a       =   0
     while True:
          j += 1
          __a       =   [0, 0, 0, 0]
          for i in range(0				,  len(__SCREAMING_SNAKE_CASE		)		):
               __a       =   get_cost_derivative(i - 1		)
               __a       =   (
                   parameter_vector[i] - LEARNING_RATE * cost_derivative
               )
          if numpy.allclose(
              __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE				,  atol=__SCREAMING_SNAKE_CASE				,  rtol=__SCREAMING_SNAKE_CASE				,  ):
               break
          __a       =   temp_parameter_vector
     print(("""Number of iterations:""", j)		)




def        __lowercase						(      )				->					List[Any]:
     """simple docstring"""




     for i in range(len(__SCREAMING_SNAKE_CASE		)		):
          print(("""Actual output value:""", output(__SCREAMING_SNAKE_CASE				,  """test"""		))		)
          print(("""Hypothesis output:""", calculate_hypothesis_value(__SCREAMING_SNAKE_CASE				,  """test"""		))		)


if __name__ == "__main__":
 run_gradient_descent()
 print('\nTesting gradient descent for a linear hypothesis function.\n')
 test_gradient_descent()






