o Gi3@s~ddlZddlmZddlZddlmZmZddlmZddl m Z ddl m Z m Z eGd d d eZGd d d e eZdS) N) dataclass) ConfigMixinregister_to_config) BaseOutput) randn_tensor)SchedulerMixinSchedulerOutputc@s&eZdZUdZejed<ejed<dS) SdeVeOutputa Output class for the scheduler's `step` function output. Args: prev_sample (`torch.Tensor` of shape `(batch_size, num_channels, height, width)` for images): Computed sample `(x_{t-1})` of previous timestep. `prev_sample` should be used as next model input in the denoising loop. prev_sample_mean (`torch.Tensor` of shape `(batch_size, num_channels, height, width)` for images): Mean averaged `prev_sample` over previous timesteps. prev_sampleprev_sample_meanN)__name__ __module__ __qualname____doc__torchTensor__annotations__rrZ/home/ubuntu/.local/lib/python3.10/site-packages/diffusers/schedulers/scheduling_sde_ve.pyr s  r c@s:eZdZdZdZe      d-ded ed ed ed ed ef ddZd.de j dede j fddZ d/ded ede e j BfddZ d0ded ed ed efddZddZ  d1de j dede j d e jdBd!edeeBf d"d#Z  d1de j de j d e jdBd!edeeBf d$d%Zd&e j d'e j d(e j de j fd)d*Zd+d,ZdS)2ScoreSdeVeSchedulera `ScoreSdeVeScheduler` is a variance exploding stochastic differential equation (SDE) scheduler. This model inherits from [`SchedulerMixin`] and [`ConfigMixin`]. Check the superclass documentation for the generic methods the library implements for all schedulers such as loading and saving. Args: num_train_timesteps (`int`, defaults to 1000): The number of diffusion steps to train the model. snr (`float`, defaults to 0.15): A coefficient weighting the step from the `model_output` sample (from the network) to the random noise. sigma_min (`float`, defaults to 0.01): The initial noise scale for the sigma sequence in the sampling procedure. The minimum sigma should mirror the distribution of the data. sigma_max (`float`, defaults to 1348.0): The maximum value used for the range of continuous timesteps passed into the model. sampling_eps (`float`, defaults to 1e-5): The end value of sampling where timesteps decrease progressively from 1 to epsilon. correct_steps (`int`, defaults to 1): The number of correction steps performed on a produced sample. r333333?{Gz?@h㈵>num_train_timestepssnr sigma_min sigma_max sampling_eps correct_stepscCs ||_d|_|||||dSN)init_noise_sigma timesteps set_sigmas)selfrrrr r!r"rrr__init__Fs zScoreSdeVeScheduler.__init__NsampletimestepreturncCs|S)a Ensures interchangeability with schedulers that need to scale the denoising model input depending on the current timestep. Args: sample (`torch.Tensor`): The input sample. timestep (`int`, *optional*): The current timestep in the diffusion chain. Returns: `torch.Tensor`: A scaled input sample. r)r'r)r*rrrscale_model_inputXsz%ScoreSdeVeScheduler.scale_model_inputnum_inference_stepsdevicecCs,|dur|n|jj}tjd|||d|_dS)aA Sets the continuous timesteps used for the diffusion chain (to be run before inference). Args: num_inference_steps (`int`): The number of diffusion steps used when generating samples with a pre-trained model. sampling_eps (`float`, *optional*): The final timestep value (overrides value given during scheduler instantiation). device (`str` or `torch.device`, *optional*): The device to which the timesteps should be moved to. If `None`, the timesteps are not moved. Nrr.)configr!rlinspacer%)r'r-r!r.rrr set_timestepsis z!ScoreSdeVeScheduler.set_timestepscsdurn|jjdurn|jj|dur|n|jj}|jdur)||||j||_tt t t ||_ t fdd|jD|_dS)a Sets the noise scales used for the diffusion chain (to be run before inference). The sigmas control the weight of the `drift` and `diffusion` components of the sample update. Args: num_inference_steps (`int`): The number of diffusion steps used when generating samples with a pre-trained model. sigma_min (`float`, optional): The initial noise scale value (overrides value given during scheduler instantiation). sigma_max (`float`, optional): The final noise scale value (overrides value given during scheduler instantiation). sampling_eps (`float`, optional): The final timestep value (overrides value given during scheduler instantiation). Ncsg|] }|qSrr).0tr rrr sz2ScoreSdeVeScheduler.set_sigmas..)r0rr r!r%r2sigmasrexpr1mathlogdiscrete_sigmastensor)r'r-rr r!rr5rr&zs  ""zScoreSdeVeScheduler.set_sigmascCs2t|dkt||j|j|d|jS)Nrr)rwhere zeros_liketor.r;)r'r%r4rrrget_adjacent_sigmas z&ScoreSdeVeScheduler.get_adjacent_sigmaT model_output generator return_dictcCs$|jdur td|tj|jd|jd}|t|jd}||j j}|j ||j}| |||j}t |} |d|dd} | } t| jt|jkrg| d} t| jt|jksX| | d|} t|j|j||j|jd } || } | | | } |s| | fSt| | d S) aD Predict the sample from the previous timestep by reversing the SDE. This function propagates the diffusion process from the learned model outputs (most often the predicted noise). Args: model_output (`torch.Tensor`): The direct output from learned diffusion model. timestep (`int`): The current discrete timestep in the diffusion chain. sample (`torch.Tensor`): A current instance of a sample created by the diffusion process. generator (`torch.Generator`, *optional*): A random number generator. return_dict (`bool`, *optional*, defaults to `True`): Whether or not to return a [`~schedulers.scheduling_sde_ve.SdeVeOutput`] or `tuple`. Returns: [`~schedulers.scheduling_sde_ve.SdeVeOutput`] or `tuple`: If return_dict is `True`, [`~schedulers.scheduling_sde_ve.SdeVeOutput`] is returned, otherwise a tuple is returned where the first element is the sample tensor. NY`self.timesteps` is not set, you need to run 'set_timesteps' after creating the schedulerrr/rr?)layoutrBr.dtype)r r )r% ValueErrorronesshaper.lenlongr?r;r@r>flatten unsqueezerrGrHr )r'rAr*r)rBrCr%sigmaadjacent_sigmadrift diffusionnoiser r rrr step_preds4      zScoreSdeVeScheduler.step_predc Cs|jdur tdt|j|j|d|j}tj| |jdddd }tj| |jdddd }|j j ||dd}|t |jd|j}|}t|jt|jkrm|d}t|jt|jks^|||} | |dd|} |s| fSt| d S) a Correct the predicted sample based on the `model_output` of the network. This is often run repeatedly after making the prediction for the previous timestep. Args: model_output (`torch.Tensor`): The direct output from learned diffusion model. sample (`torch.Tensor`): A current instance of a sample created by the diffusion process. generator (`torch.Generator`, *optional*): A random number generator. return_dict (`bool`, *optional*, defaults to `True`): Whether or not to return a [`~schedulers.scheduling_sde_ve.SdeVeOutput`] or `tuple`. Returns: [`~schedulers.scheduling_sde_ve.SdeVeOutput`] or `tuple`: If return_dict is `True`, [`~schedulers.scheduling_sde_ve.SdeVeOutput`] is returned, otherwise a tuple is returned where the first element is the sample tensor. NrD)rGrBrrF)dimrrE)r )r%rIrrKrGr?r.rnormreshapemeanr0rrJrNrLrOr ) r'rAr)rBrCrT grad_norm noise_norm step_sizer r rrr step_corrects$      z ScoreSdeVeScheduler.step_correctoriginal_samplesrTr%cCsh||j}|j|j|}|dur||dddddfnt||dddddf}||}|Sr#)r?r.r;r randn_like)r'r^rTr%r7 noisy_samplesrrr add_noises zScoreSdeVeScheduler.add_noisecCs|jjSr#)r0r)r'rrr__len__)szScoreSdeVeScheduler.__len__)rrrrrrr#)NN)NNN)NT)rrrrorderrintfloatr(rrr,strr.r2r&r@ Generatorboolr tuplerUr r]rarbrrrrr-s    H 7 r)r9 dataclassesrrconfiguration_utilsrrutilsrutils.torch_utilsrscheduling_utilsr r r rrrrrs