|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import torch.nn as nn |
|
from transformers import ViTImageProcessor |
|
from einops import rearrange, repeat |
|
from .dino import ViTModel |
|
|
|
|
|
class DinoWrapper(nn.Module): |
|
""" |
|
Dino v1 wrapper using huggingface transformer implementation. |
|
""" |
|
def __init__(self, model_name: str, freeze: bool = True): |
|
super().__init__() |
|
self.model, self.processor = self._build_dino(model_name) |
|
self.camera_embedder = nn.Sequential( |
|
nn.Linear(16, self.model.config.hidden_size, bias=True), |
|
nn.SiLU(), |
|
nn.Linear(self.model.config.hidden_size, self.model.config.hidden_size, bias=True) |
|
) |
|
if freeze: |
|
self._freeze() |
|
|
|
def forward(self, image, camera): |
|
|
|
|
|
|
|
if image.ndim == 5: |
|
image = rearrange(image, 'b n c h w -> (b n) c h w') |
|
dtype = image.dtype |
|
inputs = self.processor( |
|
images=image.float(), |
|
return_tensors="pt", |
|
do_rescale=False, |
|
do_resize=False, |
|
).to(self.model.device).to(dtype) |
|
|
|
N = camera.shape[1] |
|
camera_embeddings = self.camera_embedder(camera) |
|
camera_embeddings = rearrange(camera_embeddings, 'b n d -> (b n) d') |
|
embeddings = camera_embeddings |
|
|
|
outputs = self.model(**inputs, adaln_input=embeddings, interpolate_pos_encoding=True) |
|
last_hidden_states = outputs.last_hidden_state |
|
return last_hidden_states |
|
|
|
def _freeze(self): |
|
print(f"======== Freezing DinoWrapper ========") |
|
self.model.eval() |
|
for name, param in self.model.named_parameters(): |
|
param.requires_grad = False |
|
|
|
@staticmethod |
|
def _build_dino(model_name: str, proxy_error_retries: int = 3, proxy_error_cooldown: int = 5): |
|
import requests |
|
try: |
|
model = ViTModel.from_pretrained(model_name, add_pooling_layer=False) |
|
processor = ViTImageProcessor.from_pretrained(model_name) |
|
return model, processor |
|
except requests.exceptions.ProxyError as err: |
|
if proxy_error_retries > 0: |
|
print(f"Huggingface ProxyError: Retrying in {proxy_error_cooldown} seconds...") |
|
import time |
|
time.sleep(proxy_error_cooldown) |
|
return DinoWrapper._build_dino(model_name, proxy_error_retries - 1, proxy_error_cooldown) |
|
else: |
|
raise err |
|
|