| import torch |
| import torch.nn as nn |
|
|
| import PIL.Image |
| from typing import List |
| from friday.util import expand2square, pad_and_stack |
|
|
| from transformers import SiglipVisionModel, SiglipImageProcessor, SiglipVisionConfig |
| from friday.util.s2wrapper import forward as multiscale_forward |
|
|
|
|
| class SiglipVisionTower(nn.Module): |
| def __init__(self, pretrained_model_name_or_path, model_params={}, pad_to_square=True, **kwargs): |
| super().__init__() |
|
|
| self.is_loaded = False |
| self.pretrained_model_name_or_path = pretrained_model_name_or_path |
| self.model_params = model_params |
| self.pad_to_square = pad_to_square |
| self.select_layer = -2 |
| self.load_model() |
|
|
| @property |
| def output_dim(self): |
| return self.vision_tower.config.hidden_size if self.vision_tower else None |
| |
| def load_model(self): |
| if self.is_loaded: |
| return |
| self.image_processor = SiglipImageProcessor.from_pretrained(self.pretrained_model_name_or_path) |
| self.image_processor.crop_size = self.image_processor.size |
| self.vision_tower = SiglipVisionModel.from_pretrained( |
| self.pretrained_model_name_or_path, |
| **self.model_params, |
| ) |
| self.vision_tower.requires_grad_(False) |
|
|
| self.is_loaded = True |
| |
| def preprocess_images(self, imgs: List[PIL.Image.Image], pad_and_stack_tensors=True) -> torch.Tensor: |
| img_mean = tuple(int(x * 255) for x in self.image_processor.image_mean) |
| if self.pad_to_square: |
| imgs = [expand2square(img, img_mean) for img in imgs] |
| imgs = [self.image_processor(img, return_tensors="pt")['pixel_values'][0] for img in imgs] |
|
|
| if pad_and_stack_tensors: |
| imgs = pad_and_stack(imgs, pad_value=0.0) |
| imgs = imgs.to(dtype=torch.float32, device=self.device) |
| |
| return imgs |
|
|
| def feature_select(self, image_forward_outs): |
| image_features = image_forward_outs.hidden_states[self.select_layer] |
|
|
| return image_features |
|
|
| def forward(self, images): |
| if type(images) is list: |
| image_features = [] |
| for image in images: |
| image_forward_out = self.vision_tower(image.to(device=self.device, dtype=self.dtype).unsqueeze(0), |
| output_hidden_states=True) |
| image_feature = self.feature_select(image_forward_out).to(image.dtype) |
| image_features.append(image_feature) |
| else: |
| image_forward_outs = self.vision_tower(images.to(device=self.device, dtype=self.dtype), |
| output_hidden_states=True) |
| image_features = self.feature_select(image_forward_outs).to(images.dtype) |
|
|
| return image_features |
|
|
| @property |
| def dummy_feature(self): |
| return torch.zeros(1, self.hidden_size, device=self.device, dtype=self.dtype) |
|
|
| @property |
| def dtype(self): |
| return self.vision_tower.dtype |
|
|
| @property |
| def device(self): |
| return self.vision_tower.device |
|
|
| @property |
| def config(self): |
| if self.is_loaded: |
| return self.vision_tower.config |
| else: |
| return self.cfg_only |
|
|
| @property |
| def hidden_size(self): |
| return self.config.hidden_size |
|
|
| @property |
| def num_patches(self): |
| return (self.config.image_size // self.config.patch_size) ** 2 |
|
|
|
|
| class SiglipVisionTowerS2(SiglipVisionTower): |
| def __init__(self, pretrained_model_name_or_path, s2_scales, model_params={}, **kwargs): |
| self.s2_scales = list(map(int, s2_scales.split(','))) |
| self.s2_scales.sort() |
| self.s2_split_size = self.s2_scales[0] |
| self.s2_image_size = self.s2_scales[-1] |
|
|
| super().__init__(pretrained_model_name_or_path, model_params) |
|
|
| self.multiscale_forward = multiscale_forward |
|
|
| self.image_processor.size['height'] = self.image_processor.size['width'] = self.s2_image_size |
| self.image_processor.crop_size['height'] = self.image_processor.crop_size['width'] = self.s2_image_size |
| |
| @property |
| def output_dim(self): |
| return (2*self.vision_tower.config.hidden_size) if self.vision_tower else None |
|
|
| def load_model(self): |
| if self.is_loaded: |
| return |
| |
| super().load_model() |
| self.image_processor.size['height'] = self.image_processor.size['width'] = self.s2_image_size |
| self.image_processor.crop_size['height'] = self.image_processor.crop_size['width'] = self.s2_image_size |
|
|
| def forward_feature(self, images): |
| image_forward_outs = self.vision_tower(images.to(device=self.device, dtype=self.dtype), |
| output_hidden_states=True) |
| image_features = self.feature_select(image_forward_outs).to(images.dtype) |
| return image_features |
|
|
| def forward(self, images): |
| if type(images) is list: |
| image_features = [] |
| for image in images: |
| image_feature = self.multiscale_forward( |
| self.forward_feature, |
| image.unsqueeze(0), |
| img_sizes=self.s2_scales, |
| max_split_size=self.s2_split_size |
| ) |
| image_features.append(image_feature) |
| else: |
| image_features = self.multiscale_forward( |
| self.forward_feature, |
| images, |
| img_sizes=self.s2_scales, |
| max_split_size=self.s2_split_size |
| ) |
|
|
| return image_features |
|
|
| @property |
| def hidden_size(self): |
| return self.config.hidden_size * len(self.s2_scales) |
|
|
