



"""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  (			_lowerCAmelCase    ,						_lowerCAmelCase    ,						unittest.TestCase				):






					'''simple docstring'''





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






					def    lowerCAmelCase_				(       self:							List[Any]						)   ->	Union[str, Any]:
											torch.manual_seed(0						)
											snake_case_ :Tuple						=			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						,     )
											snake_case_ :Dict						=			DDIMScheduler()
											torch.manual_seed(0						)
											snake_case_ :Dict						=			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						)
											snake_case_ :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=1_000						,     )
											snake_case_ :Optional[Any]						=			CLIPTextModel(snake_case						)
											snake_case_ :str						=			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

											snake_case_ :int						=			{
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											    """safety_checker""": None,
											    """feature_extractor""": None,
											}
											return components






					def    lowerCAmelCase_				(       self:							int						,     snake_case:							Optional[Any]						,     snake_case:							List[str]=0						)   ->	str:
											snake_case_ :List[Any]						=			torch.manual_seed(snake_case						)
											snake_case_ :Dict						=			{
											    """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:							Union[str, Any]						)   ->	Optional[int]:
											snake_case_ :Union[str, Any]						=			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_ :int						=			self.get_dummy_components()
											snake_case_ :List[str]						=			StableDiffusionPanoramaPipeline(**snake_case						)
											snake_case_ :List[str]						=			sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :Dict						=			self.get_dummy_inputs(snake_case						)
											snake_case_ :Any						=			sd_pipe(**snake_case						).images
											snake_case_ :Optional[int]						=			image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											snake_case_ :List[Any]						=			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    lowerCAmelCase_				(       self:							Optional[int]						)   ->	Union[str, Any]:
											super().test_inference_batch_consistent(batch_sizes=[1, 2]						)






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






					def    lowerCAmelCase_				(       self:							Dict						)   ->	Optional[Any]:
											snake_case_ :str						=			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_ :str						=			self.get_dummy_components()
											snake_case_ :int						=			StableDiffusionPanoramaPipeline(**snake_case						)
											snake_case_ :Optional[int]						=			sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :Any						=			self.get_dummy_inputs(snake_case						)
											snake_case_ :Optional[Any]						=			"""french fries"""
											snake_case_ :Optional[Any]						=			sd_pipe(**snake_case						,     negative_prompt=snake_case						)
											snake_case_ :int						=			output.images
											snake_case_ :List[str]						=			image[0, -3:, -3:, -1]

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

											snake_case_ :Union[str, Any]						=			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    lowerCAmelCase_				(       self:							Tuple						)   ->	Optional[int]:
											snake_case_ :List[str]						=			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_ :Optional[int]						=			self.get_dummy_components()
											snake_case_ :List[Any]						=			StableDiffusionPanoramaPipeline(**snake_case						)
											snake_case_ :Optional[Any]						=			sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :Tuple						=			self.get_dummy_inputs(snake_case						)
											snake_case_ :Optional[int]						=			sd_pipe(**snake_case						,     view_batch_size=2						)
											snake_case_ :Dict						=			output.images
											snake_case_ :List[str]						=			image[0, -3:, -3:, -1]

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

											snake_case_ :Dict						=			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    lowerCAmelCase_				(       self:							Dict						)   ->	str:
											snake_case_ :Tuple						=			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_ :Optional[int]						=			self.get_dummy_components()
											snake_case_ :int						=			EulerAncestralDiscreteScheduler(
											    beta_start=0.0_0_0_8_5						,     beta_end=0.0_1_2						,     beta_schedule="""scaled_linear"""						)
											snake_case_ :Any						=			StableDiffusionPanoramaPipeline(**snake_case						)
											snake_case_ :Optional[int]						=			sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :str						=			self.get_dummy_inputs(snake_case						)
											snake_case_ :Tuple						=			sd_pipe(**snake_case						).images
											snake_case_ :Tuple						=			image[0, -3:, -3:, -1]

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

											snake_case_ :Any						=			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    lowerCAmelCase_				(       self:							Optional[Any]						)   ->	Any:
											snake_case_ :Union[str, Any]						=			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_ :Dict						=			self.get_dummy_components()
											snake_case_ :List[str]						=			PNDMScheduler(
											    beta_start=0.0_0_0_8_5						,     beta_end=0.0_1_2						,     beta_schedule="""scaled_linear"""						,     skip_prk_steps=snake_case						)
											snake_case_ :Optional[Any]						=			StableDiffusionPanoramaPipeline(**snake_case						)
											snake_case_ :Union[str, Any]						=			sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :int						=			self.get_dummy_inputs(snake_case						)
											snake_case_ :Tuple						=			sd_pipe(**snake_case						).images
											snake_case_ :Optional[int]						=			image[0, -3:, -3:, -1]

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

											snake_case_ :int						=			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    lowerCAmelCase_				(       self:							Dict						)   ->	Union[str, Any]:
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()






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






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

											snake_case_ :int						=			self.get_inputs()
											snake_case_ :str						=			pipe(**snake_case						).images
											snake_case_ :int						=			image[0, -3:, -3:, -1].flatten()

											assert image.shape == (1, 512, 2_048, 3)

											snake_case_ :Union[str, Any]						=			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    lowerCAmelCase_				(       self:							int						)   ->	Tuple:
											snake_case_ :Any						=			StableDiffusionPanoramaPipeline.from_pretrained(
											    """stabilityai/stable-diffusion-2-base"""						,     safety_checker=snake_case						)
											snake_case_ :Dict						=			LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
											pipe.to(snake_case						)
											pipe.set_progress_bar_config(disable=snake_case						)
											pipe.enable_attention_slicing()

											snake_case_ :str						=			self.get_inputs()
											snake_case_ :List[Any]						=			pipe(**snake_case						).images
											snake_case_ :Any						=			image[0, -3:, -3:, -1].flatten()

											assert image.shape == (1, 512, 2_048, 3)

											snake_case_ :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    lowerCAmelCase_				(       self:							List[Any]						)   ->	int:
											snake_case_ :Any						=			0

											def callback_fn(snake_case:							int						,     snake_case:							int						,     snake_case:							torch.FloatTensor						) -> None:
																	snake_case_ :Union[str, Any]						=			True
																	nonlocal number_of_steps
																	number_of_steps += 1
																	if step == 1:
																							snake_case_ :Optional[int]						=			latents.detach().cpu().numpy()
																							assert latents.shape == (1, 4, 64, 256)
																							snake_case_ :List[str]						=			latents[0, -3:, -3:, -1]

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

																							snake_case_ :Optional[int]						=			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

											snake_case_ :int						=			False

											snake_case_ :Optional[int]						=			"""stabilityai/stable-diffusion-2-base"""
											snake_case_ :List[Any]						=			DDIMScheduler.from_pretrained(snake_case						,     subfolder="""scheduler"""						)
											snake_case_ :List[str]						=			StableDiffusionPanoramaPipeline.from_pretrained(snake_case						,     scheduler=snake_case						,     safety_checker=snake_case						)
											snake_case_ :Optional[int]						=			pipe.to(snake_case						)
											pipe.set_progress_bar_config(disable=snake_case						)
											pipe.enable_attention_slicing()

											snake_case_ :Any						=			self.get_inputs()
											pipe(**snake_case						,     callback=snake_case						,     callback_steps=1						)
											assert callback_fn.has_been_called
											assert number_of_steps == 3






					def    lowerCAmelCase_				(       self:							Tuple						)   ->	int:
											torch.cuda.empty_cache()
											torch.cuda.reset_max_memory_allocated()
											torch.cuda.reset_peak_memory_stats()

											snake_case_ :List[str]						=			"""stabilityai/stable-diffusion-2-base"""
											snake_case_ :List[str]						=			DDIMScheduler.from_pretrained(snake_case						,     subfolder="""scheduler"""						)
											snake_case_ :Optional[int]						=			StableDiffusionPanoramaPipeline.from_pretrained(snake_case						,     scheduler=snake_case						,     safety_checker=snake_case						)
											snake_case_ :List[str]						=			pipe.to(snake_case						)
											pipe.set_progress_bar_config(disable=snake_case						)
											pipe.enable_attention_slicing(1						)
											pipe.enable_sequential_cpu_offload()

											snake_case_ :int						=			self.get_inputs()
											snake_case_ :List[Any]						=			pipe(**snake_case						)

											snake_case_ :int						=			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 				lowerCamelCase  :






					'''simple docstring'''





					def    lowerCAmelCase_				(       self:							Tuple						,     snake_case:							list[list[float]]						,     snake_case:							list[int]						)   ->	int:
											snake_case_ :Any						=			0.0
											snake_case_ :Tuple						=			0.0
											for i in range(len(snake_case						)						):
																	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    lowerCAmelCase_				(       self:							Optional[int]						,     snake_case:							list[list[int | float]]						,     snake_case:							list[int]						,     snake_case:							int						,     snake_case:							float						)   ->	list[list[int | float]]:
											for i in range(len(snake_case						)						):
																	weights[j][i] += alpha * (sample[i] - weights[j][i])
											return weights

def    A_  (			):







						'''simple docstring'''


						snake_case_ :Dict						=			[[1, 1, 0, 0], [0, 0, 0, 1], [1, 0, 0, 0], [0, 0, 1, 1]]

						# weight initialization ( n, C )
						snake_case_ :List[Any]						=			[[0.2, 0.6, 0.5, 0.9], [0.8, 0.4, 0.7, 0.3]]

						# training
						snake_case_ :Optional[Any]						=			SelfOrganizingMap()
						snake_case_ :Dict						=			3
						snake_case_ :Dict						=			0.5

						for _ in range(_lowercase	):
												for j in range(len(_lowercase	)	):
																		# training sample
																		snake_case_ :List[Any]						=			training_samples[j]

																		# Compute the winning vector
																		snake_case_ :Optional[int]						=			self_organizing_map.get_winner(_lowercase,					_lowercase	)

																		# Update the winning vector
																		snake_case_ :List[str]						=			self_organizing_map.update(_lowercase,					_lowercase,					_lowercase,					_lowercase	)

    # classify test sample
						snake_case_ :str						=			[0, 0, 0, 1]
						snake_case_ :List[Any]						=			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()
