o ۷i% @sddlZddlmZddlmZddlZddlmZmZddl m Z ddl m Z dd ej d ejdBd ej fd d Z  ddej dej ded ejdBd ej f ddZeGddde ZGddde eZdS)N) dataclass)Literal) ConfigMixinregister_to_config) BaseOutput)SchedulerMixint generatorreturncCsV|dur|jn|j}tj||djdd|d|j}tt|d d S)a Generate Gumbel noise for sampling. Args: t (`torch.Tensor`): Input tensor to match the shape and dtype of the output noise. generator (`torch.Generator`, *optional*): A random number generator for reproducible sampling. Returns: `torch.Tensor`: Gumbel-distributed noise with the same shape, dtype, and device as the input tensor. Ndevicerrr #B ;)rtorch zeros_likeuniform_tologclamp)r r rnoiser\/home/ubuntu/vllm_env/lib/python3.10/site-packages/diffusers/schedulers/scheduling_amused.py gumbel_noise s" r?mask_lenprobs temperaturecCsNt|d|t||d}tj|ddj}t|d|}||k}|S)a Mask tokens by selecting the top-k lowest confidence scores with temperature-based randomness. Args: mask_len (`torch.Tensor`): Number of tokens to mask per sample in the batch. probs (`torch.Tensor`): Probability scores for each token. temperature (`float`, *optional*, defaults to 1.0): Temperature parameter for controlling randomness in the masking process. generator (`torch.Generator`, *optional*): A random number generator for reproducible sampling. Returns: `torch.Tensor`: Boolean mask indicating which tokens should be masked. rrdimr)rrrrsortvaluesgatherlong)rrrr confidencesorted_confidencecut_offmaskingrrrmask_by_random_topks r*c@s.eZdZUdZejed<dZejdBed<dS)AmusedSchedulerOutputa Output class for the scheduler's `step` function output. Args: prev_sample (`torch.LongTensor` of shape `(batch_size, height, width)` or `(batch_size, sequence_length)`): Computed sample `(x_{t-1})` of previous timestep with token IDs. `prev_sample` should be used as next model input in the denoising loop. pred_original_sample (`torch.LongTensor` of shape `(batch_size, height, width)` or `(batch_size, sequence_length)`, *optional*): The predicted fully denoised sample `(x_{0})` with token IDs based on the model output from the current timestep. `pred_original_sample` can be used to preview progress or for guidance. prev_sampleNpred_original_sample) __name__ __module__ __qualname____doc__rTensor__annotations__r- Generatorrrrrr+=s  r+c@seZdZUdZdZejdBed<ejdBed<e dde de d fd d Z d d e de e e e fBe e BdeejBfddZ   d!dejde dejdedejdBdedee BfddZ d"dejde dejdBdejfddZdS)#AmusedSchedulera A scheduler for masked token generation as used in [`AmusedPipeline`]. This scheduler iteratively unmasks tokens based on their confidence scores, following either a cosine or linear schedule. Unlike traditional diffusion schedulers that work with continuous pixel values, this scheduler operates on discrete token IDs, making it suitable for autoregressive and non-autoregressive masked token generation models. This scheduler 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: mask_token_id (`int`): The token ID used to represent masked tokens in the sequence. masking_schedule (`Literal["cosine", "linear"]`, *optional*, defaults to `"cosine"`): The schedule type for determining the mask ratio at each timestep. Can be either `"cosine"` or `"linear"`. rN temperatures timestepscosine mask_token_idmasking_schedule)r8linearcCsd|_d|_dSN)r6r7)selfr9r:rrr__init__fs zAmusedScheduler.__init__rrnum_inference_stepsrrcCs\tj||dd|_t|ttfr"tj|d|d||d|_dStj|d||d|_dS)Nr rrg{Gz?) rarangeflipr7 isinstancetuplelistlinspacer6)r=r@rrrrr set_timestepsos zAmusedScheduler.set_timestepsrT model_outputtimestepsamplestarting_mask_ratior return_dictr cCst|jdko |jdk}|r)|j\}} } } ||| | }||| | | ddd}||jjk} |jdd} | j}|durB| |jn| }|jj dkrT|j t j krT| }|d| d}t j|d|d j|d }|dddfj| jdd}t | ||}|dkr|}n|jd}|j|k}|dt|j}|jjd krt |tjd}n|jjd krd|}n td |jj||}||}t | jdddd|}t t jdg|jd |}t | d|dddddfdddddf}t | |t |j j}t!|||j"||}t ||jj|}|r.||| | }||| | }|s5||fSt#||S)Nrrrrr cpurr r8r;unknown masking schedule T)r!keepdim)$ndimshapereshapepermuteconfigr9softmaxrrtypedtyperfloat32floatsize multinomialviewwherer7nonzerolenr:cosmathpi ValueErrorfloorminsummaxtensorr$finfor*r6r+)r=rHrIrJrKr rL two_dim_input batch_size codebook_sizeheightwidth unknown_maprrprobs_r-r,seq_lenstep_idxratio mask_ratiorselected_probsr)rrrstep|sN         2 zAmusedScheduler.stepc Cs|j|k}|dt|j}|jjdkr!t|tjd}n|jjdkr,d|}n t d|jjtj |j |dur@|j n|j |d |j |k}|}|jj||<|S)a Add noise to a sample by randomly masking tokens according to the masking schedule. Args: sample (`torch.LongTensor`): The input sample containing token IDs to be partially masked. timesteps (`int`): The timestep that determines how much masking to apply. Higher timesteps result in more masking. generator (`torch.Generator`, *optional*): A random number generator for reproducible masking. Returns: `torch.LongTensor`: The sample with some tokens replaced by `mask_token_id` according to the masking schedule. rr8rr;rPN)rr )r7r`rarVr:rrbrcrdrerandrSrrcloner9) r=rJr7r rtrurv mask_indices masked_samplerrr add_noises"    zAmusedScheduler.add_noise)r8)r?N)rNTr<)r.r/r0r1orderrr4r3rintrr>rDrEstrrrGr2 LongTensorr[boolr+rxr}rrrrr5Osb     Ir5r<)rN)rc dataclassesrtypingrrconfiguration_utilsrrutilsrscheduling_utilsr r2r4rr[r*r+r5rrrrs0    "