o %ݫi(T@sdZddlmZddlZddlmZddlmZddlm Z ddl m Z ddl m Z dd lmZGd d d ejZd Zd ZdZdZGdddeZGdddejZddZGdddejZGdddejZeedZedgdZedddgZdS) zAn implementation of Denoising Diffusion https://arxiv.org/pdf/2006.11239.pdf Certain parts adopted from / inspired by denoising-diffusion-pytorch https://github.com/lucidrains/denoising-diffusion-pytorch Authors * Artem Ploujnikov 2022 ) namedtupleN)nn) functional)tqdm)length_to_mask) data_utils) unsqueeze_ascsHeZdZdZd fdd Zd ddZdddZd d Zd d d ZZ S)DiffuseraA base diffusion implementation Arguments --------- model: nn.Module the underlying model timesteps: int the number of timesteps noise: callable|str the noise function/module to use The following predefined types of noise are provided "gaussian": Gaussian noise, applied to the whole sample "length_masked_gaussian": Gaussian noise applied only to the parts of the sample that is not padding NcsFt||_||_|durd}t|trt||_dS||_dS)Ngaussian)super__init__model timesteps isinstancestr_NOISE_FUNCTIONSnoise)selfr rr __class__N/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/nnet/diffusion.pyr *s   zDiffuser.__init__cCt)aAdds noise to a batch of data Arguments --------- x: torch.Tensor the original data sample timesteps: torch.Tensor a 1-D integer tensor of a length equal to the number of batches in x, where each entry corresponds to the timestep number for the batch. If omitted, timesteps will be randomly sampled NotImplementedErrorrxrrrrdistort5s zDiffuser.distortcKsl|dur t||j}|j|fd|i|\}}|dur&|j||fi|}n |j|||fi|}|||fS)aCreates a sample for the training loop with a corresponding target Arguments --------- x: torch.Tensor the original data sample timesteps: torch.Tensor a 1-D integer tensor of a length equal to the number of batches in x, where each entry corresponds to the timestep number for the batch. If omitted, timesteps will be randomly sampled condition: torch.Tensor the condition used for conditional generation Should be omitted during unconditional generation **kwargs: dict Arguments to forward to the underlying model. Returns ------- pred: torch.Tensor the model output 0 predicted noise noise: torch.Tensor the noise being applied noisy_sample: torch.Tensor the sample with the noise applied Nr)sample_timestepsrrr )rrr conditionkwargs noisy_samplerpredrrr train_sampleDs  zDiffuser.train_samplecKr)aGenerates the number of samples indicated by the count parameter Arguments --------- shape: enumerable the shape of the sample to generate **kwargs: dict Arguments to forward to the underlying model. r)rshaper rrrsampleks zDiffuser.samplecCs |||S)z*Computes the forward pass, calls distort())rrrrrforwardxs zDiffuser.forwardNNN) __name__ __module__ __qualname____doc__r rr#r%r& __classcell__rrrrr s ' r g-C6?g{Gz?izDiffusion SamplingcsbeZdZdZ       dfdd ZddZddd Zed d Z ed d Z Z S)DenoisingDiffusionaAn implementation of a classic Denoising Diffusion Probabilistic Model (DDPM) Arguments --------- model: nn.Module the underlying model timesteps: int the number of timesteps noise: str|nn.Module the type of noise being used "gaussian" will produce standard Gaussian noise beta_start: float the value of the "beta" parameter at the beginning at the end of the process (see the paper) beta_end: float the value of the "beta" parameter at the end of the process sample_min: float sample_max: float Used to clip the output. show_progress: bool whether to show progress during inference Example ------- >>> from speechbrain.nnet.unet import UNetModel >>> unet = UNetModel( ... in_channels=1, ... model_channels=16, ... norm_num_groups=4, ... out_channels=1, ... num_res_blocks=1, ... attention_resolutions=[] ... ) >>> diff = DenoisingDiffusion( ... model=unet, ... timesteps=5 ... ) >>> x = torch.randn(4, 1, 64, 64) >>> pred, noise, noisy_sample = diff.train_sample(x) >>> pred.shape torch.Size([4, 1, 64, 64]) >>> noise.shape torch.Size([4, 1, 64, 64]) >>> noisy_sample.shape torch.Size([4, 1, 64, 64]) >>> sample = diff.sample((2, 1, 64, 64)) >>> sample.shape torch.Size([2, 1, 64, 64]) NFc s|durt}tj|||d|dus|dur+t|} |dur#| t}|dur+| t}||_||_|\} } |d| |d| |j j dd} |d| t | } t d| }|d| |d |t j| dd d dd }| d|d| }|d ||d|| t |d| }d|t | d| }|d||d|d|  }|d|d| d }|d|||_||_||_dS)N)rralphasbetasr)dimalphas_cumprod?signal_coefficientsnoise_coefficients)r)valueposterior_varianceposterior_log_varianceposterior_mean_weight_startposterior_mean_weight_stepsample_pred_model_coefficientr7sample_pred_noise_coefficient)DDPM_REF_TIMESTEPSr r DDPM_DEFAULT_BETA_STARTDDPM_DEFAULT_BETA_END beta_startbeta_endcompute_coefficientsregister_bufferr/cumprodtorchsqrtFpadlog sample_min sample_max show_progress)rr rrrBrCrLrMrNscaler/r0r2r4r5alphas_cumprod_prevr9r;r<r=r>rrrr sd           zDenoisingDiffusion.__init__cCs$t|j|j|j}d|}||fS)z2Computes diffusion coefficients (alphas and betas)r3)rGlinspacerBrCr)rr0r/rrrrDsz'DenoisingDiffusion.compute_coefficientscKsf|dur t||j}|dur|j|fi|}|j|}|j|}t|||t|||}||fS)aAdds noise to the sample, in a forward diffusion process, Arguments --------- x: torch.Tensor a data sample of 2 or more dimensions, with the first dimension representing the batch noise: torch.Tensor the noise to add timesteps: torch.Tensor a 1-D integer tensor of a length equal to the number of batches in x, where each entry corresponds to the timestep number for the batch. If omitted, timesteps will be randomly sampled **kwargs: dict Arguments to forward to the underlying model. Returns ------- result: torch.Tensor a tensor of the same dimension as x N)rrrr4r5r)rrrrr r4r5r!rrrrs     zDenoisingDiffusion.distortcKs~|tj|d|jji}tt|j}|jrt |t |jd}|D]}tj |dtj |jjd|}|j ||fi|}q!|S)asGenerates the number of samples indicated by the count parameter Arguments --------- shape: enumerable the shape of the sample to generate **kwargs: dict Arguments to forward to the underlying model. Returns ------- result: torch.Tensor the generated sample(s) device)desctotalr)dtyperR)rrGzerosr/rRreversedrangerrNr DESC_SAMPLINGoneslong sample_step)rr$r r%stepstimestep_numbertimesteprrrr%szDenoisingDiffusion.samplec Ks|j||fi|}||}t|j|||t|j|||}t|j||}t|j||}||||} t|j||} | d| |} |j dusW|j dur`| j |j |j d| S)aProcesses a single timestep for the sampling process Arguments --------- sample: torch.Tensor the sample for the following timestep timestep: int the timestep number **kwargs: dict Arguments to forward to the underlying model. Returns ------- predicted_sample: torch.Tensor the predicted sample (denoised by one step`) g?N)minmax) r rrr=r>r;r<r:exprLrMclip_) rr%r_r model_outr sample_start weight_start weight_stepmean log_variancepredicted_samplerrrr\>s4     zDenoisingDiffusion.sample_step)NNNNNNFr() r)r*r+r,r rDrrGno_gradr%r\r-rrrrr.s 5@ # r.csNeZdZdZ  dfdd ZddZdd Zd d Zd d ZddZ Z S)LatentDiffusiona A latent diffusion wrapper. Latent diffusion is denoising diffusion applied to a latent space instead of the original data space Arguments --------- autoencoder: speechbrain.nnet.autoencoders.Autoencoder An autoencoder converting the original space to a latent space diffusion: speechbrain.nnet.diffusion.Diffuser A diffusion wrapper latent_downsample_factor: int The factor that latent space dimensions need to be divisible by. This is useful if the underlying model for the diffusion wrapper is based on a UNet-like architecture where the inputs are progressively downsampled and upsampled by factors of two latent_pad_dim: int|list[int] the dimension(s) along which the latent space will be padded Example ------- >>> import torch >>> from torch import nn >>> from speechbrain.nnet.CNN import Conv2d >>> from speechbrain.nnet.autoencoders import NormalizingAutoencoder >>> from speechbrain.nnet.unet import UNetModel Set up a simple autoencoder (a real autoencoder would be a deep neural network) >>> ae_enc = Conv2d( ... kernel_size=3, ... stride=4, ... in_channels=1, ... out_channels=1, ... skip_transpose=True, ... ) >>> ae_dec = nn.ConvTranspose2d( ... kernel_size=3, ... stride=4, ... in_channels=1, ... out_channels=1, ... output_padding=1 ... ) >>> ae = NormalizingAutoencoder( ... encoder=ae_enc, ... decoder=ae_dec, ... ) Construct a diffusion model with a UNet architecture >>> unet = UNetModel( ... in_channels=1, ... model_channels=16, ... norm_num_groups=4, ... out_channels=1, ... num_res_blocks=1, ... attention_resolutions=[] ... ) >>> diff = DenoisingDiffusion( ... model=unet, ... timesteps=5 ... ) >>> latent_diff = LatentDiffusion( ... autoencoder=ae, ... diffusion=diff, ... latent_downsample_factor=4, ... latent_pad_dim=2 ... ) >>> x = torch.randn(4, 1, 64, 64) >>> latent_sample = latent_diff.train_sample_latent(x) >>> diff_sample, ae_sample = latent_sample >>> pred, noise, noisy_sample = diff_sample >>> pred.shape torch.Size([4, 1, 16, 16]) >>> noise.shape torch.Size([4, 1, 16, 16]) >>> noisy_sample.shape torch.Size([4, 1, 16, 16]) >>> ae_sample.latent.shape torch.Size([4, 1, 16, 16]) Create a few samples (the shape given should be the shape of the latent space) >>> sample = latent_diff.sample((2, 1, 16, 16)) >>> sample.shape torch.Size([2, 1, 64, 64]) Nr7cs6t||_||_||_t|tr|g}||_dSr')r r autoencoder diffusionlatent_downsample_factorrintlatent_pad_dim)rrmrnrorqrrrr s   zLatentDiffusion.__init__cKs*|j|}||}|jj|fi|S)aCreates a sample for the training loop with a corresponding target Arguments --------- x: torch.Tensor the original data sample **kwargs: dict Arguments to forward to the underlying model. Returns ------- pred: torch.Tensor the model output 0 predicted noise noise: torch.Tensor the noise being applied noisy_sample the sample with the noise applied )rmencode _pad_latentrnr#)rrr latentrrrr#s  zLatentDiffusion.train_samplecCs:|jdur|jdkr|jD] }tj||j|d\}}q |S)aPads the latent space to the desired dimension Arguments --------- latent: torch.Tensor the latent representation Returns ------- result: torch.Tensor the latent representation, with padding Nr7)factorlen_dim)rorqr pad_divisible)rrtr1_rrrrss    zLatentDiffusion._pad_latentc Ks^|d}|d}|d}|jj||||d}||j}|jj|fi|}t||dS)aReturns a train sample with autoencoder output - can be used to jointly training the diffusion model and the autoencoder Arguments --------- x: torch.Tensor the original data sample **kwargs: dict Arguments to forward to the underlying model. Returns ------- LatentDiffusionTrainSample Training sample. lengthout_mask_valuelatent_mask_value)ryrzr{)rnrm)getrmr#rsrtrnLatentDiffusionTrainSample) rrr ryrzr{autoencoder_outrtdiffusion_train_samplerrrtrain_sample_latents    z#LatentDiffusion.train_sample_latentcCs|j|}|j|S)a\Adds noise to the sample, in a forward diffusion process, Arguments --------- x: torch.Tensor a data sample of 2 or more dimensions, with the first dimension representing the batch Returns ------- result: torch.Tensor a tensor of the same dimension as x )rmrrrnr)rrrtrrrr&s  zLatentDiffusion.distortcCs"|j|}||}|j|S)zObtains a sample out of the diffusion model Arguments --------- shape: torch.Tensor Returns ------- sample: torch.Tensor the sample of the specified shape )rnr%rsrmdecode)rr$rtrrrr%8s  zLatentDiffusion.sample)Nr7) r)r*r+r,r r#rsrrr%r-rrrrrlks] rlcCstj||df|jdS)a'Returns a random sample of timesteps as a 1-D tensor (one dimension only) Arguments --------- x: torch.Tensor a tensor of samples of any dimension num_timesteps: int the total number of timesteps Returns ------- Random sample of timestamps. r)rR)rGrandintsizerR)r num_timestepsrrrrJsrc@seZdZdZddZdS) GaussianNoisezAdds ordinary Gaussian noisecKs t|S)zForward pass Arguments --------- sample: the original sample **kwargs: dict Arguments to forward to the underlying model. Returns ------- Noise in shape of sample. )rG randn_like)rr%r rrrr&_s zGaussianNoise.forwardN)r)r*r+r,r&rrrrr\s rcs4eZdZdZd fdd Zd ddZdd ZZS) LengthMaskedGaussianNoisezGaussian noise applied to padded samples. No noise is added to positions that are part of padding Arguments --------- length_dim: int The time dimension for which lengths apply. r7cst||_dSr')r r length_dim)rrrrrr ys  z"LengthMaskedGaussianNoise.__init__NcKsXt|}|dur*||j}t|||}|||}||}||d|S)aCreates Gaussian noise. If a tensor of lengths is provided, no noise is added to the padding positions. Arguments --------- sample: torch.Tensor a batch of data length: torch.Tensor relative lengths **kwargs: dict Arguments to forward to the underlying model. Returns ------- Gaussian noise in shape of sample. Ng) rGrrrrbool_compute_mask_shapeview masked_fill_)rr%ryr rmax_lenmask mask_shaperrrr&}s    z!LengthMaskedGaussianNoise.forwardcCs0|jdfd|jd|fd|dS)Nrr7r7)r$rr1)rrrrrrrs  z-LengthMaskedGaussianNoise._compute_mask_shaperr')r)r*r+r,r r&rr-rrrrros  r)r length_masked_gaussianDiffusionTrainSample)r"rr!r}rnrm)r, collectionsrrGrtorch.nnrrI tqdm.autorspeechbrain.dataio.dataiorspeechbrain.utilsrspeechbrain.utils.data_utilsrModuler r@rAr?rYr.rlrrrrrr}rrrrs<       ei`2