o piY@sddlmZddlmZmZmZddlZddlZddl m m Z ddl mZmZddlmZddlmZeGd d d eZd ejd ed ejfddZdejdeejd ejfddZddefddZddefddZGdddeeZdS)) dataclass)OptionalTupleUnionN) ConfigMixinregister_to_config) BaseOutput)SchedulerMixinc@seZdZUdZejed<dS)VQDiffusionSchedulerOutputa. Output class for the scheduler's step function output. Args: prev_sample (`torch.LongTensor` of shape `(batch size, num latent pixels)`): Computed sample x_{t-1} of previous timestep. `prev_sample` should be used as next model input in the denoising loop. prev_sampleN)__name__ __module__ __qualname____doc__torch LongTensor__annotations__rrj/home/ubuntu/SoloSpeech/.venv/lib/python3.10/site-packages/diffusers/schedulers/scheduling_vq_diffusion.pyr s  r x num_classesreturncCs4t||}|ddd}t|jdd}|S)a Convert batch of vector of class indices into batch of log onehot vectors Args: x (`torch.LongTensor` of shape `(batch size, vector length)`): Batch of class indices num_classes (`int`): number of classes to be used for the onehot vectors Returns: `torch.Tensor` of shape `(batch size, num classes, vector length)`: Log onehot vectors rrr KH9)min)Fone_hotpermuterlogfloatclamp)rrx_onehotlog_xrrrindex_to_log_onehot)s r$logits generatorcCs<tj|j|j|d}tt|d d }||}|S)z( Apply gumbel noise to `logits` )devicer&r)rrandshaper'r)r%r&uniform gumbel_noisenoisedrrr gumbel_noised>sr-wJ??̔>num_diffusion_timestepscCshtd||d|||}tdg|f}|dd|dd}t|dddgf}||fS)zN Cumulative and non-cumulative alpha schedules. See section 4.1. rr Nnparange concatenate)r0alpha_cum_start alpha_cum_endattatrrralpha_schedulesHsr:cCsxtd||d|||}tdg|f}d|}|dd|dd}d|}t|dddgf}||fS)zN Cumulative and non-cumulative gamma schedules. See section 4.1. rr Nr1r2)r0gamma_cum_start gamma_cum_endctt one_minus_ctt one_minus_ctctrrrgamma_schedulesXsrAc@seZdZdZdZe     d&dededed ed ed ef d d Zd'dede e e j ffddZ  d(de jde jde jdee jdede eeff ddZddZde jde jd e jd!efd"d#Zd$d%ZdS))VQDiffusionSchedulera A scheduler for vector quantized diffusion. 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_vec_classes (`int`): The number of classes of the vector embeddings of the latent pixels. Includes the class for the masked latent pixel. num_train_timesteps (`int`, defaults to 100): The number of diffusion steps to train the model. alpha_cum_start (`float`, defaults to 0.99999): The starting cumulative alpha value. alpha_cum_end (`float`, defaults to 0.00009): The ending cumulative alpha value. gamma_cum_start (`float`, defaults to 0.00009): The starting cumulative gamma value. gamma_cum_end (`float`, defaults to 0.99999): The ending cumulative gamma value. r dr.r/num_vec_classesnum_train_timestepsr6r7r;r<cCsd||_|jd|_t|||d\}}t|||d\} } |jd} d|| | } d|| | } t|d}t| d} t| d} t|}t| }t| }t|d}t| d} t| d} t|}t| }t| }||_ ||_ ||_ ||_ ||_ ||_d|_ttd|ddd|_dS)Nr )r6r7)r;r<float64rr1) num_embed mask_classr:rArtensorastyperr log_atlog_btlog_ctlog_cumprod_atlog_cumprod_btlog_cumprod_ctnum_inference_steps from_numpyr3r4copy timesteps)selfrDrEr6r7r;r<r9r8r@r=num_non_mask_classesbtbttrKrLrMrNrOrPrrr__init__s6              &zVQDiffusionScheduler.__init__NrQr'cCs||_td|jddd}t|||_|j||_|j ||_ |j ||_ |j ||_ |j ||_ |j ||_ dS)a Sets the discrete 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. device (`str` or `torch.device`, *optional*): The device to which the timesteps and diffusion process parameters (alpha, beta, gamma) should be moved to. rNr1)rQr3r4rSrrRtorTrKrLrMrNrOrP)rUrQr'rTrrr set_timestepss z"VQDiffusionScheduler.set_timestepsT model_outputtimestepsampler& return_dictrcCsF|dkr|}n||||}t||}|jdd}|s|fSt|dS)a Predict the sample from the previous timestep by the reverse transition distribution. See [`~VQDiffusionScheduler.q_posterior`] for more details about how the distribution is computer. Args: log_p_x_0: (`torch.Tensor` of shape `(batch size, num classes - 1, num latent pixels)`): The log probabilities for the predicted classes of the initial latent pixels. Does not include a prediction for the masked class as the initial unnoised image cannot be masked. t (`torch.long`): The timestep that determines which transition matrices are used. x_t (`torch.LongTensor` of shape `(batch size, num latent pixels)`): The classes of each latent pixel at time `t`. generator (`torch.Generator`, or `None`): A random number generator for the noise applied to `p(x_{t-1} | x_t)` before it is sampled from. return_dict (`bool`, *optional*, defaults to `True`): Whether or not to return a [`~schedulers.scheduling_vq_diffusion.VQDiffusionSchedulerOutput`] or `tuple`. Returns: [`~schedulers.scheduling_vq_diffusion.VQDiffusionSchedulerOutput`] or `tuple`: If return_dict is `True`, [`~schedulers.scheduling_vq_diffusion.VQDiffusionSchedulerOutput`] is returned, otherwise a tuple is returned where the first element is the sample tensor. rr dim)r ) q_posteriorr-argmaxr )rUr\r]r^r&r_log_p_x_t_min_1 x_t_min_1rrrsteps   zVQDiffusionScheduler.stepc Cspt||j}|j|||dd}|j|||dd}||}tj|ddd}||}|||d}|||} | S)a Calculates the log probabilities for the predicted classes of the image at timestep `t-1`: ``` p(x_{t-1} | x_t) = sum( q(x_t | x_{t-1}) * q(x_{t-1} | x_0) * p(x_0) / q(x_t | x_0) ) ``` Args: log_p_x_0 (`torch.Tensor` of shape `(batch size, num classes - 1, num latent pixels)`): The log probabilities for the predicted classes of the initial latent pixels. Does not include a prediction for the masked class as the initial unnoised image cannot be masked. x_t (`torch.LongTensor` of shape `(batch size, num latent pixels)`): The classes of each latent pixel at time `t`. t (`torch.Long`): The timestep that determines which transition matrix is used. Returns: `torch.Tensor` of shape `(batch size, num classes, num latent pixels)`: The log probabilities for the predicted classes of the image at timestep `t-1`. T)tx_tlog_onehot_x_t cumulativeFr )rakeepdim)r$rG$log_Q_t_transitioning_to_known_classr logsumexpapply_cumulative_transitions) rU log_p_x_0rhrgrilog_q_x_t_given_x_0log_q_t_given_x_t_min_1q q_log_sum_exprdrrrrbs   .z VQDiffusionScheduler.q_posteriorrgrhrirjc Cs|r|j|}|j|}|j|}n|j|}|j|}|j|}|s1|dddddfd}|ddddddf}|||} ||jk} | d d|j dd} || | <|sft j | |fdd} | S)a/ Calculates the log probabilities of the rows from the (cumulative or non-cumulative) transition matrix for each latent pixel in `x_t`. Args: t (`torch.Long`): The timestep that determines which transition matrix is used. x_t (`torch.LongTensor` of shape `(batch size, num latent pixels)`): The classes of each latent pixel at time `t`. log_onehot_x_t (`torch.Tensor` of shape `(batch size, num classes, num latent pixels)`): The log one-hot vectors of `x_t`. cumulative (`bool`): If cumulative is `False`, the single step transition matrix `t-1`->`t` is used. If cumulative is `True`, the cumulative transition matrix `0`->`t` is used. Returns: `torch.Tensor` of shape `(batch size, num classes - 1, num latent pixels)`: Each _column_ of the returned matrix is a _row_ of log probabilities of the complete probability transition matrix. When non cumulative, returns `self.num_classes - 1` rows because the initial latent pixel cannot be masked. Where: - `q_n` is the probability distribution for the forward process of the `n`th latent pixel. - C_0 is a class of a latent pixel embedding - C_k is the class of the masked latent pixel non-cumulative result (omitting logarithms): ``` q_0(x_t | x_{t-1} = C_0) ... q_n(x_t | x_{t-1} = C_0) . . . . . . . . . q_0(x_t | x_{t-1} = C_k) ... q_n(x_t | x_{t-1} = C_k) ``` cumulative result (omitting logarithms): ``` q_0_cumulative(x_t | x_0 = C_0) ... q_n_cumulative(x_t | x_0 = C_0) . . . . . . . . . q_0_cumulative(x_t | x_0 = C_{k-1}) ... q_n_cumulative(x_t | x_0 = C_{k-1}) ``` Nr1r r`) rNrOrPrKrLrM unsqueeze logaddexprHexpandrGrcat) rUrgrhrirjabc(log_onehot_x_t_transitioning_from_maskedlog_Q_tmask_class_maskrrrrlds"1        z9VQDiffusionScheduler.log_Q_t_transitioning_to_known_classcCsd|jd}|j|}|j|}|j|}|jd}||d|}|||}tj||fdd}|S)Nrrr r`)r)rNrOrPrvrurrw)rUrrrgbszrxryrznum_latent_pixelsrrrrns     z1VQDiffusionScheduler.apply_cumulative_transitions)rCr.r/r/r.)N)NT)rrrrorderrintr rYrstrrr'r[Tensorlongrr Generatorboolr rrfrbrlrnrrrrrBjsb .  -o crB)r.r/)r/r.) dataclassesrtypingrrrnumpyr3rtorch.nn.functionalnn functionalrconfiguration_utilsrrutilsr scheduling_utilsr r rrrr$rr-r:rArBrrrrs