+hY`SSKJr SSKJrJrJr SSKrSSKrSSK J s J r SSK JrJr SSKJr SSKJr \"S S \55rS \R*S \S \R.4SjrS\R.S\\R2S \R.4SjrSS\4SjjrSS\4Sjjr"SS\\5rg)) dataclass)OptionalTupleUnionN) ConfigMixinregister_to_config) BaseOutput)SchedulerMixinc8\rSrSr%Sr\R \S'Srg)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____firstlineno____doc__torch LongTensor__annotations____static_attributes__rf/home/james-whalen/.local/lib/python3.13/site-packages/diffusers/schedulers/scheduling_vq_diffusion.pyrrs!!!rrx num_classesreturnc[R"X5nURSSS5n[R"UR 5R SS95nU$)aw 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+)sNyy(H1a(H IIhnn&,,,7 8E Lrlogits generatorc[R"URURUS9n[R"[R"US-5*S-5*nX0-nU$)z Apply gumbel noise to `logits` )devicer-r!)rrandshaper/r&)r,r-uniform gumbel_noisenoiseds r gumbel_noisedr5>sQjjfmmyQGIIuyy599EABBL  "F Mrnum_diffusion_timestepsc[R"SU5US- - X!- -U-n[R"S/U45nUSSUSS- n[R"USSS/45nXC4$)zB Cumulative and non-cumulative alpha schedules. See section 4.1. rr Nnparange concatenate)r6alpha_cum_start alpha_cum_endattats ralpha_schedulesrAHs !,-1H11LMQ^Qpq   ..1#s $C QR3s8 B ..#ab'A3 (C 7Nrc[R"SU5US- - X!- -U-n[R"S/U45nSU- nUSSUSS- nSU- n[R"USSS/45nXc4$)zB Cumulative and non-cumulative gamma schedules. See section 4.1. rr Nr8r9)r6gamma_cum_start gamma_cum_endctt one_minus_ctt one_minus_ctcts rgamma_schedulesrIXs !,-1H11LMQ^Qpq   ..1#s $CGM $}Sb'99L \ B ..#ab'A3 (C 7Nrcv\rSrSrSrSr\SS\S\S\S\S\S \4 S jj5r SS \S \ \ \ R44SjjrS S\ RS\ R S\ R"S\\ R&S\S\ \\44 SjjrSrS\ RS\ R"S\ RS\4SjrSrSrg )!VQDiffusionSchedulerja 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>rCrDcXlURS- Ul[X#US9upx[X%US9upURS- n SU- U - U - n SU- U - U - n [R "UR S55n[R "U R S55n [R "U R S55n [R"U5n[R"U 5n[R"U 5n[R "UR S55n[R "U R S55n [R "U R S55n [R"U5n[R"U 5n[R"U 5nUR5Ul UR5Ul UR5Ul UR5Ul UR5Ul UR5UlSUl[R "["R$"SU5SSS2R'55Ulg)Nr )r=r>)rCrDfloat64rr8) num_embed mask_classrArIrtensorastyper&r'log_atlog_btlog_ctlog_cumprod_atlog_cumprod_btlog_cumprod_ctnum_inference_steps from_numpyr:r;copy timesteps)selfrMrNr=r>rCrDr@r?rHrEnum_non_mask_classesbtbttrUrVrWrXrYrZs r__init__VQDiffusionScheduler.__init__s)..1,!"5fst!"5fst#~~1"frk1 13w} 44 \\"))I. / \\"))I. / \\"))I. /222ll3::i01ll3::i01ll3::i01333lln lln lln ,224,224,224$( ))"))A7J*KDbD*Q*V*V*XYrNr[r/cNXl[R"SUR5SSS2R5n[R "U5R U5UlURR U5UlURR U5Ul URR U5Ul URR U5Ul URR U5Ul URR U5Ul g)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. rNr8)r[r:r;r]rr\tor^rUrVrWrXrYrZ)r_r[r/r^s r set_timesteps"VQDiffusionScheduler.set_timestepss$7 IIa!9!9:4R4@EEG )))477?kknnV, kknnV, kknnV, "1144V<"1144V<"1144V<r model_outputtimestepsampler- return_dictrcUS:XaUnOURXU5n[Xd5nURSS9nU(dU4$[US9$)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_posteriorr5argmaxr)r_rirjrkr-rllog_p_x_t_min_1 x_t_min_1s rstepVQDiffusionScheduler.stepsU> q=*O"..|XNO'C#**q*1 < )i@@rc[X R5nURX2USS9nURX2USS9nX- n[R"USSS9nXx- nUR XsS- 5nXv-U-n U $)ah 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 )rokeepdim)r+rQ$log_Q_t_transitioning_to_known_classr logsumexpapply_cumulative_transitions) r_ log_p_x_0rxrwrylog_q_x_t_given_x_0log_q_t_given_x_t_min_1q q_log_sum_exprrs rrp VQDiffusionScheduler.q_posteriors*-S..A"GGDH #'"K"KE#L#   +q$?     - -aQ 75 E\rrwrxryrzcU(a.URUnURUnURUnO-URUnURUnUR UnU(dUSS2SSS24R S5nUSS2SS2SS24nX5-RU5n X R:Hn U R S5RSURS- S5n XyU 'U(d[R"U W4SS9n U $)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}) ``` Nr8r rn) rXrYrZrUrVrW unsqueeze logaddexprRexpandrQrcat) r_rwrxryrzabc(log_onehot_x_t_transitioning_from_maskedlog_Q_tmask_class_masks rr|9VQDiffusionScheduler.log_Q_t_transitioning_to_known_classds b ##A&A##A&A##A&A AA AA AA8FaQh7O7Y7YZ[7\ 4 (3B3 2"%0030)33A6==b$..STBTVXY#$ ii*R SYZ[GrcURSnURUnURUnURUnURSnUR USU5nX-R U5n[ R"X4SS9nU$)Nrrr rn)r1rXrYrZrrrr)r_rrwbszrrrnum_latent_pixelss rr~1VQDiffusionScheduler.apply_cumulative_transitionssggaj    "    "    "GGAJ HHS!. / U  a  IIqf! $r) rUrVrWrXrYrZrRrQr[r^)dwJ??̔>rr)N)NT)rrrrrorderr intr'rcrstrrr/rgTensorlongrr Generatorboolrrrtrpr|r~rrrrrKrKjsK, 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 rrK)rr)rr) dataclassesrtypingrrrnumpyr:rtorch.nn.functionalnn functionalr#configuration_utilsrr utilsr scheduling_utilsr rrrrr+rr5rArIrKrrrrs")) A,  " "  "5++#%,,*%,,8EOO3LQVQ]Q] S  S$i>;ir