prefpaintReward / InpaintReward.py
kd5678's picture
Upload InpaintReward.py
8a6cb91 verified
raw
history blame
9.01 kB
import torch
import torch.nn as nn
import clip
from torch.nn import TransformerEncoder, TransformerEncoderLayer
from torchvision import transforms
class MLP(nn.Module):
def __init__(self, input_size):
super().__init__()
self.input_size = input_size
self.layers = nn.Sequential(
nn.Linear(self.input_size, 512),
nn.Dropout(0.2),
nn.Linear(512, 256),
)
self.last_layer = nn.Linear(256, 1, bias=False)
self.last_layer_weight = self.last_layer.weight
# initial MLP param
for name, param in self.layers.named_parameters():
if 'weight' in name:
nn.init.normal_(param, mean=0.0, std=1.0/(self.input_size+1))
if 'bias' in name:
nn.init.constant_(param, val=0)
for name, param in self.last_layer.named_parameters():
if 'weight' in name:
nn.init.normal_(param, mean=0.0, std=1.0/(self.input_size+1))
if 'bias' in name:
nn.init.constant_(param, val=0)
def forward(self, input):
features = self.layers(input)
out = self.last_layer(features)
return out, features
class ViTBlock(nn.Module):
def __init__(self, feature_dim, num_heads, mlp_dim, dropout=0.1):
super(ViTBlock, self).__init__()
# Transformer encoder layer
self.encoder_layer = TransformerEncoderLayer(
d_model=feature_dim,
nhead=num_heads,
dim_feedforward=mlp_dim,
dropout=dropout,
batch_first=True # Input shape: (batch_size, seq_length, feature_dim)
)
self.transformer_encoder = TransformerEncoder(self.encoder_layer, num_layers=1)
def forward(self, x):
x = self.transformer_encoder(x)
return x
class ImageReward(nn.Module):
def __init__(self, config, device='cpu'):
super().__init__()
self.config = config
self.device = device
self.clip_model, self.preprocess = clip.load("ViT-B/32")
self.clip_model = self.clip_model.float()
self.mlp = MLP(self.config['ImageReward']['mlp_dim'])
self.vit_block = ViTBlock(self.config["ViT"]["feature_dim"], self.config["ViT"]["num_heads"], self.config["ViT"]["mlp_dim"])
self.toImage = transforms.ToPILImage()
self.mean = 0.4064 #0.65823
self.std = 2.3021 #8.5400
if self.config.fix_base:
self.clip_model.requires_grad_(False)
def score(self, inpaint_list, masks_rgb):
inpaint_embeds_bs, mask_rgb_embeds_bs = [], []
for bs in range(len(inpaint_list)):
if isinstance(inpaint_list[bs], torch.Tensor):
inpaint = self.toImage(inpaint_list[bs])
else:
inpaint = inpaint_list[bs]
inpaint = self.preprocess(inpaint).unsqueeze(0)
if isinstance(masks_rgb[bs], torch.Tensor):
mask_rgb = self.toImage(masks_rgb[bs])
else:
mask_rgb = masks_rgb[bs]
mask_rgb = self.preprocess(masks_rgb[bs]).unsqueeze(0)
inpt, msk = inpaint.to(self.device), mask_rgb.to(self.device)
inpt_embeds = self.clip_model.encode_image(inpt).to(torch.float32)
msk_embeds = self.clip_model.encode_image(msk).to(torch.float32)
inpaint_embeds_bs.append(inpt_embeds.squeeze(0))
mask_rgb_embeds_bs.append(msk_embeds.squeeze(0))
emb_inpaint = torch.stack(inpaint_embeds_bs, dim=0)
emb_mask_rgb = torch.stack(mask_rgb_embeds_bs, dim=0)
emb_feature = torch.cat((emb_inpaint, emb_mask_rgb), dim=-1)
emb_feature = emb_feature.unsqueeze(1)
emb_feature = self.vit_block(emb_feature) # 1024
scores, last_features = self.mlp(emb_feature)
scores = torch.squeeze(scores)
last_features = torch.squeeze(last_features)
if self.config.group:
scores = (scores - self.mean) / self.std
return scores.detach().cpu().numpy().tolist(), last_features.detach().cpu().numpy().tolist()
def load_model(self, model, ckpt_path = None):
print('load checkpoint from %s'%ckpt_path)
state_dict = {k: v for k, v in torch.load(ckpt_path, map_location='cpu').items()}
new_dict = {key.replace("module.", ""): value for key, value in state_dict.items()}
model.load_state_dict(new_dict)
return model
class ImageRewardGroup(nn.Module):
def __init__(self, config, device='cpu'):
super().__init__()
self.config = config
self.device = device
self.clip_model, self.preprocess = clip.load("ViT-B/32", device="cuda") #clip.load(config['clip_model'], device="cuda" if torch.cuda.is_available() else "cpu")
self.clip_model = self.clip_model.float()
self.mlp = MLP(config['ImageReward']['mlp_dim'])
self.vit_block = ViTBlock(self.config["ViT"]["feature_dim"], self.config["ViT"]["num_heads"], self.config["ViT"]["mlp_dim"])
if self.config.fix_base:
self.clip_model.requires_grad_(False)
for name, parms in self.clip_model.named_parameters():
if '_proj' in name:
parms.requires_grad_(False)
# fix certain ratio of layers
self.image_layer_num = 12
if self.config.fix_base > 0:
image_fix_num = "resblocks.{}".format(int(self.image_layer_num * self.config.fix_base))
for name, parms in self.clip_model.visual.named_parameters():
parms.requires_grad_(False)
if image_fix_num in name:
break
def loose_layer(self, fix_rate):
text_layer_id = [f"layer.{id}" for id in range(int(12 * fix_rate), 13)]
image_layer_id = [f"blocks.{id}" for id in range(int(24 * fix_rate), 25)]
for name, parms in self.blip.text_encoder.named_parameters():
for text_id in text_layer_id:
if text_id in name:
parms.requires_grad_(True)
for name, parms in self.blip.visual_encoder.named_parameters():
for image_id in image_layer_id:
if image_id in name:
parms.requires_grad_(True)
def forward(self, batch_data):
b_emb_inpt, b_emb_msk, w_emb_inpt, w_emb_msk = self.encode_pair(batch_data) # Nan
# forward
b_emb_feature = torch.cat((b_emb_inpt, b_emb_msk), dim=-1)
b_emb_feature = self.vit_block(b_emb_feature) # 1024
w_emb_feature = torch.cat((w_emb_inpt, w_emb_msk), dim=-1)
w_emb_feature = self.vit_block(w_emb_feature) # 1024
reward_better = self.mlp(b_emb_feature).squeeze(-1)
reward_worse = self.mlp(w_emb_feature).squeeze(-1)
reward = torch.concat((reward_better, reward_worse), dim=1)
return reward
def encode_pair(self, batch_data):
better_inpaint_embeds_bs, better_mask_rgb_embeds_bs = [], []
worse_inpaint_embeds_bs, worse_mask_rgb_embeds_bs = [], []
for bs in range(len(batch_data)):
better_inpt, better_msk = batch_data[bs]['better_inpt'], batch_data[bs]['better_msk']
better_inpt, better_msk = better_inpt.to(self.device), better_msk.to(self.device)
worse_inpt, worse_msk = batch_data[bs]['worse_inpt'], batch_data[bs]['worse_msk']
worse_inpt, worse_msk = worse_inpt.to(self.device), worse_msk.to(self.device)
# with torch.no_grad():
better_inpaint_embeds = self.clip_model.encode_image(better_inpt).to(torch.float32)
better_mask_rgb_embeds = self.clip_model.encode_image(better_msk).to(torch.float32)
worse_inpaint_embeds = self.clip_model.encode_image(worse_inpt).to(torch.float32)
worse_mask_rgb_embeds = self.clip_model.encode_image(worse_msk).to(torch.float32)
better_inpaint_embeds_bs.append(better_inpaint_embeds)
better_mask_rgb_embeds_bs.append(better_mask_rgb_embeds)
worse_inpaint_embeds_bs.append(worse_inpaint_embeds)
worse_mask_rgb_embeds_bs.append(worse_mask_rgb_embeds)
b_inpt = torch.stack(better_inpaint_embeds_bs, dim=0)
b_msk = torch.stack(better_mask_rgb_embeds_bs, dim=0)
w_inpt = torch.stack(worse_inpaint_embeds_bs, dim=0)
w_msk = torch.stack(worse_mask_rgb_embeds_bs, dim=0)
return b_inpt, b_msk, w_inpt, w_msk
def load_model(self, model, ckpt_path = None):
print('load checkpoint from %s'%ckpt_path)
checkpoint = torch.load(ckpt_path, map_location='cpu')
state_dict = checkpoint
msg = model.load_state_dict(state_dict,strict=False)
print("missing keys:", msg.missing_keys)
return model