




'''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 							_snake_case	(    __lowerCamelCase  ):
   _A    :       torch.FloatTensor





class 							_snake_case	(    __lowerCamelCase		,					__lowerCamelCase  ):
   @register_to_config
   def __init__(	self     : Optional[Any]       ,SCREAMING_SNAKE_CASE__     : Optional[Union[int, Tuple[int, int]]] = None       ,SCREAMING_SNAKE_CASE__     : int = 3       ,SCREAMING_SNAKE_CASE__     : int = 3       ,SCREAMING_SNAKE_CASE__     : bool = False       ,SCREAMING_SNAKE_CASE__     : str = "positional"       ,SCREAMING_SNAKE_CASE__     : int = 0       ,SCREAMING_SNAKE_CASE__     : bool = True       ,SCREAMING_SNAKE_CASE__     : Tuple[str] = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")       ,SCREAMING_SNAKE_CASE__     : Tuple[str] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")       ,SCREAMING_SNAKE_CASE__     : Tuple[int] = (224, 448, 672, 896)       ,SCREAMING_SNAKE_CASE__     : int = 2       ,SCREAMING_SNAKE_CASE__     : float = 1       ,SCREAMING_SNAKE_CASE__     : int = 1       ,SCREAMING_SNAKE_CASE__     : str = "conv"       ,SCREAMING_SNAKE_CASE__     : str = "conv"       ,SCREAMING_SNAKE_CASE__     : str = "silu"       ,SCREAMING_SNAKE_CASE__     : Optional[int] = 8       ,SCREAMING_SNAKE_CASE__     : int = 32       ,SCREAMING_SNAKE_CASE__     : float = 1e-5       ,SCREAMING_SNAKE_CASE__     : str = "default"       ,SCREAMING_SNAKE_CASE__     : bool = True       ,SCREAMING_SNAKE_CASE__     : Optional[str] = None       ,SCREAMING_SNAKE_CASE__     : Optional[int] = None       ,):
     super().__init__()

     SCREAMING_SNAKE_CASE:Any   =   sample_size
     SCREAMING_SNAKE_CASE:str   =   block_out_channels[0] * 4

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

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

     SCREAMING_SNAKE_CASE:int   =   TimestepEmbedding(_UpperCamelCase       ,_UpperCamelCase							)

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

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

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

       SCREAMING_SNAKE_CASE:List[Any]   =   get_down_block(
           _UpperCamelCase       ,num_layers=_UpperCamelCase       ,in_channels=_UpperCamelCase       ,out_channels=_UpperCamelCase       ,temb_channels=_UpperCamelCase       ,add_downsample=not is_final_block       ,resnet_eps=_UpperCamelCase       ,resnet_act_fn=_UpperCamelCase       ,resnet_groups=_UpperCamelCase       ,attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       ,downsample_padding=_UpperCamelCase       ,resnet_time_scale_shift=_UpperCamelCase       ,downsample_type=_UpperCamelCase       ,)
       self.down_blocks.append(_UpperCamelCase							)

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

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

       SCREAMING_SNAKE_CASE:Optional[Any]   =   i == len(_UpperCamelCase							) - 1

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

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







   def       __UpperCamelCase  (	self     : List[Any]       ,SCREAMING_SNAKE_CASE__     : torch.FloatTensor       ,SCREAMING_SNAKE_CASE__     : Union[torch.Tensor, float, int]       ,SCREAMING_SNAKE_CASE__     : Optional[torch.Tensor] = None       ,SCREAMING_SNAKE_CASE__     : bool = True       ,):





     if self.config.center_input_sample:
       SCREAMING_SNAKE_CASE:List[str]   =   2 * sample - 1.0

     # 1. time
     SCREAMING_SNAKE_CASE:List[Any]   =   timestep
     if not torch.is_tensor(_UpperCamelCase							):
       SCREAMING_SNAKE_CASE:Dict   =   torch.tensor([timesteps]       ,dtype=torch.long       ,device=sample.device							)
     elif torch.is_tensor(_UpperCamelCase							) and len(timesteps.shape							) == 0:
       SCREAMING_SNAKE_CASE:str   =   timesteps[None].to(sample.device							)

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

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   self.time_proj(_UpperCamelCase							)

     # 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.
     SCREAMING_SNAKE_CASE:Dict   =   t_emb.to(dtype=self.dtype							)
     SCREAMING_SNAKE_CASE:Optional[Any]   =   self.time_embedding(_UpperCamelCase							)

     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":
         SCREAMING_SNAKE_CASE:Dict   =   self.time_proj(_UpperCamelCase							)

       SCREAMING_SNAKE_CASE:Tuple   =   self.class_embedding(_UpperCamelCase							).to(dtype=self.dtype							)
       SCREAMING_SNAKE_CASE:Union[str, Any]   =   emb + class_emb

     # 2. pre-process
     SCREAMING_SNAKE_CASE:Dict   =   sample
     SCREAMING_SNAKE_CASE:Any   =   self.conv_in(_UpperCamelCase							)

     # 3. down
     SCREAMING_SNAKE_CASE:str   =   (sample,)
     for downsample_block in self.down_blocks:
       if hasattr(_UpperCamelCase       ,"skip_conv"							):
         SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE:Any   =   downsample_block(
             hidden_states=_UpperCamelCase       ,temb=_UpperCamelCase       ,skip_sample=_UpperCamelCase							)
       else:
         SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE:int   =   downsample_block(hidden_states=_UpperCamelCase       ,temb=_UpperCamelCase							)

       down_block_res_samples += res_samples

     # 4. mid
     SCREAMING_SNAKE_CASE:str   =   self.mid_block(_UpperCamelCase       ,_UpperCamelCase							)

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

       if hasattr(_UpperCamelCase       ,"skip_conv"							):
         SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE:Optional[int]   =   upsample_block(_UpperCamelCase       ,_UpperCamelCase       ,_UpperCamelCase       ,_UpperCamelCase							)
       else:
         SCREAMING_SNAKE_CASE:str   =   upsample_block(_UpperCamelCase       ,_UpperCamelCase       ,_UpperCamelCase							)

        # 6. post-process
     SCREAMING_SNAKE_CASE:int   =   self.conv_norm_out(_UpperCamelCase							)
     SCREAMING_SNAKE_CASE:Any   =   self.conv_act(_UpperCamelCase							)
     SCREAMING_SNAKE_CASE:Dict   =   self.conv_out(_UpperCamelCase							)

     if skip_sample is not None:
       sample += skip_sample

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

     if not return_dict:
       return (sample,)

     return UNetaDOutput(sample=_UpperCamelCase							)









'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


A_      =							{"configuration_yoso": ["YOSO_PRETRAINED_CONFIG_ARCHIVE_MAP", "YosoConfig"]}

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    A_      =							[
        "YOSO_PRETRAINED_MODEL_ARCHIVE_LIST",
        "YosoForMaskedLM",
        "YosoForMultipleChoice",
        "YosoForQuestionAnswering",
        "YosoForSequenceClassification",
        "YosoForTokenClassification",
        "YosoLayer",
        "YosoModel",
        "YosoPreTrainedModel",
    ]


if TYPE_CHECKING:
    from .configuration_yoso import YOSO_PRETRAINED_CONFIG_ARCHIVE_MAP, YosoConfig

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_yoso import (
            YOSO_PRETRAINED_MODEL_ARCHIVE_LIST,
            YosoForMaskedLM,
            YosoForMultipleChoice,
            YosoForQuestionAnswering,
            YosoForSequenceClassification,
            YosoForTokenClassification,
            YosoLayer,
            YosoModel,
            YosoPreTrainedModel,
        )


else:
    import sys

    A_      =							_LazyModule(__name__, globals()["__file__"], _import_structure)






