import os import torch import collections import torch.nn as nn from functools import partial from transformers import CLIPTextModel, CLIPTokenizer, logging from diffusers import AutoencoderKL, PNDMScheduler, EulerDiscreteScheduler, DPMSolverMultistepScheduler from models.unet_2d_condition import UNet2DConditionModel from utils.attention_utils import CrossAttentionLayers, SelfAttentionLayers # suppress partial model loading warning logging.set_verbosity_error() class RegionDiffusion(nn.Module): def __init__(self, device): super().__init__() self.device = device self.num_train_timesteps = 1000 self.clip_gradient = False print(f'[INFO] loading stable diffusion...') model_id = 'runwayml/stable-diffusion-v1-5' self.vae = AutoencoderKL.from_pretrained( model_id, subfolder="vae").to(self.device) self.tokenizer = CLIPTokenizer.from_pretrained( model_id, subfolder='tokenizer') self.text_encoder = CLIPTextModel.from_pretrained( model_id, subfolder='text_encoder').to(self.device) self.unet = UNet2DConditionModel.from_pretrained( model_id, subfolder="unet").to(self.device) self.scheduler = PNDMScheduler(beta_start=0.00085, beta_end=0.012, beta_schedule="scaled_linear", num_train_timesteps=self.num_train_timesteps, skip_prk_steps=True, steps_offset=1) self.alphas_cumprod = self.scheduler.alphas_cumprod.to(self.device) self.masks = [] self.attention_maps = None self.selfattn_maps = None self.crossattn_maps = None self.color_loss = torch.nn.functional.mse_loss self.forward_hooks = [] self.forward_replacement_hooks = [] print(f'[INFO] loaded stable diffusion!') def get_text_embeds(self, prompt, negative_prompt): # prompt, negative_prompt: [str] # Tokenize text and get embeddings text_input = self.tokenizer( prompt, padding='max_length', max_length=self.tokenizer.model_max_length, truncation=True, return_tensors='pt') with torch.no_grad(): text_embeddings = self.text_encoder( text_input.input_ids.to(self.device))[0] # Do the same for unconditional embeddings uncond_input = self.tokenizer(negative_prompt, padding='max_length', max_length=self.tokenizer.model_max_length, return_tensors='pt') with torch.no_grad(): uncond_embeddings = self.text_encoder( uncond_input.input_ids.to(self.device))[0] # Cat for final embeddings text_embeddings = torch.cat([uncond_embeddings, text_embeddings]) return text_embeddings def get_text_embeds_list(self, prompts): # prompts: [list] text_embeddings = [] for prompt in prompts: # Tokenize text and get embeddings text_input = self.tokenizer( [prompt], padding='max_length', max_length=self.tokenizer.model_max_length, truncation=True, return_tensors='pt') with torch.no_grad(): text_embeddings.append(self.text_encoder( text_input.input_ids.to(self.device))[0]) return text_embeddings def produce_latents(self, text_embeddings, height=512, width=512, num_inference_steps=50, guidance_scale=7.5, latents=None, use_guidance=False, text_format_dict={}, inject_selfattn=0, bg_aug_end=1000): if latents is None: latents = torch.randn( (1, self.unet.in_channels, height // 8, width // 8), device=self.device) if inject_selfattn > 0: latents_reference = latents.clone().detach() self.scheduler.set_timesteps(num_inference_steps) n_styles = text_embeddings.shape[0]-1 print(n_styles, len(self.masks)) assert n_styles == len(self.masks) with torch.autocast('cuda'): for i, t in enumerate(self.scheduler.timesteps): # predict the noise residual with torch.no_grad(): # tokens without any attributes feat_inject_step = t > (1-inject_selfattn) * 1000 noise_pred_uncond_cur = self.unet(latents, t, encoder_hidden_states=text_embeddings[:1], # text_format_dict={})['sample'] )['sample'] # tokens without any style or footnote self.register_fontsize_hooks(text_format_dict) noise_pred_text_cur = self.unet(latents, t, encoder_hidden_states=text_embeddings[-1:], # text_format_dict=text_format_dict)['sample'] )['sample'] self.remove_fontsize_hooks() if inject_selfattn > 0 or inject_background > 0: noise_pred_uncond_refer = self.unet(latents_reference, t, encoder_hidden_states=text_embeddings[:1], # text_format_dict={})['sample'] )['sample'] self.register_selfattn_hooks(feat_inject_step) noise_pred_text_refer = self.unet(latents_reference, t, encoder_hidden_states=text_embeddings[-1:], # text_format_dict={})['sample'] )['sample'] self.remove_selfattn_hooks() noise_pred_uncond = noise_pred_uncond_cur * self.masks[-1] noise_pred_text = noise_pred_text_cur * self.masks[-1] # tokens with attributes for style_i, mask in enumerate(self.masks[:-1]): if t > bg_aug_end: rand_rgb = torch.rand([1, 3, 1, 1]).cuda() black_background = torch.ones( [1, 3, height, width]).cuda()*rand_rgb black_latent = self.encode_imgs( black_background) noise = torch.randn_like(black_latent) black_latent_noisy = self.scheduler.add_noise( black_latent, noise, t) masked_latent = ( mask > 0.001) * latents + (mask < 0.001) * black_latent_noisy noise_pred_uncond_cur = self.unet(masked_latent, t, encoder_hidden_states=text_embeddings[:1], text_format_dict={})['sample'] else: masked_latent = latents self.register_replacement_hooks(feat_inject_step) noise_pred_text_cur = self.unet(latents, t, encoder_hidden_states=text_embeddings[style_i+1:style_i+2], # text_format_dict={})['sample'] )['sample'] self.remove_replacement_hooks() noise_pred_uncond = noise_pred_uncond + noise_pred_uncond_cur*mask noise_pred_text = noise_pred_text + noise_pred_text_cur*mask # perform guidance noise_pred = noise_pred_uncond + guidance_scale * \ (noise_pred_text - noise_pred_uncond) if inject_selfattn > 0: noise_pred_refer = noise_pred_uncond_refer + guidance_scale * \ (noise_pred_text_refer - noise_pred_uncond_refer) # compute the previous noisy sample x_t -> x_t-1 latents_reference = self.scheduler.step(torch.cat([noise_pred, noise_pred_refer]), t, torch.cat([latents, latents_reference]))[ 'prev_sample'] latents, latents_reference = torch.chunk( latents_reference, 2, dim=0) else: # compute the previous noisy sample x_t -> x_t-1 latents = self.scheduler.step(noise_pred, t, latents)[ 'prev_sample'] # apply guidance if use_guidance and t < text_format_dict['guidance_start_step']: with torch.enable_grad(): if not latents.requires_grad: latents.requires_grad = True latents_0 = self.predict_x0(latents, noise_pred, t) latents_inp = 1 / 0.18215 * latents_0 imgs = self.vae.decode(latents_inp).sample imgs = (imgs / 2 + 0.5).clamp(0, 1) # save_path = 'results/font_color/20230425/church_process/orange/' # os.makedirs(save_path, exist_ok=True) # torchvision.utils.save_image( # imgs, os.path.join(save_path, 'step%d.png' % t)) # loss = (((imgs - text_format_dict['target_RGB'])*text_format_dict['color_obj_atten'][:, 0])**2).mean()*100 loss_total = 0. for attn_map, rgb_val in zip(text_format_dict['color_obj_atten'], text_format_dict['target_RGB']): # loss = self.color_loss( # imgs*attn_map[:, 0], rgb_val*attn_map[:, 0])*100 avg_rgb = ( imgs*attn_map[:, 0]).sum(2).sum(2)/attn_map[:, 0].sum() loss = self.color_loss( avg_rgb, rgb_val[:, :, 0, 0])*100 # print(loss) loss_total += loss loss_total.backward() latents = ( latents - latents.grad * text_format_dict['color_guidance_weight'] * self.masks[0]).detach().clone() return latents def predict_x0(self, x_t, eps_t, t): alpha_t = self.scheduler.alphas_cumprod[t] return (x_t - eps_t * torch.sqrt(1-alpha_t)) / torch.sqrt(alpha_t) def produce_attn_maps(self, prompts, negative_prompts='', height=512, width=512, num_inference_steps=50, guidance_scale=7.5, latents=None): if isinstance(prompts, str): prompts = [prompts] if isinstance(negative_prompts, str): negative_prompts = [negative_prompts] # Prompts -> text embeds text_embeddings = self.get_text_embeds( prompts, negative_prompts) # [2, 77, 768] if latents is None: latents = torch.randn( (text_embeddings.shape[0] // 2, self.unet.in_channels, height // 8, width // 8), device=self.device) self.scheduler.set_timesteps(num_inference_steps) self.remove_replacement_hooks() with torch.autocast('cuda'): for i, t in enumerate(self.scheduler.timesteps): # expand the latents if we are doing classifier-free guidance to avoid doing two forward passes. latent_model_input = torch.cat([latents] * 2) # predict the noise residual with torch.no_grad(): noise_pred = self.unet( latent_model_input, t, encoder_hidden_states=text_embeddings)['sample'] # perform guidance noise_pred_uncond, noise_pred_text = noise_pred.chunk(2) noise_pred = noise_pred_uncond + guidance_scale * \ (noise_pred_text - noise_pred_uncond) # compute the previous noisy sample x_t -> x_t-1 latents = self.scheduler.step(noise_pred, t, latents)[ 'prev_sample'] # Img latents -> imgs imgs = self.decode_latents(latents) # [1, 3, 512, 512] # Img to Numpy imgs = imgs.detach().cpu().permute(0, 2, 3, 1).numpy() imgs = (imgs * 255).round().astype('uint8') return imgs def decode_latents(self, latents): latents = 1 / 0.18215 * latents with torch.no_grad(): imgs = self.vae.decode(latents).sample imgs = (imgs / 2 + 0.5).clamp(0, 1) return imgs def encode_imgs(self, imgs): # imgs: [B, 3, H, W] imgs = 2 * imgs - 1 posterior = self.vae.encode(imgs).latent_dist latents = posterior.sample() * 0.18215 return latents def prompt_to_img(self, prompts, negative_prompts='', height=512, width=512, num_inference_steps=50, guidance_scale=7.5, latents=None, text_format_dict={}, use_guidance=False, inject_selfattn=0, bg_aug_end=1000): if isinstance(prompts, str): prompts = [prompts] if isinstance(negative_prompts, str): negative_prompts = [negative_prompts] # Prompts -> text embeds text_embeds = self.get_text_embeds( prompts, negative_prompts) # [2, 77, 768] # else: latents = self.produce_latents(text_embeds, height=height, width=width, latents=latents, num_inference_steps=num_inference_steps, guidance_scale=guidance_scale, use_guidance=use_guidance, text_format_dict=text_format_dict, inject_selfattn=inject_selfattn, bg_aug_end=bg_aug_end) # [1, 4, 64, 64] # Img latents -> imgs imgs = self.decode_latents(latents) # [1, 3, 512, 512] # Img to Numpy imgs = imgs.detach().cpu().permute(0, 2, 3, 1).numpy() imgs = (imgs * 255).round().astype('uint8') return imgs def reset_attention_maps(self): r"""Function to reset attention maps. We reset attention maps because we append them while getting hooks to visualize attention maps for every step. """ for key in self.selfattn_maps: self.selfattn_maps[key] = [] for key in self.crossattn_maps: self.crossattn_maps[key] = [] def register_evaluation_hooks(self): r"""Function for registering hooks during evaluation. We mainly store activation maps averaged over queries. """ self.forward_hooks = [] def save_activations(activations, name, module, inp, out): r""" PyTorch Forward hook to save outputs at each forward pass. """ # out[0] - final output of attention layer # out[1] - attention probability matrix if 'attn2' in name: assert out[1].shape[-1] == 77 activations[name].append(out[1].detach().cpu()) else: assert out[1].shape[-1] != 77 attention_dict = collections.defaultdict(list) for name, module in self.unet.named_modules(): leaf_name = name.split('.')[-1] if 'attn' in leaf_name: # Register hook to obtain outputs at every attention layer. self.forward_hooks.append(module.register_forward_hook( partial(save_activations, attention_dict, name) )) # attention_dict is a dictionary containing attention maps for every attention layer self.attention_maps = attention_dict def register_selfattn_hooks(self, feat_inject_step=False): r"""Function for registering hooks during evaluation. We mainly store activation maps averaged over queries. """ self.selfattn_forward_hooks = [] def save_activations(activations, name, module, inp, out): r""" PyTorch Forward hook to save outputs at each forward pass. """ # out[0] - final output of attention layer # out[1] - attention probability matrix if 'attn2' in name: assert out[1][1].shape[-1] == 77 # cross attention injection # activations[name] = out[1][1].detach() else: assert out[1][1].shape[-1] != 77 activations[name] = out[1][1].detach() def save_resnet_activations(activations, name, module, inp, out): r""" PyTorch Forward hook to save outputs at each forward pass. """ # out[0] - final output of residual layer # out[1] - residual hidden feature # import ipdb # ipdb.set_trace() assert out[1].shape[-1] == 16 activations[name] = out[1].detach() attention_dict = collections.defaultdict(list) for name, module in self.unet.named_modules(): leaf_name = name.split('.')[-1] if 'attn' in leaf_name and feat_inject_step: # Register hook to obtain outputs at every attention layer. self.selfattn_forward_hooks.append(module.register_forward_hook( partial(save_activations, attention_dict, name) )) if name == 'up_blocks.1.resnets.1' and feat_inject_step: self.selfattn_forward_hooks.append(module.register_forward_hook( partial(save_resnet_activations, attention_dict, name) )) # attention_dict is a dictionary containing attention maps for every attention layer self.self_attention_maps_cur = attention_dict def register_replacement_hooks(self, feat_inject_step=False): r"""Function for registering hooks to replace self attention. """ self.forward_replacement_hooks = [] def replace_activations(name, module, args): r""" PyTorch Forward hook to save outputs at each forward pass. """ if 'attn1' in name: modified_args = (args[0], self.self_attention_maps_cur[name]) return modified_args # cross attention injection # elif 'attn2' in name: # modified_map = { # 'reference': self.self_attention_maps_cur[name], # 'inject_pos': self.inject_pos, # } # modified_args = (args[0], modified_map) # return modified_args def replace_resnet_activations(name, module, args): r""" PyTorch Forward hook to save outputs at each forward pass. """ modified_args = (args[0], args[1], self.self_attention_maps_cur[name]) return modified_args for name, module in self.unet.named_modules(): leaf_name = name.split('.')[-1] if 'attn' in leaf_name and feat_inject_step: # Register hook to obtain outputs at every attention layer. self.forward_replacement_hooks.append(module.register_forward_pre_hook( partial(replace_activations, name) )) if name == 'up_blocks.1.resnets.1' and feat_inject_step: # Register hook to obtain outputs at every attention layer. self.forward_replacement_hooks.append(module.register_forward_pre_hook( partial(replace_resnet_activations, name) )) def register_tokenmap_hooks(self): r"""Function for registering hooks during evaluation. We mainly store activation maps averaged over queries. """ self.forward_hooks = [] def save_activations(selfattn_maps, crossattn_maps, n_maps, name, module, inp, out): r""" PyTorch Forward hook to save outputs at each forward pass. """ # out[0] - final output of attention layer # out[1] - attention probability matrices if name in n_maps: n_maps[name] += 1 else: n_maps[name] = 1 if 'attn2' in name: assert out[1][0].shape[-1] == 77 if name in CrossAttentionLayers and n_maps[name] > 10: if name in crossattn_maps: crossattn_maps[name] += out[1][0].detach().cpu()[1:2] else: crossattn_maps[name] = out[1][0].detach().cpu()[1:2] else: assert out[1][0].shape[-1] != 77 if name in SelfAttentionLayers and n_maps[name] > 10: if name in crossattn_maps: selfattn_maps[name] += out[1][0].detach().cpu()[1:2] else: selfattn_maps[name] = out[1][0].detach().cpu()[1:2] selfattn_maps = collections.defaultdict(list) crossattn_maps = collections.defaultdict(list) n_maps = collections.defaultdict(list) for name, module in self.unet.named_modules(): leaf_name = name.split('.')[-1] if 'attn' in leaf_name: # Register hook to obtain outputs at every attention layer. self.forward_hooks.append(module.register_forward_hook( partial(save_activations, selfattn_maps, crossattn_maps, n_maps, name) )) # attention_dict is a dictionary containing attention maps for every attention layer self.selfattn_maps = selfattn_maps self.crossattn_maps = crossattn_maps self.n_maps = n_maps def remove_tokenmap_hooks(self): for hook in self.forward_hooks: hook.remove() self.selfattn_maps = None self.crossattn_maps = None self.n_maps = None def remove_evaluation_hooks(self): for hook in self.forward_hooks: hook.remove() self.attention_maps = None def remove_replacement_hooks(self): for hook in self.forward_replacement_hooks: hook.remove() def remove_selfattn_hooks(self): for hook in self.selfattn_forward_hooks: hook.remove() def register_fontsize_hooks(self, text_format_dict={}): r"""Function for registering hooks to replace self attention. """ self.forward_fontsize_hooks = [] def adjust_attn_weights(name, module, args): r""" PyTorch Forward hook to save outputs at each forward pass. """ if 'attn2' in name: modified_args = (args[0], None, attn_weights) return modified_args if text_format_dict['word_pos'] is not None and text_format_dict['font_size'] is not None: attn_weights = {'word_pos': text_format_dict['word_pos'], 'font_size': text_format_dict['font_size']} else: attn_weights = None for name, module in self.unet.named_modules(): leaf_name = name.split('.')[-1] if 'attn' in leaf_name and attn_weights is not None: # Register hook to obtain outputs at every attention layer. self.forward_fontsize_hooks.append(module.register_forward_pre_hook( partial(adjust_attn_weights, name) )) def remove_fontsize_hooks(self): for hook in self.forward_fontsize_hooks: hook.remove()