


import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch



@require_torch
class 						_UpperCAmelCase      :






		@property
		def _snake_case		(						self				:     Optional[Any]):
				return self.get_dummy_input()






		@property
		def _snake_case		(						self				:     List[str]):
				if self.block_type == "down":
						return (4, 32, 16, 16)
				elif self.block_type == "mid":
						return (4, 32, 32, 32)
				elif self.block_type == "up":
						return (4, 32, 64, 64)

				raise ValueError(F"'{self.block_type}' is not a supported block_type. Set it to 'up', 'mid', or 'down'.")






		def _snake_case		(						self				:     Any   ,  UpperCAmelCase				:     Optional[Any]=True   ,  UpperCAmelCase				:     Any=False   ,  UpperCAmelCase				:     Union[str, Any]=False   ,  UpperCAmelCase				:     Optional[int]=False   ,  ):
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						4
				SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						32
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						(32, 32)

				SCREAMING_SNAKE_CASE_				:int				     =						torch.manual_seed(0)
				SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						torch.device(UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:str				     =						(batch_size, num_channels) + sizes
				SCREAMING_SNAKE_CASE_				:int				     =						randn_tensor(UpperCAmelCase   ,  generator=UpperCAmelCase   ,  device=UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:int				     =						{"hidden_states": hidden_states}

				if include_temb:
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						1_28
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						randn_tensor((batch_size, temb_channels)   ,  generator=UpperCAmelCase   ,  device=UpperCAmelCase)

				if include_res_hidden_states_tuple:
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						torch.manual_seed(1)
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						(randn_tensor(UpperCAmelCase   ,  generator=UpperCAmelCase   ,  device=UpperCAmelCase),)

				if include_encoder_hidden_states:
						SCREAMING_SNAKE_CASE_				:Tuple				     =						floats_tensor((batch_size, 32, 32)).to(UpperCAmelCase)

				if include_skip_sample:
						SCREAMING_SNAKE_CASE_				:Optional[int]				     =						randn_tensor(((batch_size, 3) + sizes)   ,  generator=UpperCAmelCase   ,  device=UpperCAmelCase)

				return dummy_input






		def _snake_case		(						self				:     Union[str, Any]):
				SCREAMING_SNAKE_CASE_				:int				     =						{
				    "in_channels": 32,
				    "out_channels": 32,
				    "temb_channels": 1_28,
				}
				if self.block_type == "up":
						SCREAMING_SNAKE_CASE_				:List[str]				     =						32

				if self.block_type == "mid":
						init_dict.pop("out_channels")

				SCREAMING_SNAKE_CASE_				:int				     =						self.dummy_input
				return init_dict, inputs_dict






		def _snake_case		(						self				:     str   ,  UpperCAmelCase				:     int):
				SCREAMING_SNAKE_CASE_     ,				SCREAMING_SNAKE_CASE_				:Any				     =						self.prepare_init_args_and_inputs_for_common()
				SCREAMING_SNAKE_CASE_				:List[str]				     =						self.block_class(**UpperCAmelCase)
				unet_block.to(UpperCAmelCase)
				unet_block.eval()

				with torch.no_grad():
						SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						unet_block(**UpperCAmelCase)

				if isinstance(UpperCAmelCase   ,  UpperCAmelCase):
						SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						output[0]

				self.assertEqual(output.shape   ,  self.output_shape)

				SCREAMING_SNAKE_CASE_				:Tuple				     =						output[0, -1, -3:, -3:]
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						torch.tensor(UpperCAmelCase).to(UpperCAmelCase)
				assert torch_all_close(output_slice.flatten()   ,  UpperCAmelCase   ,  atol=5E-3)







		@unittest.skipIf(torch_device == "mps"   ,  "Training is not supported in mps")
		def _snake_case		(						self				:     Union[str, Any]):
				SCREAMING_SNAKE_CASE_     ,				SCREAMING_SNAKE_CASE_				:List[Any]				     =						self.prepare_init_args_and_inputs_for_common()
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						self.block_class(**UpperCAmelCase)
				model.to(UpperCAmelCase)
				model.train()
				SCREAMING_SNAKE_CASE_				:List[str]				     =						model(**UpperCAmelCase)

				if isinstance(UpperCAmelCase   ,  UpperCAmelCase):
						SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						output[0]

				SCREAMING_SNAKE_CASE_				:List[str]				     =						torch.device(UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						randn_tensor(output.shape   ,  device=UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						torch.nn.functional.mse_loss(UpperCAmelCase   ,  UpperCAmelCase)
				loss.backward()










from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
		import PIL


SCREAMING_SNAKE_CASE__     					=			logging.get_logger(__name__)


def     lowercase	(    a     , a ):



		'''simple docstring'''




		SCREAMING_SNAKE_CASE_				:Dict				     =						b.T
		SCREAMING_SNAKE_CASE_				:str				     =						np.sum(np.square(a )     , axis=1 )
		SCREAMING_SNAKE_CASE_				:List[Any]				     =						np.sum(np.square(a )     , axis=0 )
		SCREAMING_SNAKE_CASE_				:Optional[int]				     =						np.matmul(a     , a )
		SCREAMING_SNAKE_CASE_				:Any				     =						aa[:, None] - 2 * ab + ba[None, :]
		return d


def     lowercase	(    a     , a ):



		'''simple docstring'''




		SCREAMING_SNAKE_CASE_				:str				     =						x.reshape(-1     , 3 )
		SCREAMING_SNAKE_CASE_				:str				     =						squared_euclidean_distance(a     , a )
		return np.argmin(a     , axis=1 )



class 						_UpperCAmelCase      (     lowercase							):
		lowerCamelCase_	:			List[str]       			=  ["""pixel_values"""]






		def __init__(						self				:     Dict   ,  UpperCAmelCase				:     Optional[Union[List[List[int]], np.ndarray]] = None   ,  UpperCAmelCase				:     bool = True   ,  UpperCAmelCase				:     Dict[str, int] = None   ,  UpperCAmelCase				:     PILImageResampling = PILImageResampling.BILINEAR   ,  UpperCAmelCase				:     bool = True   ,  UpperCAmelCase				:     bool = True   ,  **UpperCAmelCase				:     Union[str, Any]   ,  ):
				super().__init__(**UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						size if size is not None else {"height": 2_56, "width": 2_56}
				SCREAMING_SNAKE_CASE_				:Optional[int]				     =						get_size_dict(UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:int				     =						np.array(UpperCAmelCase) if clusters is not None else None
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						do_resize
				SCREAMING_SNAKE_CASE_				:Dict				     =						size
				SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						resample
				SCREAMING_SNAKE_CASE_				:Dict				     =						do_normalize
				SCREAMING_SNAKE_CASE_				:Tuple				     =						do_color_quantize






		def _snake_case		(						self				:     Union[str, Any]   ,  UpperCAmelCase				:     np.ndarray   ,  UpperCAmelCase				:     Dict[str, int]   ,  UpperCAmelCase				:     PILImageResampling = PILImageResampling.BILINEAR   ,  UpperCAmelCase				:     Optional[Union[str, ChannelDimension]] = None   ,  **UpperCAmelCase				:     List[Any]   ,  ):
				SCREAMING_SNAKE_CASE_				:str				     =						get_size_dict(UpperCAmelCase)
				if "height" not in size or "width" not in size:
						raise ValueError(F"Size dictionary must contain both height and width keys. Got {size.keys()}")
				return resize(
				    UpperCAmelCase   ,  size=(size["height"], size["width"])   ,  resample=UpperCAmelCase   ,  data_format=UpperCAmelCase   ,  **UpperCAmelCase)






		def _snake_case		(						self				:     Union[str, Any]   ,  UpperCAmelCase				:     np.ndarray   ,  UpperCAmelCase				:     Optional[Union[str, ChannelDimension]] = None   ,  ):
				SCREAMING_SNAKE_CASE_				:int				     =						rescale(image=UpperCAmelCase   ,  scale=1 / 127.5   ,  data_format=UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:List[str]				     =						image - 1
				return image







		def _snake_case		(						self				:     List[str]   ,  UpperCAmelCase				:     ImageInput   ,  UpperCAmelCase				:     bool = None   ,  UpperCAmelCase				:     Dict[str, int] = None   ,  UpperCAmelCase				:     PILImageResampling = None   ,  UpperCAmelCase				:     bool = None   ,  UpperCAmelCase				:     Optional[bool] = None   ,  UpperCAmelCase				:     Optional[Union[List[List[int]], np.ndarray]] = None   ,  UpperCAmelCase				:     Optional[Union[str, TensorType]] = None   ,  UpperCAmelCase				:     Optional[Union[str, ChannelDimension]] = ChannelDimension.FIRST   ,  **UpperCAmelCase				:     Optional[Any]   ,  ):
				SCREAMING_SNAKE_CASE_				:Any				     =						do_resize if do_resize is not None else self.do_resize
				SCREAMING_SNAKE_CASE_				:Dict				     =						size if size is not None else self.size
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						get_size_dict(UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:Optional[int]				     =						resample if resample is not None else self.resample
				SCREAMING_SNAKE_CASE_				:Dict				     =						do_normalize if do_normalize is not None else self.do_normalize
				SCREAMING_SNAKE_CASE_				:List[str]				     =						do_color_quantize if do_color_quantize is not None else self.do_color_quantize
				SCREAMING_SNAKE_CASE_				:Tuple				     =						clusters if clusters is not None else self.clusters
				SCREAMING_SNAKE_CASE_				:Tuple				     =						np.array(UpperCAmelCase)

				SCREAMING_SNAKE_CASE_				:Dict				     =						make_list_of_images(UpperCAmelCase)

				if not valid_images(UpperCAmelCase):
						raise ValueError(
						    "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
						    "torch.Tensor, tf.Tensor or jax.ndarray.")

				if do_resize and size is None or resample is None:
						raise ValueError("Size and resample must be specified if do_resize is True.")

				if do_color_quantize and clusters is None:
						raise ValueError("Clusters must be specified if do_color_quantize is True.")

				# All transformations expect numpy arrays.
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						[to_numpy_array(UpperCAmelCase) for image in images]

				if do_resize:
						SCREAMING_SNAKE_CASE_				:Tuple				     =						[self.resize(image=UpperCAmelCase   ,  size=UpperCAmelCase   ,  resample=UpperCAmelCase) for image in images]

				if do_normalize:
						SCREAMING_SNAKE_CASE_				:int				     =						[self.normalize(image=UpperCAmelCase) for image in images]

				if do_color_quantize:
						SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						[to_channel_dimension_format(UpperCAmelCase   ,  ChannelDimension.LAST) for image in images]
						# color quantize from (batch_size, height, width, 3) to (batch_size, height, width)
						SCREAMING_SNAKE_CASE_				:Optional[int]				     =						np.array(UpperCAmelCase)
						SCREAMING_SNAKE_CASE_				:Tuple				     =						color_quantize(UpperCAmelCase   ,  UpperCAmelCase).reshape(images.shape[:-1])

						# flatten to (batch_size, height*width)
						SCREAMING_SNAKE_CASE_				:Any				     =						images.shape[0]
						SCREAMING_SNAKE_CASE_				:Dict				     =						images.reshape(UpperCAmelCase   ,  -1)

						# We need to convert back to a list of images to keep consistent behaviour across processors.
						SCREAMING_SNAKE_CASE_				:Any				     =						list(UpperCAmelCase)
				else:
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						[to_channel_dimension_format(UpperCAmelCase   ,  UpperCAmelCase) for image in images]

				SCREAMING_SNAKE_CASE_				:str				     =						{"input_ids": images}
				return BatchFeature(data=UpperCAmelCase   ,  tensor_type=UpperCAmelCase)






