

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   __lowercase      (_UpperCAmelCase ):
 _UpperCamelCase			  =  42







class   __lowercase      (nn.Module ):
 def __init__(							self      ,							A_=3      ,							A_=3      ,							A_=("DownEncoderBlock2D",)      ,							A_=(64,)      ,							A_=2      ,							A_=32      ,							A_="silu"      ,							A_=True      ,							)					->Union[str, Any]:



       '''simple docstring'''





       super().__init__()
       __lowerCAmelCase					:						Any					=					layers_per_block

       __lowerCAmelCase					:						str					=					torch.nn.Convad(
           A_      ,							block_out_channels[0]      ,							kernel_size=3      ,							stride=1      ,							padding=1      ,							)

       __lowerCAmelCase					:						Union[str, Any]					=					None
       __lowerCAmelCase					:						str					=					nn.ModuleList([]					)

       # down
       __lowerCAmelCase					:						Tuple					=					block_out_channels[0]
       for i, down_block_type in enumerate(A_					):
             __lowerCAmelCase					:						List[Any]					=					output_channel
             __lowerCAmelCase					:						str					=					block_out_channels[i]
             __lowerCAmelCase					:						List[str]					=					i == len(A_					) - 1

             __lowerCAmelCase					:						List[Any]					=					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
       __lowerCAmelCase					:						Tuple					=					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
       __lowerCAmelCase					:						Tuple					=					nn.GroupNorm(num_channels=block_out_channels[-1]      ,							num_groups=A_      ,							eps=1e-6					)
       __lowerCAmelCase					:						Optional[Any]					=					nn.SiLU()

       __lowerCAmelCase					:						Any					=					2 * out_channels if double_z else out_channels
       __lowerCAmelCase					:						Optional[Any]					=					nn.Convad(block_out_channels[-1]      ,							A_      ,							3      ,							padding=1					)

       __lowerCAmelCase					:						List[Any]					=					False





 def 	UpperCamelCase__  (							self      ,							A_					)					->Tuple:



       '''simple docstring'''





       __lowerCAmelCase					:						int					=					x
       __lowerCAmelCase					:						Optional[int]					=					self.conv_in(A_					)

       if self.training and self.gradient_checkpointing:

             def create_custom_forward(A_					):
                   def custom_forward(*A_					):
                         return module(*A_					)

                   return custom_forward

             # down
             if is_torch_version('''>='''      ,							'''1.11.0'''					):
                   for down_block in self.down_blocks:
                         __lowerCAmelCase					:						Optional[Any]					=					torch.utils.checkpoint.checkpoint(
                             create_custom_forward(A_					)      ,							A_      ,							use_reentrant=A_					)
                   # middle
                   __lowerCAmelCase					:						Any					=					torch.utils.checkpoint.checkpoint(
                       create_custom_forward(self.mid_block					)      ,							A_      ,							use_reentrant=A_					)
             else:
                   for down_block in self.down_blocks:
                         __lowerCAmelCase					:						Optional[int]					=					torch.utils.checkpoint.checkpoint(create_custom_forward(A_					)      ,							A_					)
                   # middle
                   __lowerCAmelCase					:						Optional[Any]					=					torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block					)      ,							A_					)

       else:
             # down
             for down_block in self.down_blocks:
                   __lowerCAmelCase					:						str					=					down_block(A_					)

             # middle
             __lowerCAmelCase					:						List[Any]					=					self.mid_block(A_					)

       # post-process
       __lowerCAmelCase					:						Union[str, Any]					=					self.conv_norm_out(A_					)
       __lowerCAmelCase					:						Dict					=					self.conv_act(A_					)
       __lowerCAmelCase					:						List[Any]					=					self.conv_out(A_					)

       return sample







class   __lowercase      (nn.Module ):
 def __init__(							self      ,							A_=3      ,							A_=3      ,							A_=("UpDecoderBlock2D",)      ,							A_=(64,)      ,							A_=2      ,							A_=32      ,							A_="silu"      ,							A_="group"      ,							)					->int:



       '''simple docstring'''





       super().__init__()
       __lowerCAmelCase					:						int					=					layers_per_block

       __lowerCAmelCase					:						int					=					nn.Convad(
           A_      ,							block_out_channels[-1]      ,							kernel_size=3      ,							stride=1      ,							padding=1      ,							)

       __lowerCAmelCase					:						Tuple					=					None
       __lowerCAmelCase					:						Any					=					nn.ModuleList([]					)

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

       # mid
       __lowerCAmelCase					:						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
       __lowerCAmelCase					:						List[str]					=					list(reversed(A_					)					)
       __lowerCAmelCase					:						int					=					reversed_block_out_channels[0]
       for i, up_block_type in enumerate(A_					):
             __lowerCAmelCase					:						int					=					output_channel
             __lowerCAmelCase					:						List[str]					=					reversed_block_out_channels[i]

             __lowerCAmelCase					:						Tuple					=					i == len(A_					) - 1

             __lowerCAmelCase					:						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_					)
             __lowerCAmelCase					:						str					=					output_channel

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

       __lowerCAmelCase					:						Tuple					=					False





 def 	UpperCamelCase__  (							self      ,							A_      ,							A_=None					)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						List[Any]					=					z
       __lowerCAmelCase					:						Any					=					self.conv_in(A_					)

       __lowerCAmelCase					:						Tuple					=					next(iter(self.up_blocks.parameters()					)					).dtype
       if self.training and self.gradient_checkpointing:

             def create_custom_forward(A_					):
                   def custom_forward(*A_					):
                         return module(*A_					)

                   return custom_forward

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

                   # up
                   for up_block in self.up_blocks:
                         __lowerCAmelCase					:						List[Any]					=					torch.utils.checkpoint.checkpoint(
                             create_custom_forward(A_					)      ,							A_      ,							A_      ,							use_reentrant=A_					)
             else:
                   # middle
                   __lowerCAmelCase					:						str					=					torch.utils.checkpoint.checkpoint(
                       create_custom_forward(self.mid_block					)      ,							A_      ,							A_					)
                   __lowerCAmelCase					:						List[Any]					=					sample.to(A_					)

                   # up
                   for up_block in self.up_blocks:
                         __lowerCAmelCase					:						Any					=					torch.utils.checkpoint.checkpoint(create_custom_forward(A_					)      ,							A_      ,							A_					)
       else:
             # middle
             __lowerCAmelCase					:						List[str]					=					self.mid_block(A_      ,							A_					)
             __lowerCAmelCase					:						List[str]					=					sample.to(A_					)

             # up
             for up_block in self.up_blocks:
                   __lowerCAmelCase					:						int					=					up_block(A_      ,							A_					)

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

       return sample







class   __lowercase      (nn.Module ):
 def __init__(							self      ,							A_      ,							A_      ,							A_      ,							A_=None      ,							A_="random"      ,							A_=False      ,							A_=True					)					->List[str]:



       '''simple docstring'''





       super().__init__()
       __lowerCAmelCase					:						List[Any]					=					n_e
       __lowerCAmelCase					:						Dict					=					vq_embed_dim
       __lowerCAmelCase					:						Tuple					=					beta
       __lowerCAmelCase					:						Union[str, Any]					=					legacy

       __lowerCAmelCase					:						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					)

       __lowerCAmelCase					:						Union[str, Any]					=					remap
       if self.remap is not None:
             self.register_buffer('''used'''      ,							torch.tensor(np.load(self.remap					)					)					)
             __lowerCAmelCase					:						Optional[Any]					=					self.used.shape[0]
             __lowerCAmelCase					:						List[Any]					=					unknown_index  # "random" or "extra" or integer
             if self.unknown_index == "extra":
                   __lowerCAmelCase					:						Optional[Any]					=					self.re_embed
                   __lowerCAmelCase					:						Optional[int]					=					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:
             __lowerCAmelCase					:						Optional[Any]					=					n_e

       __lowerCAmelCase					:						List[str]					=					sane_index_shape
 def 	UpperCamelCase__  (							self      ,							A_					)					->Optional[Any]:



       '''simple docstring'''





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



       '''simple docstring'''





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



       '''simple docstring'''





       __lowerCAmelCase					:						str					=					z.permute(0      ,							2      ,							3      ,							1					).contiguous()
       __lowerCAmelCase					:						Any					=					z.view(-1      ,							self.vq_embed_dim					)

       # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
       __lowerCAmelCase					:						Optional[int]					=					torch.argmin(torch.cdist(A_      ,							self.embedding.weight					)      ,							dim=1					)

       __lowerCAmelCase					:						Dict					=					self.embedding(A_					).view(z.shape					)
       __lowerCAmelCase					:						str					=					None
       __lowerCAmelCase					:						Optional[Any]					=					None

       # compute loss for embedding
       if not self.legacy:
             __lowerCAmelCase					:						Dict					=					self.beta * torch.mean((z_q.detach() - z) ** 2					) + torch.mean((z_q - z.detach()) ** 2					)
       else:
             __lowerCAmelCase					:						Any					=					torch.mean((z_q.detach() - z) ** 2					) + self.beta * torch.mean((z_q - z.detach()) ** 2					)

       # preserve gradients
       __lowerCAmelCase					:						Dict					=					z + (z_q - z).detach()

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

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

       if self.sane_index_shape:
             __lowerCAmelCase					:						Union[str, Any]					=					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 	UpperCamelCase__  (							self      ,							A_      ,							A_					)					->List[Any]:



       '''simple docstring'''





       if self.remap is not None:
             __lowerCAmelCase					:						Tuple					=					indices.reshape(shape[0]      ,							-1					)  # add batch axis
             __lowerCAmelCase					:						Optional[int]					=					self.unmap_to_all(A_					)
             __lowerCAmelCase					:						str					=					indices.reshape(-1					)  # flatten again

       # get quantized latent vectors
       __lowerCAmelCase					:						int					=					self.embedding(A_					)

       if shape is not None:
             __lowerCAmelCase					:						Any					=					z_q.view(A_					)
             # reshape back to match original input shape
             __lowerCAmelCase					:						List[str]					=					z_q.permute(0      ,							3      ,							1      ,							2					).contiguous()

       return z_q







class   __lowercase      (_UpperCAmelCase ):
 def __init__(							self      ,							A_      ,							A_=False					)					->List[str]:



       '''simple docstring'''





       __lowerCAmelCase					:						Tuple					=					parameters
       __lowerCAmelCase,							__lowerCAmelCase					:						Any					=					torch.chunk(A_      ,							2      ,							dim=1					)
       __lowerCAmelCase					:						str					=					torch.clamp(self.logvar      ,							-30.0      ,							20.0					)
       __lowerCAmelCase					:						List[str]					=					deterministic
       __lowerCAmelCase					:						str					=					torch.exp(0.5 * self.logvar					)
       __lowerCAmelCase					:						List[Any]					=					torch.exp(self.logvar					)
       if self.deterministic:
             __lowerCAmelCase					:						Optional[int]					=					torch.zeros_like(
                 self.mean      ,							device=self.parameters.device      ,							dtype=self.parameters.dtype					)
 def 	UpperCamelCase__  (							self      ,							A_ = None					)					->torch.FloatTensor:



       '''simple docstring'''





       __lowerCAmelCase					:						Tuple					=					randn_tensor(
           self.mean.shape      ,							generator=A_      ,							device=self.parameters.device      ,							dtype=self.parameters.dtype					)
       __lowerCAmelCase					:						Optional[int]					=					self.mean + self.std * sample
       return x
 def 	UpperCamelCase__  (							self      ,							A_=None					)					->Dict:



       '''simple docstring'''





       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 	UpperCamelCase__  (							self      ,							A_      ,							A_=[1, 2, 3]					)					->Optional[int]:



       '''simple docstring'''





       if self.deterministic:
             return torch.Tensor([0.0]					)
       __lowerCAmelCase					:						Union[str, Any]					=					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 	UpperCamelCase__  (							self					)					->Optional[Any]:



       '''simple docstring'''





       return self.mean





from dataclasses import dataclass
from typing import Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin







@dataclass
class   __lowercase      (_UpperCAmelCase ):
 _UpperCamelCase			  =  42
 _UpperCamelCase			  =  42
 _UpperCamelCase			  =  None







class   __lowercase      (_UpperCAmelCase		,   _UpperCAmelCase ):
 _UpperCamelCase			  =  2
 @register_to_config
 def __init__(							self      ,							A_ = 0.02      ,							A_ = 100      ,							A_ = 1.007      ,							A_ = 80      ,							A_ = 0.05      ,							A_ = 50      ,							)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					sigma_max

       # setable values
       __lowerCAmelCase					:						int					=					None
       __lowerCAmelCase					:						np.IntTensor					=					None
       __lowerCAmelCase					:						torch.FloatTensor					=					None  # sigma(t_i)
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_ = None					)					->torch.FloatTensor:



       '''simple docstring'''





       return sample
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_ = None					)					->List[str]:



       '''simple docstring'''





       __lowerCAmelCase					:						str					=					num_inference_steps
       __lowerCAmelCase					:						Dict					=					np.arange(0      ,							self.num_inference_steps					)[::-1].copy()
       __lowerCAmelCase					:						Optional[Any]					=					torch.from_numpy(A_					).to(A_					)
       __lowerCAmelCase					:						Tuple					=					[
           (
               self.config.sigma_max**2
               * (self.config.sigma_min**2 / self.config.sigma_max**2) ** (i / (num_inference_steps - 1))
           )
           for i in self.timesteps
       ]
       __lowerCAmelCase					:						Optional[int]					=					torch.tensor(A_      ,							dtype=torch.floataa      ,							device=A_					)
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_      ,							A_ = None					)					->Tuple[torch.FloatTensor, float]:



       '''simple docstring'''





       if self.config.s_min <= sigma <= self.config.s_max:
             __lowerCAmelCase					:						List[str]					=					min(self.config.s_churn / self.num_inference_steps      ,							2**0.5 - 1					)
       else:
             __lowerCAmelCase					:						List[str]					=					0

       # sample eps ~ N(0, S_noise^2 * I)
       __lowerCAmelCase					:						int					=					self.config.s_noise * randn_tensor(sample.shape      ,							generator=A_					).to(sample.device					)
       __lowerCAmelCase					:						str					=					sigma + gamma * sigma
       __lowerCAmelCase					:						Any					=					sample + ((sigma_hat**2 - sigma**2) ** 0.5 * eps)

       return sample_hat, sigma_hat
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_      ,							A_      ,							A_      ,							A_ = True      ,							)					->Union[KarrasVeOutput, Tuple]:



       '''simple docstring'''





       __lowerCAmelCase					:						Union[str, Any]					=					sample_hat + sigma_hat * model_output
       __lowerCAmelCase					:						int					=					(sample_hat - pred_original_sample) / sigma_hat
       __lowerCAmelCase					:						Tuple					=					sample_hat + (sigma_prev - sigma_hat) * derivative

       if not return_dict:
             return (sample_prev, derivative)

       return KarrasVeOutput(
           prev_sample=A_      ,							derivative=A_      ,							pred_original_sample=A_					)
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_      ,							A_      ,							A_      ,							A_      ,							A_      ,							A_ = True      ,							)					->Union[KarrasVeOutput, Tuple]:



       '''simple docstring'''





       __lowerCAmelCase					:						str					=					sample_prev + sigma_prev * model_output
       __lowerCAmelCase					:						List[Any]					=					(sample_prev - pred_original_sample) / sigma_prev
       __lowerCAmelCase					:						Any					=					sample_hat + (sigma_prev - sigma_hat) * (0.5 * derivative + 0.5 * derivative_corr)

       if not return_dict:
             return (sample_prev, derivative)

       return KarrasVeOutput(
           prev_sample=A_      ,							derivative=A_      ,							pred_original_sample=A_					)





 def 	UpperCamelCase__  (							self      ,							A_      ,							A_      ,							A_					)					->Any:



       '''simple docstring'''





       raise NotImplementedError()


