




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 		UpperCamelCase__     (lowerCAmelCase__					):

    '''simple docstring'''



    lowerCamelCase_				:     torch.FloatTensor




class 		UpperCamelCase__     (lowerCAmelCase__   ,    lowerCAmelCase__					):

    '''simple docstring'''



    @register_to_config
    def __init__(   self					,  UpperCamelCase__ = None					,  UpperCamelCase__ = 3					,  UpperCamelCase__ = 3					,  UpperCamelCase__ = False					,  UpperCamelCase__ = "positional"					,  UpperCamelCase__ = 0					,  UpperCamelCase__ = True					,  UpperCamelCase__ = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")					,  UpperCamelCase__ = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")					,  UpperCamelCase__ = (224, 448, 672, 896)					,  UpperCamelCase__ = 2					,  UpperCamelCase__ = 1					,  UpperCamelCase__ = 1					,  UpperCamelCase__ = "conv"					,  UpperCamelCase__ = "conv"					,  UpperCamelCase__ = "silu"					,  UpperCamelCase__ = 8					,  UpperCamelCase__ = 32					,  UpperCamelCase__ = 1e-5					,  UpperCamelCase__ = "default"					,  UpperCamelCase__ = True					,  UpperCamelCase__ = None					,  UpperCamelCase__ = None					,  )					->				List[Any]:
           super().__init__()

           lowerCamelCase       :				Optional[Any]      			= sample_size
           lowerCamelCase       :				List[Any]      			= 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
           lowerCamelCase       :				int      			= nn.Convad(UpperCamelCase__					,  block_out_channels[0]					,  kernel_size=3					,  padding=(1, 1)					)

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

           lowerCamelCase       :				Optional[int]      			= TimestepEmbedding(UpperCamelCase__					,  UpperCamelCase__					)

           # class embedding
           if class_embed_type is None and num_class_embeds is not None:
                  lowerCamelCase       :				List[str]      			= nn.Embedding(UpperCamelCase__					,  UpperCamelCase__					)
           elif class_embed_type == "timestep":
                  lowerCamelCase       :				Dict      			= TimestepEmbedding(UpperCamelCase__					,  UpperCamelCase__					)
           elif class_embed_type == "identity":
                  lowerCamelCase       :				str      			= nn.Identity(UpperCamelCase__					,  UpperCamelCase__					)
           else:
                  lowerCamelCase       :				Dict      			= None

           lowerCamelCase       :				Optional[Any]      			= nn.ModuleList([]					)
           lowerCamelCase       :				List[str]      			= None
           lowerCamelCase       :				int      			= nn.ModuleList([]					)

           # down
           lowerCamelCase       :				List[Any]      			= block_out_channels[0]
           for i, down_block_type in enumerate(UpperCamelCase__					):
                  lowerCamelCase       :				Union[str, Any]      			= output_channel
                  lowerCamelCase       :				int      			= block_out_channels[i]
                  lowerCamelCase       :				str      			= i == len(UpperCamelCase__					) - 1

                  lowerCamelCase       :				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
           lowerCamelCase       :				Optional[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
           lowerCamelCase       :				Dict      			= list(reversed(UpperCamelCase__					)					)
           lowerCamelCase       :				List[str]      			= reversed_block_out_channels[0]
           for i, up_block_type in enumerate(UpperCamelCase__					):
                  lowerCamelCase       :				Any      			= output_channel
                  lowerCamelCase       :				Optional[Any]      			= reversed_block_out_channels[i]
                  lowerCamelCase       :				Optional[int]      			= reversed_block_out_channels[min(i + 1					,  len(UpperCamelCase__					) - 1					)]

                  lowerCamelCase       :				Optional[Any]      			= i == len(UpperCamelCase__					) - 1

                  lowerCamelCase       :				Tuple      			= 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__					)
                  lowerCamelCase       :				Optional[Any]      			= output_channel

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







    def    _lowercase					(   self					,  UpperCamelCase__					,  UpperCamelCase__					,  UpperCamelCase__ = None					,  UpperCamelCase__ = True					,  )					->				Union[UNetaDOutput, Tuple]:
           # 0. center input if necessary
           if self.config.center_input_sample:
                  lowerCamelCase       :				Dict      			= 2 * sample - 1.0

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

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

           lowerCamelCase       :				Dict      			= 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.
           lowerCamelCase       :				Optional[int]      			= t_emb.to(dtype=self.dtype					)
           lowerCamelCase       :				Dict      			= 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":
                         lowerCamelCase       :				List[Any]      			= self.time_proj(UpperCamelCase__					)

                  lowerCamelCase       :				Tuple      			= self.class_embedding(UpperCamelCase__					).to(dtype=self.dtype					)
                  lowerCamelCase       :				Tuple      			= emb + class_emb

           # 2. pre-process
           lowerCamelCase       :				str      			= sample
           lowerCamelCase       :				Union[str, Any]      			= self.conv_in(UpperCamelCase__					)

           # 3. down
           lowerCamelCase       :				int      			= (sample,)
           for downsample_block in self.down_blocks:
                  if hasattr(UpperCamelCase__					,  "skip_conv"					):
                         lowerCamelCase       :				Optional[int]      			= downsample_block(
                             hidden_states=UpperCamelCase__					,  temb=UpperCamelCase__					,  skip_sample=UpperCamelCase__					)
                  else:
                         lowerCamelCase       :				Union[str, Any]      			= downsample_block(hidden_states=UpperCamelCase__					,  temb=UpperCamelCase__					)

                  down_block_res_samples += res_samples

           # 4. mid
           lowerCamelCase       :				Dict      			= self.mid_block(UpperCamelCase__					,  UpperCamelCase__					)

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

                  if hasattr(UpperCamelCase__					,  "skip_conv"					):
                         lowerCamelCase       :				List[str]      			= upsample_block(UpperCamelCase__					,  UpperCamelCase__					,  UpperCamelCase__					,  UpperCamelCase__					)
                  else:
                         lowerCamelCase       :				Union[str, Any]      			= upsample_block(UpperCamelCase__					,  UpperCamelCase__					,  UpperCamelCase__					)

        # 6. post-process
           lowerCamelCase       :				Optional[int]      			= self.conv_norm_out(UpperCamelCase__					)
           lowerCamelCase       :				Tuple      			= self.conv_act(UpperCamelCase__					)
           lowerCamelCase       :				Optional[Any]      			= self.conv_out(UpperCamelCase__					)

           if skip_sample is not None:
                  sample += skip_sample

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

           if not return_dict:
                  return (sample,)

           return UNetaDOutput(sample=UpperCamelCase__					)












import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig


SCREAMING_SNAKE_CASE__	:				int  =       logging.get_logger(__name__)




class 		UpperCamelCase__     (lowerCAmelCase__					):

    '''simple docstring'''



    lowerCamelCase_				:     str     					=				"""speech-encoder-decoder"""
    lowerCamelCase_				:     List[str]     					=				True







    def __init__(   self					,  **UpperCamelCase__					)					->				Dict:
           super().__init__(**UpperCamelCase__					)
           if "encoder" not in kwargs or "decoder" not in kwargs:
                  raise ValueError(
                      F'''A configuraton of type {self.model_type} cannot be instantiated because not both `encoder` and'''
                      F''' `decoder` sub-configurations are passed, but only {kwargs}'''					)

           lowerCamelCase       :				Dict      			= kwargs.pop("encoder"					)
           lowerCamelCase       :				Tuple      			= encoder_config.pop("model_type"					)
           lowerCamelCase       :				int      			= kwargs.pop("decoder"					)
           lowerCamelCase       :				List[Any]      			= decoder_config.pop("model_type"					)

           lowerCamelCase       :				str      			= AutoConfig.for_model(UpperCamelCase__					,  **UpperCamelCase__					)
           lowerCamelCase       :				str      			= AutoConfig.for_model(UpperCamelCase__					,  **UpperCamelCase__					)
           lowerCamelCase       :				Tuple      			= True







    @classmethod
    def    _lowercase					(   cls					,  UpperCamelCase__					,  UpperCamelCase__					,  **UpperCamelCase__					)					->				PretrainedConfig:
           logger.info("Setting `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config"					)
           lowerCamelCase       :				Dict      			= True
           lowerCamelCase       :				Union[str, Any]      			= True

           return cls(encoder=encoder_config.to_dict()					,  decoder=decoder_config.to_dict()					,  **UpperCamelCase__					)







    def    _lowercase					(   self					)					->				Tuple:
           lowerCamelCase       :				Tuple      			= copy.deepcopy(self.__dict__					)
           lowerCamelCase       :				Optional[int]      			= self.encoder.to_dict()
           lowerCamelCase       :				List[str]      			= self.decoder.to_dict()
           lowerCamelCase       :				Dict      			= self.__class__.model_type
           return output






