

'''simple docstring'''







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

import torch

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


@dataclass
class 					A__      (      UpperCAmelCase__			):
    lowercase						=		42
    lowercase						=		42


class 					A__      (      UpperCAmelCase__						,						UpperCAmelCase__			):
    lowercase						=		1



    @register_to_config
    def __init__(					self		,				UpperCamelCase__ = 2000		,				UpperCamelCase__ = 0.15		,				UpperCamelCase__ = 0.01		,				UpperCamelCase__ = 1348.0		,				UpperCamelCase__ = 1e-5		,				UpperCamelCase__ = 1		,				)						->      Optional[Any]:





           '''simple docstring'''
           A_						=      sigma_max

           # setable values
           A_						=      None

           self.set_sigmas(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__       )



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__ = None       )						->      torch.FloatTensor:





           '''simple docstring'''
           return sample



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__ = None		,				UpperCamelCase__ = None       )						->      str:





           '''simple docstring'''
           A_						=      sampling_eps if sampling_eps is not None else self.config.sampling_eps

           A_						=      torch.linspace(1		,				UpperCamelCase__		,				UpperCamelCase__		,				device=UpperCamelCase__       )



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__ = None		,				UpperCamelCase__ = None		,				UpperCamelCase__ = None       )						->      Optional[int]:





           '''simple docstring'''
           A_						=      sigma_min if sigma_min is not None else self.config.sigma_min
           A_						=      sigma_max if sigma_max is not None else self.config.sigma_max
           A_						=      sampling_eps if sampling_eps is not None else self.config.sampling_eps
           if self.timesteps is None:
                  self.set_timesteps(UpperCamelCase__		,				UpperCamelCase__       )

           A_						=      sigma_min * (sigma_max / sigma_min) ** (self.timesteps / sampling_eps)
           A_						=      torch.exp(torch.linspace(math.log(UpperCamelCase__       )		,				math.log(UpperCamelCase__       )		,				UpperCamelCase__       )       )
           A_						=      torch.tensor([sigma_min * (sigma_max / sigma_min) ** t for t in self.timesteps]       )



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__       )						->      Tuple:





           '''simple docstring'''
           return torch.where(
               timesteps == 0		,				torch.zeros_like(t.to(timesteps.device       )       )		,				self.discrete_sigmas[timesteps - 1].to(timesteps.device       )		,				)



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__ = None		,				UpperCamelCase__ = True		,				)						->      Union[SdeVeOutput, Tuple]:





           '''simple docstring'''




           if self.timesteps is None:
                  raise ValueError(
                      """`self.timesteps` is not set, you need to run 'set_timesteps' after creating the scheduler"""       )

           A_						=      timestep * torch.ones(
               sample.shape[0]		,				device=sample.device       )  # torch.repeat_interleave(timestep, sample.shape[0])
           A_						=      (timestep * (len(self.timesteps       ) - 1)).long()

           # mps requires indices to be in the same device, so we use cpu as is the default with cuda
           A_						=      timesteps.to(self.discrete_sigmas.device       )

           A_						=      self.discrete_sigmas[timesteps].to(sample.device       )
           A_						=      self.get_adjacent_sigma(UpperCamelCase__		,				UpperCamelCase__       ).to(sample.device       )
           A_						=      torch.zeros_like(UpperCamelCase__       )
           A_						=      (sigma**2 - adjacent_sigma**2) ** 0.5

           # equation 6 in the paper: the model_output modeled by the network is grad_x log pt(x)
           # also equation 47 shows the analog from SDE models to ancestral sampling methods
           A_						=      diffusion.flatten()
           while len(diffusion.shape       ) < len(sample.shape       ):
                  A_						=      diffusion.unsqueeze(-1       )
           A_						=      drift - diffusion**2 * model_output

           #  equation 6: sample noise for the diffusion term of
           A_						=      randn_tensor(
               sample.shape		,				layout=sample.layout		,				generator=UpperCamelCase__		,				device=sample.device		,				dtype=sample.dtype       )
           A_						=      sample - drift  # subtract because `dt` is a small negative timestep
           # TODO is the variable diffusion the correct scaling term for the noise?
           A_						=      prev_sample_mean + diffusion * noise  # add impact of diffusion field g

           if not return_dict:
                  return (prev_sample, prev_sample_mean)

           return SdeVeOutput(prev_sample=UpperCamelCase__		,				prev_sample_mean=UpperCamelCase__       )



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__ = None		,				UpperCamelCase__ = True		,				)						->      Union[SchedulerOutput, Tuple]:





           '''simple docstring'''




           if self.timesteps is None:
                  raise ValueError(
                      """`self.timesteps` is not set, you need to run 'set_timesteps' after creating the scheduler"""       )

           # For small batch sizes, the paper "suggest replacing norm(z) with sqrt(d), where d is the dim. of z"
           # sample noise for correction
           A_						=      randn_tensor(sample.shape		,				layout=sample.layout		,				generator=UpperCamelCase__       ).to(sample.device       )

           # compute step size from the model_output, the noise, and the snr
           A_						=      torch.norm(model_output.reshape(model_output.shape[0]		,				-1       )		,				dim=-1       ).mean()
           A_						=      torch.norm(noise.reshape(noise.shape[0]		,				-1       )		,				dim=-1       ).mean()
           A_						=      (self.config.snr * noise_norm / grad_norm) ** 2 * 2
           A_						=      step_size * torch.ones(sample.shape[0]       ).to(sample.device       )
           # self.repeat_scalar(step_size, sample.shape[0])

           # compute corrected sample: model_output term and noise term
           A_						=      step_size.flatten()
           while len(step_size.shape       ) < len(sample.shape       ):
                  A_						=      step_size.unsqueeze(-1       )
           A_						=      sample + step_size * model_output
           A_						=      prev_sample_mean + ((step_size * 2) ** 0.5) * noise

           if not return_dict:
                  return (prev_sample,)

           return SchedulerOutput(prev_sample=UpperCamelCase__       )



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				)						->      torch.FloatTensor:





           '''simple docstring'''
           A_						=      timesteps.to(original_samples.device       )
           A_						=      self.discrete_sigmas.to(original_samples.device       )[timesteps]
           A_						=      (
               noise * sigmas[:, None, None, None]
               if noise is not None
               else torch.randn_like(UpperCamelCase__       ) * sigmas[:, None, None, None]
           )
           A_						=      noise + original_samples
           return noisy_samples







    def __len__(					self       )						->      Dict:





           '''simple docstring'''
           return self.config.num_train_timesteps



'''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 					A__      (      _snake_case			):
    lowercase						=		42


class 					A__      (      nn.Module			):



    def __init__(					self		,				UpperCamelCase__=3		,				UpperCamelCase__=3		,				UpperCamelCase__=("DownEncoderBlock2D",)		,				UpperCamelCase__=(64,)		,				UpperCamelCase__=2		,				UpperCamelCase__=32		,				UpperCamelCase__="silu"		,				UpperCamelCase__=True		,				)						->      Union[str, Any]:





           '''simple docstring'''
           super().__init__()
           A_						=      layers_per_block

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

           A_						=      None
           A_						=      nn.ModuleList([]       )

           # down
           A_						=      block_out_channels[0]
           for i, down_block_type in enumerate(UpperCamelCase__       ):
                  A_						=      output_channel
                  A_						=      block_out_channels[i]
                  A_						=      i == len(UpperCamelCase__       ) - 1

                  A_						=      get_down_block(
                      UpperCamelCase__		,				num_layers=self.layers_per_block		,				in_channels=UpperCamelCase__		,				out_channels=UpperCamelCase__		,				add_downsample=not is_final_block		,				resnet_eps=1e-6		,				downsample_padding=0		,				resnet_act_fn=UpperCamelCase__		,				resnet_groups=UpperCamelCase__		,				attention_head_dim=UpperCamelCase__		,				temb_channels=UpperCamelCase__		,				)
                  self.down_blocks.append(UpperCamelCase__       )

           # mid
           A_						=      UNetMidBlockaD(
               in_channels=block_out_channels[-1]		,				resnet_eps=1e-6		,				resnet_act_fn=UpperCamelCase__		,				output_scale_factor=1		,				resnet_time_scale_shift="""default"""		,				attention_head_dim=block_out_channels[-1]		,				resnet_groups=UpperCamelCase__		,				temb_channels=UpperCamelCase__		,				)

           # out
           A_						=      nn.GroupNorm(num_channels=block_out_channels[-1]		,				num_groups=UpperCamelCase__		,				eps=1e-6       )
           A_						=      nn.SiLU()

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

           A_						=      False







    def 					snake_case_			(					self		,				UpperCamelCase__       )						->      str:





           '''simple docstring'''
           A_						=      x
           A_						=      self.conv_in(UpperCamelCase__       )

           if self.training and self.gradient_checkpointing:

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

                         return custom_forward

                  # down
                  if is_torch_version(""">="""		,				"""1.11.0"""       ):
                         for down_block in self.down_blocks:
                                A_						=      torch.utils.checkpoint.checkpoint(
                                    create_custom_forward(UpperCamelCase__       )		,				UpperCamelCase__		,				use_reentrant=UpperCamelCase__       )
                         # middle
                         A_						=      torch.utils.checkpoint.checkpoint(
                             create_custom_forward(self.mid_block       )		,				UpperCamelCase__		,				use_reentrant=UpperCamelCase__       )
                  else:
                         for down_block in self.down_blocks:
                                A_						=      torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCamelCase__       )		,				UpperCamelCase__       )
                         # middle
                         A_						=      torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block       )		,				UpperCamelCase__       )

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

                  # middle
                  A_						=      self.mid_block(UpperCamelCase__       )

           # post-process
           A_						=      self.conv_norm_out(UpperCamelCase__       )
           A_						=      self.conv_act(UpperCamelCase__       )
           A_						=      self.conv_out(UpperCamelCase__       )

           return sample


class 					A__      (      nn.Module			):



    def __init__(					self		,				UpperCamelCase__=3		,				UpperCamelCase__=3		,				UpperCamelCase__=("UpDecoderBlock2D",)		,				UpperCamelCase__=(64,)		,				UpperCamelCase__=2		,				UpperCamelCase__=32		,				UpperCamelCase__="silu"		,				UpperCamelCase__="group"		,				)						->      List[Any]:





           '''simple docstring'''
           super().__init__()
           A_						=      layers_per_block

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

           A_						=      None
           A_						=      nn.ModuleList([]       )

           A_						=      in_channels if norm_type == """spatial""" else None

           # mid
           A_						=      UNetMidBlockaD(
               in_channels=block_out_channels[-1]		,				resnet_eps=1e-6		,				resnet_act_fn=UpperCamelCase__		,				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=UpperCamelCase__		,				temb_channels=UpperCamelCase__		,				)

           # up
           A_						=      list(reversed(UpperCamelCase__       )       )
           A_						=      reversed_block_out_channels[0]
           for i, up_block_type in enumerate(UpperCamelCase__       ):
                  A_						=      output_channel
                  A_						=      reversed_block_out_channels[i]

                  A_						=      i == len(UpperCamelCase__       ) - 1

                  A_						=      get_up_block(
                      UpperCamelCase__		,				num_layers=self.layers_per_block + 1		,				in_channels=UpperCamelCase__		,				out_channels=UpperCamelCase__		,				prev_output_channel=UpperCamelCase__		,				add_upsample=not is_final_block		,				resnet_eps=1e-6		,				resnet_act_fn=UpperCamelCase__		,				resnet_groups=UpperCamelCase__		,				attention_head_dim=UpperCamelCase__		,				temb_channels=UpperCamelCase__		,				resnet_time_scale_shift=UpperCamelCase__		,				)
                  self.up_blocks.append(UpperCamelCase__       )
                  A_						=      output_channel

           # out
           if norm_type == "spatial":
                  A_						=      SpatialNorm(block_out_channels[0]		,				UpperCamelCase__       )
           else:
                  A_						=      nn.GroupNorm(num_channels=block_out_channels[0]		,				num_groups=UpperCamelCase__		,				eps=1e-6       )
           A_						=      nn.SiLU()
           A_						=      nn.Convad(block_out_channels[0]		,				UpperCamelCase__		,				3		,				padding=1       )

           A_						=      False







    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__=None       )						->      Optional[Any]:





           '''simple docstring'''
           A_						=      z
           A_						=      self.conv_in(UpperCamelCase__       )

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

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

                         return custom_forward

                  if is_torch_version(""">="""		,				"""1.11.0"""       ):
                         # middle
                         A_						=      torch.utils.checkpoint.checkpoint(
                             create_custom_forward(self.mid_block       )		,				UpperCamelCase__		,				UpperCamelCase__		,				use_reentrant=UpperCamelCase__       )
                         A_						=      sample.to(UpperCamelCase__       )

                         # up
                         for up_block in self.up_blocks:
                                A_						=      torch.utils.checkpoint.checkpoint(
                                    create_custom_forward(UpperCamelCase__       )		,				UpperCamelCase__		,				UpperCamelCase__		,				use_reentrant=UpperCamelCase__       )
                  else:
                         # middle
                         A_						=      torch.utils.checkpoint.checkpoint(
                             create_custom_forward(self.mid_block       )		,				UpperCamelCase__		,				UpperCamelCase__       )
                         A_						=      sample.to(UpperCamelCase__       )

                         # up
                         for up_block in self.up_blocks:
                                A_						=      torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCamelCase__       )		,				UpperCamelCase__		,				UpperCamelCase__       )
           else:
                  # middle
                  A_						=      self.mid_block(UpperCamelCase__		,				UpperCamelCase__       )
                  A_						=      sample.to(UpperCamelCase__       )

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

        # post-process
           if latent_embeds is None:
                  A_						=      self.conv_norm_out(UpperCamelCase__       )
           else:
                  A_						=      self.conv_norm_out(UpperCamelCase__		,				UpperCamelCase__       )
           A_						=      self.conv_act(UpperCamelCase__       )
           A_						=      self.conv_out(UpperCamelCase__       )

           return sample


class 					A__      (      nn.Module			):



    def __init__(					self		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__=None		,				UpperCamelCase__="random"		,				UpperCamelCase__=False		,				UpperCamelCase__=True       )						->      str:





           '''simple docstring'''
           super().__init__()
           A_						=      n_e
           A_						=      vq_embed_dim
           A_						=      beta
           A_						=      legacy

           A_						=      nn.Embedding(self.n_e		,				self.vq_embed_dim       )
           self.embedding.weight.data.uniform_(-1.0 / self.n_e		,				1.0 / self.n_e       )

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

           A_						=      sane_index_shape



    def 					snake_case_			(					self		,				UpperCamelCase__       )						->      Optional[Any]:





           '''simple docstring'''
           A_						=      inds.shape
           assert len(UpperCamelCase__       ) > 1
           A_						=      inds.reshape(ishape[0]		,				-1       )
           A_						=      self.used.to(UpperCamelCase__       )
           A_						=      (inds[:, :, None] == used[None, None, ...]).long()
           A_						=      match.argmax(-1       )
           A_						=      match.sum(2       ) < 1
           if self.unknown_index == "random":
                  A_						=      torch.randint(0		,				self.re_embed		,				size=new[unknown].shape       ).to(device=new.device       )
           else:
                  A_						=      self.unknown_index
           return new.reshape(UpperCamelCase__       )



    def 					snake_case_			(					self		,				UpperCamelCase__       )						->      Dict:





           '''simple docstring'''
           A_						=      inds.shape
           assert len(UpperCamelCase__       ) > 1
           A_						=      inds.reshape(ishape[0]		,				-1       )
           A_						=      self.used.to(UpperCamelCase__       )
           if self.re_embed > self.used.shape[0]:  # extra token
                  A_						=      0  # simply set to zero
           A_						=      torch.gather(used[None, :][inds.shape[0] * [0], :]		,				1		,				UpperCamelCase__       )
           return back.reshape(UpperCamelCase__       )



    def 					snake_case_			(					self		,				UpperCamelCase__       )						->      str:





           '''simple docstring'''
           # reshape z -> (batch, height, width, channel) and flatten
           A_						=      z.permute(0		,				2		,				3		,				1       ).contiguous()
           A_						=      z.view(-1		,				self.vq_embed_dim       )

           # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
           A_						=      torch.argmin(torch.cdist(UpperCamelCase__		,				self.embedding.weight       )		,				dim=1       )

           A_						=      self.embedding(UpperCamelCase__       ).view(z.shape       )
           A_						=      None
           A_						=      None

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

           # preserve gradients
           A_						=      z + (z_q - z).detach()

           # reshape back to match original input shape
           A_						=      z_q.permute(0		,				3		,				1		,				2       ).contiguous()

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

           if self.sane_index_shape:
                  A_						=      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 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__       )						->      Optional[int]:





           '''simple docstring'''
           # shape specifying (batch, height, width, channel)
           if self.remap is not None:
                  A_						=      indices.reshape(shape[0]		,				-1       )  # add batch axis
                  A_						=      self.unmap_to_all(UpperCamelCase__       )
                  A_						=      indices.reshape(-1       )  # flatten again

           # get quantized latent vectors
           A_						=      self.embedding(UpperCamelCase__       )

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

           return z_q


class 					A__      (      _snake_case			):



    def __init__(					self		,				UpperCamelCase__		,				UpperCamelCase__=False       )						->      Dict:





           '''simple docstring'''
           A_						=      parameters
           A_					,			A_						=      torch.chunk(UpperCamelCase__		,				2		,				dim=1       )
           A_						=      torch.clamp(self.logvar		,				-30.0		,				20.0       )
           A_						=      deterministic
           A_						=      torch.exp(0.5 * self.logvar       )
           A_						=      torch.exp(self.logvar       )
           if self.deterministic:
                  A_						=      A_						=      torch.zeros_like(
                      self.mean		,				device=self.parameters.device		,				dtype=self.parameters.dtype       )



    def 					snake_case_			(					self		,				UpperCamelCase__ = None       )						->      torch.FloatTensor:





           '''simple docstring'''
           # make sure sample is on the same device as the parameters and has same dtype
           A_						=      randn_tensor(
               self.mean.shape		,				generator=UpperCamelCase__		,				device=self.parameters.device		,				dtype=self.parameters.dtype       )
           A_						=      self.mean + self.std * sample
           return x



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





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





           '''simple docstring'''
           if self.deterministic:
                  return torch.Tensor([0.0]       )
           A_						=      np.log(2.0 * np.pi       )
           return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean		,				2       ) / self.var		,				dim=UpperCamelCase__       )







    def 					snake_case_			(					self       )						->      List[Any]:





           '''simple docstring'''
           return self.mean
