


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      (				lowerCamelCase_ ):


				"""simple docstring"""







				a_:		torch.FloatTensor


class        __lowerCamelCase      (				lowerCamelCase_				,		lowerCamelCase_ ):


				"""simple docstring"""







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

								_lowerCAmelCase  						=sample_size
								_lowerCAmelCase  						=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
								_lowerCAmelCase  						=nn.Convad(lowerCamelCase_	,							block_out_channels[0]	,							kernel_size=3	,							padding=(1, 1)					)

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

								_lowerCAmelCase  						=TimestepEmbedding(lowerCamelCase_	,							lowerCamelCase_					)

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

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

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

												_lowerCAmelCase  						=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
								_lowerCAmelCase  						=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
								_lowerCAmelCase  						=list(reversed(lowerCamelCase_					)					)
								_lowerCAmelCase  						=reversed_block_out_channels[0]
								for i, up_block_type in enumerate(lowerCamelCase_					):
												_lowerCAmelCase  						=output_channel
												_lowerCAmelCase  						=reversed_block_out_channels[i]
												_lowerCAmelCase  						=reversed_block_out_channels[min(i + 1	,							len(lowerCamelCase_					) - 1					)]

												_lowerCAmelCase  						=i == len(lowerCamelCase_					) - 1

												_lowerCAmelCase  						=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_					)
												_lowerCAmelCase  						=output_channel

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






				def        lowerCAmelCase__			(				self     :    List[Any]	,							lowerCamelCase_     :    torch.FloatTensor	,							lowerCamelCase_     :    Union[torch.Tensor, float, int]	,							lowerCamelCase_     :    Optional[torch.Tensor] = None	,							lowerCamelCase_     :    bool = True	,							):
								# 0. center input if necessary
								if self.config.center_input_sample:
												_lowerCAmelCase  						=2 * sample - 1.0

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

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

								_lowerCAmelCase  						=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.
								_lowerCAmelCase  						=t_emb.to(dtype=self.dtype					)
								_lowerCAmelCase  						=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":
																_lowerCAmelCase  						=self.time_proj(lowerCamelCase_					)

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

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

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

												down_block_res_samples += res_samples

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

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

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

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

								if skip_sample is not None:
												sample += skip_sample

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

								if not return_dict:
												return (sample,)

								return UNetaDOutput(sample=lowerCamelCase_					)





from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_sentencepiece_available, is_torch_available


__SCREAMING_SNAKE_CASE	:							str           =  {
    '''configuration_ernie_m''': ['''ERNIE_M_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''ErnieMConfig'''],
}

try:
		if not is_sentencepiece_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		__SCREAMING_SNAKE_CASE	:							Any           =  ['''ErnieMTokenizer''']

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		__SCREAMING_SNAKE_CASE	:							Tuple           =  [
		    '''ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''ErnieMForMultipleChoice''',
		    '''ErnieMForQuestionAnswering''',
		    '''ErnieMForSequenceClassification''',
		    '''ErnieMForTokenClassification''',
		    '''ErnieMModel''',
		    '''ErnieMPreTrainedModel''',
		    '''ErnieMForInformationExtraction''',
		]


if TYPE_CHECKING:
		from .configuration_ernie_m import ERNIE_M_PRETRAINED_CONFIG_ARCHIVE_MAP, ErnieMConfig

		try:
				if not is_sentencepiece_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .tokenization_ernie_m import ErnieMTokenizer

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_ernie_m import (
				    ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST,
				    ErnieMForInformationExtraction,
				    ErnieMForMultipleChoice,
				    ErnieMForQuestionAnswering,
				    ErnieMForSequenceClassification,
				    ErnieMForTokenClassification,
				    ErnieMModel,
				    ErnieMPreTrainedModel,
				)


else:
		import sys

		__SCREAMING_SNAKE_CASE	:							str           =  _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)

