




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


     """simple docstring"""


     _A    :   torch.FloatTensor






class 				__a (			__lowerCamelCase							,					__lowerCamelCase					):


     """simple docstring"""


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


         '''simple docstring'''





         super().__init__()

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

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

         SCREAMING_SNAKE_CASE__       =TimestepEmbedding(_UpperCamelCase    ,_UpperCamelCase						)

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

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

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

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

             SCREAMING_SNAKE_CASE__       =i == len(_UpperCamelCase						) - 1

             SCREAMING_SNAKE_CASE__       =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__       =output_channel

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




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


         '''simple docstring'''





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

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

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

         SCREAMING_SNAKE_CASE__       =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__       =t_emb.to(dtype=self.dtype						)
         SCREAMING_SNAKE_CASE__       =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__       =self.time_proj(_UpperCamelCase						)

             SCREAMING_SNAKE_CASE__       =self.class_embedding(_UpperCamelCase						).to(dtype=self.dtype						)
             SCREAMING_SNAKE_CASE__       =emb + class_emb

         # 2. pre-process
         SCREAMING_SNAKE_CASE__       =sample
         SCREAMING_SNAKE_CASE__       =self.conv_in(_UpperCamelCase						)

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

             down_block_res_samples += res_samples

         # 4. mid
         SCREAMING_SNAKE_CASE__       =self.mid_block(_UpperCamelCase    ,_UpperCamelCase						)

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

             if hasattr(_UpperCamelCase    ,"""skip_conv"""						):
                 SCREAMING_SNAKE_CASE__      ,		SCREAMING_SNAKE_CASE__       =upsample_block(_UpperCamelCase    ,_UpperCamelCase    ,_UpperCamelCase    ,_UpperCamelCase						)
             else:
                 SCREAMING_SNAKE_CASE__       =upsample_block(_UpperCamelCase    ,_UpperCamelCase    ,_UpperCamelCase						)

        # 6. post-process
         SCREAMING_SNAKE_CASE__       =self.conv_norm_out(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =self.conv_act(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =self.conv_out(_UpperCamelCase						)

         if skip_sample is not None:
             sample += skip_sample

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

         if not return_dict:
             return (sample,)

         return UNetaDOutput(sample=_UpperCamelCase						)












import tempfile
import unittest

from diffusers import (
    DDIMScheduler,
    DDPMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    PNDMScheduler,
    logging,
)
from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.utils.testing_utils import CaptureLogger
class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			int    ,_UpperCamelCase      :			int=2    ,_UpperCamelCase      :			Optional[Any]=5    ,_UpperCamelCase      :			List[str]=(2, 5)    ,_UpperCamelCase      :			List[Any]="for diffusion"    ,_UpperCamelCase      :			Optional[int]=[1, 3]    ,)		->       Any:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[str]     =  "config.json"




     @register_to_config
     def __init__(     self      :			List[Any]    ,_UpperCamelCase      :			str=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			Optional[Any]=(2, 5)    ,_UpperCamelCase      :			Union[str, Any]="for diffusion"    ,_UpperCamelCase      :			Tuple=[1, 3]    ,)		->       List[str]:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   Optional[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			str    ,_UpperCamelCase      :			Optional[int]=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			List[Any]=(2, 5)    ,_UpperCamelCase      :			Dict="for diffusion"    ,_UpperCamelCase      :			Union[str, Any]=[1, 3]    ,_UpperCamelCase      :			Dict=[1, 3]    ,)		->       Any:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			Union[str, Any]    ,_UpperCamelCase      :			str=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			List[Any]=(2, 5)    ,_UpperCamelCase      :			int="for diffusion"    ,_UpperCamelCase      :			Any=[1, 5]    ,_UpperCamelCase      :			Optional[Any]=[5, 4]    ,)		->       List[Any]:


         '''simple docstring'''





         pass






class 				__a (			unittest.TestCase					):


     """simple docstring"""


     def   __A    (     self      :			Dict						)		->       Optional[Any]:


         '''simple docstring'''





         with self.assertRaises(_UpperCamelCase						):
             ConfigMixin.load_config("""dummy_path"""						)




     def   __A    (     self      :			int						)		->       List[str]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # init ignore private arguments
         SCREAMING_SNAKE_CASE__       =SampleObject(_name_or_path="""lalala"""						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # can override default
         SCREAMING_SNAKE_CASE__       =SampleObject(c=6						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == 6
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # can use positional arguments.
         SCREAMING_SNAKE_CASE__       =SampleObject(1    ,c=6						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 1
         assert config["b"] == 5
         assert config["c"] == 6
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]




     def   __A    (     self      :			List[Any]						)		->       Tuple:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()
         SCREAMING_SNAKE_CASE__       =obj.config

         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         with tempfile.TemporaryDirectory() as tmpdirname:
             obj.save_config(_UpperCamelCase						)
             SCREAMING_SNAKE_CASE__       =SampleObject.from_config(SampleObject.load_config(_UpperCamelCase						)						)
             SCREAMING_SNAKE_CASE__       =new_obj.config

         # unfreeze configs
         SCREAMING_SNAKE_CASE__       =dict(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =dict(_UpperCamelCase						)

         assert config.pop("""c"""						) == (2, 5)  # instantiated as tuple
         assert new_config.pop("""c"""						) == [2, 5]  # saved & loaded as list because of json
         config.pop("""_use_default_values"""						)
         assert config == new_config




     def   __A    (     self      :			Union[str, Any]						)		->       Union[str, Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DDIMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert ddim.__class__ == DDIMScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			Dict						)		->       Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =EulerDiscreteScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert euler.__class__ == EulerDiscreteScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       int:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =EulerAncestralDiscreteScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert euler.__class__ == EulerAncestralDiscreteScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       Optional[int]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =PNDMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert pndm.__class__ == PNDMScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			Optional[int]						)		->       List[str]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DDPMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""    ,prediction_type="""sample"""    ,beta_end=8    ,)

         with CaptureLogger(_UpperCamelCase						) as cap_logger_a:
             SCREAMING_SNAKE_CASE__       =DDPMScheduler.from_pretrained("""google/ddpm-celebahq-256"""    ,beta_start=8_8						)

         assert ddpm.__class__ == DDPMScheduler
         assert ddpm.config.prediction_type == "sample"
         assert ddpm.config.beta_end == 8
         assert ddpm_a.config.beta_start == 8_8

         # no warning should be thrown
         assert cap_logger.out == ""
         assert cap_logger_a.out == ""




     def   __A    (     self      :			Any						)		->       Optional[int]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DPMSolverMultistepScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert dpm.__class__ == DPMSolverMultistepScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()

         SCREAMING_SNAKE_CASE__       ={k: v for k, v in config.config.items() if not k.startswith("""_"""						)}

         # make sure that default config has all keys in `_use_default_values`
         assert set(config_dict.keys()						) == set(config.config._use_default_values						)

         with tempfile.TemporaryDirectory() as tmpdirname:
             config.save_config(_UpperCamelCase						)

             # now loading it with SampleObject2 should put f into `_use_default_values`
             SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(_UpperCamelCase						)

             assert "f" in config._use_default_values
             assert config.f == [1, 3]

         # now loading the config, should **NOT** use [1, 3] for `f`, but the default [1, 4] value
         # **BECAUSE** it is part of `config._use_default_values`
         SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(config.config						)
         assert new_config.f == [5, 4]

         config.config._use_default_values.pop()
         SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(config.config						)
         assert new_config_a.f == [1, 3]

         # Nevertheless "e" should still be correctly loaded to [1, 3] from SampleObject2 instead of defaulting to [1, 5]
         assert new_config_a.e == [1, 3]






