

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        UpperCAmelCase__		(       lowerCamelCase_						,     lowerCamelCase_						,     unittest.TestCase       ):






  """simple docstring"""


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

       SCREAMING_SNAKE_CASE__    =    {
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''safety_checker''': None,
           '''feature_extractor''': None,
       }
       return components






  def       lowercase_		(							self						:		int	,   __lowerCamelCase						:		Tuple	,   __lowerCamelCase						:		List[str]=0					)		->							List[Any]:
       SCREAMING_SNAKE_CASE__    =    torch.manual_seed(__a					)
       SCREAMING_SNAKE_CASE__    =    {
           '''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						:		Dict					)		->							int:
       SCREAMING_SNAKE_CASE__    =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__    =    self.get_dummy_components()
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline(**__a					)
       SCREAMING_SNAKE_CASE__    =    sd_pipe.to(__a					)
       sd_pipe.set_progress_bar_config(disable=__a					)

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

       SCREAMING_SNAKE_CASE__    =    np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757]					)

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






  def       lowercase_		(							self						:		Dict					)		->							Optional[Any]:
       super().test_inference_batch_consistent(batch_sizes=[1, 2]					)






  def       lowercase_		(							self						:		List[str]					)		->							int:
       super().test_inference_batch_single_identical(batch_size=2	,   expected_max_diff=3.25e-3					)






  def       lowercase_		(							self						:		Union[str, Any]					)		->							Optional[Any]:
       SCREAMING_SNAKE_CASE__    =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__    =    self.get_dummy_components()
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline(**__a					)
       SCREAMING_SNAKE_CASE__    =    sd_pipe.to(__a					)
       sd_pipe.set_progress_bar_config(disable=__a					)

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

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

       SCREAMING_SNAKE_CASE__    =    np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]					)

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






  def       lowercase_		(							self						:		int					)		->							Optional[int]:
       SCREAMING_SNAKE_CASE__    =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__    =    self.get_dummy_components()
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline(**__a					)
       SCREAMING_SNAKE_CASE__    =    sd_pipe.to(__a					)
       sd_pipe.set_progress_bar_config(disable=__a					)

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

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

       SCREAMING_SNAKE_CASE__    =    np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]					)

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






  def       lowercase_		(							self						:		str					)		->							int:
       SCREAMING_SNAKE_CASE__    =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__    =    self.get_dummy_components()
       SCREAMING_SNAKE_CASE__    =    EulerAncestralDiscreteScheduler(
           beta_start=0.00085	,   beta_end=0.012	,   beta_schedule='''scaled_linear'''					)
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline(**__a					)
       SCREAMING_SNAKE_CASE__    =    sd_pipe.to(__a					)
       sd_pipe.set_progress_bar_config(disable=__a					)

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

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

       SCREAMING_SNAKE_CASE__    =    np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952]					)

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







  def       lowercase_		(							self						:		int					)		->							Tuple:
       SCREAMING_SNAKE_CASE__    =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__    =    self.get_dummy_components()
       SCREAMING_SNAKE_CASE__    =    PNDMScheduler(
           beta_start=0.00085	,   beta_end=0.012	,   beta_schedule='''scaled_linear'''	,   skip_prk_steps=__a					)
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline(**__a					)
       SCREAMING_SNAKE_CASE__    =    sd_pipe.to(__a					)
       sd_pipe.set_progress_bar_config(disable=__a					)

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

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

       SCREAMING_SNAKE_CASE__    =    np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539]					)

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






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






  """simple docstring"""


  def       lowercase_		(							self						:		Any					)		->							Union[str, Any]:
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()






  def       lowercase_		(							self						:		str	,   __lowerCamelCase						:		Optional[int]=0					)		->							Optional[Any]:
       SCREAMING_SNAKE_CASE__    =    torch.manual_seed(__a					)
       SCREAMING_SNAKE_CASE__    =    {
           '''prompt''': '''a photo of the dolomites''',
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs






  def       lowercase_		(							self						:		str					)		->							str:
       SCREAMING_SNAKE_CASE__    =    '''stabilityai/stable-diffusion-2-base'''
       SCREAMING_SNAKE_CASE__    =    DDIMScheduler.from_pretrained(__a	,   subfolder='''scheduler'''					)
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline.from_pretrained(__a	,   scheduler=__a	,   safety_checker=__a					)
       pipe.to(__a					)
       pipe.set_progress_bar_config(disable=__a					)
       pipe.enable_attention_slicing()

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

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

       SCREAMING_SNAKE_CASE__    =    np.array(
           [
               0.36968392,
               0.27025372,
               0.32446766,
               0.28379387,
               0.36363274,
               0.30733347,
               0.27100027,
               0.27054125,
               0.25536096,
           ]					)

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






  def       lowercase_		(							self						:		Any					)		->							Tuple:
       SCREAMING_SNAKE_CASE__    =    StableDiffusionPanoramaPipeline.from_pretrained(
           '''stabilityai/stable-diffusion-2-base'''	,   safety_checker=__a					)
       SCREAMING_SNAKE_CASE__    =    LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
       pipe.to(__a					)
       pipe.set_progress_bar_config(disable=__a					)
       pipe.enable_attention_slicing()

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

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

       SCREAMING_SNAKE_CASE__    =    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						:		Optional[int]					)		->							Any:
       SCREAMING_SNAKE_CASE__    =    0

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

                 SCREAMING_SNAKE_CASE__    =    np.array(
                     [
                         0.18681869,
                         0.33907816,
                         0.5361276,
                         0.14432865,
                         -0.02856611,
                         -0.73941123,
                         0.23397987,
                         0.47322682,
                         -0.37823164,
                     ]					)
                 assert np.abs(latents_slice.flatten() - expected_slice					).max() < 5e-2
            elif step == 2:
                 SCREAMING_SNAKE_CASE__    =    latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 64, 256)
                 SCREAMING_SNAKE_CASE__    =    latents[0, -3:, -3:, -1]

                 SCREAMING_SNAKE_CASE__    =    np.array(
                     [
                         0.18539645,
                         0.33987248,
                         0.5378559,
                         0.14437142,
                         -0.02455261,
                         -0.7338317,
                         0.23990755,
                         0.47356272,
                         -0.3786505,
                     ]					)

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

       SCREAMING_SNAKE_CASE__    =    False

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

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







  def       lowercase_		(							self						:		Tuple					)		->							Union[str, Any]:
       torch.cuda.empty_cache()
       torch.cuda.reset_max_memory_allocated()
       torch.cuda.reset_peak_memory_stats()

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

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

       SCREAMING_SNAKE_CASE__    =    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 math
class 				__SCREAMING_SNAKE_CASE  :






				"""simple docstring"""







				def   UpperCamelCase__   (  self     :					List[str]					,				__a     :					list[list[float]]					,				__a     :					list[int]					):
								_a								= 0.0
								_a								= 0.0
								for i in range(len(__a					)					):
												da += math.pow((sample[i] - weights[0][i])					,				2					)
												da += math.pow((sample[i] - weights[1][i])					,				2					)
												return 0 if da > da else 1
								return 0




				def   UpperCamelCase__   (  self     :					List[Any]					,				__a     :					list[list[int | float]]					,				__a     :					list[int]					,				__a     :					int					,				__a     :					float					):
								for i in range(len(__a					)					):
												weights[j][i] += alpha * (sample[i] - weights[j][i])
								return weights






def  _lowerCamelCase				( )	->	None:
				# Training Examples ( m, n )
				_a								= [[1, 1, 0, 0], [0, 0, 0, 1], [1, 0, 0, 0], [0, 0, 1, 1]]

				# weight initialization ( n, C )
				_a								= [[0.2, 0.6, 0.5, 0.9], [0.8, 0.4, 0.7, 0.3]]

				# training
				_a								= SelfOrganizingMap()
				_a								= 3
				_a								= 0.5

				for _ in range(lowercase						):
								for j in range(len(lowercase						)						):
												# training sample
												_a								= training_samples[j]

												# Compute the winning vector
												_a								= self_organizing_map.get_winner(lowercase    ,							lowercase						)

												# Update the winning vector
												_a								= self_organizing_map.update(lowercase    ,							lowercase    ,							lowercase    ,							lowercase						)

    # classify test sample
				_a								= [0, 0, 0, 1]
				_a								= self_organizing_map.get_winner(lowercase    ,							lowercase						)

				# results
				print(F'Clusters that the test sample belongs to : {winner}'						)
				print(F'Weights that have been trained : {weights}'						)


# running the main() function
if __name__ == "__main__":
					main()
