from typing import Optional, Tuple from einops import rearrange import torch import torch.nn.functional as F from PIL import Image import torchvision.transforms as transforms from torch import nn import numpy as np import os import time import gradio as gr use_cuda = torch.cuda.is_available() # use_cuda = False print("CUDA is available:", use_cuda) class MobileSAM(nn.Module): def __init__(self, **kwargs): super().__init__(**kwargs) from mobile_sam import sam_model_registry url = 'https://raw.githubusercontent.com/ChaoningZhang/MobileSAM/master/weights/mobile_sam.pt' model_type = "vit_t" sam_checkpoint = "mobile_sam.pt" if not os.path.exists(sam_checkpoint): import requests r = requests.get(url) with open(sam_checkpoint, 'wb') as f: f.write(r.content) device = 'cuda' if use_cuda else 'cpu' mobile_sam = sam_model_registry[model_type](checkpoint=sam_checkpoint) def new_forward_fn(self, x): shortcut = x x = self.conv1(x) x = self.act1(x) x = self.conv2(x) x = self.act2(x) self.attn_output = rearrange(x.clone(), "b c h w -> b h w c") x = self.conv3(x) self.mlp_output = rearrange(x.clone(), "b c h w -> b h w c") x = self.drop_path(x) x += shortcut x = self.act3(x) self.block_output = rearrange(x.clone(), "b c h w -> b h w c") return x setattr(mobile_sam.image_encoder.layers[0].blocks[0].__class__, "forward", new_forward_fn) def new_forward_fn2(self, x): H, W = self.input_resolution B, L, C = x.shape assert L == H * W, "input feature has wrong size" res_x = x if H == self.window_size and W == self.window_size: x = self.attn(x) else: x = x.view(B, H, W, C) pad_b = (self.window_size - H % self.window_size) % self.window_size pad_r = (self.window_size - W % self.window_size) % self.window_size padding = pad_b > 0 or pad_r > 0 if padding: x = F.pad(x, (0, 0, 0, pad_r, 0, pad_b)) pH, pW = H + pad_b, W + pad_r nH = pH // self.window_size nW = pW // self.window_size # window partition x = x.view(B, nH, self.window_size, nW, self.window_size, C).transpose(2, 3).reshape( B * nH * nW, self.window_size * self.window_size, C) x = self.attn(x) # window reverse x = x.view(B, nH, nW, self.window_size, self.window_size, C).transpose(2, 3).reshape(B, pH, pW, C) if padding: x = x[:, :H, :W].contiguous() x = x.view(B, L, C) hw = np.sqrt(x.shape[1]).astype(int) self.attn_output = rearrange(x.clone(), "b (h w) c -> b h w c", h=hw) x = res_x + self.drop_path(x) x = x.transpose(1, 2).reshape(B, C, H, W) x = self.local_conv(x) x = x.view(B, C, L).transpose(1, 2) mlp_output = self.mlp(x) self.mlp_output = rearrange(mlp_output.clone(), "b (h w) c -> b h w c", h=hw) x = x + self.drop_path(mlp_output) self.block_output = rearrange(x.clone(), "b (h w) c -> b h w c", h=hw) return x setattr(mobile_sam.image_encoder.layers[1].blocks[0].__class__, "forward", new_forward_fn2) mobile_sam.to(device=device) mobile_sam.eval() self.image_encoder = mobile_sam.image_encoder @torch.no_grad() def forward(self, x): with torch.no_grad(): x = torch.nn.functional.interpolate(x, size=(1024, 1024), mode="bilinear") out = self.image_encoder(x) attn_outputs, mlp_outputs, block_outputs = [], [], [] for i_layer in range(len(self.image_encoder.layers)): for i_block in range(len(self.image_encoder.layers[i_layer].blocks)): blk = self.image_encoder.layers[i_layer].blocks[i_block] attn_outputs.append(blk.attn_output) mlp_outputs.append(blk.mlp_output) block_outputs.append(blk.block_output) return attn_outputs, mlp_outputs, block_outputs def image_mobilesam_feature( images, resolution=(1024, 1024), node_type="block", layer=-1, ): transform = transforms.Compose( [ transforms.Resize(resolution), transforms.ToTensor(), transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), ] ) feat_extractor = MobileSAM() # attn_outputs, mlp_outputs, block_outputs = [], [], [] outputs = [] for i, image in enumerate(images): torch_image = transform(image) if use_cuda: torch_image = torch_image.cuda() attn_output, mlp_output, block_output = feat_extractor( torch_image.unsqueeze(0) ) out_dict = { "attn": attn_output, "mlp": mlp_output, "block": block_output, } out = out_dict[node_type] out = out[layer] outputs.append(out.cpu()) outputs = torch.cat(outputs, dim=0) return outputs class SAM(torch.nn.Module): def __init__(self, checkpoint="/data/sam_model/sam_vit_b_01ec64.pth", **kwargs): super().__init__(**kwargs) from segment_anything import sam_model_registry, SamPredictor from segment_anything.modeling.sam import Sam sam: Sam = sam_model_registry["vit_b"](checkpoint=checkpoint) from segment_anything.modeling.image_encoder import ( window_partition, window_unpartition, ) def new_block_forward(self, x: torch.Tensor) -> torch.Tensor: shortcut = x x = self.norm1(x) # Window partition if self.window_size > 0: H, W = x.shape[1], x.shape[2] x, pad_hw = window_partition(x, self.window_size) x = self.attn(x) # Reverse window partition if self.window_size > 0: x = window_unpartition(x, self.window_size, pad_hw, (H, W)) self.attn_output = x.clone() x = shortcut + x mlp_outout = self.mlp(self.norm2(x)) self.mlp_output = mlp_outout.clone() x = x + mlp_outout self.block_output = x.clone() return x setattr(sam.image_encoder.blocks[0].__class__, "forward", new_block_forward) self.image_encoder = sam.image_encoder self.image_encoder.eval() if use_cuda: self.image_encoder = self.image_encoder.cuda() @torch.no_grad() def forward(self, x: torch.Tensor) -> torch.Tensor: with torch.no_grad(): x = torch.nn.functional.interpolate(x, size=(1024, 1024), mode="bilinear") out = self.image_encoder(x) attn_outputs, mlp_outputs, block_outputs = [], [], [] for i, blk in enumerate(self.image_encoder.blocks): attn_outputs.append(blk.attn_output) mlp_outputs.append(blk.mlp_output) block_outputs.append(blk.block_output) attn_outputs = torch.stack(attn_outputs) mlp_outputs = torch.stack(mlp_outputs) block_outputs = torch.stack(block_outputs) return attn_outputs, mlp_outputs, block_outputs def image_sam_feature( images, resolution=(1024, 1024), node_type="block", layer=-1, ): transform = transforms.Compose( [ transforms.Resize(resolution), transforms.ToTensor(), transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), ] ) checkpoint = "sam_vit_b_01ec64.pth" if not os.path.exists(checkpoint): checkpoint_url = 'https://dl.fbaipublicfiles.com/segment_anything/sam_vit_b_01ec64.pth' import requests r = requests.get(checkpoint_url) with open(checkpoint, 'wb') as f: f.write(r.content) feat_extractor = SAM(checkpoint=checkpoint) # attn_outputs, mlp_outputs, block_outputs = [], [], [] outputs = [] for i, image in enumerate(images): torch_image = transform(image) if use_cuda: torch_image = torch_image.cuda() attn_output, mlp_output, block_output = feat_extractor( torch_image.unsqueeze(0) ) out_dict = { "attn": attn_output, "mlp": mlp_output, "block": block_output, } out = out_dict[node_type] out = out[layer] outputs.append(out.cpu()) outputs = torch.cat(outputs, dim=0) return outputs class DiNOv2(torch.nn.Module): def __init__(self, ver="dinov2_vitb14_reg"): super().__init__() self.dinov2 = torch.hub.load("facebookresearch/dinov2", ver) self.dinov2.requires_grad_(False) self.dinov2.eval() if use_cuda: self.dinov2 = self.dinov2.cuda() def new_block_forward(self, x: torch.Tensor) -> torch.Tensor: def attn_residual_func(x): return self.ls1(self.attn(self.norm1(x))) def ffn_residual_func(x): return self.ls2(self.mlp(self.norm2(x))) attn_output = attn_residual_func(x) self.attn_output = attn_output.clone() x = x + attn_output mlp_output = ffn_residual_func(x) self.mlp_output = mlp_output.clone() x = x + mlp_output block_output = x self.block_output = block_output.clone() return x setattr(self.dinov2.blocks[0].__class__, "forward", new_block_forward) @torch.no_grad() def forward(self, x): out = self.dinov2(x) attn_outputs, mlp_outputs, block_outputs = [], [], [] for i, blk in enumerate(self.dinov2.blocks): attn_outputs.append(blk.attn_output) mlp_outputs.append(blk.mlp_output) block_outputs.append(blk.block_output) attn_outputs = torch.stack(attn_outputs) mlp_outputs = torch.stack(mlp_outputs) block_outputs = torch.stack(block_outputs) return attn_outputs, mlp_outputs, block_outputs def image_dino_feature(images, resolution=(448, 448), node_type="block", layer=-1): transform = transforms.Compose( [ transforms.Resize(resolution), transforms.ToTensor(), transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), ] ) feat_extractor = DiNOv2() outputs = [] for i, image in enumerate(images): torch_image = transform(image) if use_cuda: torch_image = torch_image.cuda() attn_output, mlp_output, block_output = feat_extractor( torch_image.unsqueeze(0) ) out_dict = { "attn": attn_output, "mlp": mlp_output, "block": block_output, } out = out_dict[node_type] out = out[layer] outputs.append(out.cpu()) outputs = torch.cat(outputs, dim=0) outputs = rearrange(outputs[:, 5:, :], "b (h w) c -> b h w c", h=32, w=32) return outputs class CLIP(torch.nn.Module): def __init__(self): super().__init__() from transformers import CLIPProcessor, CLIPModel model = CLIPModel.from_pretrained("openai/clip-vit-base-patch16") # processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch16") self.model = model.eval() if use_cuda: self.model = self.model.cuda() def new_forward( self, hidden_states: torch.Tensor, attention_mask: torch.Tensor, causal_attention_mask: torch.Tensor, output_attentions: Optional[bool] = False, ) -> Tuple[torch.FloatTensor]: residual = hidden_states hidden_states = self.layer_norm1(hidden_states) hidden_states, attn_weights = self.self_attn( hidden_states=hidden_states, attention_mask=attention_mask, causal_attention_mask=causal_attention_mask, output_attentions=output_attentions, ) self.attn_output = hidden_states.clone() hidden_states = residual + hidden_states residual = hidden_states hidden_states = self.layer_norm2(hidden_states) hidden_states = self.mlp(hidden_states) self.mlp_output = hidden_states.clone() hidden_states = residual + hidden_states outputs = (hidden_states,) if output_attentions: outputs += (attn_weights,) self.block_output = hidden_states.clone() return outputs setattr(self.model.vision_model.encoder.layers[0].__class__, "forward", new_forward) @torch.no_grad() def forward(self, x): out = self.model.vision_model(x) attn_outputs, mlp_outputs, block_outputs = [], [], [] for i, blk in enumerate(self.model.vision_model.encoder.layers): attn_outputs.append(blk.attn_output) mlp_outputs.append(blk.mlp_output) block_outputs.append(blk.block_output) attn_outputs = torch.stack(attn_outputs) mlp_outputs = torch.stack(mlp_outputs) block_outputs = torch.stack(block_outputs) return attn_outputs, mlp_outputs, block_outputs def image_clip_feature( images, resolution=(224, 224), node_type="block", layer=-1 ): if isinstance(images, list): assert isinstance(images[0], Image.Image), "Input must be a list of PIL images." else: assert isinstance(images, Image.Image), "Input must be a PIL image." images = [images] transform = transforms.Compose( [ transforms.Resize(resolution), transforms.ToTensor(), transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), ] ) feat_extractor = CLIP() outputs = [] for i, image in enumerate(images): torch_image = transform(image) if use_cuda: torch_image = torch_image.cuda() attn_output, mlp_output, block_output = feat_extractor( torch_image.unsqueeze(0) ) out_dict = { "attn": attn_output, "mlp": mlp_output, "block": block_output, } out = out_dict[node_type] out = out[layer] outputs.append(out.cpu()) outputs = torch.cat(outputs, dim=0) return outputs def extract_features(images, model_name="sam", node_type="block", layer=-1): if model_name == "SAM(sam_vit_b)": return image_sam_feature(images, node_type=node_type, layer=layer) elif model_name == 'MobileSAM': return image_mobilesam_feature(images, node_type=node_type, layer=layer) elif model_name == "DiNO(dinov2_vitb14_reg)": return image_dino_feature(images, node_type=node_type, layer=layer) elif model_name == "CLIP(openai/clip-vit-base-patch16)": return image_clip_feature(images, node_type=node_type, layer=layer) else: raise ValueError(f"Model {model_name} not supported.") def compute_ncut( features, num_eig=100, num_sample_ncut=10000, affinity_focal_gamma=0.3, knn_ncut=10, knn_tsne=10, num_sample_tsne=1000, perplexity=500, ): from ncut_pytorch import NCUT, rgb_from_tsne_3d start = time.time() eigvecs, eigvals = NCUT( num_eig=num_eig, num_sample=num_sample_ncut, device="cuda" if use_cuda else "cpu", affinity_focal_gamma=affinity_focal_gamma, knn=knn_ncut, ).fit_transform(features.reshape(-1, features.shape[-1])) print(f"NCUT time: {time.time() - start:.2f}s") start = time.time() X_3d, rgb = rgb_from_tsne_3d( eigvecs, num_sample=num_sample_tsne, perplexity=perplexity, knn=knn_tsne, ) print(f"t-SNE time: {time.time() - start:.2f}s") # print("input shape:", features.shape) # print("output shape:", rgb.shape) rgb = rgb.reshape(features.shape[:3] + (3,)) return rgb def dont_use_too_much_green(image_rgb): # make sure the foval 40% of the image is red leading x1, x2 = int(image_rgb.shape[1] * 0.3), int(image_rgb.shape[1] * 0.7) y1, y2 = int(image_rgb.shape[2] * 0.3), int(image_rgb.shape[2] * 0.7) sum_values = image_rgb[:, x1:x2, y1:y2].mean((0, 1, 2)) sorted_indices = sum_values.argsort(descending=True) image_rgb = image_rgb[:, :, :, sorted_indices] return image_rgb def to_pil_images(images): return [ Image.fromarray((image * 255).cpu().numpy().astype(np.uint8)).resize((256, 256), Image.NEAREST) for image in images ] def main_fn( images, model_name="SAM(sam_vit_b)", node_type="block", layer=-1, num_eig=100, affinity_focal_gamma=0.3, num_sample_ncut=10000, knn_ncut=10, num_sample_tsne=1000, knn_tsne=10, perplexity=500, ): if perplexity >= num_sample_tsne: # raise gr.Error("Perplexity must be less than the number of samples for t-SNE.") gr.Warning("Perplexity must be less than the number of samples for t-SNE.\n" f"Setting perplexity to {num_sample_tsne-1}.") perplexity = num_sample_tsne - 1 images = [image[0] for image in images] start = time.time() features = extract_features( images, model_name=model_name, node_type=node_type, layer=layer ) print(f"Feature extraction time: {time.time() - start:.2f}s") rgb = compute_ncut( features, num_eig=num_eig, num_sample_ncut=num_sample_ncut, affinity_focal_gamma=affinity_focal_gamma, knn_ncut=knn_ncut, knn_tsne=knn_tsne, num_sample_tsne=num_sample_tsne, perplexity=perplexity, ) rgb = dont_use_too_much_green(rgb) return to_pil_images(rgb) default_images = ['./images/image_0.jpg', './images/image_1.jpg', './images/image_2.jpg', './images/image_3.jpg', './images/image_4.jpg', './images/image_5.jpg'] default_outputs = ['./images/ncut_0.jpg', './images/ncut_1.jpg', './images/ncut_2.jpg', './images/ncut_3.jpg', './images/ncut_4.jpg', './images/ncut_5.jpg'] demo = gr.Interface( main_fn, [ gr.Gallery(value=default_images, label="Select images", show_label=False, elem_id="images", columns=[3], rows=[1], object_fit="contain", height="auto", type="pil"), gr.Dropdown(["MobileSAM", "SAM(sam_vit_b)", "DiNO(dinov2_vitb14_reg)", "CLIP(openai/clip-vit-base-patch16)"], label="Model", value="MobileSAM", elem_id="model_name"), gr.Dropdown(["attn", "mlp", "block"], label="Node type", value="block", elem_id="node_type", info="attn: attention output, mlp: mlp output, block: sum of residual stream"), gr.Slider(0, 11, step=1, label="Layer", value=11, elem_id="layer", info="which layer of the image backbone features"), gr.Slider(1, 1000, step=1, label="Number of eigenvectors", value=100, elem_id="num_eig", info='increase for more object parts, decrease for whole object'), gr.Slider(0.01, 1, step=0.01, label="Affinity focal gamma", value=0.3, elem_id="affinity_focal_gamma", info="decrease for more aggressive cleaning on the affinity matrix"), ], gr.Gallery(value=default_outputs, label="NCUT Embedding", show_label=False, elem_id="ncut", columns=[3], rows=[1], object_fit="contain", height="auto"), additional_inputs=[ gr.Slider(100, 30000, step=100, label="num_sample (NCUT)", value=10000, elem_id="num_sample_ncut", info="for Nyström approximation"), gr.Slider(1, 100, step=1, label="KNN (NCUT)", value=10, elem_id="knn_ncut", info="for Nyström approximation"), gr.Slider(100, 10000, step=100, label="num_sample (t-SNE)", value=1000, elem_id="num_sample_tsne", info="for Nyström approximation. Adding will slow down t-SNE quite a lot"), gr.Slider(1, 100, step=1, label="KNN (t-SNE)", value=10, elem_id="knn_tsne", info="for Nyström approximation"), gr.Slider(10, 1000, step=10, label="Perplexity (t-SNE)", value=500, elem_id="perplexity", info="for t-SNE"), ] ) demo.launch()