import functools import os import random import numpy as np import imagesize import torch import torch.nn.functional as F from PIL import Image import cv2 def seed_everything(seed): # np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) random.seed(seed) def find_nearest(array, value): array = np.asarray(array) idx = (np.abs(array/value - 1)).argmin() return idx def get_sup_mask(mask_list): or_mask = np.zeros_like(mask_list[0]) for mask in mask_list: or_mask += mask or_mask[or_mask >= 1] = 1 sup_mask = 1 - or_mask return sup_mask def get_masks(obboxes, height, width, device): # Construct Instance Guidance Mask guidance_masks, in_box = [], [] for obbox in obboxes[0]: guidance_mask = np.zeros((height, width)) if np.count_nonzero(obbox): pts = np.array(obbox).reshape(-1, 1, 2) pts[..., 0] = pts[..., 0] * width pts[..., 1] = pts[..., 1] * height pts = np.int32(pts) guidance_masks.append(cv2.fillPoly(guidance_mask, [pts], 1)[None, ...]) else: guidance_masks.append(guidance_mask[None, ...]) in_box.append([obbox[0], obbox[2], obbox[4], obbox[6], obbox[1], obbox[3], obbox[5], obbox[7]]) # Construct Background Guidance Mask sup_mask = get_sup_mask(guidance_masks) supplement_mask = torch.from_numpy(sup_mask[None, ...]) supplement_mask = F.interpolate(supplement_mask, (height//8, width//8), mode='bilinear').float() supplement_mask = supplement_mask.to(device) # (1, 1, H, W) guidance_masks = np.concatenate(guidance_masks, axis=0) guidance_masks = guidance_masks[None, ...] guidance_masks = torch.from_numpy(guidance_masks).float().to(device) guidance_masks = F.interpolate(guidance_masks, (height//8, width//8), mode='bilinear') # (1, instance_num, H, W) # guidance_masks.shape [1, 15, 64, 64] ; supplement_mask.shape [1, 1, 64, 64] in_box = torch.from_numpy(np.array(in_box))[None, ...].float().to(device) # (1, instance_num, 4) return guidance_masks, supplement_mask, in_box def get_sigmoid(bboxes, height, width, device): sigmoid_values = [] for w_min, h_min, w_max, h_max in bboxes[0]: H, W = height//8, width // 8 x = torch.linspace(0, W - 1, W) y = torch.linspace(0, H - 1, H) yy, xx = torch.meshgrid(y, x, indexing='ij') xx, yy = xx / H, yy / W mu1 = (w_min + w_max) / 2 mu2 = (h_min + h_max) / 2 sigma1 = ((w_max - w_min) ** 2) / 4 sigma2 = ((h_max - h_min) ** 2) / 4 if sigma1 == 0 or sigma2 == 0: sigmoid_values.append(torch.zeros_like(xx)) continue exponent = -10 * (1 - ((xx - mu1) ** 2) / sigma1 - ((yy - mu2) ** 2) / sigma2) sigmoid_value = 1 / (1 + torch.exp(exponent)) sigmoid_values.append(sigmoid_value) sigmoid_values = torch.stack(sigmoid_values, dim=0)[None, ...].to(device) # sigmoid_values.shape [1, 15, 64, 64] ; in_box.shape [1, 15, 8] return sigmoid_values class ExemplarPool: def __init__(self, data_embeds_dict_path, exemplar_pool_path, image_processor): self.data_embeds_dict = torch.load(data_embeds_dict_path, map_location='cpu') self.all_img_names = np.array(list(self.data_embeds_dict.keys())) self.all_txt_embs = torch.cat([self.data_embeds_dict[name]['txt_emb'] for name in self.all_img_names], dim=0) self.all_img_embs = torch.cat([self.data_embeds_dict[name]['img_emb'] for name in self.all_img_names], dim=0) self.exemplar_pool_path = exemplar_pool_path self.image_processor = image_processor def to(self, device, dtype=None): self.device = device self.all_txt_embs = self.all_txt_embs.to(device=device, dtype=dtype) self.all_img_embs = self.all_img_embs.to(device=device, dtype=dtype) def get_similar_exemplars_names(self, prompt_emb, topk, sim_mode): prompt_emb = F.normalize(prompt_emb, dim=-1).detach() if sim_mode == 'text2text': sim_vals = torch.matmul(prompt_emb, self.all_txt_embs.T) elif sim_mode == 'text2img': sim_vals = torch.matmul(prompt_emb, self.all_img_embs.T) elif sim_mode == 'both': txt_sim_vals = torch.matmul(prompt_emb, self.all_txt_embs.T) img_sim_vals = torch.matmul(prompt_emb, self.all_img_embs.T) sim_vals = (txt_sim_vals + img_sim_vals) * 0.5 else: raise ValueError('Invalid mode for similarity computation!') _, topk_indices = torch.topk(sim_vals, k=topk, dim=1) topk_img_names = self.all_img_names[topk_indices.cpu().numpy()].tolist() return topk_img_names def names_to_tensors(self, topk_img_names): topk_img_tensors = [] for names_for_one_prompt in topk_img_names: topk_tensors_per_prompt = [] for name in names_for_one_prompt: img = Image.open(os.path.join(self.exemplar_pool_path, name)).convert('RGB') # tensor = self.image_processor(images=img, return_tensors="pt", do_normalize=False)['pixel_values'].squeeze(0) tensor = self.image_processor(images=img, return_tensors="pt")['pixel_values'].squeeze(0) topk_tensors_per_prompt.append(tensor) stacked_k_tensors = torch.stack(topk_tensors_per_prompt) topk_img_tensors.append(stacked_k_tensors) topk_img_tensors = torch.stack(topk_img_tensors) return topk_img_tensors def get_similar_exemplars(self, prompt_emb, topk=1, sim_mode='text2img'): topk_img_names = self.get_similar_exemplars_names(prompt_emb, topk, sim_mode) topk_img_tensors = self.names_to_tensors(topk_img_names) return topk_img_tensors.to(self.device)