




"""simple docstring"""


import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest






class    lowercase(	__a					):







 '''simple docstring'''





 lowercase__					   =   (DEISMultistepScheduler,)
 lowercase__					   =   (("num_inference_steps", 25),)


 def 					UpperCamelCase_  (							self:						Optional[Any], **a_:						int						):




   '''simple docstring'''



   _snake_case				:						List[Any]								=    {
       """num_train_timesteps""": 1_000,
       """beta_start""": 0.0_001,
       """beta_end""": 0.02,
       """beta_schedule""": """linear""",
       """solver_order""": 2,
   }

   config.update(**lowerCamelCase__						)
   return config


 def 					UpperCamelCase_  (							self:						Any, a_:						Optional[Any]=0, **a_:						Union[str, Any]						):




   '''simple docstring'''



   _snake_case				:						List[str]								=    dict(self.forward_default_kwargs						)
   _snake_case				:						str								=    kwargs.pop("""num_inference_steps""", lowerCamelCase__						)
   _snake_case				:						Tuple								=    self.dummy_sample
   _snake_case				:						Optional[int]								=    0.1 * sample
   _snake_case				:						Optional[Any]								=    [residual + 0.2, residual + 0.15, residual + 0.10]

   for scheduler_class in self.scheduler_classes:
     _snake_case				:						int								=    self.get_scheduler_config(**lowerCamelCase__						)
     _snake_case				:						Union[str, Any]								=    scheduler_class(**lowerCamelCase__						)
     scheduler.set_timesteps(lowerCamelCase__						)
     # copy over dummy past residuals
     _snake_case				:						Any								=    dummy_past_residuals[: scheduler.config.solver_order]

     with tempfile.TemporaryDirectory() as tmpdirname:
       scheduler.save_config(lowerCamelCase__						)
       _snake_case				:						List[Any]								=    scheduler_class.from_pretrained(lowerCamelCase__						)
       new_scheduler.set_timesteps(lowerCamelCase__						)
       # copy over dummy past residuals
       _snake_case				:						Optional[int]								=    dummy_past_residuals[: new_scheduler.config.solver_order]

     _snake_case						,   _snake_case				:						Any								=    sample, sample
     for t in range(lowerCamelCase__, time_step + scheduler.config.solver_order + 1						):
       _snake_case				:						Optional[int]								=    scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__, **lowerCamelCase__						).prev_sample
       _snake_case				:						Optional[int]								=    new_scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__, **lowerCamelCase__						).prev_sample

       assert torch.sum(torch.abs(output - new_output						)						) < 1E-5, "Scheduler outputs are not identical"


 def 					UpperCamelCase_  (							self:						Any						):




   '''simple docstring'''



   pass


 def 					UpperCamelCase_  (							self:						Dict, a_:						str=0, **a_:						Any						):




   '''simple docstring'''



   _snake_case				:						Optional[int]								=    dict(self.forward_default_kwargs						)
   _snake_case				:						Dict								=    kwargs.pop("""num_inference_steps""", lowerCamelCase__						)
   _snake_case				:						Dict								=    self.dummy_sample
   _snake_case				:						Union[str, Any]								=    0.1 * sample
   _snake_case				:						List[Any]								=    [residual + 0.2, residual + 0.15, residual + 0.10]

   for scheduler_class in self.scheduler_classes:
     _snake_case				:						List[str]								=    self.get_scheduler_config()
     _snake_case				:						List[str]								=    scheduler_class(**lowerCamelCase__						)
     scheduler.set_timesteps(lowerCamelCase__						)

     # copy over dummy past residuals (must be after setting timesteps)
     _snake_case				:						Dict								=    dummy_past_residuals[: scheduler.config.solver_order]

     with tempfile.TemporaryDirectory() as tmpdirname:
       scheduler.save_config(lowerCamelCase__						)
       _snake_case				:						List[str]								=    scheduler_class.from_pretrained(lowerCamelCase__						)
       # copy over dummy past residuals
       new_scheduler.set_timesteps(lowerCamelCase__						)

       # copy over dummy past residual (must be after setting timesteps)
       _snake_case				:						Tuple								=    dummy_past_residuals[: new_scheduler.config.solver_order]

     _snake_case				:						Dict								=    scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__, **lowerCamelCase__						).prev_sample
     _snake_case				:						Tuple								=    new_scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__, **lowerCamelCase__						).prev_sample

     assert torch.sum(torch.abs(output - new_output						)						) < 1E-5, "Scheduler outputs are not identical"


 def 					UpperCamelCase_  (							self:						Union[str, Any], a_:						Dict=None, **a_:						Optional[int]						):




   '''simple docstring'''



   if scheduler is None:
     _snake_case				:						List[Any]								=    self.scheduler_classes[0]
     _snake_case				:						Dict								=    self.get_scheduler_config(**lowerCamelCase__						)
     _snake_case				:						Union[str, Any]								=    scheduler_class(**lowerCamelCase__						)

   _snake_case				:						List[Any]								=    self.scheduler_classes[0]
   _snake_case				:						Any								=    self.get_scheduler_config(**lowerCamelCase__						)
   _snake_case				:						Optional[int]								=    scheduler_class(**lowerCamelCase__						)

   _snake_case				:						Any								=    10
   _snake_case				:						List[Any]								=    self.dummy_model()
   _snake_case				:						int								=    self.dummy_sample_deter
   scheduler.set_timesteps(lowerCamelCase__						)

   for i, t in enumerate(scheduler.timesteps						):
     _snake_case				:						str								=    model(lowerCamelCase__, lowerCamelCase__						)
     _snake_case				:						Dict								=    scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__						).prev_sample

   return sample


 def 					UpperCamelCase_  (							self:						int						):




   '''simple docstring'''



   _snake_case				:						List[str]								=    dict(self.forward_default_kwargs						)

   _snake_case				:						Union[str, Any]								=    kwargs.pop("""num_inference_steps""", lowerCamelCase__						)

   for scheduler_class in self.scheduler_classes:
     _snake_case				:						Any								=    self.get_scheduler_config()
     _snake_case				:						Any								=    scheduler_class(**lowerCamelCase__						)

     _snake_case				:						Dict								=    self.dummy_sample
     _snake_case				:						int								=    0.1 * sample

     if num_inference_steps is not None and hasattr(lowerCamelCase__, """set_timesteps"""						):
       scheduler.set_timesteps(lowerCamelCase__						)
     elif num_inference_steps is not None and not hasattr(lowerCamelCase__, """set_timesteps"""						):
       _snake_case				:						Tuple								=    num_inference_steps

     # copy over dummy past residuals (must be done after set_timesteps)
     _snake_case				:						Optional[Any]								=    [residual + 0.2, residual + 0.15, residual + 0.10]
     _snake_case				:						Tuple								=    dummy_past_residuals[: scheduler.config.solver_order]

     _snake_case				:						int								=    scheduler.timesteps[5]
     _snake_case				:						Union[str, Any]								=    scheduler.timesteps[6]

     _snake_case				:						Optional[Any]								=    scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__, **lowerCamelCase__						).prev_sample
     _snake_case				:						str								=    scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__, **lowerCamelCase__						).prev_sample

     self.assertEqual(output_a.shape, sample.shape						)
     self.assertEqual(output_a.shape, output_a.shape						)


 def 					UpperCamelCase_  (							self:						Any						):




   '''simple docstring'''



   _snake_case				:						Optional[Any]								=    DEISMultistepScheduler(**self.get_scheduler_config()						)
   _snake_case				:						List[Any]								=    self.full_loop(scheduler=lowerCamelCase__						)
   _snake_case				:						List[str]								=    torch.mean(torch.abs(lowerCamelCase__						)						)

   assert abs(result_mean.item() - 0.23_916						) < 1E-3

   _snake_case				:						Any								=    DPMSolverSinglestepScheduler.from_config(scheduler.config						)
   _snake_case				:						Optional[Any]								=    DPMSolverMultistepScheduler.from_config(scheduler.config						)
   _snake_case				:						int								=    UniPCMultistepScheduler.from_config(scheduler.config						)
   _snake_case				:						int								=    DEISMultistepScheduler.from_config(scheduler.config						)

   _snake_case				:						Optional[int]								=    self.full_loop(scheduler=lowerCamelCase__						)
   _snake_case				:						Optional[Any]								=    torch.mean(torch.abs(lowerCamelCase__						)						)

   assert abs(result_mean.item() - 0.23_916						) < 1E-3


 def 					UpperCamelCase_  (							self:						int						):




   '''simple docstring'''



   for timesteps in [25, 50, 100, 999, 1_000]:
     self.check_over_configs(num_train_timesteps=lowerCamelCase__						)


 def 					UpperCamelCase_  (							self:						Optional[Any]						):




   '''simple docstring'''



   self.check_over_configs(thresholding=lowerCamelCase__						)
   for order in [1, 2, 3]:
     for solver_type in ["logrho"]:
       for threshold in [0.5, 1.0, 2.0]:
         for prediction_type in ["epsilon", "sample"]:
           self.check_over_configs(
               thresholding=lowerCamelCase__, prediction_type=lowerCamelCase__, sample_max_value=lowerCamelCase__, algorithm_type="""deis""", solver_order=lowerCamelCase__, solver_type=lowerCamelCase__, )


 def 					UpperCamelCase_  (							self:						List[str]						):




   '''simple docstring'''



   for prediction_type in ["epsilon", "v_prediction"]:
     self.check_over_configs(prediction_type=lowerCamelCase__						)


 def 					UpperCamelCase_  (							self:						str						):




   '''simple docstring'''



   for algorithm_type in ["deis"]:
     for solver_type in ["logrho"]:
       for order in [1, 2, 3]:
         for prediction_type in ["epsilon", "sample"]:
           self.check_over_configs(
               solver_order=lowerCamelCase__, solver_type=lowerCamelCase__, prediction_type=lowerCamelCase__, algorithm_type=lowerCamelCase__, )
           _snake_case				:						Dict								=    self.full_loop(
               solver_order=lowerCamelCase__, solver_type=lowerCamelCase__, prediction_type=lowerCamelCase__, algorithm_type=lowerCamelCase__, )
           assert not torch.isnan(lowerCamelCase__						).any(), "Samples have nan numbers"


 def 					UpperCamelCase_  (							self:						Tuple						):




   '''simple docstring'''



   self.check_over_configs(lower_order_final=lowerCamelCase__						)
   self.check_over_configs(lower_order_final=lowerCamelCase__						)


 def 					UpperCamelCase_  (							self:						Union[str, Any]						):




   '''simple docstring'''



   for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1_000]:
     self.check_over_forward(num_inference_steps=lowerCamelCase__, time_step=0						)


 def 					UpperCamelCase_  (							self:						int						):




   '''simple docstring'''



   _snake_case				:						Tuple								=    self.full_loop()
   _snake_case				:						Optional[int]								=    torch.mean(torch.abs(lowerCamelCase__						)						)

   assert abs(result_mean.item() - 0.23_916						) < 1E-3


 def 					UpperCamelCase_  (							self:						Optional[Any]						):




   '''simple docstring'''



   _snake_case				:						Optional[int]								=    self.full_loop(prediction_type="""v_prediction"""						)
   _snake_case				:						List[Any]								=    torch.mean(torch.abs(lowerCamelCase__						)						)

   assert abs(result_mean.item() - 0.091						) < 1E-3


 def 					UpperCamelCase_  (							self:						Dict						):




   '''simple docstring'''



   _snake_case				:						List[str]								=    self.scheduler_classes[0]
   _snake_case				:						Any								=    self.get_scheduler_config(thresholding=lowerCamelCase__, dynamic_thresholding_ratio=0						)
   _snake_case				:						str								=    scheduler_class(**lowerCamelCase__						)

   _snake_case				:						Any								=    10
   _snake_case				:						str								=    self.dummy_model()
   _snake_case				:						List[Any]								=    self.dummy_sample_deter.half()
   scheduler.set_timesteps(lowerCamelCase__						)

   for i, t in enumerate(scheduler.timesteps						):
     _snake_case				:						int								=    model(lowerCamelCase__, lowerCamelCase__						)
     _snake_case				:						Tuple								=    scheduler.step(lowerCamelCase__, lowerCamelCase__, lowerCamelCase__						).prev_sample

   assert sample.dtype == torch.floataa








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 get_down_block, get_mid_block, get_out_block, get_up_block







@dataclass
class        __lowerCAmelCase					(			__magic_name__					):

			"""simple docstring"""







			snake_case_  						=			42







class        __lowerCAmelCase					(			__magic_name__     ,							__magic_name__					):

			"""simple docstring"""







			@register_to_config
			def __init__( self     ,		lowerCamelCase__ = 65_536     ,		lowerCamelCase__ = None     ,		lowerCamelCase__ = 2     ,		lowerCamelCase__ = 2     ,		lowerCamelCase__ = 0     ,		lowerCamelCase__ = "fourier"     ,		lowerCamelCase__ = True     ,		lowerCamelCase__ = False     ,		lowerCamelCase__ = 0.0     ,		lowerCamelCase__ = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")     ,		lowerCamelCase__ = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")     ,		lowerCamelCase__ = "UNetMidBlock1D"     ,		lowerCamelCase__ = None     ,		lowerCamelCase__ = (32, 32, 64)     ,		lowerCamelCase__ = None     ,		lowerCamelCase__ = 8     ,		lowerCamelCase__ = 1     ,		lowerCamelCase__ = False     ,		)       -> Dict:



							'''simple docstring'''
							super().__init__()
							__lowerCamelCase									=       sample_size

							# time
							if time_embedding_type == "fourier":
											__lowerCamelCase									=       GaussianFourierProjection(
											    embedding_size=8     ,		set_W_to_weight=lowerCamelCase__     ,		log=lowerCamelCase__     ,		flip_sin_to_cos=lowerCamelCase__				)
											__lowerCamelCase									=       2 * block_out_channels[0]
							elif time_embedding_type == "positional":
											__lowerCamelCase									=       Timesteps(
											    block_out_channels[0]     ,		flip_sin_to_cos=lowerCamelCase__     ,		downscale_freq_shift=lowerCamelCase__				)
											__lowerCamelCase									=       block_out_channels[0]

							if use_timestep_embedding:
											__lowerCamelCase									=       block_out_channels[0] * 4
											__lowerCamelCase									=       TimestepEmbedding(
											    in_channels=lowerCamelCase__     ,		time_embed_dim=lowerCamelCase__     ,		act_fn=lowerCamelCase__     ,		out_dim=block_out_channels[0]     ,		)

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

							# down
							__lowerCamelCase									=       in_channels
							for i, down_block_type in enumerate(lowerCamelCase__				):
											__lowerCamelCase									=       output_channel
											__lowerCamelCase									=       block_out_channels[i]

											if i == 0:
															input_channel += extra_in_channels

											__lowerCamelCase									=       i == len(lowerCamelCase__				) - 1

											__lowerCamelCase									=       get_down_block(
											    lowerCamelCase__     ,		num_layers=lowerCamelCase__     ,		in_channels=lowerCamelCase__     ,		out_channels=lowerCamelCase__     ,		temb_channels=block_out_channels[0]     ,		add_downsample=not is_final_block or downsample_each_block     ,		)
											self.down_blocks.append(lowerCamelCase__				)

							# mid
							__lowerCamelCase									=       get_mid_block(
							    lowerCamelCase__     ,		in_channels=block_out_channels[-1]     ,		mid_channels=block_out_channels[-1]     ,		out_channels=block_out_channels[-1]     ,		embed_dim=block_out_channels[0]     ,		num_layers=lowerCamelCase__     ,		add_downsample=lowerCamelCase__     ,		)

							# up
							__lowerCamelCase									=       list(reversed(lowerCamelCase__				)				)
							__lowerCamelCase									=       reversed_block_out_channels[0]
							if out_block_type is None:
											__lowerCamelCase									=       out_channels
							else:
											__lowerCamelCase									=       block_out_channels[0]

							for i, up_block_type in enumerate(lowerCamelCase__				):
											__lowerCamelCase									=       output_channel
											__lowerCamelCase									=       (
											    reversed_block_out_channels[i + 1] if i < len(lowerCamelCase__				) - 1 else final_upsample_channels
											)

											__lowerCamelCase									=       i == len(lowerCamelCase__				) - 1

											__lowerCamelCase									=       get_up_block(
											    lowerCamelCase__     ,		num_layers=lowerCamelCase__     ,		in_channels=lowerCamelCase__     ,		out_channels=lowerCamelCase__     ,		temb_channels=block_out_channels[0]     ,		add_upsample=not is_final_block     ,		)
											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									=       get_out_block(
							    out_block_type=lowerCamelCase__     ,		num_groups_out=lowerCamelCase__     ,		embed_dim=block_out_channels[0]     ,		out_channels=lowerCamelCase__     ,		act_fn=lowerCamelCase__     ,		fc_dim=block_out_channels[-1] // 4     ,		)






			def  lowercase_       ( self     ,		lowerCamelCase__     ,		lowerCamelCase__     ,		lowerCamelCase__ = True     ,		)       -> Union[UNetaDOutput, Tuple]:



							'''simple docstring'''
							__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				)

							__lowerCamelCase									=       self.time_proj(lowerCamelCase__				)
							if self.config.use_timestep_embedding:
											__lowerCamelCase									=       self.time_mlp(lowerCamelCase__				)
							else:
											__lowerCamelCase									=       timestep_embed[..., None]
											__lowerCamelCase									=       timestep_embed.repeat([1, 1, sample.shape[2]]				).to(sample.dtype				)
											__lowerCamelCase									=       timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:])				)

							# 2. down
							__lowerCamelCase									=       ()
							for downsample_block in self.down_blocks:
											__lowerCamelCase			,		__lowerCamelCase									=       downsample_block(hidden_states=lowerCamelCase__     ,		temb=lowerCamelCase__				)
											down_block_res_samples += res_samples

							# 3. mid
							if self.mid_block:
											__lowerCamelCase									=       self.mid_block(lowerCamelCase__     ,		lowerCamelCase__				)

							# 4. up
							for i, upsample_block in enumerate(self.up_blocks				):
											__lowerCamelCase									=       down_block_res_samples[-1:]
											__lowerCamelCase									=       down_block_res_samples[:-1]
											__lowerCamelCase									=       upsample_block(lowerCamelCase__     ,		res_hidden_states_tuple=lowerCamelCase__     ,		temb=lowerCamelCase__				)

							# 5. post-process
							if self.out_block:
											__lowerCamelCase									=       self.out_block(lowerCamelCase__     ,		lowerCamelCase__				)

							if not return_dict:
											return (sample,)

							return UNetaDOutput(sample=lowerCamelCase__				)


