Erasing-Concepts-In-Diffusion / StableDiffuser.py
JadenFK's picture
Inference
843b14b
raw
history blame
8.32 kB
import argparse
import torch
from baukit import TraceDict
from diffusers import AutoencoderKL, UNet2DConditionModel
from PIL import Image
from tqdm.auto import tqdm
from transformers import CLIPTextModel, CLIPTokenizer
import util
from LMSDiscreteScheduler import LMSDiscreteScheduler
def default_parser():
parser = argparse.ArgumentParser()
parser.add_argument('prompts', type=str, nargs='+')
parser.add_argument('outpath', type=str)
parser.add_argument('--images', type=str, nargs='+', default=None)
parser.add_argument('--nsteps', type=int, default=1000)
parser.add_argument('--nimgs', type=int, default=1)
parser.add_argument('--start_itr', type=int, default=0)
parser.add_argument('--return_steps', action='store_true', default=False)
parser.add_argument('--pred_x0', action='store_true', default=False)
parser.add_argument('--device', type=str, default='cuda:0')
parser.add_argument('--seed', type=int, default=42)
return parser
class StableDiffuser(torch.nn.Module):
def __init__(self,
seed=None
):
super().__init__()
self._seed = seed
# Load the autoencoder model which will be used to decode the latents into image space.
self.vae = AutoencoderKL.from_pretrained(
"CompVis/stable-diffusion-v1-4", subfolder="vae")
# Load the tokenizer and text encoder to tokenize and encode the text.
self.tokenizer = CLIPTokenizer.from_pretrained(
"openai/clip-vit-large-patch14")
self.text_encoder = CLIPTextModel.from_pretrained(
"openai/clip-vit-large-patch14")
# The UNet model for generating the latents.
self.unet = UNet2DConditionModel.from_pretrained(
"CompVis/stable-diffusion-v1-4", subfolder="unet")
self.scheduler = LMSDiscreteScheduler(
beta_start=0.00085, beta_end=0.012, beta_schedule="scaled_linear", num_train_timesteps=1000)
self.generator = torch.Generator()
if self._seed is not None:
self.seed(seed)
self.eval()
def seed(self, seed):
self.generator = torch.manual_seed(seed)
def get_noise(self, batch_size, img_size):
param = list(self.parameters())[0]
return torch.randn(
(batch_size, self.unet.in_channels, img_size // 8, img_size // 8),
generator=self.generator).type(param.dtype).to(param.device)
def add_noise(self, latents, noise, step):
return self.scheduler.add_noise(latents, noise, torch.tensor([self.scheduler.timesteps[step]]))
def text_tokenize(self, prompts):
return self.tokenizer(prompts, padding="max_length", max_length=self.tokenizer.model_max_length, truncation=True, return_tensors="pt")
def text_detokenize(self, tokens):
return [self.tokenizer.decode(token) for token in tokens if token != self.tokenizer.vocab_size - 1]
def text_encode(self, tokens):
return self.text_encoder(tokens.input_ids.to(self.unet.device))[0]
def decode(self, latents):
return self.vae.decode(1 / 0.18215 * latents).sample
def encode(self, tensors):
return self.vae.encode(tensors).latent_dist.mode() * 0.18215
def to_image(self, image):
image = (image / 2 + 0.5).clamp(0, 1)
image = image.detach().cpu().permute(0, 2, 3, 1).numpy()
images = (image * 255).round().astype("uint8")
pil_images = [Image.fromarray(image) for image in images]
return pil_images
def set_scheduler_timesteps(self, n_steps):
self.scheduler.set_timesteps(n_steps, device=self.unet.device)
def get_initial_latents(self, n_imgs, img_size, n_prompts):
noise = self.get_noise(n_imgs, img_size).repeat(n_prompts, 1, 1, 1)
latents = noise * self.scheduler.init_noise_sigma
return latents
def get_text_embeddings(self, prompts, n_imgs):
text_tokens = self.text_tokenize(prompts)
text_embeddings = self.text_encode(text_tokens)
unconditional_tokens = self.text_tokenize([""] * len(prompts))
unconditional_embeddings = self.text_encode(unconditional_tokens)
text_embeddings = torch.cat([unconditional_embeddings, text_embeddings]).repeat_interleave(n_imgs, dim=0)
return text_embeddings
def predict_noise(self,
iteration,
latents,
text_embeddings,
guidance_scale=7.5
):
# expand the latents if we are doing classifier-free guidance to avoid doing two forward passes.
latents = torch.cat([latents] * 2)
latents = self.scheduler.scale_model_input(
latents, iteration)
# predict the noise residual
noise_prediction = self.unet(
latents, self.scheduler.timesteps[iteration], encoder_hidden_states=text_embeddings).sample
# perform guidance
noise_prediction_uncond, noise_prediction_text = noise_prediction.chunk(2)
noise_prediction = noise_prediction_uncond + guidance_scale * \
(noise_prediction_text - noise_prediction_uncond)
return noise_prediction
@torch.no_grad()
def diffusion(self,
latents,
text_embeddings,
end_iteration=1000,
start_iteration=0,
return_steps=False,
pred_x0=False,
trace_args=None,
show_progress=True,
**kwargs):
latents_steps = []
trace_steps = []
trace = None
for iteration in tqdm(range(start_iteration, end_iteration), disable=not show_progress):
if trace_args:
trace = TraceDict(self, **trace_args)
noise_pred = self.predict_noise(
iteration,
latents,
text_embeddings,
**kwargs)
# compute the previous noisy sample x_t -> x_t-1
output = self.scheduler.step(noise_pred, iteration, latents)
if trace_args:
trace.close()
trace_steps.append(trace)
latents = output.prev_sample
if return_steps or iteration == end_iteration - 1:
output = output.pred_original_sample if pred_x0 else latents
if return_steps:
latents_steps.append(output.cpu())
else:
latents_steps.append(output)
return latents_steps, trace_steps
@torch.no_grad()
def __call__(self,
prompts,
img_size=512,
n_steps=50,
n_imgs=1,
end_iteration=None,
reseed=False,
**kwargs
):
assert 0 <= n_steps <= 1000
if not isinstance(prompts, list):
prompts = [prompts]
self.set_scheduler_timesteps(n_steps)
if reseed:
self.seed(self._seed)
latents = self.get_initial_latents(n_imgs, img_size, len(prompts))
text_embeddings = self.get_text_embeddings(prompts,n_imgs=n_imgs)
end_iteration = end_iteration or n_steps
latents_steps, trace_steps = self.diffusion(
latents,
text_embeddings,
end_iteration=end_iteration,
**kwargs
)
latents_steps = [self.decode(latents.to(self.unet.device)) for latents in latents_steps]
images_steps = [self.to_image(latents) for latents in latents_steps]
images_steps = list(zip(*images_steps))
if trace_steps:
return images_steps, trace_steps
return images_steps
if __name__ == '__main__':
parser = default_parser()
args = parser.parse_args()
diffuser = StableDiffuser(seed=args.seed).to(torch.device(args.device)).half()
images = diffuser(args.prompts,
n_steps=args.nsteps,
n_imgs=args.nimgs,
start_iteration=args.start_itr,
return_steps=args.return_steps,
pred_x0=args.pred_x0
)
util.image_grid(images, args.outpath)