



import torch

from diffusers import KDPMaAncestralDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest



class        lowerCAmelCase_	(						A__							):
       UpperCAmelCase__       :      List[str]       =						(KDPMaAncestralDiscreteScheduler,)
       UpperCAmelCase__       :      Tuple       =						10
       def       snake_case_ (    self,    **SCREAMING_SNAKE_CASE_ )  ->	List[str]:
          UpperCamelCase					:						int					=			{
              'num_train_timesteps': 1100,
              'beta_start': 0.00_01,
              'beta_end': 0.02,
              'beta_schedule': 'linear',
          }

          config.update(**_lowerCAmelCase )
          return config
       def       snake_case_ (    self )  ->	Tuple:
          for timesteps in [10, 50, 100, 1000]:
             self.check_over_configs(num_train_timesteps=_lowerCAmelCase )
       def       snake_case_ (    self )  ->	Tuple:
          for beta_start, beta_end in zip([0.0_00_01, 0.00_01, 0.0_01],    [0.00_02, 0.0_02, 0.02] ):
             self.check_over_configs(beta_start=_lowerCAmelCase,    beta_end=_lowerCAmelCase )
       def       snake_case_ (    self )  ->	Any:
          for schedule in ["linear", "scaled_linear"]:
             self.check_over_configs(beta_schedule=_lowerCAmelCase )
       def       snake_case_ (    self )  ->	str:
          if torch_device == "mps":
             return
          UpperCamelCase					:						List[str]					=			self.scheduler_classes[0]
          UpperCamelCase					:						Tuple					=			self.get_scheduler_config()
          UpperCamelCase					:						List[str]					=			scheduler_class(**_lowerCAmelCase )

          scheduler.set_timesteps(self.num_inference_steps )

          UpperCamelCase					:						Optional[int]					=			torch.manual_seed(0 )

          UpperCamelCase					:						Dict					=			self.dummy_model()
          UpperCamelCase					:						List[Any]					=			self.dummy_sample_deter * scheduler.init_noise_sigma
          UpperCamelCase					:						Dict					=			sample.to(_lowerCAmelCase )

          for i, t in enumerate(scheduler.timesteps ):
             UpperCamelCase					:						Any					=			scheduler.scale_model_input(_lowerCAmelCase,    _lowerCAmelCase )

             UpperCamelCase					:						List[Any]					=			model(_lowerCAmelCase,    _lowerCAmelCase )

             UpperCamelCase					:						Dict					=			scheduler.step(_lowerCAmelCase,    _lowerCAmelCase,    _lowerCAmelCase,    generator=_lowerCAmelCase )
             UpperCamelCase					:						List[Any]					=			output.prev_sample

          UpperCamelCase					:						List[Any]					=			torch.sum(torch.abs(_lowerCAmelCase ) )
          UpperCamelCase					:						Any					=			torch.mean(torch.abs(_lowerCAmelCase ) )

          assert abs(result_sum.item() - 1_3849.3877 ) < 1e-2
          assert abs(result_mean.item() - 18.03_31 ) < 5e-3
       def       snake_case_ (    self )  ->	List[str]:
          for prediction_type in ["epsilon", "v_prediction"]:
             self.check_over_configs(prediction_type=_lowerCAmelCase )
       def       snake_case_ (    self )  ->	List[str]:
          if torch_device == "mps":
             return
          UpperCamelCase					:						int					=			self.scheduler_classes[0]
          UpperCamelCase					:						Union[str, Any]					=			self.get_scheduler_config(prediction_type='v_prediction' )
          UpperCamelCase					:						List[Any]					=			scheduler_class(**_lowerCAmelCase )

          scheduler.set_timesteps(self.num_inference_steps )

          UpperCamelCase					:						Dict					=			self.dummy_model()
          UpperCamelCase					:						Dict					=			self.dummy_sample_deter * scheduler.init_noise_sigma
          UpperCamelCase					:						List[Any]					=			sample.to(_lowerCAmelCase )

          UpperCamelCase					:						Optional[int]					=			torch.manual_seed(0 )

          for i, t in enumerate(scheduler.timesteps ):
             UpperCamelCase					:						int					=			scheduler.scale_model_input(_lowerCAmelCase,    _lowerCAmelCase )

             UpperCamelCase					:						List[Any]					=			model(_lowerCAmelCase,    _lowerCAmelCase )

             UpperCamelCase					:						Any					=			scheduler.step(_lowerCAmelCase,    _lowerCAmelCase,    _lowerCAmelCase,    generator=_lowerCAmelCase )
             UpperCamelCase					:						Union[str, Any]					=			output.prev_sample

          UpperCamelCase					:						Union[str, Any]					=			torch.sum(torch.abs(_lowerCAmelCase ) )
          UpperCamelCase					:						int					=			torch.mean(torch.abs(_lowerCAmelCase ) )

          assert abs(result_sum.item() - 3_28.99_70 ) < 1e-2
          assert abs(result_mean.item() - 0.42_84 ) < 1e-3






       def       snake_case_ (    self )  ->	Optional[Any]:
          if torch_device == "mps":
             return
          UpperCamelCase					:						Dict					=			self.scheduler_classes[0]
          UpperCamelCase					:						Optional[int]					=			self.get_scheduler_config()
          UpperCamelCase					:						Dict					=			scheduler_class(**_lowerCAmelCase )

          scheduler.set_timesteps(self.num_inference_steps,    device=_lowerCAmelCase )
          UpperCamelCase					:						Optional[Any]					=			torch.manual_seed(0 )

          UpperCamelCase					:						Optional[Any]					=			self.dummy_model()
          UpperCamelCase					:						Dict					=			self.dummy_sample_deter.to(_lowerCAmelCase ) * scheduler.init_noise_sigma

          for t in scheduler.timesteps:
             UpperCamelCase					:						Optional[int]					=			scheduler.scale_model_input(_lowerCAmelCase,    _lowerCAmelCase )

             UpperCamelCase					:						Union[str, Any]					=			model(_lowerCAmelCase,    _lowerCAmelCase )

             UpperCamelCase					:						int					=			scheduler.step(_lowerCAmelCase,    _lowerCAmelCase,    _lowerCAmelCase,    generator=_lowerCAmelCase )
             UpperCamelCase					:						int					=			output.prev_sample

          UpperCamelCase					:						Optional[int]					=			torch.sum(torch.abs(_lowerCAmelCase ) )
          UpperCamelCase					:						int					=			torch.mean(torch.abs(_lowerCAmelCase ) )

          assert abs(result_sum.item() - 1_3849.3818 ) < 1e-1
          assert abs(result_mean.item() - 18.03_31 ) < 1e-3








from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin



@dataclass
class        lowerCAmelCase_	(						a__							):
       UpperCAmelCase__       :      torch.FloatTensor






class        lowerCAmelCase_	(						a__ ,					a__							):
       @register_to_config
       def __init__(    self,    SCREAMING_SNAKE_CASE_ = 16,    SCREAMING_SNAKE_CASE_ = 88,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = 1,    SCREAMING_SNAKE_CASE_ = 0.0,    SCREAMING_SNAKE_CASE_ = 32,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = False,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = "geglu",    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = False,    SCREAMING_SNAKE_CASE_ = False,    SCREAMING_SNAKE_CASE_ = False,    SCREAMING_SNAKE_CASE_ = "layer_norm",    SCREAMING_SNAKE_CASE_ = True,    )  ->	Union[str, Any]:
          super().__init__()
          UpperCamelCase					:						Tuple					=			use_linear_projection
          UpperCamelCase					:						int					=			num_attention_heads
          UpperCamelCase					:						Dict					=			attention_head_dim
          UpperCamelCase					:						Optional[int]					=			num_attention_heads * attention_head_dim

          # 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
          # Define whether input is continuous or discrete depending on configuration
          UpperCamelCase					:						Optional[Any]					=			(in_channels is not None) and (patch_size is None)
          UpperCamelCase					:						Tuple					=			num_vector_embeds is not None
          UpperCamelCase					:						Optional[int]					=			in_channels is not None and patch_size is not None

          if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
             UpperCamelCase					:						Any					=			(
                 F"""The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or"""
                 ' incorrectly set to `\'layer_norm\'`.Make sure to set `norm_type` to `\'ada_norm\'` in the config.'
                 ' Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect'
                 ' results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it'
                 ' would be very nice if you could open a Pull request for the `transformer/config.json` file'
             )
             deprecate('norm_type!=num_embeds_ada_norm',    '1.0.0',    SCREAMING_SNAKE_CASE_,    standard_warn=SCREAMING_SNAKE_CASE_ )
             UpperCamelCase					:						Any					=			'ada_norm'

          if self.is_input_continuous and self.is_input_vectorized:
             raise ValueError(
                 F"""Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make"""
                 ' sure that either `in_channels` or `num_vector_embeds` is None.' )
          elif self.is_input_vectorized and self.is_input_patches:
             raise ValueError(
                 F"""Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make"""
                 ' sure that either `num_vector_embeds` or `num_patches` is None.' )
          elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
             raise ValueError(
                 F"""Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:"""
                 F""" {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None.""" )

          # 2. Define input layers
          if self.is_input_continuous:
             UpperCamelCase					:						Optional[int]					=			in_channels

             UpperCamelCase					:						str					=			torch.nn.GroupNorm(num_groups=SCREAMING_SNAKE_CASE_,    num_channels=SCREAMING_SNAKE_CASE_,    eps=1e-6,    affine=SCREAMING_SNAKE_CASE_ )
             if use_linear_projection:
                UpperCamelCase					:						Dict					=			nn.Linear(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_ )
             else:
                UpperCamelCase					:						Any					=			nn.Convad(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    kernel_size=1,    stride=1,    padding=0 )
          elif self.is_input_vectorized:
             assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
             assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

             UpperCamelCase					:						Union[str, Any]					=			sample_size
             UpperCamelCase					:						int					=			sample_size
             UpperCamelCase					:						Optional[Any]					=			num_vector_embeds
             UpperCamelCase					:						str					=			self.height * self.width

             UpperCamelCase					:						Union[str, Any]					=			ImagePositionalEmbeddings(
                 num_embed=SCREAMING_SNAKE_CASE_,    embed_dim=SCREAMING_SNAKE_CASE_,    height=self.height,    width=self.width )
          elif self.is_input_patches:
             assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

             UpperCamelCase					:						int					=			sample_size
             UpperCamelCase					:						Optional[Any]					=			sample_size

             UpperCamelCase					:						List[str]					=			patch_size
             UpperCamelCase					:						Tuple					=			PatchEmbed(
                 height=SCREAMING_SNAKE_CASE_,    width=SCREAMING_SNAKE_CASE_,    patch_size=SCREAMING_SNAKE_CASE_,    in_channels=SCREAMING_SNAKE_CASE_,    embed_dim=SCREAMING_SNAKE_CASE_,    )

          # 3. Define transformers blocks
          UpperCamelCase					:						Any					=			nn.ModuleList(
              [
                  BasicTransformerBlock(
                      SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    dropout=SCREAMING_SNAKE_CASE_,    cross_attention_dim=SCREAMING_SNAKE_CASE_,    activation_fn=SCREAMING_SNAKE_CASE_,    num_embeds_ada_norm=SCREAMING_SNAKE_CASE_,    attention_bias=SCREAMING_SNAKE_CASE_,    only_cross_attention=SCREAMING_SNAKE_CASE_,    upcast_attention=SCREAMING_SNAKE_CASE_,    norm_type=SCREAMING_SNAKE_CASE_,    norm_elementwise_affine=SCREAMING_SNAKE_CASE_,    )
                  for d in range(SCREAMING_SNAKE_CASE_ )
              ] )

          # 4. Define output layers
          UpperCamelCase					:						Optional[int]					=			in_channels if out_channels is None else out_channels
          if self.is_input_continuous:
             # TODO: should use out_channels for continuous projections
             if use_linear_projection:
                UpperCamelCase					:						Optional[Any]					=			nn.Linear(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_ )
             else:
                UpperCamelCase					:						List[str]					=			nn.Convad(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    kernel_size=1,    stride=1,    padding=0 )
          elif self.is_input_vectorized:
             UpperCamelCase					:						Optional[Any]					=			nn.LayerNorm(SCREAMING_SNAKE_CASE_ )
             UpperCamelCase					:						Any					=			nn.Linear(SCREAMING_SNAKE_CASE_,    self.num_vector_embeds - 1 )
          elif self.is_input_patches:
             UpperCamelCase					:						List[str]					=			nn.LayerNorm(SCREAMING_SNAKE_CASE_,    elementwise_affine=SCREAMING_SNAKE_CASE_,    eps=1e-6 )
             UpperCamelCase					:						Tuple					=			nn.Linear(SCREAMING_SNAKE_CASE_,    2 * inner_dim )
             UpperCamelCase					:						Any					=			nn.Linear(SCREAMING_SNAKE_CASE_,    patch_size * patch_size * self.out_channels )






       def       snake_case_ (    self,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = None,    SCREAMING_SNAKE_CASE_ = True,    )  ->	Optional[Any]:
          # ensure attention_mask is a bias, and give it a singleton query_tokens dimension.
          #   we may have done this conversion already, e.g. if we came here via UNet2DConditionModel#forward.
          #   we can tell by counting dims; if ndim == 2: it's a mask rather than a bias.
          # expects mask of shape:
          #   [batch, key_tokens]
          # adds singleton query_tokens dimension:
          #   [batch,                    1, key_tokens]
          # this helps to broadcast it as a bias over attention scores, which will be in one of the following shapes:
          #   [batch,  heads, query_tokens, key_tokens] (e.g. torch sdp attn)
          #   [batch * heads, query_tokens, key_tokens] (e.g. xformers or classic attn)
          if attention_mask is not None and attention_mask.ndim == 2:
             # assume that mask is expressed as:
             #   (1 = keep,      0 = discard)
             # convert mask into a bias that can be added to attention scores:
             #       (keep = +0,     discard = -10000.0)
             UpperCamelCase					:						List[str]					=			(1 - attention_mask.to(hidden_states.dtype )) * -1_00_00.0
             UpperCamelCase					:						Tuple					=			attention_mask.unsqueeze(1 )

          # convert encoder_attention_mask to a bias the same way we do for attention_mask
          if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
             UpperCamelCase					:						int					=			(1 - encoder_attention_mask.to(hidden_states.dtype )) * -1_00_00.0
             UpperCamelCase					:						Dict					=			encoder_attention_mask.unsqueeze(1 )

          # 1. Input
          if self.is_input_continuous:
             UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase					:						Union[str, Any]					=			hidden_states.shape
             UpperCamelCase					:						str					=			hidden_states

             UpperCamelCase					:						Union[str, Any]					=			self.norm(SCREAMING_SNAKE_CASE_ )
             if not self.use_linear_projection:
                UpperCamelCase					:						Tuple					=			self.proj_in(SCREAMING_SNAKE_CASE_ )
                UpperCamelCase					:						Optional[int]					=			hidden_states.shape[1]
                UpperCamelCase					:						Tuple					=			hidden_states.permute(0,    2,    3,    1 ).reshape(SCREAMING_SNAKE_CASE_,    height * width,    SCREAMING_SNAKE_CASE_ )
             else:
                UpperCamelCase					:						Optional[int]					=			hidden_states.shape[1]
                UpperCamelCase					:						Dict					=			hidden_states.permute(0,    2,    3,    1 ).reshape(SCREAMING_SNAKE_CASE_,    height * width,    SCREAMING_SNAKE_CASE_ )
                UpperCamelCase					:						Union[str, Any]					=			self.proj_in(SCREAMING_SNAKE_CASE_ )
          elif self.is_input_vectorized:
             UpperCamelCase					:						Union[str, Any]					=			self.latent_image_embedding(SCREAMING_SNAKE_CASE_ )
          elif self.is_input_patches:
             UpperCamelCase					:						Any					=			self.pos_embed(SCREAMING_SNAKE_CASE_ )

          # 2. Blocks
          for block in self.transformer_blocks:
             UpperCamelCase					:						Optional[int]					=			block(
                 SCREAMING_SNAKE_CASE_,    attention_mask=SCREAMING_SNAKE_CASE_,    encoder_hidden_states=SCREAMING_SNAKE_CASE_,    encoder_attention_mask=SCREAMING_SNAKE_CASE_,    timestep=SCREAMING_SNAKE_CASE_,    cross_attention_kwargs=SCREAMING_SNAKE_CASE_,    class_labels=SCREAMING_SNAKE_CASE_,    )

          # 3. Output
          if self.is_input_continuous:
             if not self.use_linear_projection:
                UpperCamelCase					:						Tuple					=			hidden_states.reshape(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_ ).permute(0,    3,    1,    2 ).contiguous()
                UpperCamelCase					:						int					=			self.proj_out(SCREAMING_SNAKE_CASE_ )
             else:
                UpperCamelCase					:						Optional[int]					=			self.proj_out(SCREAMING_SNAKE_CASE_ )
                UpperCamelCase					:						Optional[int]					=			hidden_states.reshape(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_ ).permute(0,    3,    1,    2 ).contiguous()

             UpperCamelCase					:						Any					=			hidden_states + residual
          elif self.is_input_vectorized:
             UpperCamelCase					:						List[Any]					=			self.norm_out(SCREAMING_SNAKE_CASE_ )
             UpperCamelCase					:						Optional[int]					=			self.out(SCREAMING_SNAKE_CASE_ )
             # (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
             UpperCamelCase					:						Tuple					=			logits.permute(0,    2,    1 )

             # log(p(x_0))
             UpperCamelCase					:						List[str]					=			F.log_softmax(logits.double(),    dim=1 ).float()
          elif self.is_input_patches:
             # TODO: cleanup!
             UpperCamelCase					:						int					=			self.transformer_blocks[0].norma.emb(
                 SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    hidden_dtype=hidden_states.dtype )
             UpperCamelCase			,  UpperCamelCase					:						int					=			self.proj_out_a(F.silu(SCREAMING_SNAKE_CASE_ ) ).chunk(2,    dim=1 )
             UpperCamelCase					:						Dict					=			self.norm_out(SCREAMING_SNAKE_CASE_ ) * (1 + scale[:, None]) + shift[:, None]
             UpperCamelCase					:						Any					=			self.proj_out_a(SCREAMING_SNAKE_CASE_ )

             # unpatchify
             UpperCamelCase					:						Union[str, Any]					=			int(hidden_states.shape[1] ** 0.5 )
             UpperCamelCase					:						Tuple					=			hidden_states.reshape(
                 shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels) )
             UpperCamelCase					:						int					=			torch.einsum('nhwpqc->nchpwq',    SCREAMING_SNAKE_CASE_ )
             UpperCamelCase					:						Any					=			hidden_states.reshape(
                 shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size) )

          if not return_dict:
             return (output,)

          return TransformeraDModelOutput(sample=SCREAMING_SNAKE_CASE_ )



