#!/usr/bin/env python3 # Code by Kat Crowson in k-diffusion repo, modified by Scott H Hawley (SHH) """Samples from k-diffusion models.""" import argparse from pathlib import Path import accelerate import safetensors.torch as safetorch import torch from tqdm import trange, tqdm from PIL import Image from torchvision import transforms import k_diffusion as K from pom.v_diffusion import DDPM, LogSchedule, CrashSchedule #CHORD_BORDER = 8 # chord border size in pixels from pom.chords import CHORD_BORDER, img_batch_to_seq_emb, ChordSeqEncoder # ---- my mangled sampler that includes repaint import torchsde class BatchedBrownianTree: """A wrapper around torchsde.BrownianTree that enables batches of entropy.""" def __init__(self, x, t0, t1, seed=None, **kwargs): t0, t1, self.sign = self.sort(t0, t1) w0 = kwargs.get('w0', torch.zeros_like(x)) if seed is None: seed = torch.randint(0, 2 ** 63 - 1, []).item() self.batched = True try: assert len(seed) == x.shape[0] w0 = w0[0] except TypeError: seed = [seed] self.batched = False self.trees = [torchsde.BrownianTree(t0, w0, t1, entropy=s, **kwargs) for s in seed] @staticmethod def sort(a, b): return (a, b, 1) if a < b else (b, a, -1) def __call__(self, t0, t1): t0, t1, sign = self.sort(t0, t1) w = torch.stack([tree(t0, t1) for tree in self.trees]) * (self.sign * sign) return w if self.batched else w[0] class BrownianTreeNoiseSampler: """A noise sampler backed by a torchsde.BrownianTree. Args: x (Tensor): The tensor whose shape, device and dtype to use to generate random samples. sigma_min (float): The low end of the valid interval. sigma_max (float): The high end of the valid interval. seed (int or List[int]): The random seed. If a list of seeds is supplied instead of a single integer, then the noise sampler will use one BrownianTree per batch item, each with its own seed. transform (callable): A function that maps sigma to the sampler's internal timestep. """ def __init__(self, x, sigma_min, sigma_max, seed=None, transform=lambda x: x): self.transform = transform t0, t1 = self.transform(torch.as_tensor(sigma_min)), self.transform(torch.as_tensor(sigma_max)) self.tree = BatchedBrownianTree(x, t0, t1, seed) def __call__(self, sigma, sigma_next): t0, t1 = self.transform(torch.as_tensor(sigma)), self.transform(torch.as_tensor(sigma_next)) return self.tree(t0, t1) / (t1 - t0).abs().sqrt() def append_dims(x, target_dims): """Appends dimensions to the end of a tensor until it has target_dims dimensions.""" dims_to_append = target_dims - x.ndim if dims_to_append < 0: raise ValueError(f'input has {x.ndim} dims but target_dims is {target_dims}, which is less') return x[(...,) + (None,) * dims_to_append] def to_d(x, sigma, denoised): """Converts a denoiser output to a Karras ODE derivative.""" return (x - denoised) / append_dims(sigma, x.ndim) @torch.no_grad() def my_sample_euler(model, x, sigmas, extra_args=None, callback=None, disable=None, s_churn=0., s_tmin=0., s_tmax=float('inf'), s_noise=1., repaint=1): """Implements Algorithm 2 (Euler steps) from Karras et al. (2022).""" extra_args = {} if extra_args is None else extra_args s_in = x.new_ones([x.shape[0]]) for i in trange(len(sigmas) - 1, disable=disable): for u in range(repaint): gamma = min(s_churn / (len(sigmas) - 1), 2 ** 0.5 - 1) if s_tmin <= sigmas[i] <= s_tmax else 0. eps = torch.randn_like(x) * s_noise sigma_hat = sigmas[i] * (gamma + 1) if gamma > 0: x = x + eps * (sigma_hat ** 2 - sigmas[i] ** 2) ** 0.5 denoised = model(x, sigma_hat * s_in, **extra_args) d = to_d(x, sigma_hat, denoised) if callback is not None: callback({'x': x, 'i': i, 'sigma': sigmas[i], 'sigma_hat': sigma_hat, 'denoised': denoised}) dt = sigmas[i + 1] - sigma_hat # Euler method x = x + d * dt if x.isnan().any(): assert False, f"x has NaNs, i = {i}, u = {u}, repaint = {repaint}" if u < repaint - 1: beta = (sigmas[i + 1] / sigmas[-1]) ** 2 x = torch.sqrt(1 - beta) * x + torch.sqrt(beta) * torch.randn_like(x) return x def get_scalings(sigma, sigma_data=0.5): c_skip = sigma_data ** 2 / (sigma ** 2 + sigma_data ** 2) c_out = sigma * sigma_data / (sigma ** 2 + sigma_data ** 2) ** 0.5 c_in = 1 / (sigma ** 2 + sigma_data ** 2) ** 0.5 return c_skip, c_out, c_in @torch.no_grad() def my_dpmpp_2m_sde(model, x, sigmas, extra_args=None, callback=None, disable=None, eta=1., s_noise=1., noise_sampler=None, solver_type='midpoint', repaint=4): """DPM-Solver++(2M) SDE. but with repaint added""" if solver_type not in {'heun', 'midpoint'}: raise ValueError('solver_type must be \'heun\' or \'midpoint\'') sigma_min, sigma_max = sigmas[sigmas > 0].min(), sigmas.max() noise_sampler = BrownianTreeNoiseSampler(x, sigma_min, sigma_max) if noise_sampler is None else noise_sampler extra_args = {} if extra_args is None else extra_args s_in = x.new_ones([x.shape[0]]) old_denoised = None h_last = None old_x = None for i in trange(len(sigmas) - 1, disable=disable): # time loop for u in range(repaint): denoised = model(x, sigmas[i] * s_in, **extra_args) if callback is not None: callback({'x': x, 'i': i, 'sigma': sigmas[i], 'sigma_hat': sigmas[i], 'denoised': denoised}) #print("i, u, sigmas[i], sigmas[i + 1] = ", i, u, sigmas[i], sigmas[i + 1]) if sigmas[i + 1] == 0: # Denoising step x = denoised else: # DPM-Solver++(2M) SDE t, s = -sigmas[i].log(), -sigmas[i + 1].log() h = s - t eta_h = eta * h x = sigmas[i + 1] / sigmas[i] * (-eta_h).exp() * x + (-h - eta_h).expm1().neg() * denoised if old_denoised is not None: r = h_last / h if solver_type == 'heun': x = x + ((-h - eta_h).expm1().neg() / (-h - eta_h) + 1) * (1 / r) * (denoised - old_denoised) elif solver_type == 'midpoint': x = x + 0.5 * (-h - eta_h).expm1().neg() * (1 / r) * (denoised - old_denoised) if eta: x = x + noise_sampler(sigmas[i], sigmas[i + 1]) * sigmas[i + 1] * (-2 * eta_h).expm1().neg().sqrt() * s_noise if callback is not None: callback({'x': x, 'i': i, 'sigma': sigmas[i], 'sigma_hat': sigmas[i], 'denoised': denoised}) if x.isnan().any(): assert False, f"x has NaNs, i = {i}, u = {u}, repaint = {repaint}" if u < repaint - 1: # RePaint: go "back" in integration via the "forward" process, by adding a little noise to x # ...but scaled properly! # But how to convert from original RePaint to k-diffusion? I'll try a few variants repaint_choice = 'orig' # ['orig','var1','var2', etc...] sigma_diff = (sigmas[i] - sigmas[i+1]).abs() sigma_ratio = ( sigmas[i+1] / sigma_max ) # use i+1 or i? if repaint_choice == 'orig': # attempt at original RePaint algorithm, which used betas # if sigmas are the std devs, then betas are variances? but beta_max = 1, so how to get that? ratio? beta = sigma_ratio**2 x = torch.sqrt(1-beta)*x + torch.sqrt(beta)*torch.randn_like(x) # this is from RePaint Paper elif repaint_choice == 'var1': # or maybe this...? # worse than orig x = x + sigma_diff*torch.randn_like(x) elif repaint_choice == 'var2': # or this...? # yields NaNs x = (1-sigma_diff)*x + sigma_diff*torch.randn_like(x) elif repaint_choice == 'var3': # results similar to var1 x = (1.0-sigma_ratio)*x + sigmas[i+1]*torch.randn_like(x) elif repaint_choice == 'var4': # NaNs # stealing code from elsewhere, no idea WTF I'm doing. #Invert this: target = (input - c_skip * noised_input) / c_out, where target = model_output x_tm1, x_t = x, old_x # x_tm1 = ( x_0 - c_skip * noised_x0 ) / c_out # So x_tm1*c_out = x_0 - c_skip * noised_x0 input, noise = x_tm1, torch.randn_like(x) noised_input = input + noise * append_dims(sigma_diff, input.ndim) c_skip, c_out, c_in = [append_dims(x, input.ndim) for x in get_scalings(sigmas[i])] model_output = x_tm1 renoised_x = c_out * model_output + c_skip * noised_input x = renoised_x elif repaint_choice == 'var5': x = torch.sqrt((1-(sigma_diff/sigma_max)**2))*x + sigma_diff*torch.randn_like(x) # include this? guessing no. #old_denoised = denoised #h_last = h old_denoised = denoised h_last = h old_x = x return x # -----from stable-audio-tools # Define the noise schedule and sampling loop def get_alphas_sigmas(t): """Returns the scaling factors for the clean image (alpha) and for the noise (sigma), given a timestep.""" return torch.cos(t * math.pi / 2), torch.sin(t * math.pi / 2) def alpha_sigma_to_t(alpha, sigma): """Returns a timestep, given the scaling factors for the clean image and for the noise.""" return torch.atan2(sigma, alpha) / math.pi * 2 def t_to_alpha_sigma(t): """Returns the scaling factors for the clean image and for the noise, given a timestep.""" return torch.cos(t * math.pi / 2), torch.sin(t * math.pi / 2) @torch.no_grad() def sample(model, x, steps, eta, **extra_args): """Draws samples from a model given starting noise. v-diffusion""" ts = x.new_ones([x.shape[0]]) # Create the noise schedule t = torch.linspace(1, 0, steps + 1)[:-1] alphas, sigmas = get_alphas_sigmas(t) # The sampling loop for i in trange(steps): # Get the model output (v, the predicted velocity) with torch.cuda.amp.autocast(): v = model(x, ts * t[i], **extra_args).float() # Predict the noise and the denoised image pred = x * alphas[i] - v * sigmas[i] eps = x * sigmas[i] + v * alphas[i] # If we are not on the last timestep, compute the noisy image for the # next timestep. if i < steps - 1: # If eta > 0, adjust the scaling factor for the predicted noise # downward according to the amount of additional noise to add ddim_sigma = eta * (sigmas[i + 1]**2 / sigmas[i]**2).sqrt() * \ (1 - alphas[i]**2 / alphas[i + 1]**2).sqrt() adjusted_sigma = (sigmas[i + 1]**2 - ddim_sigma**2).sqrt() # Recombine the predicted noise and predicted denoised image in the # correct proportions for the next step x = pred * alphas[i + 1] + eps * adjusted_sigma # Add the correct amount of fresh noise if eta: x += torch.randn_like(x) * ddim_sigma # If we are on the last timestep, output the denoised image return pred # Soft mask inpainting is just shrinking hard (binary) mask inpainting # Given a float-valued soft mask (values between 0 and 1), get the binary mask for this particular step def get_bmask(i, steps, mask): strength = (i+1)/(steps) # convert to binary mask bmask = torch.where(mask<=strength,1,0) return bmask def make_cond_model_fn(model, cond_fn): def cond_model_fn(x, sigma, **kwargs): with torch.enable_grad(): x = x.detach().requires_grad_() denoised = model(x, sigma, **kwargs) cond_grad = cond_fn(x, sigma, denoised=denoised, **kwargs).detach() cond_denoised = denoised.detach() + cond_grad * K.utils.append_dims(sigma**2, x.ndim) return cond_denoised return cond_model_fn # Uses k-diffusion from https://github.com/crowsonkb/k-diffusion # init_data is init_audio as latents (if this is latent diffusion) # For sampling, set both init_data and mask to None # For variations, set init_data # For inpainting, set both init_data & mask def sample_k( model_fn, noise, init_data=None, mask=None, steps=100, sampler_type="dpmpp-2m-sde", sigma_min=0.5, sigma_max=50, rho=1.0, device="cuda", callback=None, cond_fn=None, model_config=None, repaint=1, **extra_args ): #denoiser = K.external.VDenoiser(model_fn) denoiser = K.Denoiser(model_fn, sigma_data=model_config['sigma_data']) if cond_fn is not None: denoiser = make_cond_model_fn(denoiser, cond_fn) # Make the list of sigmas. Sigma values are scalars related to the amount of noise each denoising step has #sigmas = K.sampling.get_sigmas_polyexponential(steps, sigma_min, sigma_max, rho, device=device) sigmas = K.sampling.get_sigmas_karras(steps, sigma_min, sigma_max, rho=7., device=device) print("sigmas[0] = ", sigmas[0]) # Scale the initial noise by sigma noise = noise * sigmas[0] wrapped_callback = callback if mask is None and init_data is not None: # VARIATION (no inpainting) # set the initial latent to the init_data, and noise it with initial sigma x = init_data + noise elif mask is not None and init_data is not None: # INPAINTING bmask = get_bmask(0, steps, mask) # initial noising input_noised = init_data + noise # set the initial latent to a mix of init_data and noise, based on step 0's binary mask x = input_noised * bmask + noise * (1-bmask) # define the inpainting callback function (Note: side effects, it mutates x) # See https://github.com/crowsonkb/k-diffusion/blob/master/k_diffusion/sampling.py#L596C13-L596C105 # callback({'x': x, 'i': i, 'sigma': sigmas[i], 'sigma_hat': sigmas[i], 'denoised': denoised}) # This is called immediately after `denoised = model(x, sigmas[i] * s_in, **extra_args)` def inpainting_callback(args): i = args["i"] x = args["x"] sigma = args["sigma"] #denoised = args["denoised"] # noise the init_data input with this step's appropriate amount of noise input_noised = init_data + torch.randn_like(init_data) * sigma # shrinking hard mask bmask = get_bmask(i, steps, mask) # mix input_noise with x, using binary mask new_x = input_noised * bmask + x * (1-bmask) # mutate x x[:,:,:] = new_x[:,:,:] # wrap together the inpainting callback and the user-submitted callback. if callback is None: wrapped_callback = inpainting_callback else: wrapped_callback = lambda args: (inpainting_callback(args), callback(args)) else: # SAMPLING # set the initial latent to noise x = noise print("sample_k: x.min, x.max = ", x.min(), x.max()) print(f"sample_k: key, val.dtype = ",[ (key, val.dtype if val is not None else val) for key,val in extra_args.items()]) with torch.cuda.amp.autocast(): if sampler_type == "k-heun": return K.sampling.sample_heun(denoiser, x, sigmas, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "k-lms": return K.sampling.sample_lms(denoiser, x, sigmas, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "k-dpmpp-2s-ancestral": return K.sampling.sample_dpmpp_2s_ancestral(denoiser, x, sigmas, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "k-dpm-2": return K.sampling.sample_dpm_2(denoiser, x, sigmas, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "k-dpm-fast": return K.sampling.sample_dpm_fast(denoiser, x, sigma_min, sigma_max, steps, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "k-dpm-adaptive": return K.sampling.sample_dpm_adaptive(denoiser, x, sigma_min, sigma_max, rtol=0.01, atol=0.01, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "dpmpp-2m-sde": return K.sampling.sample_dpmpp_2m_sde(denoiser, x, sigmas, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "my-dpmpp-2m-sde": return my_dpmpp_2m_sde(denoiser, x, sigmas, disable=False, callback=wrapped_callback, repaint=repaint, extra_args=extra_args) elif sampler_type == "dpmpp-3m-sde": return K.sampling.sample_dpmpp_3m_sde(denoiser, x, sigmas, disable=False, callback=wrapped_callback, extra_args=extra_args) elif sampler_type == "my-sample-euler": return my_sample_euler(denoiser, x, sigmas, disable=False, callback=wrapped_callback, repaint=repaint, extra_args=extra_args) ## ---- end stable-audio-tools def infer_mask_from_init_img(img, mask_with='white'): """given an image with mask areas marked, extract the mask itself note, this works whether image is normalized on 0..1 or -1..1, but not 0..255""" print("Inferring mask from init_img") assert mask_with in ['blue','white'] if not torch.is_tensor(img): img = ToTensor()(img) mask = torch.zeros(img.shape[-2:]) if mask_with == 'white': mask[ (img[0,:,:]==1) & (img[1,:,:]==1) & (img[2,:,:]==1)] = 1 elif mask_with == 'blue': mask[img[2,:,:]==1] = 1 # blue return mask*1.0 def grow_mask(init_mask, grow_by=2): "adds a border of grow_by pixels to the mask, by growing it grow_by times. If grow_by=0, does nothing" new_mask = init_mask.clone() for c in range(grow_by): # wherever mask is bordered by a 1, set it to 1 new_mask[1:-1,1:-1] = (new_mask[1:-1,1:-1] + new_mask[0:-2,1:-1] + new_mask[2:,1:-1] + new_mask[1:-1,0:-2] + new_mask[1:-1,2:]) > 0 return new_mask def add_seeding(init_image, init_mask, grow_by=0, seed_scale=1.0): "adds extra noise inside mask" init_mask = grow_mask(init_mask, grow_by=grow_by) # make the mask bigger if not torch.is_tensor(init_image): init_image = ToTensor()(init_image) init_image = init_image.clone() # wherever mask is 1, set first set init_image to min value init_image[:,init_mask == 1] = init_image.min() init_image = init_image + seed_scale*torch.randn_like(init_image) * (init_mask) # add noise where mask is 1 # wherever the mask is 1, set the blue channel to -1.0, otherwise leave it alone init_image[2,:,:] = init_image[2,:,:] * (1-init_mask) - 1.0*init_mask return init_image def get_init_image_and_mask(args, device): convert_tensor = transforms.ToTensor() init_image = Image.open(args.init_image).convert('RGB') init_image = convert_tensor(init_image) #normalize image from 0..1 to -1..1 init_image = (2.0 * init_image) - 1.0 init_mask = torch.ones(init_image.shape[-2:]) # ones are where stuff will change, zeros will stay the same inpaint_task = 'infer' # infer mask from init_image assert inpaint_task in ['accomp','chords','melody','nucleation','notes','continue','infer'] if inpaint_task in ['melody','accomp']: init_mask[0:70,:] = 0 # zero out a melody strip of image near top init_mask[128+0:128+70,:] = 0 # zero out a melody strip of image along bottom row if inpaint_task == 'melody': init_mask = 1 - init_mask elif inpaint_task in ['notes','chords']: # keep chords only #init_mask = torch.ones_like(x) init_mask[0:CHORD_BORDER,:] = 0 # top row of 256x256 init_mask[128-CHORD_BORDER:128+CHORD_BORDER,:] = 0 # middle rows of 256x256 init_mask[-CHORD_BORDER:,:] = 0 # bottom row of 256x256 if inpaint_task == 'chords': init_mask = 1 - init_mask # inverse: genereate chords given notes elif inpaint_task == 'continue': init_mask[0:128,:] = 0 # remember it's a square, so just mask out the bottom half elif inpaint_task == 'nucleation': # set mask to wherever the blue channel is >= 0.9 init_mask = (init_image[2,:,:] > 0.0)*1.0 # zero out init mask in top and bottom borders init_mask[0:CHORD_BORDER,:] = 0 init_mask[-CHORD_BORDER:,:] = 0 init_mask[128-CHORD_BORDER:128+CHORD_BORDER,:] = 0 # remove all blue in init_image between the borders init_image[2,CHORD_BORDER:128-CHORD_BORDER,:] = -1.0 init_image[2,128+CHORD_BORDER:-CHORD_BORDER,:] = -1.0 # grow the sides of the mask by one pixel: # wherever mask is zero but is bordered by a 1, set it to 1 init_mask[1:-1,1:-1] = (init_mask[1:-1,1:-1] + init_mask[0:-2,1:-1] + init_mask[2:,1:-1] + init_mask[1:-1,0:-2] + init_mask[1:-1,2:]) > 0 #init_mask[1:-1,1:-1] = (init_mask[1:-1,1:-1] + init_mask[0:-2,1:-1] + init_mask[2:,1:-1] + init_mask[1:-1,0:-2] + init_mask[1:-1,2:]) > 0 elif inpaint_task == 'infer': init_mask = infer_mask_from_init_img(init_image, mask_with='white') # Also black out init_image wherever init mask is 1 init_image[:,init_mask == 1] = init_image.min() if args.seed_scale > 0: # driving nucleation print("Seeding nucleation, seed_scale = ", args.seed_scale) init_image = add_seeding(init_image, init_mask, grow_by=0, seed_scale=args.seed_scale) # remove any blue in middle of init image print("init_image.shape = ", init_image.shape) init_image[2,CHORD_BORDER:128-CHORD_BORDER,:] = -1.0 init_image[2,128+CHORD_BORDER:-CHORD_BORDER,:] = -1.0 # Debugging: output some images so we can see what's going on init_mask_t = init_mask.float()*255 # convert mask to 0..255 for writing as image # Convert to NumPy array and rearrange dimensions init_mask_img_numpy = init_mask_t.byte().cpu().numpy()#.transpose(1, 2, 0) init_mask_debug_img = Image.fromarray(init_mask_img_numpy) init_mask_debug_img.save("init_mask_debug.png") init_image_debug_img = Image.fromarray((init_image*127.5+127.5).byte().cpu().numpy().transpose(1,2,0)) init_image_debug_img.save("init_image_debug.png") # reshape image and mask to be 4D tensors init_image = init_image.unsqueeze(0).repeat(args.batch_size, 1, 1, 1) init_mask = init_mask.unsqueeze(0).unsqueeze(1).repeat(args.batch_size,3,1,1).float() return init_image.to(device), init_mask.to(device) def main(): global init_image, init_mask p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter) p.add_argument('--batch-size', type=int, default=64, help='the batch size') p.add_argument('--checkpoint', type=Path, required=True, help='the checkpoint to use') p.add_argument('--config', type=Path, help='the model config') p.add_argument('-n', type=int, default=64, help='the number of images to sample') p.add_argument('--prefix', type=str, default='out', help='the output prefix') p.add_argument('--repaint', type=int, default=1, help='number of (re)paint steps') p.add_argument('--steps', type=int, default=50, help='the number of denoising steps') p.add_argument('--seed-scale', type=float, default=0.0, help='strength of nucleation seeding') p.add_argument('--init-image', type=Path, default=None, help='the initial image') p.add_argument('--init-strength', type=float, default=1., help='strength of init image') args = p.parse_args() print("args =", args, flush=True) config = K.config.load_config(args.config if args.config else args.checkpoint) model_config = config['model'] # TODO: allow non-square input sizes assert len(model_config['input_size']) == 2 and model_config['input_size'][0] == model_config['input_size'][1] size = model_config['input_size'] accelerator = accelerate.Accelerator() device = accelerator.device print('Using device:', device, flush=True) inner_model = K.config.make_model(config).eval().requires_grad_(False).to(device) cse = None # ChordSeqEncoder().eval().requires_grad_(False).to(device) # add chord embedding-maker to main model if cse is not None: inner_model.cse = cse try: inner_model.load_state_dict(safetorch.load_file(args.checkpoint)) except: #ckpt = torch.load(args.checkpoint).to(device) ckpt = torch.load(args.checkpoint, map_location='cpu') inner_model.load_state_dict(ckpt['model']) accelerator.print('Parameters:', K.utils.n_params(inner_model)) model = K.Denoiser(inner_model, sigma_data=model_config['sigma_data']) sigma_min = model_config['sigma_min'] sigma_max = model_config['sigma_max'] # SHH modified torch.set_float32_matmul_precision('high') #class_cond = torch.tensor([0]).to(device) #num_classes = 10 #class_cond = torch.remainder(torch.arange(0, args.n), num_classes).int().to(device) #extra_args = {'class_cond':class_cond} extra_args = {} init_image, init_mask = None, None if args.init_image is not None: init_image, init_mask = get_init_image_and_mask(args, device) init_image = init_image.to(device) init_mask = init_mask.to(device) @torch.no_grad() @K.utils.eval_mode(model) def run(): global init_image, init_mask if accelerator.is_local_main_process: tqdm.write('Sampling...') sigmas = K.sampling.get_sigmas_karras(args.steps, sigma_min, sigma_max, rho=7., device=device) #ddpm_sampler = DDPM(model) #model_fn = model #ddpm_sampler = K.external.VDenoiser(model_fn) def sample_fn(n, debug=True): x = torch.randn([n, model_config['input_channels'], size[0], size[1]], device=device) * sigma_max print("n, sigma_max, x.min, x.max = ", n, sigma_max, x.min(), x.max()) if args.init_image is not None: init_data, mask = get_init_image_and_mask(args, device) init_data = args.seed_scale*x*mask + (1-mask)*init_data # extra nucleation? if cse is not None: chord_cond = img_batch_to_seq_emb(init_data, inner_model.cse).to(device) else: chord_cond = None #print("init_data.shape, init_data.min, init_data.max = ", init_data.shape, init_data.min(), init_data.max()) else: init_data, mask, chord_cond = None, None, None # chord_cond doesn't work anyway so f it: chord_cond = None print("chord_cond = ", chord_cond) if chord_cond is not None: extra_args['chord_cond'] = chord_cond # these two work: #x_0 = K.sampling.sample_lms(model, x, sigmas, disable=not accelerator.is_local_main_process, extra_args=extra_args) #x_0 = K.sampling.sample_dpmpp_2m_sde(model, x, sigmas, disable=not accelerator.is_local_main_process, extra_args=extra_args) noise = torch.randn([n, model_config['input_channels'], size[0], size[1]], device=device) sampler_type="my-dpmpp-2m-sde" # "k-lms" #sampler_type="my-sample-euler" #sampler_type="dpmpp-2m-sde" #sampler_type = "dpmpp-3m-sde" #sampler_type = "k-dpmpp-2s-ancestral" print("dtypes:", [x.dtype if x is not None else None for x in [noise, init_data, mask, chord_cond]]) x_0 = sample_k(inner_model, noise, sampler_type=sampler_type, init_data=init_data, mask=mask, steps=args.steps, sigma_min=sigma_min, sigma_max=sigma_max, rho=7., device=device, model_config=model_config, repaint=args.repaint, **extra_args) #x_0 = sample_k(inner_model, noise, sampler_type="dpmpp-2m-sde", steps=100, sigma_min=0.5, sigma_max=50, rho=1., device=device, model_config=model_config, **extra_args) print("x_0.min, x_0.max = ", x_0.min(), x_0.max()) if x_0.isnan().any(): assert False, "x_0 has NaNs" # do gpu garbage collection before proceeding torch.cuda.empty_cache() return x_0 x_0 = K.evaluation.compute_features(accelerator, sample_fn, lambda x: x, args.n, args.batch_size) if accelerator.is_main_process: for i, out in enumerate(x_0): filename = f'{args.prefix}_{i:05}.png' K.utils.to_pil_image(out).save(filename) try: run() except KeyboardInterrupt: pass if __name__ == '__main__': main()