File size: 4,420 Bytes
186701e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
# Copyright (c) Lin Song. All rights reserved.
import os
import json
import argparse
import os.path as osp
import torch
import torch.nn.functional as F
from mmengine.config import Config, DictAction
from mmengine.runner import Runner
from mmengine.dataset import Compose
from mmyolo.registry import RUNNERS
def get_caption_embed(runner, caption, prompt_template):
captions = json.load(open(caption, 'r'))
captions = [[prompt_template.format(c[0])] for c in captions]
with torch.no_grad():
embed = runner.model.backbone.text_model(captions)
embed = F.normalize(embed[:, 0, :], dim=1, p=2)
embed = embed.detach().cpu()
embed = embed[:, :, None, None]
return embed
def convert(runner, caption, checkpoint, prompt_template):
checkpoint = torch.load(checkpoint, map_location='cpu')
state_dict = checkpoint['state_dict']
embed = get_caption_embed(runner, caption, prompt_template)
import ipdb; ipdb.set_trace()
new_state_dict = {}
for key in list(state_dict.keys()):
if key.startswith('backbone.text_model'):
continue
elif key.startswith('backbone.image_model'):
new_key = key.replace('backbone.image_model', 'backbone')
new_state_dict[new_key] = state_dict[key].clone()
elif key.startswith('bbox_head.head_module.cls_contrasts'):
module_key = '.'.join(key.split('.')[:4])
logit_scale = state_dict[module_key + '.logit_scale']
bias = state_dict[module_key + '.bias']
conv_weight = embed * logit_scale.exp()
conv_bias = bias.repeat(conv_weight.shape[0])
new_state_dict[module_key + '.conv.weight'] = conv_weight
new_state_dict[module_key + '.conv.bias'] = conv_bias
else:
new_state_dict[key] = state_dict[key].clone()
new_checkpoint = {'state_dict': new_state_dict}
return new_checkpoint
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument('config', type=str)
parser.add_argument('checkpoint', type=str)
parser.add_argument('caption', type=str)
parser.add_argument('output', type=str)
parser.add_argument('--prompt-template', type=str,
default='{}')
parser.add_argument(
'--work-dir',
help='the directory to save the file containing evaluation metrics')
parser.add_argument(
'--cfg-options',
nargs='+',
action=DictAction,
help='override some settings in the used config, the key-value pair '
'in xxx=yyy format will be merged into config file. If the value to '
'be overwritten is a list, it should be like key="[a,b]" or key=a,b '
'It also allows nested list/tuple values, e.g. key="[(a,b),(c,d)]" '
'Note that the quotation marks are necessary and that no white space '
'is allowed.')
args = parser.parse_args()
return args
if __name__ == '__main__':
args = parse_args()
# load config
cfg = Config.fromfile(args.config)
# replace the ${key} with the value of cfg.key
# cfg = replace_cfg_vals(cfg)
if args.cfg_options is not None:
cfg.merge_from_dict(args.cfg_options)
# work_dir is determined in this priority: CLI > segment in file > filename
if args.work_dir is not None:
# update configs according to CLI args if args.work_dir is not None
cfg.work_dir = args.work_dir
elif cfg.get('work_dir', None) is None:
# use config filename as default work_dir if cfg.work_dir is None
cfg.work_dir = osp.join('./work_dirs',
osp.splitext(osp.basename(args.config))[0])
cfg.load_from = args.checkpoint
# build the runner from config
if 'runner_type' not in cfg:
# build the default runner
runner = Runner.from_cfg(cfg)
else:
# build customized runner from the registry
# if 'runner_type' is set in the cfg
runner = RUNNERS.build(cfg)
runner.call_hook('before_run')
runner.load_or_resume()
pipeline = cfg.test_dataloader.dataset.pipeline
runner.pipeline = Compose(pipeline)
runner.model.eval()
new_checkpoint = convert(runner, args.caption, args.checkpoint,
args.prompt_template)
os.makedirs(os.path.dirname(args.output), exist_ok=True)
torch.save(new_checkpoint, args.output)
|