Solution Author: dauzhan beketov
Solution Setup
Copied from baseline
#Made by DauJohn for you to solve his task.
import json
from pathlib import Path
import numpy as np
import pandas as pd
from tqdm.auto import tqdm
from PIL import Image
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt
from transformers import CLIPModel, CLIPProcessor, ViTImageProcessor, ViTModel
TRAIN_ROOT = Path("/kaggle/input/competitions/what-i-want-and-what-i-dont-aicc-round-10/train")
TEST_ROOT = Path("/kaggle/input/competitions/what-i-want-and-what-i-dont-aicc-round-10/test")
def load_tasks(root):
with (root / "tasks.json").open("r", encoding="utf-8") as f:
rows = json.load(f)
return {str(row["id"]): row for row in rows}
TRAIN_TASKS = load_tasks(TRAIN_ROOT)
TEST_TASKS = load_tasks(TEST_ROOT)
print("Train:", len(TRAIN_TASKS))
print("Test:", len(TEST_TASKS))
The history saving thread hit an unexpected error (OperationalError('attempt to write a readonly database')).History will not be written to the database.
Train: 15
Test: 150
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch16")
clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch16").to(DEVICE).eval()
dino_processor = ViTImageProcessor.from_pretrained("facebook/dino-vitb16")
dino_model = ViTModel.from_pretrained("facebook/dino-vitb16").to(DEVICE).eval()
@torch.inference_mode()
def get_clip_image_embedding(image_path):
image = Image.open(image_path).resize((224, 224))
inputs = clip_processor(images=[image], return_tensors="pt", do_center_crop=False).to(DEVICE)
embedding = clip_model.get_image_features(**inputs)['pooler_output']
return F.normalize(embedding, dim=-1).cpu()
@torch.inference_mode()
def get_clip_text_embedding(text):
inputs = clip_processor(text=[text], return_tensors="pt", padding=True, truncation=True).to(DEVICE)
embedding = clip_model.get_text_features(**inputs)['pooler_output']
return F.normalize(embedding, dim=-1).cpu()
def get_clip_texts_embedding(texts):
inputs = clip_processor(text=texts, return_tensors="pt", padding=True, truncation=True).to(DEVICE)
embedding = clip_model.get_text_features(**inputs)['pooler_output']
return F.normalize(embedding, dim=-1).cpu().detach()
@torch.inference_mode()
def get_dino_image_embedding(image_path):
image = Image.open(image_path).resize((224, 224))
inputs = dino_processor(images=[image], return_tensors="pt", do_center_crop=False).to(DEVICE)
output = dino_model(**inputs)
embedding = output['pooler_output']
return F.normalize(embedding, dim=-1).cpu()
s1 = get_clip_image_embedding('/kaggle/input/competitions/what-i-want-and-what-i-dont-aicc-round-10/train/images/000.png')
print(s1.shape)
s2 = get_clip_text_embedding('banana')
print(s2.shape)
s3 = get_dino_image_embedding('/kaggle/input/competitions/what-i-want-and-what-i-dont-aicc-round-10/train/images/000.png')
print(s3.shape)
def load_mask_mapping(sample_id):
mask_dir = TRAIN_ROOT / "masks" / str(sample_id)
with (mask_dir / "masks.json").open("r", encoding="utf-8") as f:
manifest = json.load(f)
return {
item["label"]: mask_dir / item["file"]
for item in manifest["masks"]
}
def score_sample(row, bbox):
mask_mapping = load_mask_mapping(str(row["id"]))
x1, y1, x2, y2 = bbox
tp = fp = fn = 0
labels = (
[(label, 1) for label in row["objects_a"]]
+ [(label, 0) for label in row["objects_b"]]
)
for label, truth in labels:
with Image.open(mask_mapping[label]) as image:
mask = np.asarray(image.convert("L")) > 0
total = int(mask.sum())
inside = int(mask[y1:y2, x1:x2].sum())
if truth:
correct = inside >= total * 0.8
tp += int(correct)
fn += int(not correct)
else:
fp += int(inside > total * 0.2)
precision = tp / (tp + fp) if tp + fp else 0.0
recall = tp / (tp + fn) if tp + fn else 0.0
f1 = 2 * tp / (2 * tp + fp + fn) if 2 * tp + fp + fn else 0.0
return precision, recall, f1
Simple Solution (No MaskCLIP, or other CLIP tricks) ~0.84
def get_clip_images_embedding(imagese):
inputs = clip_processor(images=imagese, return_tensors="pt", do_center_crop=False).to(DEVICE)
embedding = clip_model.get_image_features(**inputs)['pooler_output'].detach()
return F.normalize(embedding, dim=-1).cpu()
B = 28
def get_patch_embs(image_path):
K = 3
KK = 5
image = Image.open(image_path).convert("RGB").resize((224*K, 224*K))
npimage = np.array(image)
emb = []
imagese = []
for g1 in range(0, 224*K-224+1, 112):
# print(g1)
for g2 in range(0, 224*K-224+1, 112):
pimage = Image.fromarray(npimage[g1:g1+224,g2:g2+224])
imagese.append(pimage)
for l in range(0, len(imagese), 32):
r = min(l + 32, len(imagese))
emb.append(get_clip_images_embedding(imagese[l:r]))
emb = torch.concat(emb).reshape((KK, KK, -1))
emb = emb.permute(2, 0, 1).unsqueeze(0)
emb = F.interpolate(emb, size=(B, B), mode="bilinear", align_corners=False)
emb = emb.squeeze(0).permute(1, 2, 0).reshape((B*B, -1))
return emb
def predict_bbox(sample_id, split="test"):
tasks = TRAIN_TASKS if split == "train" else TEST_TASKS
root = TRAIN_ROOT if split == "train" else TEST_ROOT
row = tasks[str(sample_id)]
objects_a = row["objects_a"]
objects_b = row["objects_b"]
emb = get_patch_embs(root / row["image"])
oba = []
for id1 in range(len(objects_a)):
oba.append(objects_a[id1].split(',')[0].replace('_', ' '))
obb = []
for id2 in range(len(objects_b)):
obb.append(objects_b[id2].split(',')[0].replace('_', ' '))
emba = get_clip_texts_embedding(oba)
embb = get_clip_texts_embedding(obb)
scoresA = (emba @ emb.T)
scoresB = (embb @ emb.T)
for i1 in range(len(oba)):
scoresA[i1] -= scoresA[i1].min()
scoresA[i1] /= scoresA[i1].sum()
scoresA = scoresA.mean(dim = 0)
for i2 in range(len(obb)):
scoresB[i2] -= scoresB[i2].min()
scoresB[i2] /= scoresB[i2].sum()
scoresB = scoresB.mean(dim = 0)
scoresC = scoresA - scoresB
scoresC = scoresC.reshape((B, B)).detach().numpy()
bst = -10000000
bbox = [0, 0, 0, 0]
for d1 in range(B):
for d2 in range(B):
for d3 in range(d1, B):
for d4 in range(d2, B):
cur = scoresC[d1:d3+1, d2:d4+1].sum()
# print(cur)
if cur > bst:
bst = cur
bbox = [d1*8/224, d2*8/224, (d3*8+8)/224, (d4*8+8)/224]
img = Image.open(root / row["image"])
bbox[0] *= img.size[1]
bbox[2] *= img.size[1]
bbox[1] *= img.size[0]
bbox[3] *= img.size[0]
bbox[0] = round(int(bbox[0]))
bbox[1] = round(int(bbox[1]))
bbox[2] = round(int(bbox[2])) - 1
bbox[3] = round(int(bbox[3])) - 1
return [bbox[1], bbox[0], bbox[3], bbox[2]]
Solution 1 (MaskCLIP to extract embeddings map) ~0.89
def get_patch_embs(image_path):
image = Image.open(image_path).convert("RGB").resize((224, 224))
inputs = clip_processor(images = image, return_tensors = "pt", do_center_crop=False).to(DEVICE)
x = clip_model.vision_model.embeddings(**inputs)
x = clip_model.vision_model.pre_layrnorm(x)
for layer in clip_model.vision_model.encoder.layers[:-1]:
x = layer(x, attention_mask=None)
last = clip_model.vision_model.encoder.layers[-1]
x = last.layer_norm1(x)
x = last.self_attn.v_proj(x)
x = last.self_attn.out_proj(x)
x = clip_model.vision_model.post_layernorm(x)
x = clip_model.visual_projection(x)
x = x[:, 1:]
x = F.normalize(x, dim=-1)
return x.detach().cpu()
def predict_bbox(sample_id, split="test"):
tasks = TRAIN_TASKS if split == "train" else TEST_TASKS
root = TRAIN_ROOT if split == "train" else TEST_ROOT
row = tasks[str(sample_id)]
objects_a = row["objects_a"]
objects_b = row["objects_b"]
emb = get_patch_embs(root / row["image"])[0]
oba = []
for id1 in range(len(objects_a)):
oba.append(objects_a[id1].split(',')[0].replace('_', ' '))
obb = []
for id2 in range(len(objects_b)):
obb.append(objects_b[id2].split(',')[0].replace('_', ' '))
emba = get_clip_texts_embedding(oba)
embb = get_clip_texts_embedding(obb)
scoresA = (emba @ emb.T)
scoresB = (embb @ emb.T)
for i1 in range(len(oba)):
scoresA[i1] -= scoresA[i1].min()
scoresA[i1] /= scoresA[i1].sum()
scoresA = scoresA.mean(dim = 0)
for i2 in range(len(obb)):
scoresB[i2] -= scoresB[i2].min()
scoresB[i2] /= scoresB[i2].sum()
scoresB = scoresB.mean(dim = 0)
scoresC = scoresA - scoresB
scoresC = scoresC.reshape((14, 14)).detach().numpy()
bst = -10000000
bbox = [0, 0, 0, 0]
for d1 in range(14):
for d2 in range(14):
for d3 in range(d1, 14):
for d4 in range(d2, 14):
cur = scoresC[d1:d3+1, d2:d4+1].sum()
if cur > bst:
bst = cur
bbox = [d1*16/224, d2*16/224, (d3*16+16)/224, (d4*16+16)/224]
img = Image.open(root / row["image"])
bbox[0] *= img.size[1]
bbox[2] *= img.size[1]
bbox[1] *= img.size[0]
bbox[3] *= img.size[0]
bbox[0] = round(int(bbox[0]))
bbox[1] = round(int(bbox[1]))
bbox[2] = round(int(bbox[2])) - 1
bbox[3] = round(int(bbox[3])) - 1
return [bbox[1], bbox[0], bbox[3], bbox[2]]
Solution 2 ~0.94
clip_mean = torch.tensor(clip_processor.image_processor.image_mean, device=DEVICE).view(3, 1, 1)
clip_std = torch.tensor(clip_processor.image_processor.image_std, device=DEVICE).view(3, 1, 1)
dino_mean = torch.tensor(dino_processor.image_mean, device=DEVICE).view(3, 1, 1)
dino_std = torch.tensor(dino_processor.image_std, device=DEVICE).view(3, 1, 1)
@torch.inference_mode()
def get_patch_embs(image):
pixels = torch.tensor(np.asarray(image.resize((448, 448))), device=DEVICE).permute(2, 0, 1) / 255
x = clip_model.vision_model.embeddings(((pixels - clip_mean) / clip_std)[None], interpolate_pos_encoding=True)
x = clip_model.vision_model.pre_layrnorm(x)
for layer in clip_model.vision_model.encoder.layers[:-1]:
x = layer(x, attention_mask=None)
last = clip_model.vision_model.encoder.layers[-1]
values = last.self_attn.v_proj(last.layer_norm1(x))[0, 1:]
patches = F.normalize(dino_model(pixel_values=((pixels - dino_mean) / dino_std)[None], interpolate_pos_encoding=True).last_hidden_state[0, 1:], dim=-1)
affinity = patches @ patches.T
affinity = (affinity * 3).masked_fill(affinity < affinity.mean(), -torch.inf).softmax(-1)
x = last.self_attn.out_proj(affinity @ values)
x = clip_model.visual_projection(clip_model.vision_model.post_layernorm(x))
return F.normalize(x, dim=-1).reshape(28, 28, -1)
samples = []
for sample_id, row in TRAIN_TASKS.items():
emb = get_patch_embs(Image.open(TRAIN_ROOT / row["image"])).clone()
mask_files = load_mask_mapping(sample_id)
masks = torch.stack([torch.tensor(np.asarray(Image.open(path)) > 0, dtype=torch.float32, device=DEVICE) for path in mask_files.values()])
targets = F.adaptive_avg_pool2d(masks[None], emb.shape[:2])[0].flatten(1)
positive = (targets > 0.5).float()
weight = (positive / positive.sum(1, keepdim=True).clamp_min(1) + (1 - positive) / (1 - positive).sum(1, keepdim=True)) / 2
text = get_clip_texts_embedding(list(mask_files)).detach().to(DEVICE)
samples.append((emb.flatten(0, 1), text, targets, weight))
W = torch.eye(512, device=DEVICE, requires_grad=True)
opt = torch.optim.Adam([W])
for _ in tqdm(range(300)):
loss = 0.0
for emb, text, targets, weight in samples:
logits = 25 * text @ F.normalize(emb @ W.T, dim=-1).T - 5
loss = loss + F.binary_cross_entropy_with_logits(logits, targets, weight=weight, reduction="sum") / len(text)
opt.zero_grad()
(loss / len(samples)).backward()
opt.step()
0%| | 0/300 [00:00<?, ?it/s]
from skimage.filters import threshold_otsu
@torch.inference_mode()
def predict_bbox(sample_id):
row = TEST_TASKS[sample_id]
image = Image.open(TEST_ROOT / row["image"])
emb = F.normalize(get_patch_embs(image) @ W.T, dim=-1)
prompts = [f"a photo of a {label}." for label in row["objects_a"] + row["objects_b"]]
sim = torch.einsum("ld,hwd->lhw", get_clip_texts_embedding(prompts).to(DEVICE), emb)
mass = torch.stack([(s - threshold_otsu(s.cpu().numpy())).clamp_min(0) for s in sim])
mass = mass / mass.sum((1, 2), keepdim=True)
integral = F.pad(mass.cumsum(1).cumsum(2), (1, 0, 1, 0))
lo, hi = torch.triu_indices(29, 29, offset=1, device=DEVICE)
rows = integral[:, hi] - integral[:, lo]
frac = (rows[:, :, hi] - rows[:, :, lo]).flatten(1)
wanted = len(row["objects_a"])
top, left = divmod(int((frac[:wanted].mean(0) - frac[wanted:].mean(0)).argmax()), len(lo))
scale_x, scale_y = image.width / 28, image.height / 28
x1, y1 = int(lo[left].item() * scale_x), int(lo[top].item() * scale_y)
x2, y2 = int(np.ceil(hi[left].item() * scale_x)), int(np.ceil(hi[top].item() * scale_y))
return [x1, y1, min(x2, image.width), min(y2, image.height)]
submission = pd.DataFrame([
{
"id": sample_id,
"bbox": json.dumps(
predict_bbox(sample_id)
),
}
for sample_id in tqdm(TEST_TASKS, desc='Answering test tasks')
])
submission.to_csv(
"/kaggle/working/submission.csv",
index=False,
)
display(submission.head())
print("Saved: /kaggle/working/submission.csv")
Answering test tasks: 0%| | 0/150 [00:00<?, ?it/s] id bbox
0 000 [0, 658, 1683, 1536]
1 001 [0, 742, 1200, 1600]
2 002 [0, 91, 165, 256]
3 003 [487, 0, 871, 697]
4 004 [0, 183, 618, 428]Saved: /kaggle/working/submission.csv