biYTddlmZddlmZmZmZddlZddlZddl m cm Z ddl mZmZddlmZddlmZeGd d eZd ej*d ed ej.fdZdej.deej2d ej.fdZddefdZddefdZGddeeZy)) dataclass)OptionalTupleUnionN) ConfigMixinregister_to_config) BaseOutput)SchedulerMixinc0eZdZUdZej ed<y)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__g/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/diffusers/schedulers/scheduling_vq_diffusion.pyrrs!!!rrx num_classesreturnctj||}|jddd}tj|j j d}|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_xs rindex_to_log_onehotr()sPyyK(H1a(H IIhnn&,,,7 8E Lrlogits generatorctj|j|j|}tjtj|dz dz }||z}|S)z( Apply gumbel noise to `logits` )devicer*r)rrandshaper,r#)r)r*uniform gumbel_noisenoiseds r gumbel_noisedr2>sSjjfmmyQGIIuyy599EABBL F "F Mrnum_diffusion_timestepsctjd||dz z ||z z|z}tjdg|f}|dd|ddz }tj|dddgf}||fS)zN Cumulative and non-cumulative alpha schedules. See section 4.1. rr Nnparange concatenate)r3alpha_cum_start alpha_cum_endattats ralpha_schedulesr>Hs !,-1H11LMQ^apQpq   ..1#s $C QR3s8 B ..#ab'A3 (C s7Nrctjd||dz z ||z z|z}tjdg|f}d|z }|dd|ddz }d|z }tj|dddgf}||fS)zN Cumulative and non-cumulative gamma schedules. See section 4.1. rr Nr5r6)r3gamma_cum_start gamma_cum_endctt one_minus_ctt one_minus_ctcts rgamma_schedulesrFXs !,-1H11LMQ^apQpq   ..1#s $CGM $}Sb'99L \ B ..#ab'A3 (C s7Nrc`eZdZdZdZe ddedededededef d Zdd ed e e e jffd Z dde jde jde j dee j$dede eeff dZdZde j de j de jdefdZdZy )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 num_vec_classesnum_train_timestepsr:r;r@rAc||_|jdz |_t|||\}}t|||\} } |jdz } d|z | z | z } d|z | z | z } t j |j d}t j | j d} t j | j d} t j|}t j| }t j| }t j |j d}t j | j d} t j | j d} t j|}t j| }t j| }|j|_ |j|_ |j|_ |j|_ |j|_ |j|_d|_t j t#j$d|dddj'|_y)Nr )r:r;)r@rAfloat64rr5) num_embed mask_classr>rFrtensorastyper#r$log_atlog_btlog_ctlog_cumprod_atlog_cumprod_btlog_cumprod_ctnum_inference_steps from_numpyr7r8copy timesteps)selfrIrJr:r;r@rAr=r<rErBnum_non_mask_classesbtbttrQrRrSrTrUrVs r__init__zVQDiffusionScheduler.__init__s)..1,!"5fstC!"5fstC#~~1"frk1 13w} 44 \\"))I. / \\"))I. / \\"))I. /222ll3::i01ll3::i01ll3::i01333lln lln lln ,224,224,224$( ))"))A7J*KDbD*Q*V*V*XYrNrWr,cL||_tjd|jdddj}t j |j ||_|jj ||_|jj ||_ |jj ||_ |jj ||_ |jj ||_ |jj ||_ y)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. rNr5)rWr7r8rYrrXtorZrQrRrSrTrUrV)r[rWr,rZs r set_timestepsz"VQDiffusionScheduler.set_timestepss$7 IIa!9!9:4R4@EEG )))477?kknnV, kknnV, kknnV, "1144V<"1144V<"1144V<r model_outputtimestepsampler* return_dictrc|dk(r|}n|j|||}t||}|jd}|s|fSt|S)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_posteriorr2argmaxr)r[rcrdrer*rflog_p_x_t_min_1 x_t_min_1s rstepzVQDiffusionScheduler.stepsY> q=*O"..|VXNO'C#**q*1 < )i@@rct||j}|j|||d}|j|||d}||z }tj|dd}||z }|j ||dz }||z|z} | 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 )rikeepdim)r(rM$log_Q_t_transitioning_to_known_classr logsumexpapply_cumulative_transitions) r[ log_p_x_0rqrprrlog_q_x_t_given_x_0log_q_t_given_x_t_min_1q q_log_sum_exprls rrjz VQDiffusionScheduler.q_posteriors*-S$..A"GGSDH #'"K"KSE#L#  + +q$?     - -aQ 755 E\rrprqrrrsc|r.|j|}|j|}|j|}n-|j|}|j|}|j |}|s|dddddfj d}|ddddddf}||zj|} ||jk(} | j djd|jdz d} || | <|stj| fd} | 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}) ``` Nr5r rh) rTrUrVrQrRrS unsqueeze logaddexprNexpandrMrcat) r[rprqrrrsabc(log_onehot_x_t_transitioning_from_maskedlog_Q_tmask_class_masks rruz9VQDiffusionScheduler.log_Q_t_transitioning_to_known_classdsb ##A&A##A&A##A&A AA AA AA8FaQh7O7Y7YZ[7\ 4 (3B3 2"A%0030)33A6==b$..STBTVXY#$ ii*R SYZ[Grc|jd}|j|}|j|}|j|}|jd}|j |d|}||zj |}t j||fd}|S)Nrrr rh)r.rTrUrVrrrr)r[r{rpbszrrrnum_latent_pixelss rrwz1VQDiffusionScheduler.apply_cumulative_transitionssggaj    "    "    "GGAJ HHS!. / U  a  IIq!f! $r)dwJ??̔>rr)N)NT)rrrrorderr intr$r_rstrrr,rbTensorlongrr GeneratorboolrrrnrjrurwrrrrHrHjsK, E$'!('!)&,Z,Z!,Z ,Z  ,Z  ,Z,Z,Z\==eCDU>V=604 +All+A**+A  +A EOO, +A  +A )50 1+AZm^aIIa$)$4$4aFKlla`daF rrH)rr)rr) dataclassesrtypingrrrnumpyr7rtorch.nn.functionalnn functionalr configuration_utilsrr utilsr scheduling_utilsr rrrrr(rr2r>rFrHrrrrs")) A,  " "  "5++#%,,*%,,8EOO3LQVQ]Q] S  S$i>;ir