mirror of
https://github.com/lllyasviel/stable-diffusion-webui-forge.git
synced 2026-07-21 21:01:24 +08:00
130 lines
5.0 KiB
Python
130 lines
5.0 KiB
Python
import k_diffusion.sampling
|
|
import torch
|
|
import tqdm
|
|
|
|
from modules.uni_pc import uni_pc as unipc_impl
|
|
|
|
|
|
@torch.no_grad()
|
|
def restart_sampler(model, x, sigmas, extra_args=None, callback=None, disable=None, s_noise=1.0, restart_list=None):
|
|
"""
|
|
Implements restart sampling in Restart Sampling for Improving Generative Processes (2023)
|
|
Restart_list format: {min_sigma: [ restart_steps, restart_times, max_sigma]}
|
|
If restart_list is None: will choose restart_list automatically, otherwise will use the given restart_list
|
|
"""
|
|
extra_args = {} if extra_args is None else extra_args
|
|
s_in = x.new_ones([x.shape[0]])
|
|
step_id = 0
|
|
from k_diffusion.sampling import get_sigmas_karras, to_d
|
|
|
|
def heun_step(x, old_sigma, new_sigma, second_order=True):
|
|
nonlocal step_id
|
|
denoised = model(x, old_sigma * s_in, **extra_args)
|
|
d = to_d(x, old_sigma, denoised)
|
|
if callback is not None:
|
|
callback({"x": x, "i": step_id, "sigma": new_sigma, "sigma_hat": old_sigma, "denoised": denoised})
|
|
dt = new_sigma - old_sigma
|
|
if new_sigma == 0 or not second_order:
|
|
# Euler method
|
|
x = x + d * dt
|
|
else:
|
|
# Heun's method
|
|
x_2 = x + d * dt
|
|
denoised_2 = model(x_2, new_sigma * s_in, **extra_args)
|
|
d_2 = to_d(x_2, new_sigma, denoised_2)
|
|
d_prime = (d + d_2) / 2
|
|
x = x + d_prime * dt
|
|
step_id += 1
|
|
return x
|
|
|
|
steps = sigmas.shape[0] - 1
|
|
if restart_list is None:
|
|
if steps >= 20:
|
|
restart_steps = 9
|
|
restart_times = 1
|
|
if steps >= 36:
|
|
restart_steps = steps // 4
|
|
restart_times = 2
|
|
sigmas = get_sigmas_karras(steps - restart_steps * restart_times, sigmas[-2].item(), sigmas[0].item(), device=sigmas.device)
|
|
restart_list = {0.1: [restart_steps + 1, restart_times, 2]}
|
|
else:
|
|
restart_list = {}
|
|
|
|
restart_list = {int(torch.argmin(abs(sigmas - key), dim=0)): value for key, value in restart_list.items()}
|
|
|
|
step_list = []
|
|
for i in range(len(sigmas) - 1):
|
|
step_list.append((sigmas[i], sigmas[i + 1]))
|
|
if i + 1 in restart_list:
|
|
restart_steps, restart_times, restart_max = restart_list[i + 1]
|
|
min_idx = i + 1
|
|
max_idx = int(torch.argmin(abs(sigmas - restart_max), dim=0))
|
|
if max_idx < min_idx:
|
|
sigma_restart = get_sigmas_karras(restart_steps, sigmas[min_idx].item(), sigmas[max_idx].item(), device=sigmas.device)[:-1]
|
|
while restart_times > 0:
|
|
restart_times -= 1
|
|
step_list.extend(zip(sigma_restart[:-1], sigma_restart[1:]))
|
|
|
|
last_sigma = None
|
|
for old_sigma, new_sigma in tqdm.tqdm(step_list, disable=disable):
|
|
if last_sigma is None:
|
|
last_sigma = old_sigma
|
|
elif last_sigma < old_sigma:
|
|
x = x + k_diffusion.sampling.torch.randn_like(x) * s_noise * (old_sigma**2 - last_sigma**2) ** 0.5
|
|
x = heun_step(x, old_sigma, new_sigma)
|
|
last_sigma = new_sigma
|
|
|
|
return x
|
|
|
|
|
|
class SigmaConvert:
|
|
schedule = ""
|
|
|
|
def marginal_log_mean_coeff(self, sigma):
|
|
return 0.5 * torch.log(1 / ((sigma * sigma) + 1))
|
|
|
|
def marginal_alpha(self, t):
|
|
return torch.exp(self.marginal_log_mean_coeff(t))
|
|
|
|
def marginal_std(self, t):
|
|
return torch.sqrt(1.0 - torch.exp(2.0 * self.marginal_log_mean_coeff(t)))
|
|
|
|
def marginal_lambda(self, t):
|
|
"""
|
|
Compute lambda_t = log(alpha_t) - log(sigma_t) of a given continuous-time label t in [0, T]
|
|
"""
|
|
log_mean_coeff = self.marginal_log_mean_coeff(t)
|
|
log_std = 0.5 * torch.log(1.0 - torch.exp(2.0 * log_mean_coeff))
|
|
return log_mean_coeff - log_std
|
|
|
|
|
|
def predict_eps_sigma(model, input, sigma_in, **kwargs):
|
|
sigma = sigma_in.view(sigma_in.shape[:1] + (1,) * (input.ndim - 1))
|
|
input = input * ((sigma**2 + 1.0) ** 0.5)
|
|
return (input - model(input, sigma_in, **kwargs)) / sigma
|
|
|
|
|
|
def sample_unipc(model, x, sigmas, extra_args=None, callback=None, disable=False, variant="bh1"):
|
|
timesteps = sigmas.clone()
|
|
if sigmas[-1] < 0.001:
|
|
timesteps[-1] = 0.001
|
|
|
|
ns = SigmaConvert()
|
|
|
|
x = x / torch.sqrt(1.0 + timesteps[0] ** 2.0)
|
|
model_type = "noise"
|
|
|
|
model_fn = unipc_impl.model_wrapper(
|
|
lambda input, sigma, **kwargs: predict_eps_sigma(model, input, sigma, **kwargs),
|
|
ns,
|
|
model_type=model_type,
|
|
guidance_type="uncond",
|
|
model_kwargs=extra_args,
|
|
)
|
|
|
|
order = min(3, len(timesteps) - 2)
|
|
uni_pc = unipc_impl.UniPC(model_fn, ns, predict_x0=True, thresholding=False, variant=variant)
|
|
x = uni_pc.sample(x, timesteps=timesteps, skip_type="time_uniform", method="multistep", order=order, lower_order_final=True, callback=callback, disable_pbar=disable)
|
|
x /= ns.marginal_alpha(timesteps[-1])
|
|
return x
|