


'''simple docstring'''

from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block
@dataclass
class _lowerCamelCase	(						lowercase__							):



		'''simple docstring'''







		A_     :      torch.FloatTensor







class _lowerCamelCase	(						nn.Module							):



		'''simple docstring'''







		def __init__( self		:						Union[str, Any]  ,					_A		:						str=3  ,					_A		:						int=3  ,					_A		:						Tuple=("DownEncoderBlock2D",)  ,					_A		:						str=(64,)  ,					_A		:						Optional[int]=2  ,					_A		:						List[Any]=32  ,					_A		:						List[str]="silu"  ,					_A		:						Any=True  ,					)  ->			Tuple:
			super().__init__()
			__magic_name__	:   Union[str, Any]					     =					layers_per_block

			__magic_name__	:   Any					     =					torch.nn.Convad(
			    _A  ,					block_out_channels[0]  ,					kernel_size=3  ,					stride=1  ,					padding=1  ,					)

			__magic_name__	:   Any					     =					None
			__magic_name__	:   List[Any]					     =					nn.ModuleList([]      )

			# down
			__magic_name__	:   Union[str, Any]					     =					block_out_channels[0]
			for i, down_block_type in enumerate(_A      ):
				__magic_name__	:   List[Any]					     =					output_channel
				__magic_name__	:   Union[str, Any]					     =					block_out_channels[i]
				__magic_name__	:   int					     =					i == len(_A      ) - 1

				__magic_name__	:   str					     =					get_down_block(
				    _A  ,					num_layers=self.layers_per_block  ,					in_channels=_A  ,					out_channels=_A  ,					add_downsample=not is_final_block  ,					resnet_eps=1E-6  ,					downsample_padding=0  ,					resnet_act_fn=_A  ,					resnet_groups=_A  ,					attention_head_dim=_A  ,					temb_channels=_A  ,					)
				self.down_blocks.append(_A      )

			# mid
			__magic_name__	:   Union[str, Any]					     =					UNetMidBlockaD(
			    in_channels=block_out_channels[-1]  ,					resnet_eps=1E-6  ,					resnet_act_fn=_A  ,					output_scale_factor=1  ,					resnet_time_scale_shift='default'  ,					attention_head_dim=block_out_channels[-1]  ,					resnet_groups=_A  ,					temb_channels=_A  ,					)

			# out
			__magic_name__	:   Dict					     =					nn.GroupNorm(num_channels=block_out_channels[-1]  ,					num_groups=_A  ,					eps=1E-6      )
			__magic_name__	:   Any					     =					nn.SiLU()

			__magic_name__	:   int					     =					2 * out_channels if double_z else out_channels
			__magic_name__	:   str					     =					nn.Convad(block_out_channels[-1]  ,					_A  ,					3  ,					padding=1      )

			__magic_name__	:   str					     =					False







		def        __lowerCAmelCase		( self		:						Union[str, Any]  ,					_A		:						List[Any]      )  ->			List[str]:
			__magic_name__	:   Union[str, Any]					     =					x
			__magic_name__	:   Union[str, Any]					     =					self.conv_in(_A      )

			if self.training and self.gradient_checkpointing:

				def create_custom_forward(_A		:						List[str]      ):
					def custom_forward(*_A		:						List[Any]      ):
						return module(*_A      )

					return custom_forward

				# down
				if is_torch_version('>='  ,					'1.11.0'      ):
					for down_block in self.down_blocks:
						__magic_name__	:   List[Any]					     =					torch.utils.checkpoint.checkpoint(
						    create_custom_forward(_A      )  ,					_A  ,					use_reentrant=_A      )
					# middle
					__magic_name__	:   List[Any]					     =					torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block      )  ,					_A  ,					use_reentrant=_A      )
				else:
					for down_block in self.down_blocks:
						__magic_name__	:   List[str]					     =					torch.utils.checkpoint.checkpoint(create_custom_forward(_A      )  ,					_A      )
					# middle
					__magic_name__	:   Tuple					     =					torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block      )  ,					_A      )

			else:
				# down
				for down_block in self.down_blocks:
					__magic_name__	:   str					     =					down_block(_A      )

				# middle
				__magic_name__	:   Optional[int]					     =					self.mid_block(_A      )

			# post-process
			__magic_name__	:   int					     =					self.conv_norm_out(_A      )
			__magic_name__	:   List[str]					     =					self.conv_act(_A      )
			__magic_name__	:   Optional[int]					     =					self.conv_out(_A      )

			return sample







class _lowerCamelCase	(						nn.Module							):



		'''simple docstring'''







		def __init__( self		:						Optional[Any]  ,					_A		:						Optional[Any]=3  ,					_A		:						Dict=3  ,					_A		:						int=("UpDecoderBlock2D",)  ,					_A		:						Dict=(64,)  ,					_A		:						Tuple=2  ,					_A		:						List[Any]=32  ,					_A		:						Optional[Any]="silu"  ,					_A		:						str="group"  ,					)  ->			Any:
			super().__init__()
			__magic_name__	:   List[str]					     =					layers_per_block

			__magic_name__	:   List[Any]					     =					nn.Convad(
			    _A  ,					block_out_channels[-1]  ,					kernel_size=3  ,					stride=1  ,					padding=1  ,					)

			__magic_name__	:   Optional[int]					     =					None
			__magic_name__	:   Tuple					     =					nn.ModuleList([]      )

			__magic_name__	:   Union[str, Any]					     =					in_channels if norm_type == 'spatial' else None

			# mid
			__magic_name__	:   List[Any]					     =					UNetMidBlockaD(
			    in_channels=block_out_channels[-1]  ,					resnet_eps=1E-6  ,					resnet_act_fn=_A  ,					output_scale_factor=1  ,					resnet_time_scale_shift='default' if norm_type == 'group' else norm_type  ,					attention_head_dim=block_out_channels[-1]  ,					resnet_groups=_A  ,					temb_channels=_A  ,					)

			# up
			__magic_name__	:   Union[str, Any]					     =					list(reversed(_A      )      )
			__magic_name__	:   Optional[Any]					     =					reversed_block_out_channels[0]
			for i, up_block_type in enumerate(_A      ):
				__magic_name__	:   List[str]					     =					output_channel
				__magic_name__	:   int					     =					reversed_block_out_channels[i]

				__magic_name__	:   List[Any]					     =					i == len(_A      ) - 1

				__magic_name__	:   Optional[Any]					     =					get_up_block(
				    _A  ,					num_layers=self.layers_per_block + 1  ,					in_channels=_A  ,					out_channels=_A  ,					prev_output_channel=_A  ,					add_upsample=not is_final_block  ,					resnet_eps=1E-6  ,					resnet_act_fn=_A  ,					resnet_groups=_A  ,					attention_head_dim=_A  ,					temb_channels=_A  ,					resnet_time_scale_shift=_A  ,					)
				self.up_blocks.append(_A      )
				__magic_name__	:   Optional[int]					     =					output_channel

			# out
			if norm_type == "spatial":
				__magic_name__	:   Union[str, Any]					     =					SpatialNorm(block_out_channels[0]  ,					_A      )
			else:
				__magic_name__	:   Any					     =					nn.GroupNorm(num_channels=block_out_channels[0]  ,					num_groups=_A  ,					eps=1E-6      )
			__magic_name__	:   Optional[int]					     =					nn.SiLU()
			__magic_name__	:   List[str]					     =					nn.Convad(block_out_channels[0]  ,					_A  ,					3  ,					padding=1      )

			__magic_name__	:   Union[str, Any]					     =					False







		def        __lowerCAmelCase		( self		:						int  ,					_A		:						Optional[Any]  ,					_A		:						List[str]=None      )  ->			Tuple:
			__magic_name__	:   Optional[Any]					     =					z
			__magic_name__	:   Union[str, Any]					     =					self.conv_in(_A      )

			__magic_name__	:   int					     =					next(iter(self.up_blocks.parameters()      )      ).dtype
			if self.training and self.gradient_checkpointing:

				def create_custom_forward(_A		:						Optional[int]      ):
					def custom_forward(*_A		:						str      ):
						return module(*_A      )

					return custom_forward

				if is_torch_version('>='  ,					'1.11.0'      ):
					# middle
					__magic_name__	:   Any					     =					torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block      )  ,					_A  ,					_A  ,					use_reentrant=_A      )
					__magic_name__	:   Union[str, Any]					     =					sample.to(_A      )

					# up
					for up_block in self.up_blocks:
						__magic_name__	:   Tuple					     =					torch.utils.checkpoint.checkpoint(
						    create_custom_forward(_A      )  ,					_A  ,					_A  ,					use_reentrant=_A      )
				else:
					# middle
					__magic_name__	:   Tuple					     =					torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block      )  ,					_A  ,					_A      )
					__magic_name__	:   Optional[int]					     =					sample.to(_A      )

					# up
					for up_block in self.up_blocks:
						__magic_name__	:   Dict					     =					torch.utils.checkpoint.checkpoint(create_custom_forward(_A      )  ,					_A  ,					_A      )
			else:
				# middle
				__magic_name__	:   int					     =					self.mid_block(_A  ,					_A      )
				__magic_name__	:   List[str]					     =					sample.to(_A      )

				# up
				for up_block in self.up_blocks:
					__magic_name__	:   str					     =					up_block(_A  ,					_A      )

        # post-process
			if latent_embeds is None:
				__magic_name__	:   Optional[Any]					     =					self.conv_norm_out(_A      )
			else:
				__magic_name__	:   List[str]					     =					self.conv_norm_out(_A  ,					_A      )
			__magic_name__	:   Optional[int]					     =					self.conv_act(_A      )
			__magic_name__	:   List[str]					     =					self.conv_out(_A      )

			return sample







class _lowerCamelCase	(						nn.Module							):



		'''simple docstring'''







		def __init__( self		:						Any  ,					_A		:						List[Any]  ,					_A		:						Union[str, Any]  ,					_A		:						Tuple  ,					_A		:						Union[str, Any]=None  ,					_A		:						str="random"  ,					_A		:						Dict=False  ,					_A		:						List[Any]=True      )  ->			List[Any]:
			super().__init__()
			__magic_name__	:   Any					     =					n_e
			__magic_name__	:   int					     =					vq_embed_dim
			__magic_name__	:   str					     =					beta
			__magic_name__	:   int					     =					legacy

			__magic_name__	:   List[str]					     =					nn.Embedding(self.n_e  ,					self.vq_embed_dim      )
			self.embedding.weight.data.uniform_(-1.0 / self.n_e  ,					1.0 / self.n_e      )

			__magic_name__	:   List[str]					     =					remap
			if self.remap is not None:
				self.register_buffer('used'  ,					torch.tensor(np.load(self.remap      )      )      )
				__magic_name__	:   List[Any]					     =					self.used.shape[0]
				__magic_name__	:   Dict					     =					unknown_index  # "random" or "extra" or integer
				if self.unknown_index == "extra":
					__magic_name__	:   List[str]					     =					self.re_embed
					__magic_name__	:   Union[str, Any]					     =					self.re_embed + 1
				print(
				    F'Remapping {self.n_e} indices to {self.re_embed} indices. '
				    F'Using {self.unknown_index} for unknown indices.'      )
			else:
				__magic_name__	:   Dict					     =					n_e

			__magic_name__	:   Union[str, Any]					     =					sane_index_shape







		def        __lowerCAmelCase		( self		:						Any  ,					_A		:						List[Any]      )  ->			Tuple:
			__magic_name__	:   List[str]					     =					inds.shape
			assert len(_A      ) > 1
			__magic_name__	:   Any					     =					inds.reshape(ishape[0]  ,					-1      )
			__magic_name__	:   Optional[int]					     =					self.used.to(_A      )
			__magic_name__	:   Union[str, Any]					     =					(inds[:, :, None] == used[None, None, ...]).long()
			__magic_name__	:   Optional[Any]					     =					match.argmax(-1      )
			__magic_name__	:   Optional[int]					     =					match.sum(2      ) < 1
			if self.unknown_index == "random":
				__magic_name__	:   Any					     =					torch.randint(0  ,					self.re_embed  ,					size=new[unknown].shape      ).to(device=new.device      )
			else:
				__magic_name__	:   Optional[Any]					     =					self.unknown_index
			return new.reshape(_A      )







		def        __lowerCAmelCase		( self		:						str  ,					_A		:						Optional[int]      )  ->			Optional[int]:
			__magic_name__	:   Union[str, Any]					     =					inds.shape
			assert len(_A      ) > 1
			__magic_name__	:   Any					     =					inds.reshape(ishape[0]  ,					-1      )
			__magic_name__	:   Any					     =					self.used.to(_A      )
			if self.re_embed > self.used.shape[0]:  # extra token
				__magic_name__	:   str					     =					0  # simply set to zero
			__magic_name__	:   Tuple					     =					torch.gather(used[None, :][inds.shape[0] * [0], :]  ,					1  ,					_A      )
			return back.reshape(_A      )







		def        __lowerCAmelCase		( self		:						Optional[Any]  ,					_A		:						Union[str, Any]      )  ->			Any:
			# reshape z -> (batch, height, width, channel) and flatten
			__magic_name__	:   int					     =					z.permute(0  ,					2  ,					3  ,					1      ).contiguous()
			__magic_name__	:   int					     =					z.view(-1  ,					self.vq_embed_dim      )

			# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
			__magic_name__	:   List[str]					     =					torch.argmin(torch.cdist(_A  ,					self.embedding.weight      )  ,					dim=1      )

			__magic_name__	:   Dict					     =					self.embedding(_A      ).view(z.shape      )
			__magic_name__	:   List[Any]					     =					None
			__magic_name__	:   Any					     =					None

			# compute loss for embedding
			if not self.legacy:
				__magic_name__	:   Dict					     =					self.beta * torch.mean((z_q.detach() - z) ** 2      ) + torch.mean((z_q - z.detach()) ** 2      )
			else:
				__magic_name__	:   Optional[int]					     =					torch.mean((z_q.detach() - z) ** 2      ) + self.beta * torch.mean((z_q - z.detach()) ** 2      )

			# preserve gradients
			__magic_name__	:   Any					     =					z + (z_q - z).detach()

			# reshape back to match original input shape
			__magic_name__	:   List[Any]					     =					z_q.permute(0  ,					3  ,					1  ,					2      ).contiguous()

			if self.remap is not None:
				__magic_name__	:   List[str]					     =					min_encoding_indices.reshape(z.shape[0]  ,					-1      )  # add batch axis
				__magic_name__	:   Any					     =					self.remap_to_used(_A      )
				__magic_name__	:   str					     =					min_encoding_indices.reshape(-1  ,					1      )  # flatten

			if self.sane_index_shape:
				__magic_name__	:   Tuple					     =					min_encoding_indices.reshape(z_q.shape[0]  ,					z_q.shape[2]  ,					z_q.shape[3]      )

			return z_q, loss, (perplexity, min_encodings, min_encoding_indices)







		def        __lowerCAmelCase		( self		:						Tuple  ,					_A		:						Optional[int]  ,					_A		:						List[Any]      )  ->			Optional[int]:
			# shape specifying (batch, height, width, channel)
			if self.remap is not None:
				__magic_name__	:   str					     =					indices.reshape(shape[0]  ,					-1      )  # add batch axis
				__magic_name__	:   Optional[Any]					     =					self.unmap_to_all(_A      )
				__magic_name__	:   Optional[Any]					     =					indices.reshape(-1      )  # flatten again

			# get quantized latent vectors
			__magic_name__	:   Union[str, Any]					     =					self.embedding(_A      )

			if shape is not None:
				__magic_name__	:   Any					     =					z_q.view(_A      )
				# reshape back to match original input shape
				__magic_name__	:   Tuple					     =					z_q.permute(0  ,					3  ,					1  ,					2      ).contiguous()

			return z_q







class _lowerCamelCase	(						lowercase__							):



		'''simple docstring'''







		def __init__( self		:						Optional[int]  ,					_A		:						str  ,					_A		:						Any=False      )  ->			Dict:
			__magic_name__	:   List[str]					     =					parameters
			__magic_name__     ,    __magic_name__	:   Dict					     =					torch.chunk(_A  ,					2  ,					dim=1      )
			__magic_name__	:   Any					     =					torch.clamp(self.logvar  ,					-30.0  ,					20.0      )
			__magic_name__	:   int					     =					deterministic
			__magic_name__	:   str					     =					torch.exp(0.5 * self.logvar      )
			__magic_name__	:   Tuple					     =					torch.exp(self.logvar      )
			if self.deterministic:
				__magic_name__	:   Optional[int]					     =					torch.zeros_like(
				    self.mean  ,					device=self.parameters.device  ,					dtype=self.parameters.dtype      )







		def        __lowerCAmelCase		( self		:						str  ,					_A		:						Optional[torch.Generator] = None      )  ->			torch.FloatTensor:
			# make sure sample is on the same device as the parameters and has same dtype
			__magic_name__	:   List[Any]					     =					randn_tensor(
			    self.mean.shape  ,					generator=_A  ,					device=self.parameters.device  ,					dtype=self.parameters.dtype      )
			__magic_name__	:   Any					     =					self.mean + self.std * sample
			return x







		def        __lowerCAmelCase		( self		:						List[Any]  ,					_A		:						Tuple=None      )  ->			str:
			if self.deterministic:
				return torch.Tensor([0.0]      )
			else:
				if other is None:
					return 0.5 * torch.sum(torch.pow(self.mean  ,					2      ) + self.var - 1.0 - self.logvar  ,					dim=[1, 2, 3]      )
				else:
					return 0.5 * torch.sum(
					    torch.pow(self.mean - other.mean  ,					2      ) / other.var
					    + self.var / other.var
					    - 1.0
					    - self.logvar
					    + other.logvar  ,					dim=[1, 2, 3]  ,					)







		def        __lowerCAmelCase		( self		:						str  ,					_A		:						Tuple  ,					_A		:						List[Any]=[1, 2, 3]      )  ->			Any:
			if self.deterministic:
				return torch.Tensor([0.0]      )
			__magic_name__	:   List[str]					     =					np.log(2.0 * np.pi      )
			return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean  ,					2      ) / self.var  ,					dim=_A      )







		def        __lowerCAmelCase		( self		:						Optional[int]      )  ->			Union[str, Any]:
			return self.mean



'''simple docstring'''

from typing import List, Optional, Tuple, Union

import PIL
import torch
from torchvision import transforms

from diffusers.pipeline_utils import DiffusionPipeline, ImagePipelineOutput
from diffusers.schedulers import DDIMScheduler
from diffusers.utils import randn_tensor


lowerCAmelCase				:Optional[Any]        =      transforms.Compose(
    [
        transforms.Resize((2_5_6, 2_5_6)),
        transforms.ToTensor(),
        transforms.Normalize([0.5], [0.5]),
    ]
)





def lowerCamelCase      (      lowerCAmelCase    :     Tuple		):





	"""simple docstring"""







	if isinstance(lowerCAmelCase		,		torch.Tensor		):
		return image
	elif isinstance(lowerCAmelCase		,		PIL.Image.Image		):
		__magic_name__	:   List[Any]					     =					[image]

	__magic_name__	:   List[Any]					     =					[trans(img.convert('RGB'		)		) for img in image]
	__magic_name__	:   Dict					     =					torch.stack(lowerCAmelCase		)
	return image





class _lowerCamelCase	(						lowercase__							):



		'''simple docstring'''







		def __init__( self		:						Optional[Any]  ,					_A		:						str  ,					_A		:						int      )  ->			Dict:
			super().__init__()

			# make sure scheduler can always be converted to DDIM
			__magic_name__	:   Optional[int]					     =					DDIMScheduler.from_config(scheduler.config      )

			self.register_modules(unet=_A  ,					scheduler=_A      )







		def        __lowerCAmelCase		( self		:						Union[str, Any]  ,					_A		:						Union[str, Any]      )  ->			Optional[int]:
			if strength < 0 or strength > 1:
				raise ValueError(F'The value of strength should in [0.0, 1.0] but is {strength}'      )







		def        __lowerCAmelCase		( self		:						Any  ,					_A		:						List[str]  ,					_A		:						Optional[Any]  ,					_A		:						int      )  ->			List[Any]:
			# get the original timestep using init_timestep
			__magic_name__	:   Tuple					     =					min(int(num_inference_steps * strength      )  ,					_A      )

			__magic_name__	:   Any					     =					max(num_inference_steps - init_timestep  ,					0      )
			__magic_name__	:   List[str]					     =					self.scheduler.timesteps[t_start:]

			return timesteps, num_inference_steps - t_start







		def        __lowerCAmelCase		( self		:						Any  ,					_A		:						str  ,					_A		:						Optional[int]  ,					_A		:						Tuple  ,					_A		:						List[str]  ,					_A		:						str  ,					_A		:						Optional[int]=None      )  ->			Dict:
			if not isinstance(_A  ,					(torch.Tensor, PIL.Image.Image, list)      ):
				raise ValueError(
				    F'`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(_A      )}'      )

			__magic_name__	:   Union[str, Any]					     =					image.to(device=_A  ,					dtype=_A      )

			if isinstance(_A  ,					_A      ) and len(_A      ) != batch_size:
				raise ValueError(
				    F'You have passed a list of generators of length {len(_A      )}, but requested an effective batch'
				    F' size of {batch_size}. Make sure the batch size matches the length of the generators.'      )

			__magic_name__	:   Tuple					     =					init_latents.shape
			__magic_name__	:   Any					     =					randn_tensor(_A  ,					generator=_A  ,					device=_A  ,					dtype=_A      )

			# get latents
			print('add noise to latents at timestep'  ,					_A      )
			__magic_name__	:   List[str]					     =					self.scheduler.add_noise(_A  ,					_A  ,					_A      )
			__magic_name__	:   List[str]					     =					init_latents

			return latents







		@torch.no_grad()
		def __call__( self		:						Tuple  ,					_A		:						Union[torch.FloatTensor, PIL.Image.Image] = None  ,					_A		:						float = 0.8  ,					_A		:						int = 1  ,					_A		:						Optional[Union[torch.Generator, List[torch.Generator]]] = None  ,					_A		:						float = 0.0  ,					_A		:						int = 50  ,					_A		:						Optional[bool] = None  ,					_A		:						Optional[str] = "pil"  ,					_A		:						bool = True  ,					)  ->			Union[ImagePipelineOutput, Tuple]:
			self.check_inputs(_A      )

			# 2. Preprocess image
			__magic_name__	:   int					     =					preprocess(_A      )

			# 3. set timesteps
			self.scheduler.set_timesteps(_A  ,					device=self.device      )
			__magic_name__     ,    __magic_name__	:   Dict					     =					self.get_timesteps(_A  ,					_A  ,					self.device      )
			__magic_name__	:   Dict					     =					timesteps[:1].repeat(_A      )

			# 4. Prepare latent variables
			__magic_name__	:   Optional[Any]					     =					self.prepare_latents(_A  ,					_A  ,					_A  ,					self.unet.dtype  ,					self.device  ,					_A      )
			__magic_name__	:   Optional[Any]					     =					latents

			# 5. Denoising loop
			for t in self.progress_bar(_A      ):
				# 1. predict noise model_output
				__magic_name__	:   Dict					     =					self.unet(_A  ,					_A      ).sample

				# 2. predict previous mean of image x_t-1 and add variance depending on eta
				# eta corresponds to η in paper and should be between [0, 1]
				# do x_t -> x_t-1
				__magic_name__	:   List[Any]					     =					self.scheduler.step(
				    _A  ,					_A  ,					_A  ,					eta=_A  ,					use_clipped_model_output=_A  ,					generator=_A  ,					).prev_sample

			__magic_name__	:   Union[str, Any]					     =					(image / 2 + 0.5).clamp(0  ,					1      )
			__magic_name__	:   List[Any]					     =					image.cpu().permute(0  ,					2  ,					3  ,					1      ).numpy()
			if output_type == "pil":
				__magic_name__	:   Dict					     =					self.numpy_to_pil(_A      )

			if not return_dict:
				return (image, latent_timestep.item())

			return ImagePipelineOutput(images=_A      )