Delete clip

clip/build_text_index.py

@@ -1,105 +0,0 @@
-import sys
-import torch
-import numpy as np
-import progressbar
-import os
-
-def parse_config():
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--clip_name", type=str, default="openai/clip-vit-base-patch32")
-    parser.add_argument("--text_file_path", type=str)
-    # save configuration
-    parser.add_argument("--save_index_prefix", type=str, help='where to save the mips index')
-    parser.add_argument("--save_index_name", type=str)
-    parser.add_argument("--save_mapping_dict_name", type=str, 
-        help="a json file that stores a dictory. The dictory contains mapping between mips index and caption text")
-    # inference configuration
-    parser.add_argument("--batch_size", type=int, help="the batch size used to conduct inference with CLIP")
-    return parser.parse_args()
-
-def load_batch_text(text_file_path, batch_size):
-    import json
-    with open(text_file_path) as f:
-        item_list = json.load(f)
-
-    text_list = []
-    for item in item_list:
-        captions = item["captions"]
-        for cap in captions:
-            text_list.append(cap)
-    print ('Number of text instances is {}'.format(len(text_list)))
-
-    data_num = len(text_list)
-    batch_num = data_num // batch_size
-    batch_text_list = []
-    s_idx, e_idx = 0, batch_size
-    for p_idx in range(batch_num):
-        one_batch_text_list = []
-        for idx in range(s_idx, e_idx):
-            one_batch_text_list.append(text_list[idx])
-        batch_text_list.append(one_batch_text_list)
-    return batch_text_list
-
-
-import argparse
-if __name__ == '__main__':
-    if torch.cuda.is_available():
-        print ('Cuda is available.')
-    cuda_available = torch.cuda.is_available()
-    args = parse_config()
-    device = torch.device('cuda')
-
-    import os
-    if os.path.exists(args.save_index_prefix):
-        pass
-    else: # recursively construct directory
-        os.makedirs(args.save_index_prefix, exist_ok=True)
-
-    print ('Loading CLIP...')
-    from clip import CLIP
-    model = CLIP(args.clip_name)
-    if cuda_available:
-        model = model.cuda(device)
-    model.eval()
-    print ('CLIP loaded!')
-
-    print ('Loading text data...')
-    batch_text_list = load_batch_text(args.text_file_path, args.batch_size)
-    print ('Text data loaded.')
-
-    res_text_vec_list, res_text_list = [], []
-    batch_num = len(batch_text_list)
-    print ('Number of batches is {}'.format(batch_num))
-    print ('Start inference...')
-    p = progressbar.ProgressBar(batch_num)
-    p.start()
-    with torch.no_grad():
-        for p_idx in range(batch_num):
-            p.update(p_idx)
-            one_text_batch = batch_text_list[p_idx]
-            one_batch_vec = model.compute_batch_index_text_representation(one_text_batch).detach().cpu()
-            one_batch_vec_list = one_batch_vec.unbind(dim=0)
-            bsz = len(one_batch_vec_list)
-            for k in range(bsz):
-                res_text_vec_list.append(one_batch_vec_list[k].numpy())
-                res_text_list.append(one_text_batch[k])
-    p.finish()
-    assert len(res_text_vec_list) == len(res_text_list)
-    print ('Inference completed!')
-
-    index_text_mapping_dict = {}
-    for k in range(len(res_text_list)):
-        index_text_mapping_dict[k] = res_text_list[k]
-    mapping_list_save_path = args.save_index_prefix + '/' + args.save_mapping_dict_name
-    import json
-    with open(mapping_list_save_path, 'w') as outfile:
-        json.dump(index_text_mapping_dict, outfile, indent=4)
-    print ('Mapping dictionary saved!')
-
-    print ('Start buiding index...')
-    index_save_path = args.save_index_prefix + '/' + args.save_index_name
-    with open(index_save_path, 'w', encoding = 'utf8') as o:
-        for vec in res_text_vec_list:
-            one_text = ' '.join([str(num) for num in vec]).strip()
-            o.writelines(one_text + '\n')
-    print ('Index completed!')

clip/clip.py

@@ -1,146 +0,0 @@
-import torch
-import requests
-from torch import nn
-from PIL import Image
-
-class CLIP(nn.Module):
-    def __init__(self, model_name):
-        super(CLIP, self).__init__()
-        # model name: e.g. openai/clip-vit-base-patch32
-        print ('Initializing CLIP model...')
-        from transformers import CLIPProcessor, CLIPModel
-        self.model = CLIPModel.from_pretrained(model_name)
-        self.model.eval()
-        self.processor = CLIPProcessor.from_pretrained(model_name)
-        from transformers import CLIPTokenizer
-        self.tokenizer = CLIPTokenizer.from_pretrained(model_name)
-        self.cuda_has_been_checked = False
-        print ('CLIP model initialized.')
-
-    def check_cuda(self):
-        self.cuda_available = next(self.model.parameters()).is_cuda
-        self.device = next(self.model.parameters()).get_device()
-        if self.cuda_available:
-            print ('Cuda is available.')
-            print ('Device is {}'.format(self.device))
-        else:
-            print ('Cuda is not available.')
-            print ('Device is {}'.format(self.device))
-
-    @torch.no_grad()
-    def compute_image_representation_from_image_path(self, image_path):
-        if not self.cuda_has_been_checked:
-            self.check_cuda()
-            self.cuda_has_been_checked = True
-        else:
-            pass
-        # image_path: the path of the image
-        image = Image.open(image_path)
-        inputs = self.processor(images=image, return_tensors="pt")
-        pixel_values = inputs['pixel_values']
-        if self.cuda_available:
-            pixel_values = pixel_values.cuda(self.device)
-        visual_outputs = self.model.vision_model(pixel_values=pixel_values)
-        image_embeds = visual_outputs[1]
-        image_embeds = self.model.visual_projection(image_embeds) # [1 x embed_dim]
-        return image_embeds
-
-    def compute_image_representation_from_image_instance(self, image):
-        if not self.cuda_has_been_checked:
-            self.check_cuda()
-            self.cuda_has_been_checked = True
-        else:
-            pass
-        # image_path: the path of the image
-        inputs = self.processor(images=image, return_tensors="pt")
-        pixel_values = inputs['pixel_values']
-        if self.cuda_available:
-            pixel_values = pixel_values.cuda(self.device)
-        visual_outputs = self.model.vision_model(pixel_values=pixel_values)
-        image_embeds = visual_outputs[1]
-        image_embeds = self.model.visual_projection(image_embeds) # [1 x embed_dim]
-        return image_embeds
-
-    def compute_text_representation(self, text_list):
-        if not self.cuda_has_been_checked:
-            self.check_cuda()
-            self.cuda_has_been_checked = True
-        else:
-            pass
-        # text_list: a list of text
-        text_inputs = self.tokenizer(text_list, padding=True, return_tensors="pt",
-            max_length=self.tokenizer.max_len_single_sentence + 2, truncation=True)
-        # self.tokenizer.max_len_single_sentence + 2 = 77
-        input_ids, attention_mask = text_inputs['input_ids'], text_inputs['attention_mask']
-        if self.cuda_available:
-            input_ids = input_ids.cuda(self.device)
-            attention_mask = attention_mask.cuda(self.device)
-        text_outputs = self.model.text_model(
-            input_ids=input_ids,
-            attention_mask=attention_mask
-        )
-        text_embeds = text_outputs[1]
-        text_embeds = self.model.text_projection(text_embeds)
-        return text_embeds
-
-    def compute_image_text_similarity_via_embeddings(self, image_embeds, text_embeds):
-        '''
-            image_embeds: 1 x embed_dim
-            text_embeds: len(text_list) x embed_dim
-        '''
-        image_embeds = image_embeds / image_embeds.norm(dim=-1, keepdim=True)
-        text_embeds = text_embeds / text_embeds.norm(dim=-1, keepdim=True)
-        logit_scale = self.model.logit_scale.exp()
-        logits_per_text = torch.matmul(text_embeds, image_embeds.t()) * logit_scale
-        logits_per_image = logits_per_text.T
-        return logits_per_image.softmax(dim=1), logits_per_image/logit_scale # 1 x len(text_list)
-
-    def compute_image_text_similarity_via_raw_text(self, image_embeds, text_list):
-        text_embeds = self.compute_text_representation(text_list)
-        return self.compute_image_text_similarity_via_embeddings(image_embeds, text_embeds)
-
-    ### -------------------- functions for building index ---------------------- ###
-    def compute_batch_index_image_features(self, image_list):
-        '''
-            # list of image instances
-        '''
-        if not self.cuda_has_been_checked:
-            self.check_cuda()
-            self.cuda_has_been_checked = True
-        else:
-            pass
-        # image_path: the path of the image
-        inputs = self.processor(images=image_list, return_tensors="pt")
-        pixel_values = inputs['pixel_values']
-        if self.cuda_available:
-            pixel_values = pixel_values.cuda(self.device)
-        visual_outputs = self.model.vision_model(pixel_values=pixel_values)
-        image_embeds = visual_outputs[1]
-        image_embeds = self.model.visual_projection(image_embeds) # [1 x embed_dim]
-        return image_embeds # len(image_list) x embed_dim
-
-    def compute_batch_index_text_representation(self, text_list):
-        if not self.cuda_has_been_checked:
-            self.check_cuda()
-            self.cuda_has_been_checked = True
-        else:
-            pass
-        # text_list: a list of text
-        #text_inputs = self.tokenizer(text_list, padding=True, return_tensors="pt")
-        text_inputs = self.tokenizer(text_list, padding=True, return_tensors="pt",
-            max_length=self.tokenizer.max_len_single_sentence + 2, truncation=True)
-        input_ids, attention_mask = text_inputs['input_ids'], text_inputs['attention_mask']
-        if self.cuda_available:
-            input_ids = input_ids.cuda(self.device)
-            attention_mask = attention_mask.cuda(self.device)
-        text_outputs = self.model.text_model(
-            input_ids=input_ids,
-            attention_mask=attention_mask
-        )
-        text_embeds = text_outputs[1]
-        text_embeds = self.model.text_projection(text_embeds)
-        return text_embeds
-        #logit_scale = self.model.logit_scale.exp()
-        #text_embeds = text_embeds * logit_scale
-        #return text_embeds
-

clip/clipretrieval.py

@@ -1,135 +0,0 @@
-import json
-import copy
-import torch
-import progressbar
-import numpy as np
-from PIL import Image
-
-class CLIPIndex:
-    def __init__(self, index_matrix_path, mapping_dict_path, clip):
-        '''
-            index_path: the pre-trained index
-            mapping_dict_path: the pre-indexed mapping dictionary
-            clip: the pre-trained clip model
-        '''
-        print ('Loading index...')
-        self.index_matrix = self.normalization(self.load_matrix(index_matrix_path))
-        print ('Index loaded.')
-        print (self.index_matrix.shape)
-        with open(mapping_dict_path) as f:
-            self.mapping_dict = json.load(f)
-        self.clip = clip
-
-    def load_matrix(self, in_f):
-        matrix_list = []
-        with open(in_f, 'r', encoding = 'utf8') as i:
-            lines = i.readlines()
-            for l in lines:
-                one_vec = [float(num) for num in l.strip('\n').split()]
-                matrix_list.append(one_vec)
-        return np.array(matrix_list)
-
-    def normalization(self, matrix):
-        '''
-            matrix: num_instance x num_feature
-        '''
-        return matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
-
-    def get_image_representation(self, image_path):
-        image_instance = Image.open(image_path)
-        image_vec = self.clip.compute_batch_index_image_features([image_instance]).detach().cpu().numpy()
-        image_vec = self.normalization(image_vec)
-        return image_vec
-
-    def search_text(self, image_path):
-        image_vec = self.get_image_representation(image_path)
-        sort_idx_list = np.matmul(image_vec, self.index_matrix.transpose())[0].argsort()[::-1]
-        top_idx = sort_idx_list[0]
-        return self.mapping_dict[str(top_idx)]
-
-
-def parse_config():
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--clip_name", type=str)
-    parser.add_argument("--test_image_prefix_path", type=str, help="the folder that stores all test images")
-    parser.add_argument("--test_path", type=str)
-    # index configuration
-    parser.add_argument("--index_matrix_path", type=str)
-    parser.add_argument("--mapping_dict_path", type=str)
-    # save configuration
-    parser.add_argument("--save_path_prefix", type=str, help="save the result in which directory")
-    parser.add_argument("--save_name", type=str, help="the name of the saved file")
-    return parser.parse_args()
-
-import argparse
-if __name__ == '__main__':
-    if torch.cuda.is_available():
-        print ('Cuda is available.')
-    cuda_available = torch.cuda.is_available()
-    args = parse_config()
-    device = torch.device('cuda')
-
-    save_path_prefix = args.save_path_prefix
-    import os
-    if os.path.exists(save_path_prefix):
-        pass
-    else: # recursively construct directory
-        os.makedirs(save_path_prefix, exist_ok=True)
-    # parse save name
-    save_name = args.save_name
-    full_save_path = save_path_prefix + '/' + save_name
-    print ('full save path is {}'.format(full_save_path))
-
-    print ('Loading CLIP...')
-    from clip import CLIP
-    clip = CLIP(args.clip_name)
-    if cuda_available:
-        clip = clip.cuda(device)
-    clip.eval()
-    print ('CLIP loaded!')
-
-    clipindex = CLIPIndex(args.index_matrix_path, args.mapping_dict_path, clip)
-
-    print ('Loading data...')
-    import json
-    with open(args.test_path) as f:
-        item_list = json.load(f)
-    print ('Data loaded.')
-    print ('Number of test instances is {}'.format(len(item_list)))
-
-    result_list = []
-    invalid_num = 0
-    print ('----------------------------------------------------------------')
-    with torch.no_grad():
-        test_num = len(item_list)
-        #test_num = 10
-        print ('Number of inference instances is {}'.format(test_num))
-        p = progressbar.ProgressBar(test_num)
-        p.start()
-        for p_idx in range(test_num):
-            p.update(p_idx)
-            one_test_dict = item_list[p_idx]
-
-            one_res_dict = {
-                'split':one_test_dict['split'],
-                'image_name':one_test_dict['image_name'],
-                #'file_path':one_test_dict['file_path'],
-                'captions':one_test_dict['captions']
-            }
-
-            image_full_path = args.test_image_prefix_path + '/' + one_test_dict['image_name']
-            try:
-                output_text = clipindex.search_text(image_full_path)
-                one_res_dict['prediction'] = output_text
-                result_list.append(one_res_dict)
-            except:
-                invalid_num += 1
-                print ('invalid number is {}'.format(invalid_num))
-                continue
-        p.finish()
-    print ('Inference completed!')
-
-    import json
-    with open(full_save_path, 'w') as outfile:
-        json.dump(result_list, outfile, indent=4)
-