





"""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		( _lowerCAmelCase							):
				UpperCamelCase								=       42




class 					_a		( _lowerCAmelCase      ,							_lowerCAmelCase							):






				@register_to_config
				def __init__(					self						:       Dict,					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] = (2_2_4, 4_4_8, 6_7_2, 8_9_6),					lowerCAmelCase__						:       int = 2,					lowerCAmelCase__						:       float = 1,					lowerCAmelCase__						:       int = 1,					lowerCAmelCase__						:       str = "conv",					lowerCAmelCase__						:       str = "conv",					lowerCAmelCase__						:       str = "silu",					lowerCAmelCase__						:       Optional[int] = 8,					lowerCAmelCase__						:       int = 3_2,					lowerCAmelCase__						:       float = 1e-5,					lowerCAmelCase__						:       str = "default",					lowerCAmelCase__						:       bool = True,					lowerCAmelCase__						:       Optional[str] = None,					lowerCAmelCase__						:       Optional[int] = None,					)      -> List[Any]:




										'''simple docstring'''



										super().__init__()

										_UpperCamelCase     :   List[Any]      =       sample_size
										_UpperCamelCase     :   List[str]      =       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
										_UpperCamelCase     :   Tuple      =       nn.Convad(lowerCAmelCase__,					block_out_channels[0],					kernel_size=3,					padding=(1, 1)				)

										# time
										if time_embedding_type == "fourier":
																_UpperCamelCase     :   str      =       GaussianFourierProjection(embedding_size=block_out_channels[0],					scale=1_6				)
																_UpperCamelCase     :   Dict      =       2 * block_out_channels[0]
										elif time_embedding_type == "positional":
																_UpperCamelCase     :   Tuple      =       Timesteps(block_out_channels[0],					lowerCAmelCase__,					lowerCAmelCase__				)
																_UpperCamelCase     :   Optional[int]      =       block_out_channels[0]

										_UpperCamelCase     :   str      =       TimestepEmbedding(lowerCAmelCase__,					lowerCAmelCase__				)

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

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

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

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

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

																_UpperCamelCase     :   List[str]      =       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__				)
																_UpperCamelCase     :   List[Any]      =       output_channel

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






				def 							snake_case      (					self						:       List[str],					lowerCAmelCase__						:       torch.FloatTensor,					lowerCAmelCase__						:       Union[torch.Tensor, float, int],					lowerCAmelCase__						:       Optional[torch.Tensor] = None,					lowerCAmelCase__						:       bool = True,					)      -> Union[UNetaDOutput, Tuple]:




										'''simple docstring'''



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

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

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

										_UpperCamelCase     :   int      =       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.
										_UpperCamelCase     :   Tuple      =       t_emb.to(dtype=self.dtype				)
										_UpperCamelCase     :   Union[str, Any]      =       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":
																						_UpperCamelCase     :   List[Any]      =       self.time_proj(lowerCAmelCase__				)

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

										# 2. pre-process
										_UpperCamelCase     :   Dict      =       sample
										_UpperCamelCase     :   Tuple      =       self.conv_in(lowerCAmelCase__				)

										# 3. down
										_UpperCamelCase     :   Tuple      =       (sample,)
										for downsample_block in self.down_blocks:
																if hasattr(lowerCAmelCase__,					'''skip_conv'''				):
																						_UpperCamelCase  ,				_UpperCamelCase  ,				_UpperCamelCase     :   List[Any]      =       downsample_block(
																						    hidden_states=lowerCAmelCase__,					temb=lowerCAmelCase__,					skip_sample=lowerCAmelCase__				)
																else:
																						_UpperCamelCase  ,				_UpperCamelCase     :   Any      =       downsample_block(hidden_states=lowerCAmelCase__,					temb=lowerCAmelCase__				)

																down_block_res_samples += res_samples

										# 4. mid
										_UpperCamelCase     :   Dict      =       self.mid_block(lowerCAmelCase__,					lowerCAmelCase__				)

										# 5. up
										_UpperCamelCase     :   Dict      =       None
										for upsample_block in self.up_blocks:
																_UpperCamelCase     :   Union[str, Any]      =       down_block_res_samples[-len(upsample_block.resnets				) :]
																_UpperCamelCase     :   Any      =       down_block_res_samples[: -len(upsample_block.resnets				)]

																if hasattr(lowerCAmelCase__,					'''skip_conv'''				):
																						_UpperCamelCase  ,				_UpperCamelCase     :   Dict      =       upsample_block(lowerCAmelCase__,					lowerCAmelCase__,					lowerCAmelCase__,					lowerCAmelCase__				)
																else:
																						_UpperCamelCase     :   str      =       upsample_block(lowerCAmelCase__,					lowerCAmelCase__,					lowerCAmelCase__				)

        # 6. post-process
										_UpperCamelCase     :   Union[str, Any]      =       self.conv_norm_out(lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       self.conv_act(lowerCAmelCase__				)
										_UpperCamelCase     :   Optional[Any]      =       self.conv_out(lowerCAmelCase__				)

										if skip_sample is not None:
																sample += skip_sample

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

										if not return_dict:
																return (sample,)

										return UNetaDOutput(sample=lowerCAmelCase__				)









from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
)


__SCREAMING_SNAKE_CASE     : Tuple      ={
    """configuration_open_llama""": ["""OPEN_LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP""", """OpenLlamaConfig"""],
}

try:
      if not is_sentencepiece_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __SCREAMING_SNAKE_CASE     : Dict      =["""LlamaTokenizer"""]

try:
      if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __SCREAMING_SNAKE_CASE     : int      =["""LlamaTokenizerFast"""]

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __SCREAMING_SNAKE_CASE     : Optional[Any]      =[
          """OpenLlamaForCausalLM""",
          """OpenLlamaModel""",
          """OpenLlamaPreTrainedModel""",
          """OpenLlamaForSequenceClassification""",
      ]


if TYPE_CHECKING:
      from .configuration_open_llama import OPEN_LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP, OpenLlamaConfig

      try:
            if not is_sentencepiece_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from transformers import LlamaTokenizer

      try:
            if not is_tokenizers_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from transformers import LlamaTokenizerFast

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_open_llama import (
                OpenLlamaForCausalLM,
                OpenLlamaForSequenceClassification,
                OpenLlamaModel,
                OpenLlamaPreTrainedModel,
            )


else:
      import sys

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