add app file

app.py

@@ -0,0 +1,189 @@
+import gradio as gr
+import torch
+import copy
+import time
+import requests
+import io
+import numpy as np
+import re
+from einops import rearrange
+
+import ipdb
+
+from PIL import Image
+
+from vilt.config import ex
+from vilt.modules import ViLTransformerSS
+
+from vilt.modules.objectives import cost_matrix_cosine, ipot
+from vilt.transforms import pixelbert_transform
+from vilt.datamodules.datamodule_base import get_pretrained_tokenizer
+
+
+@ex.automain
+def main(_config):
+    _config = copy.deepcopy(_config)
+
+    loss_names = {
+        "itm": 1,
+        "mlm": 0.5,
+        "mpp": 0,
+        "vqa": 0,
+        "imgcls": 0,
+        "nlvr2": 0,
+        "irtr": 1,
+        "arc": 0,
+    }
+    tokenizer = get_pretrained_tokenizer(_config["tokenizer"])
+
+    _config.update(
+        {
+            "loss_names": loss_names,
+        }
+    )
+
+    model = ViLTransformerSS(_config)
+    model.setup("test")
+    model.eval()
+
+    device = "cuda:0" if _config["num_gpus"] > 0 else "cpu"
+    model.to(device)
+    lst_imgs = [f"C:\\Users\\alimh\\PycharmProjects\\ViLT\\assets\\database\\{i}.jpg" for i in range(1,10)]
+
+
+    def infer( mp_text, hidx =0 ):
+        def get_image(path):
+            image = Image.open(path).convert("RGB")
+            img = pixelbert_transform(size=384)(image)
+            return img.unsqueeze(0).to(device)
+
+        imgs = [get_image(pth) for pth in lst_imgs]
+
+        batch = []
+        for img in imgs:
+            batch.append({"text": [mp_text], "image": [img]})
+
+        for dic in batch:
+            encoded = tokenizer(dic["text"])
+
+            dic["text_ids"] = torch.tensor(encoded["input_ids"]).to(device)
+            dic["text_labels"] = torch.tensor(encoded["input_ids"]).to(device)
+            dic["text_masks"] = torch.tensor(encoded["attention_mask"]).to(device)
+
+        scores = []
+        with torch.no_grad():
+
+            for dic in batch:
+                s = time.time()
+                infer = model(dic)
+
+                e = time.time()
+                print("time ", round(e - s, 2))
+
+                score = model.rank_output(infer["cls_feats"])
+                scores.append(score.item())
+            print(scores)
+            img_idx =np.argmax(scores)
+            print(np.argmax(scores) + 1 )
+            selected_image =  Image.open(lst_imgs[img_idx]).convert("RGB")
+            selected_image = np.asarray(selected_image)
+            print(selected_image.shape)
+            selected_token =""
+            if hidx > 0 and hidx < len(encoded["input_ids"][0][:-1]):
+                image = Image.open(lst_imgs[img_idx]).convert("RGB")
+                selected_batch  = batch[img_idx]
+                with torch.no_grad():
+                    infer = model(selected_batch)
+                    txt_emb, img_emb = infer["text_feats"], infer["image_feats"]
+                    txt_mask, img_mask = (
+                        infer["text_masks"].bool(),
+                        infer["image_masks"].bool(),
+                    )
+                    for i, _len in enumerate(txt_mask.sum(dim=1)):
+                        txt_mask[i, _len - 1] = False
+                    txt_mask[:, 0] = False
+                    img_mask[:, 0] = False
+                    txt_pad, img_pad = ~txt_mask, ~img_mask
+
+                    cost = cost_matrix_cosine(txt_emb.float(), img_emb.float())
+                    joint_pad = txt_pad.unsqueeze(-1) | img_pad.unsqueeze(-2)
+                    cost.masked_fill_(joint_pad, 0)
+
+                    txt_len = (txt_pad.size(1) - txt_pad.sum(dim=1, keepdim=False)).to(
+                        dtype=cost.dtype
+                    )
+                    img_len = (img_pad.size(1) - img_pad.sum(dim=1, keepdim=False)).to(
+                        dtype=cost.dtype
+                    )
+                    T = ipot(
+                        cost.detach(),
+                        txt_len,
+                        txt_pad,
+                        img_len,
+                        img_pad,
+                        joint_pad,
+                        0.1,
+                        1000,
+                        1,
+                    )
+
+                    plan = T[0]
+                    plan_single = plan * len(txt_emb)
+                    cost_ = plan_single.t()
+
+                    cost_ = cost_[hidx][1:].cpu()
+
+                    patch_index, (H, W) = infer["patch_index"]
+                    heatmap = torch.zeros(H, W)
+                    for i, pidx in enumerate(patch_index[0]):
+                        h, w = pidx[0].item(), pidx[1].item()
+                        heatmap[h, w] = cost_[i]
+
+                    heatmap = (heatmap - heatmap.mean()) / heatmap.std()
+                    heatmap = np.clip(heatmap, 1.0, 3.0)
+                    heatmap = (heatmap - heatmap.min()) / (heatmap.max() - heatmap.min())
+
+                    _w, _h = image.size
+                    overlay = Image.fromarray(np.uint8(heatmap * 255), "L").resize(
+                        (_w, _h), resample=Image.NEAREST
+                    )
+                    image_rgba = image.copy()
+                    image_rgba.putalpha(overlay)
+                    selected_image =  image_rgba
+
+                    selected_token = tokenizer.convert_ids_to_tokens(
+                        encoded["input_ids"][0][hidx]
+                    )
+
+
+        return [selected_image,hidx]
+
+    imgs = [Image.open(pth).convert("RGB") for pth in lst_imgs]
+    inputs = [
+
+        gr.inputs.Textbox(label="Caption with [MASK] tokens to be filled.", lines=5),
+        gr.inputs.Slider(
+            minimum=0,
+            maximum=38,
+            step=1,
+            label="Index of token for heatmap visualization (ignored if zero)",
+        ),
+    ]
+    outputs = [
+        gr.outputs.Image(label="Image"),
+
+
+        gr.outputs.Textbox(label="matching index "),
+    ]
+
+
+    interface = gr.Interface(
+        fn=infer,
+        inputs=inputs,
+        outputs=outputs,
+        server_name="localhost",
+        server_port=8888,
+
+    )
+
+    interface.launch(debug=True,share=False)