

"""simple docstring"""

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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block






@dataclass
class  __a				(	__magic_name__					):





       """simple docstring"""


       __UpperCamelCase				:			torch.FloatTensor






class  __a				(	__magic_name__					,  __magic_name__					):





       """simple docstring"""


       @register_to_config
       def __init__(      self ,				snake_case = None ,				snake_case = 3 ,				snake_case = 3 ,				snake_case = False ,				snake_case = "positional" ,				snake_case = 0 ,				snake_case = True ,				snake_case = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D") ,				snake_case = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D") ,				snake_case = (224, 448, 672, 896) ,				snake_case = 2 ,				snake_case = 1 ,				snake_case = 1 ,				snake_case = "conv" ,				snake_case = "conv" ,				snake_case = "silu" ,				snake_case = 8 ,				snake_case = 32 ,				snake_case = 1e-5 ,				snake_case = "default" ,				snake_case = True ,				snake_case = None ,				snake_case = None ,				):





        """simple docstring"""

        super().__init__()

        lowerCAmelCase__      :							List[str]    =							sample_size
        lowerCAmelCase__      :							Union[str, Any]    =							block_out_channels[0] * 4

        # Check inputs
        if len(snake_case      ) != len(snake_case      ):
         raise ValueError(
             F"""Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}."""      )

        if len(snake_case      ) != len(snake_case      ):
         raise ValueError(
             F"""Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}."""      )

        # input
        lowerCAmelCase__      :							str    =							nn.Convad(snake_case ,				block_out_channels[0] ,				kernel_size=3 ,				padding=(1, 1)      )

        # time
        if time_embedding_type == "fourier":
         lowerCAmelCase__      :							Optional[Any]    =							GaussianFourierProjection(embedding_size=block_out_channels[0] ,				scale=16      )
         lowerCAmelCase__      :							Optional[int]    =							2 * block_out_channels[0]
        elif time_embedding_type == "positional":
         lowerCAmelCase__      :							List[Any]    =							Timesteps(block_out_channels[0] ,				snake_case ,				snake_case      )
         lowerCAmelCase__      :							Any    =							block_out_channels[0]

        lowerCAmelCase__      :							Union[str, Any]    =							TimestepEmbedding(snake_case ,				snake_case      )

        # class embedding
        if class_embed_type is None and num_class_embeds is not None:
         lowerCAmelCase__      :							int    =							nn.Embedding(snake_case ,				snake_case      )
        elif class_embed_type == "timestep":
         lowerCAmelCase__      :							Dict    =							TimestepEmbedding(snake_case ,				snake_case      )
        elif class_embed_type == "identity":
         lowerCAmelCase__      :							Tuple    =							nn.Identity(snake_case ,				snake_case      )
        else:
         lowerCAmelCase__      :							Union[str, Any]    =							None

        lowerCAmelCase__      :							int    =							nn.ModuleList([]      )
        lowerCAmelCase__      :							List[str]    =							None
        lowerCAmelCase__      :							Union[str, Any]    =							nn.ModuleList([]      )

        # down
        lowerCAmelCase__      :							Optional[Any]    =							block_out_channels[0]
        for i, down_block_type in enumerate(snake_case      ):
         lowerCAmelCase__      :							Optional[Any]    =							output_channel
         lowerCAmelCase__      :							List[Any]    =							block_out_channels[i]
         lowerCAmelCase__      :							Tuple    =							i == len(snake_case      ) - 1

         lowerCAmelCase__      :							Optional[int]    =							get_down_block(
             snake_case ,				num_layers=snake_case ,				in_channels=snake_case ,				out_channels=snake_case ,				temb_channels=snake_case ,				add_downsample=not is_final_block ,				resnet_eps=snake_case ,				resnet_act_fn=snake_case ,				resnet_groups=snake_case ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel ,				downsample_padding=snake_case ,				resnet_time_scale_shift=snake_case ,				downsample_type=snake_case ,				)
         self.down_blocks.append(snake_case      )

        # mid
        lowerCAmelCase__      :							Tuple    =							UNetMidBlockaD(
            in_channels=block_out_channels[-1] ,				temb_channels=snake_case ,				resnet_eps=snake_case ,				resnet_act_fn=snake_case ,				output_scale_factor=snake_case ,				resnet_time_scale_shift=snake_case ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1] ,				resnet_groups=snake_case ,				add_attention=snake_case ,				)

        # up
        lowerCAmelCase__      :							str    =							list(reversed(snake_case      )      )
        lowerCAmelCase__      :							Optional[Any]    =							reversed_block_out_channels[0]
        for i, up_block_type in enumerate(snake_case      ):
         lowerCAmelCase__      :							List[str]    =							output_channel
         lowerCAmelCase__      :							str    =							reversed_block_out_channels[i]
         lowerCAmelCase__      :							List[Any]    =							reversed_block_out_channels[min(i + 1 ,				len(snake_case      ) - 1      )]

         lowerCAmelCase__      :							Union[str, Any]    =							i == len(snake_case      ) - 1

         lowerCAmelCase__      :							List[str]    =							get_up_block(
             snake_case ,				num_layers=layers_per_block + 1 ,				in_channels=snake_case ,				out_channels=snake_case ,				prev_output_channel=snake_case ,				temb_channels=snake_case ,				add_upsample=not is_final_block ,				resnet_eps=snake_case ,				resnet_act_fn=snake_case ,				resnet_groups=snake_case ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel ,				resnet_time_scale_shift=snake_case ,				upsample_type=snake_case ,				)
         self.up_blocks.append(snake_case      )
         lowerCAmelCase__      :							Tuple    =							output_channel

        # out
        lowerCAmelCase__      :							Dict    =							norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4 ,				32      )
        lowerCAmelCase__      :							List[Any]    =							nn.GroupNorm(num_channels=block_out_channels[0] ,				num_groups=snake_case ,				eps=snake_case      )
        lowerCAmelCase__      :							Optional[Any]    =							nn.SiLU()
        lowerCAmelCase__      :							Optional[Any]    =							nn.Convad(block_out_channels[0] ,				snake_case ,				kernel_size=3 ,				padding=1      )






       def  SCREAMING_SNAKE_CASE_  (      self ,				snake_case ,				snake_case ,				snake_case = None ,				snake_case = True ,				):





        """simple docstring"""

        if self.config.center_input_sample:
         lowerCAmelCase__      :							Optional[Any]    =							2 * sample - 1.0

        # 1. time
        lowerCAmelCase__      :							Any    =							timestep
        if not torch.is_tensor(snake_case      ):
         lowerCAmelCase__      :							Optional[int]    =							torch.tensor([timesteps] ,				dtype=torch.long ,				device=sample.device      )
        elif torch.is_tensor(snake_case      ) and len(timesteps.shape      ) == 0:
         lowerCAmelCase__      :							Tuple    =							timesteps[None].to(sample.device      )

        # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
        lowerCAmelCase__      :							Optional[Any]    =							timesteps * torch.ones(sample.shape[0] ,				dtype=timesteps.dtype ,				device=timesteps.device      )

        lowerCAmelCase__      :							Optional[Any]    =							self.time_proj(snake_case      )

        # timesteps does not contain any weights and will always return f32 tensors
        # but time_embedding might actually be running in fp16. so we need to cast here.
        # there might be better ways to encapsulate this.
        lowerCAmelCase__      :							Union[str, Any]    =							t_emb.to(dtype=self.dtype      )
        lowerCAmelCase__      :							Dict    =							self.time_embedding(snake_case      )

        if self.class_embedding is not None:
         if class_labels is None:
          raise ValueError("class_labels should be provided when doing class conditioning"      )

         if self.config.class_embed_type == "timestep":
          lowerCAmelCase__      :							Union[str, Any]    =							self.time_proj(snake_case      )

         lowerCAmelCase__      :							Any    =							self.class_embedding(snake_case      ).to(dtype=self.dtype      )
         lowerCAmelCase__      :							str    =							emb + class_emb

        # 2. pre-process
        lowerCAmelCase__      :							List[Any]    =							sample
        lowerCAmelCase__      :							List[str]    =							self.conv_in(snake_case      )

        # 3. down
        lowerCAmelCase__      :							str    =							(sample,)
        for downsample_block in self.down_blocks:
         if hasattr(snake_case ,				"skip_conv"      ):
          lowerCAmelCase__ ,	lowerCAmelCase__ ,	lowerCAmelCase__      :							List[str]    =							downsample_block(
              hidden_states=snake_case ,				temb=snake_case ,				skip_sample=snake_case      )
         else:
          lowerCAmelCase__ ,	lowerCAmelCase__      :							Dict    =							downsample_block(hidden_states=snake_case ,				temb=snake_case      )

         down_block_res_samples += res_samples

        # 4. mid
        lowerCAmelCase__      :							Any    =							self.mid_block(snake_case ,				snake_case      )

        # 5. up
        lowerCAmelCase__      :							Optional[Any]    =							None
        for upsample_block in self.up_blocks:
         lowerCAmelCase__      :							Any    =							down_block_res_samples[-len(upsample_block.resnets      ) :]
         lowerCAmelCase__      :							Optional[int]    =							down_block_res_samples[: -len(upsample_block.resnets      )]

         if hasattr(snake_case ,				"skip_conv"      ):
          lowerCAmelCase__ ,	lowerCAmelCase__      :							Optional[Any]    =							upsample_block(snake_case ,				snake_case ,				snake_case ,				snake_case      )
         else:
          lowerCAmelCase__      :							List[str]    =							upsample_block(snake_case ,				snake_case ,				snake_case      )

        # 6. post-process
        lowerCAmelCase__      :							int    =							self.conv_norm_out(snake_case      )
        lowerCAmelCase__      :							Optional[int]    =							self.conv_act(snake_case      )
        lowerCAmelCase__      :							int    =							self.conv_out(snake_case      )

        if skip_sample is not None:
         sample += skip_sample

        if self.config.time_embedding_type == "fourier":
         lowerCAmelCase__      :							Optional[Any]    =							timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]      )))      )
         lowerCAmelCase__      :							Union[str, Any]    =							sample / timesteps

        if not return_dict:
         return (sample,)

        return UNetaDOutput(sample=snake_case      )









"""simple docstring"""

import torch
import torch.nn.functional as F

from diffusers import VQDiffusionScheduler

from .test_schedulers import SchedulerCommonTest






class  __a				(	__magic_name__					):





       """simple docstring"""


       __UpperCamelCase				:			Optional[int]          =	(VQDiffusionScheduler,)






       def  SCREAMING_SNAKE_CASE_  (      self ,				**snake_case      ):





        """simple docstring"""

        lowerCAmelCase__      :							str    =							{
            "num_vec_classes": 4_097,
            "num_train_timesteps": 100,
        }

        config.update(**snake_case      )
        return config






       def  SCREAMING_SNAKE_CASE_  (      self ,				snake_case      ):





        """simple docstring"""

        lowerCAmelCase__      :							Optional[int]    =							4
        lowerCAmelCase__      :							Optional[Any]    =							8
        lowerCAmelCase__      :							Union[str, Any]    =							8

        lowerCAmelCase__      :							Optional[Any]    =							torch.randint(0 ,				snake_case ,				(batch_size, height * width)      )

        return sample






       @property
       def  SCREAMING_SNAKE_CASE_  (      self      ):





        """simple docstring"""

        assert False






       def  SCREAMING_SNAKE_CASE_  (      self ,				snake_case      ):





        """simple docstring"""

        def model(snake_case ,				snake_case ,				*snake_case      ):
         lowerCAmelCase__ ,	lowerCAmelCase__      :							str    =							sample.shape
         lowerCAmelCase__      :							Dict    =							torch.rand((batch_size, num_vec_classes - 1, num_latent_pixels)      )
         lowerCAmelCase__      :							int    =							F.log_softmax(logits.double() ,				dim=1      ).float()
         return return_value

        return model






       def  SCREAMING_SNAKE_CASE_  (      self      ):





        """simple docstring"""

        for timesteps in [2, 5, 100, 1_000]:
         self.check_over_configs(num_train_timesteps=snake_case      )






       def  SCREAMING_SNAKE_CASE_  (      self      ):





        """simple docstring"""

        for num_vec_classes in [5, 100, 1_000, 4_000]:
         self.check_over_configs(num_vec_classes=snake_case      )






       def  SCREAMING_SNAKE_CASE_  (      self      ):





        """simple docstring"""

        for t in [0, 50, 99]:
         self.check_over_forward(time_step=snake_case      )






       def  SCREAMING_SNAKE_CASE_  (      self      ):





        """simple docstring"""

        pass






