





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






  """simple docstring"""





  _SCREAMING_SNAKE_CASE		    =       42






class     __lowerCAmelCase     ( a ,  a					):






  """simple docstring"""





  @register_to_config
  def __init__(     self       :       Tuple    ,				_lowerCAmelCase       :       int = None    ,				_lowerCAmelCase       :       Optional[int] = 3    ,				_lowerCAmelCase       :       Any = 3    ,				_lowerCAmelCase       :       Dict = False    ,				_lowerCAmelCase       :       List[Any] = "positional"    ,				_lowerCAmelCase       :       List[str] = 0    ,				_lowerCAmelCase       :       Union[str, Any] = True    ,				_lowerCAmelCase       :       Tuple = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")    ,				_lowerCAmelCase       :       Union[str, Any] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")    ,				_lowerCAmelCase       :       List[Any] = (2_2_4, 4_4_8, 6_7_2, 8_9_6)    ,				_lowerCAmelCase       :       Dict = 2    ,				_lowerCAmelCase       :       str = 1    ,				_lowerCAmelCase       :       List[Any] = 1    ,				_lowerCAmelCase       :       str = "conv"    ,				_lowerCAmelCase       :       Dict = "conv"    ,				_lowerCAmelCase       :       List[Any] = "silu"    ,				_lowerCAmelCase       :       Union[str, Any] = 8    ,				_lowerCAmelCase       :       Optional[int] = 3_2    ,				_lowerCAmelCase       :       Optional[Any] = 1e-5    ,				_lowerCAmelCase       :       Optional[int] = "default"    ,				_lowerCAmelCase       :       int = True    ,				_lowerCAmelCase       :       str = None    ,				_lowerCAmelCase       :       int = None    ,				)   -> Any:




     """simple docstring"""
     super().__init__()

     snake_case_					    =					sample_size
     snake_case_					    =					block_out_channels[0] * 4

     # Check inputs
     if len(_lowerCAmelCase  ) != len(_lowerCAmelCase  ):
        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(_lowerCAmelCase  ) != len(_lowerCAmelCase  ):
        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
     snake_case_					    =					nn.Convad(_lowerCAmelCase    ,				block_out_channels[0]    ,				kernel_size=3    ,				padding=(1, 1)  )

     # time
     if time_embedding_type == "fourier":
        snake_case_					    =					GaussianFourierProjection(embedding_size=block_out_channels[0]    ,				scale=1_6  )
        snake_case_					    =					2 * block_out_channels[0]
     elif time_embedding_type == "positional":
        snake_case_					    =					Timesteps(block_out_channels[0]    ,				_lowerCAmelCase    ,				_lowerCAmelCase  )
        snake_case_					    =					block_out_channels[0]

     snake_case_					    =					TimestepEmbedding(_lowerCAmelCase    ,				_lowerCAmelCase  )

     # class embedding
     if class_embed_type is None and num_class_embeds is not None:
        snake_case_					    =					nn.Embedding(_lowerCAmelCase    ,				_lowerCAmelCase  )
     elif class_embed_type == "timestep":
        snake_case_					    =					TimestepEmbedding(_lowerCAmelCase    ,				_lowerCAmelCase  )
     elif class_embed_type == "identity":
        snake_case_					    =					nn.Identity(_lowerCAmelCase    ,				_lowerCAmelCase  )
     else:
        snake_case_					    =					None

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

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

        snake_case_					    =					get_down_block(
            _lowerCAmelCase    ,				num_layers=_lowerCAmelCase    ,				in_channels=_lowerCAmelCase    ,				out_channels=_lowerCAmelCase    ,				temb_channels=_lowerCAmelCase    ,				add_downsample=not is_final_block    ,				resnet_eps=_lowerCAmelCase    ,				resnet_act_fn=_lowerCAmelCase    ,				resnet_groups=_lowerCAmelCase    ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel    ,				downsample_padding=_lowerCAmelCase    ,				resnet_time_scale_shift=_lowerCAmelCase    ,				downsample_type=_lowerCAmelCase    ,				)
        self.down_blocks.append(_lowerCAmelCase  )

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

     # up
     snake_case_					    =					list(reversed(_lowerCAmelCase  )  )
     snake_case_					    =					reversed_block_out_channels[0]
     for i, up_block_type in enumerate(_lowerCAmelCase  ):
        snake_case_					    =					output_channel
        snake_case_					    =					reversed_block_out_channels[i]
        snake_case_					    =					reversed_block_out_channels[min(i + 1    ,				len(_lowerCAmelCase  ) - 1  )]

        snake_case_					    =					i == len(_lowerCAmelCase  ) - 1

        snake_case_					    =					get_up_block(
            _lowerCAmelCase    ,				num_layers=layers_per_block + 1    ,				in_channels=_lowerCAmelCase    ,				out_channels=_lowerCAmelCase    ,				prev_output_channel=_lowerCAmelCase    ,				temb_channels=_lowerCAmelCase    ,				add_upsample=not is_final_block    ,				resnet_eps=_lowerCAmelCase    ,				resnet_act_fn=_lowerCAmelCase    ,				resnet_groups=_lowerCAmelCase    ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel    ,				resnet_time_scale_shift=_lowerCAmelCase    ,				upsample_type=_lowerCAmelCase    ,				)
        self.up_blocks.append(_lowerCAmelCase  )
        snake_case_					    =					output_channel

     # out
     snake_case_					    =					norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4    ,				3_2  )
     snake_case_					    =					nn.GroupNorm(num_channels=block_out_channels[0]    ,				num_groups=_lowerCAmelCase    ,				eps=_lowerCAmelCase  )
     snake_case_					    =					nn.SiLU()
     snake_case_					    =					nn.Convad(block_out_channels[0]    ,				_lowerCAmelCase    ,				kernel_size=3    ,				padding=1  )



  def 					lowerCAmelCase__  (     self       :       Tuple    ,				_lowerCAmelCase       :       Tuple    ,				_lowerCAmelCase       :       Optional[int]    ,				_lowerCAmelCase       :       Any = None    ,				_lowerCAmelCase       :       List[Any] = True    ,				)   -> Union[UNetaDOutput, Tuple]:




     """simple docstring"""






     if self.config.center_input_sample:
        snake_case_					    =					2 * sample - 1.0

     # 1. time
     snake_case_					    =					timestep
     if not torch.is_tensor(_lowerCAmelCase  ):
        snake_case_					    =					torch.tensor([timesteps]    ,				dtype=torch.long    ,				device=sample.device  )
     elif torch.is_tensor(_lowerCAmelCase  ) and len(timesteps.shape  ) == 0:
        snake_case_					    =					timesteps[None].to(sample.device  )

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

     snake_case_					    =					self.time_proj(_lowerCAmelCase  )

     # 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.
     snake_case_					    =					t_emb.to(dtype=self.dtype  )
     snake_case_					    =					self.time_embedding(_lowerCAmelCase  )

     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":
           snake_case_					    =					self.time_proj(_lowerCAmelCase  )

        snake_case_					    =					self.class_embedding(_lowerCAmelCase  ).to(dtype=self.dtype  )
        snake_case_					    =					emb + class_emb

     # 2. pre-process
     snake_case_					    =					sample
     snake_case_					    =					self.conv_in(_lowerCAmelCase  )

     # 3. down
     snake_case_					    =					(sample,)
     for downsample_block in self.down_blocks:
        if hasattr(_lowerCAmelCase    ,				"skip_conv"  ):
           snake_case_					    =					downsample_block(
               hidden_states=_lowerCAmelCase    ,				temb=_lowerCAmelCase    ,				skip_sample=_lowerCAmelCase  )
        else:
           snake_case_					    =					downsample_block(hidden_states=_lowerCAmelCase    ,				temb=_lowerCAmelCase  )

        down_block_res_samples += res_samples

     # 4. mid
     snake_case_					    =					self.mid_block(_lowerCAmelCase    ,				_lowerCAmelCase  )

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

        if hasattr(_lowerCAmelCase    ,				"skip_conv"  ):
           snake_case_					    =					upsample_block(_lowerCAmelCase    ,				_lowerCAmelCase    ,				_lowerCAmelCase    ,				_lowerCAmelCase  )
        else:
           snake_case_					    =					upsample_block(_lowerCAmelCase    ,				_lowerCAmelCase    ,				_lowerCAmelCase  )

        # 6. post-process
     snake_case_					    =					self.conv_norm_out(_lowerCAmelCase  )
     snake_case_					    =					self.conv_act(_lowerCAmelCase  )
     snake_case_					    =					self.conv_out(_lowerCAmelCase  )

     if skip_sample is not None:
        sample += skip_sample

     if self.config.time_embedding_type == "fourier":
        snake_case_					    =					timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]  )))  )
        snake_case_					    =					sample / timesteps

     if not return_dict:
        return (sample,)

     return UNetaDOutput(sample=_lowerCAmelCase  )








from typing import List, Optional, Tuple, Union

import torch

from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


class     __lowerCAmelCase     ( a					):






  """simple docstring"""





  def __init__(     self       :       List[str]    ,				_lowerCAmelCase       :       Optional[Any]    ,				_lowerCAmelCase       :       Tuple  )   -> str:




     """simple docstring"""
     super().__init__()
     self.register_modules(unet=_lowerCAmelCase    ,				scheduler=_lowerCAmelCase  )



  @torch.no_grad()
  def __call__(     self       :       Tuple    ,				_lowerCAmelCase       :       int = 1    ,				_lowerCAmelCase       :       Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,				_lowerCAmelCase       :       int = 1_0_0_0    ,				_lowerCAmelCase       :       Optional[str] = "pil"    ,				_lowerCAmelCase       :       bool = True    ,				)   -> Union[ImagePipelineOutput, Tuple]:




     """simple docstring"""
     # Sample gaussian noise to begin loop
     if isinstance(self.unet.config.sample_size    ,				_lowerCAmelCase  ):
        snake_case_					    =					(
            batch_size,
            self.unet.config.in_channels,
            self.unet.config.sample_size,
            self.unet.config.sample_size,
        )
     else:
        snake_case_					    =					(batch_size, self.unet.config.in_channels, *self.unet.config.sample_size)

     if self.device.type == "mps":
        # randn does not work reproducibly on mps
        snake_case_					    =					randn_tensor(_lowerCAmelCase    ,				generator=_lowerCAmelCase  )
        snake_case_					    =					image.to(self.device  )
     else:
        snake_case_					    =					randn_tensor(_lowerCAmelCase    ,				generator=_lowerCAmelCase    ,				device=self.device  )

     # set step values
     self.scheduler.set_timesteps(_lowerCAmelCase  )

     for t in self.progress_bar(self.scheduler.timesteps  ):
        # 1. predict noise model_output
        snake_case_					    =					self.unet(_lowerCAmelCase    ,				_lowerCAmelCase  ).sample

        # 2. compute previous image: x_t -> x_t-1
        snake_case_					    =					self.scheduler.step(_lowerCAmelCase    ,				_lowerCAmelCase    ,				_lowerCAmelCase    ,				generator=_lowerCAmelCase  ).prev_sample

     snake_case_					    =					(image / 2 + 0.5).clamp(0    ,				1  )
     snake_case_					    =					image.cpu().permute(0    ,				2    ,				3    ,				1  ).numpy()
     if output_type == "pil":
        snake_case_					    =					self.numpy_to_pil(_lowerCAmelCase  )

     if not return_dict:
        return (image,)

     return ImagePipelineOutput(images=_lowerCAmelCase  )




