Spaces:
Running
Running
| # Ultralytics π AGPL-3.0 License - https://ultralytics.com/license | |
| from typing import List, Optional, Tuple, Type | |
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| from ultralytics.nn.modules import LayerNorm2d | |
| from .blocks import ( | |
| Block, | |
| CXBlock, | |
| Fuser, | |
| MaskDownSampler, | |
| MultiScaleBlock, | |
| PatchEmbed, | |
| PositionEmbeddingRandom, | |
| PositionEmbeddingSine, | |
| ) | |
| class ImageEncoderViT(nn.Module): | |
| """ | |
| An image encoder using Vision Transformer (ViT) architecture for encoding images into a compact latent space. | |
| This class processes images by splitting them into patches, applying transformer blocks, and generating a final | |
| encoded representation through a neck module. | |
| Attributes: | |
| img_size (int): Dimension of input images, assumed to be square. | |
| patch_embed (PatchEmbed): Module for patch embedding. | |
| pos_embed (nn.Parameter | None): Absolute positional embedding for patches. | |
| blocks (nn.ModuleList): List of transformer blocks for processing patch embeddings. | |
| neck (nn.Sequential): Neck module to further process the output. | |
| Methods: | |
| forward: Processes input through patch embedding, positional embedding, blocks, and neck. | |
| Examples: | |
| >>> import torch | |
| >>> encoder = ImageEncoderViT(img_size=224, patch_size=16, embed_dim=768, depth=12, num_heads=12) | |
| >>> input_image = torch.randn(1, 3, 224, 224) | |
| >>> output = encoder(input_image) | |
| >>> print(output.shape) | |
| """ | |
| def __init__( | |
| self, | |
| img_size: int = 1024, | |
| patch_size: int = 16, | |
| in_chans: int = 3, | |
| embed_dim: int = 768, | |
| depth: int = 12, | |
| num_heads: int = 12, | |
| mlp_ratio: float = 4.0, | |
| out_chans: int = 256, | |
| qkv_bias: bool = True, | |
| norm_layer: Type[nn.Module] = nn.LayerNorm, | |
| act_layer: Type[nn.Module] = nn.GELU, | |
| use_abs_pos: bool = True, | |
| use_rel_pos: bool = False, | |
| rel_pos_zero_init: bool = True, | |
| window_size: int = 0, | |
| global_attn_indexes: Tuple[int, ...] = (), | |
| ) -> None: | |
| """ | |
| Initializes an ImageEncoderViT instance for encoding images using Vision Transformer architecture. | |
| Args: | |
| img_size (int): Input image size, assumed to be square. | |
| patch_size (int): Size of image patches. | |
| in_chans (int): Number of input image channels. | |
| embed_dim (int): Dimension of patch embeddings. | |
| depth (int): Number of transformer blocks. | |
| num_heads (int): Number of attention heads in each block. | |
| mlp_ratio (float): Ratio of MLP hidden dimension to embedding dimension. | |
| out_chans (int): Number of output channels from the neck module. | |
| qkv_bias (bool): If True, adds learnable bias to query, key, value projections. | |
| norm_layer (Type[nn.Module]): Type of normalization layer to use. | |
| act_layer (Type[nn.Module]): Type of activation layer to use. | |
| use_abs_pos (bool): If True, uses absolute positional embeddings. | |
| use_rel_pos (bool): If True, adds relative positional embeddings to attention maps. | |
| rel_pos_zero_init (bool): If True, initializes relative positional parameters to zero. | |
| window_size (int): Size of attention window for windowed attention blocks. | |
| global_attn_indexes (Tuple[int, ...]): Indices of blocks that use global attention. | |
| Attributes: | |
| img_size (int): Dimension of input images. | |
| patch_embed (PatchEmbed): Module for patch embedding. | |
| pos_embed (nn.Parameter | None): Absolute positional embedding for patches. | |
| blocks (nn.ModuleList): List of transformer blocks. | |
| neck (nn.Sequential): Neck module for final processing. | |
| Examples: | |
| >>> encoder = ImageEncoderViT(img_size=224, patch_size=16, embed_dim=768, depth=12, num_heads=12) | |
| >>> input_image = torch.randn(1, 3, 224, 224) | |
| >>> output = encoder(input_image) | |
| >>> print(output.shape) | |
| """ | |
| super().__init__() | |
| self.img_size = img_size | |
| self.patch_embed = PatchEmbed( | |
| kernel_size=(patch_size, patch_size), | |
| stride=(patch_size, patch_size), | |
| in_chans=in_chans, | |
| embed_dim=embed_dim, | |
| ) | |
| self.pos_embed: Optional[nn.Parameter] = None | |
| if use_abs_pos: | |
| # Initialize absolute positional embedding with pretrain image size. | |
| self.pos_embed = nn.Parameter(torch.zeros(1, img_size // patch_size, img_size // patch_size, embed_dim)) | |
| self.blocks = nn.ModuleList() | |
| for i in range(depth): | |
| block = Block( | |
| dim=embed_dim, | |
| num_heads=num_heads, | |
| mlp_ratio=mlp_ratio, | |
| qkv_bias=qkv_bias, | |
| norm_layer=norm_layer, | |
| act_layer=act_layer, | |
| use_rel_pos=use_rel_pos, | |
| rel_pos_zero_init=rel_pos_zero_init, | |
| window_size=window_size if i not in global_attn_indexes else 0, | |
| input_size=(img_size // patch_size, img_size // patch_size), | |
| ) | |
| self.blocks.append(block) | |
| self.neck = nn.Sequential( | |
| nn.Conv2d( | |
| embed_dim, | |
| out_chans, | |
| kernel_size=1, | |
| bias=False, | |
| ), | |
| LayerNorm2d(out_chans), | |
| nn.Conv2d( | |
| out_chans, | |
| out_chans, | |
| kernel_size=3, | |
| padding=1, | |
| bias=False, | |
| ), | |
| LayerNorm2d(out_chans), | |
| ) | |
| def forward(self, x: torch.Tensor) -> torch.Tensor: | |
| """Processes input through patch embedding, positional embedding, transformer blocks, and neck module.""" | |
| x = self.patch_embed(x) | |
| if self.pos_embed is not None: | |
| pos_embed = ( | |
| F.interpolate(self.pos_embed.permute(0, 3, 1, 2), scale_factor=self.img_size / 1024).permute(0, 2, 3, 1) | |
| if self.img_size != 1024 | |
| else self.pos_embed | |
| ) | |
| x = x + pos_embed | |
| for blk in self.blocks: | |
| x = blk(x) | |
| return self.neck(x.permute(0, 3, 1, 2)) | |
| class PromptEncoder(nn.Module): | |
| """ | |
| Encodes different types of prompts for input to SAM's mask decoder, producing sparse and dense embeddings. | |
| Attributes: | |
| embed_dim (int): Dimension of the embeddings. | |
| input_image_size (Tuple[int, int]): Size of the input image as (H, W). | |
| image_embedding_size (Tuple[int, int]): Spatial size of the image embedding as (H, W). | |
| pe_layer (PositionEmbeddingRandom): Module for random position embedding. | |
| num_point_embeddings (int): Number of point embeddings for different types of points. | |
| point_embeddings (nn.ModuleList): List of point embeddings. | |
| not_a_point_embed (nn.Embedding): Embedding for points that are not part of any label. | |
| mask_input_size (Tuple[int, int]): Size of the input mask. | |
| mask_downscaling (nn.Sequential): Neural network for downscaling the mask. | |
| no_mask_embed (nn.Embedding): Embedding for cases where no mask is provided. | |
| Methods: | |
| get_dense_pe: Returns the positional encoding used to encode point prompts. | |
| forward: Embeds different types of prompts, returning both sparse and dense embeddings. | |
| Examples: | |
| >>> prompt_encoder = PromptEncoder(256, (64, 64), (1024, 1024), 16) | |
| >>> points = (torch.rand(1, 5, 2), torch.randint(0, 4, (1, 5))) | |
| >>> boxes = torch.rand(1, 2, 2) | |
| >>> masks = torch.rand(1, 1, 256, 256) | |
| >>> sparse_embeddings, dense_embeddings = prompt_encoder(points, boxes, masks) | |
| >>> print(sparse_embeddings.shape, dense_embeddings.shape) | |
| torch.Size([1, 7, 256]) torch.Size([1, 256, 64, 64]) | |
| """ | |
| def __init__( | |
| self, | |
| embed_dim: int, | |
| image_embedding_size: Tuple[int, int], | |
| input_image_size: Tuple[int, int], | |
| mask_in_chans: int, | |
| activation: Type[nn.Module] = nn.GELU, | |
| ) -> None: | |
| """ | |
| Initializes the PromptEncoder module for encoding various types of prompts. | |
| This module encodes different types of prompts (points, boxes, masks) for input to SAM's mask decoder, | |
| producing both sparse and dense embeddings. | |
| Args: | |
| embed_dim (int): The dimension of the embeddings. | |
| image_embedding_size (Tuple[int, int]): The spatial size of the image embedding as (H, W). | |
| input_image_size (Tuple[int, int]): The padded size of the input image as (H, W). | |
| mask_in_chans (int): The number of hidden channels used for encoding input masks. | |
| activation (Type[nn.Module]): The activation function to use when encoding input masks. | |
| Attributes: | |
| embed_dim (int): Dimension of the embeddings. | |
| input_image_size (Tuple[int, int]): Size of the input image as (H, W). | |
| image_embedding_size (Tuple[int, int]): Spatial size of the image embedding as (H, W). | |
| pe_layer (PositionEmbeddingRandom): Module for random position embedding. | |
| num_point_embeddings (int): Number of point embeddings for different types of points. | |
| point_embeddings (nn.ModuleList): List of point embeddings. | |
| not_a_point_embed (nn.Embedding): Embedding for points that are not part of any label. | |
| mask_input_size (Tuple[int, int]): Size of the input mask. | |
| mask_downscaling (nn.Sequential): Neural network for downscaling the mask. | |
| Examples: | |
| >>> prompt_encoder = PromptEncoder(256, (64, 64), (1024, 1024), 16) | |
| >>> points = (torch.rand(1, 5, 2), torch.randint(0, 4, (1, 5))) | |
| >>> boxes = torch.rand(1, 2, 2) | |
| >>> masks = torch.rand(1, 1, 256, 256) | |
| >>> sparse_embeddings, dense_embeddings = prompt_encoder(points, boxes, masks) | |
| >>> print(sparse_embeddings.shape, dense_embeddings.shape) | |
| torch.Size([1, 7, 256]) torch.Size([1, 256, 64, 64]) | |
| """ | |
| super().__init__() | |
| self.embed_dim = embed_dim | |
| self.input_image_size = input_image_size | |
| self.image_embedding_size = image_embedding_size | |
| self.pe_layer = PositionEmbeddingRandom(embed_dim // 2) | |
| self.num_point_embeddings: int = 4 # pos/neg point + 2 box corners | |
| point_embeddings = [nn.Embedding(1, embed_dim) for _ in range(self.num_point_embeddings)] | |
| self.point_embeddings = nn.ModuleList(point_embeddings) | |
| self.not_a_point_embed = nn.Embedding(1, embed_dim) | |
| self.mask_input_size = (4 * image_embedding_size[0], 4 * image_embedding_size[1]) | |
| self.mask_downscaling = nn.Sequential( | |
| nn.Conv2d(1, mask_in_chans // 4, kernel_size=2, stride=2), | |
| LayerNorm2d(mask_in_chans // 4), | |
| activation(), | |
| nn.Conv2d(mask_in_chans // 4, mask_in_chans, kernel_size=2, stride=2), | |
| LayerNorm2d(mask_in_chans), | |
| activation(), | |
| nn.Conv2d(mask_in_chans, embed_dim, kernel_size=1), | |
| ) | |
| self.no_mask_embed = nn.Embedding(1, embed_dim) | |
| def get_dense_pe(self) -> torch.Tensor: | |
| """ | |
| Returns the dense positional encoding used for encoding point prompts. | |
| This method generates a positional encoding for a dense set of points matching the shape of the image | |
| encoding. The encoding is used to provide spatial information to the model when processing point prompts. | |
| Returns: | |
| (torch.Tensor): Positional encoding tensor with shape (1, embed_dim, H, W), where H and W are the | |
| height and width of the image embedding size, respectively. | |
| Examples: | |
| >>> prompt_encoder = PromptEncoder(256, (64, 64), (1024, 1024), 16) | |
| >>> dense_pe = prompt_encoder.get_dense_pe() | |
| >>> print(dense_pe.shape) | |
| torch.Size([1, 256, 64, 64]) | |
| """ | |
| return self.pe_layer(self.image_embedding_size).unsqueeze(0) | |
| def _embed_points(self, points: torch.Tensor, labels: torch.Tensor, pad: bool) -> torch.Tensor: | |
| """Embeds point prompts by applying positional encoding and label-specific embeddings.""" | |
| points = points + 0.5 # Shift to center of pixel | |
| if pad: | |
| padding_point = torch.zeros((points.shape[0], 1, 2), device=points.device) | |
| padding_label = -torch.ones((labels.shape[0], 1), device=labels.device) | |
| points = torch.cat([points, padding_point], dim=1) | |
| labels = torch.cat([labels, padding_label], dim=1) | |
| point_embedding = self.pe_layer.forward_with_coords(points, self.input_image_size) | |
| point_embedding[labels == -1] = 0.0 | |
| point_embedding[labels == -1] += self.not_a_point_embed.weight | |
| point_embedding[labels == 0] += self.point_embeddings[0].weight | |
| point_embedding[labels == 1] += self.point_embeddings[1].weight | |
| point_embedding[labels == 2] += self.point_embeddings[2].weight | |
| point_embedding[labels == 3] += self.point_embeddings[3].weight | |
| return point_embedding | |
| def _embed_boxes(self, boxes: torch.Tensor) -> torch.Tensor: | |
| """Embeds box prompts by applying positional encoding and adding corner embeddings.""" | |
| boxes = boxes + 0.5 # Shift to center of pixel | |
| coords = boxes.reshape(-1, 2, 2) | |
| corner_embedding = self.pe_layer.forward_with_coords(coords, self.input_image_size) | |
| corner_embedding[:, 0, :] += self.point_embeddings[2].weight | |
| corner_embedding[:, 1, :] += self.point_embeddings[3].weight | |
| return corner_embedding | |
| def _embed_masks(self, masks: torch.Tensor) -> torch.Tensor: | |
| """Embeds mask inputs by downscaling and processing through convolutional layers.""" | |
| return self.mask_downscaling(masks) | |
| def _get_batch_size( | |
| points: Optional[Tuple[torch.Tensor, torch.Tensor]], | |
| boxes: Optional[torch.Tensor], | |
| masks: Optional[torch.Tensor], | |
| ) -> int: | |
| """Gets the batch size of the output given the batch size of the input prompts.""" | |
| if points is not None: | |
| return points[0].shape[0] | |
| elif boxes is not None: | |
| return boxes.shape[0] | |
| elif masks is not None: | |
| return masks.shape[0] | |
| else: | |
| return 1 | |
| def _get_device(self) -> torch.device: | |
| """Returns the device of the first point embedding's weight tensor.""" | |
| return self.point_embeddings[0].weight.device | |
| def forward( | |
| self, | |
| points: Optional[Tuple[torch.Tensor, torch.Tensor]], | |
| boxes: Optional[torch.Tensor], | |
| masks: Optional[torch.Tensor], | |
| ) -> Tuple[torch.Tensor, torch.Tensor]: | |
| """ | |
| Embeds different types of prompts, returning both sparse and dense embeddings. | |
| Args: | |
| points (Tuple[torch.Tensor, torch.Tensor] | None): Point coordinates and labels to embed. The first | |
| tensor contains coordinates with shape (B, N, 2), and the second tensor contains labels with | |
| shape (B, N). | |
| boxes (torch.Tensor | None): Boxes to embed with shape (B, M, 2, 2), where M is the number of boxes. | |
| masks (torch.Tensor | None): Masks to embed with shape (B, 1, H, W). | |
| Returns: | |
| (Tuple[torch.Tensor, torch.Tensor]): A tuple containing: | |
| - sparse_embeddings (torch.Tensor): Sparse embeddings for points and boxes with shape (B, N, embed_dim). | |
| - dense_embeddings (torch.Tensor): Dense embeddings for masks of shape (B, embed_dim, embed_H, embed_W). | |
| Examples: | |
| >>> encoder = PromptEncoder(256, (64, 64), (1024, 1024), 16) | |
| >>> points = (torch.rand(1, 5, 2), torch.randint(0, 4, (1, 5))) | |
| >>> boxes = torch.rand(1, 2, 2, 2) | |
| >>> masks = torch.rand(1, 1, 256, 256) | |
| >>> sparse_emb, dense_emb = encoder(points, boxes, masks) | |
| >>> print(sparse_emb.shape, dense_emb.shape) | |
| torch.Size([1, 7, 256]) torch.Size([1, 256, 64, 64]) | |
| """ | |
| bs = self._get_batch_size(points, boxes, masks) | |
| sparse_embeddings = torch.empty((bs, 0, self.embed_dim), device=self._get_device()) | |
| if points is not None: | |
| coords, labels = points | |
| point_embeddings = self._embed_points(coords, labels, pad=(boxes is None)) | |
| sparse_embeddings = torch.cat([sparse_embeddings, point_embeddings], dim=1) | |
| if boxes is not None: | |
| box_embeddings = self._embed_boxes(boxes) | |
| sparse_embeddings = torch.cat([sparse_embeddings, box_embeddings], dim=1) | |
| if masks is not None: | |
| dense_embeddings = self._embed_masks(masks) | |
| else: | |
| dense_embeddings = self.no_mask_embed.weight.reshape(1, -1, 1, 1).expand( | |
| bs, -1, self.image_embedding_size[0], self.image_embedding_size[1] | |
| ) | |
| return sparse_embeddings, dense_embeddings | |
| class MemoryEncoder(nn.Module): | |
| """ | |
| Encodes pixel features and masks into a memory representation for efficient image segmentation. | |
| This class processes pixel-level features and masks, fusing them to generate encoded memory representations | |
| suitable for downstream tasks in image segmentation models like SAM (Segment Anything Model). | |
| Attributes: | |
| mask_downsampler (MaskDownSampler): Module for downsampling input masks. | |
| pix_feat_proj (nn.Conv2d): Convolutional layer for projecting pixel features. | |
| fuser (Fuser): Module for fusing pixel features and masks. | |
| position_encoding (PositionEmbeddingSine): Module for adding positional encoding to features. | |
| out_proj (nn.Module): Output projection layer, either nn.Identity or nn.Conv2d. | |
| Methods: | |
| forward: Processes input pixel features and masks to generate encoded memory representations. | |
| Examples: | |
| >>> import torch | |
| >>> encoder = MemoryEncoder(out_dim=256, in_dim=256) | |
| >>> pix_feat = torch.randn(1, 256, 64, 64) | |
| >>> masks = torch.randn(1, 1, 64, 64) | |
| >>> encoded_feat, pos = encoder(pix_feat, masks) | |
| >>> print(encoded_feat.shape, pos.shape) | |
| torch.Size([1, 256, 64, 64]) torch.Size([1, 128, 64, 64]) | |
| """ | |
| def __init__( | |
| self, | |
| out_dim, | |
| in_dim=256, # in_dim of pix_feats | |
| ): | |
| """Initializes the MemoryEncoder for encoding pixel features and masks into memory representations.""" | |
| super().__init__() | |
| self.mask_downsampler = MaskDownSampler(kernel_size=3, stride=2, padding=1) | |
| self.pix_feat_proj = nn.Conv2d(in_dim, in_dim, kernel_size=1) | |
| self.fuser = Fuser(CXBlock(dim=256), num_layers=2) | |
| self.position_encoding = PositionEmbeddingSine(num_pos_feats=64) | |
| self.out_proj = nn.Identity() | |
| if out_dim != in_dim: | |
| self.out_proj = nn.Conv2d(in_dim, out_dim, kernel_size=1) | |
| def forward( | |
| self, | |
| pix_feat: torch.Tensor, | |
| masks: torch.Tensor, | |
| skip_mask_sigmoid: bool = False, | |
| ) -> Tuple[torch.Tensor, torch.Tensor]: | |
| """Processes pixel features and masks to generate encoded memory representations for segmentation.""" | |
| if not skip_mask_sigmoid: | |
| masks = F.sigmoid(masks) | |
| masks = self.mask_downsampler(masks) | |
| # Fuse pix_feats and downsampled masks, in case the visual features are on CPU, cast them to CUDA | |
| pix_feat = pix_feat.to(masks.device) | |
| x = self.pix_feat_proj(pix_feat) | |
| x = x + masks | |
| x = self.fuser(x) | |
| x = self.out_proj(x) | |
| pos = self.position_encoding(x).to(x.dtype) | |
| return {"vision_features": x, "vision_pos_enc": [pos]} | |
| class ImageEncoder(nn.Module): | |
| """ | |
| Encodes images using a trunk-neck architecture, producing multiscale features and positional encodings. | |
| This class combines a trunk network for feature extraction with a neck network for feature refinement | |
| and positional encoding generation. It can optionally discard the lowest resolution features. | |
| Attributes: | |
| trunk (nn.Module): The trunk network for initial feature extraction. | |
| neck (nn.Module): The neck network for feature refinement and positional encoding generation. | |
| scalp (int): Number of lowest resolution feature levels to discard. | |
| Methods: | |
| forward: Processes the input image through the trunk and neck networks. | |
| Examples: | |
| >>> trunk = SomeTrunkNetwork() | |
| >>> neck = SomeNeckNetwork() | |
| >>> encoder = ImageEncoder(trunk, neck, scalp=1) | |
| >>> image = torch.randn(1, 3, 224, 224) | |
| >>> output = encoder(image) | |
| >>> print(output.keys()) | |
| dict_keys(['vision_features', 'vision_pos_enc', 'backbone_fpn']) | |
| """ | |
| def __init__( | |
| self, | |
| trunk: nn.Module, | |
| neck: nn.Module, | |
| scalp: int = 0, | |
| ): | |
| """Initializes the ImageEncoder with trunk and neck networks for feature extraction and refinement.""" | |
| super().__init__() | |
| self.trunk = trunk | |
| self.neck = neck | |
| self.scalp = scalp | |
| assert self.trunk.channel_list == self.neck.backbone_channel_list, ( | |
| f"Channel dims of trunk {self.trunk.channel_list} and neck {self.neck.backbone_channel_list} do not match." | |
| ) | |
| def forward(self, sample: torch.Tensor): | |
| """Encodes input through patch embedding, positional embedding, transformer blocks, and neck module.""" | |
| features, pos = self.neck(self.trunk(sample)) | |
| if self.scalp > 0: | |
| # Discard the lowest resolution features | |
| features, pos = features[: -self.scalp], pos[: -self.scalp] | |
| src = features[-1] | |
| return { | |
| "vision_features": src, | |
| "vision_pos_enc": pos, | |
| "backbone_fpn": features, | |
| } | |
| class FpnNeck(nn.Module): | |
| """ | |
| A Feature Pyramid Network (FPN) neck variant for multiscale feature fusion in object detection models. | |
| This FPN variant removes the output convolution and uses bicubic interpolation for feature resizing, | |
| similar to ViT positional embedding interpolation. | |
| Attributes: | |
| position_encoding (PositionEmbeddingSine): Sinusoidal positional encoding module. | |
| convs (nn.ModuleList): List of convolutional layers for each backbone level. | |
| backbone_channel_list (List[int]): List of channel dimensions from the backbone. | |
| fpn_interp_model (str): Interpolation mode for FPN feature resizing. | |
| fuse_type (str): Type of feature fusion, either 'sum' or 'avg'. | |
| fpn_top_down_levels (List[int]): Levels to have top-down features in outputs. | |
| Methods: | |
| forward: Performs forward pass through the FPN neck. | |
| Examples: | |
| >>> backbone_channels = [64, 128, 256, 512] | |
| >>> fpn_neck = FpnNeck(256, backbone_channels) | |
| >>> inputs = [torch.rand(1, c, 32, 32) for c in backbone_channels] | |
| >>> outputs, positions = fpn_neck(inputs) | |
| >>> print(len(outputs), len(positions)) | |
| 4 4 | |
| """ | |
| def __init__( | |
| self, | |
| d_model: int, | |
| backbone_channel_list: List[int], | |
| kernel_size: int = 1, | |
| stride: int = 1, | |
| padding: int = 0, | |
| fpn_interp_model: str = "bilinear", | |
| fuse_type: str = "sum", | |
| fpn_top_down_levels: Optional[List[int]] = None, | |
| ): | |
| """ | |
| Initializes a modified Feature Pyramid Network (FPN) neck. | |
| This FPN variant removes the output convolution and uses bicubic interpolation for feature resizing, | |
| similar to ViT positional embedding interpolation. | |
| Args: | |
| d_model (int): Dimension of the model. | |
| backbone_channel_list (List[int]): List of channel dimensions from the backbone. | |
| kernel_size (int): Kernel size for the convolutional layers. | |
| stride (int): Stride for the convolutional layers. | |
| padding (int): Padding for the convolutional layers. | |
| fpn_interp_model (str): Interpolation mode for FPN feature resizing. | |
| fuse_type (str): Type of feature fusion, either 'sum' or 'avg'. | |
| fpn_top_down_levels (Optional[List[int]]): Levels to have top-down features in outputs. | |
| Examples: | |
| >>> backbone_channels = [64, 128, 256, 512] | |
| >>> fpn_neck = FpnNeck(256, backbone_channels) | |
| >>> print(fpn_neck) | |
| """ | |
| super().__init__() | |
| self.position_encoding = PositionEmbeddingSine(num_pos_feats=256) | |
| self.convs = nn.ModuleList() | |
| self.backbone_channel_list = backbone_channel_list | |
| for dim in backbone_channel_list: | |
| current = nn.Sequential() | |
| current.add_module( | |
| "conv", | |
| nn.Conv2d( | |
| in_channels=dim, | |
| out_channels=d_model, | |
| kernel_size=kernel_size, | |
| stride=stride, | |
| padding=padding, | |
| ), | |
| ) | |
| self.convs.append(current) | |
| self.fpn_interp_model = fpn_interp_model | |
| assert fuse_type in {"sum", "avg"} | |
| self.fuse_type = fuse_type | |
| # levels to have top-down features in its outputs | |
| # e.g. if fpn_top_down_levels is [2, 3], then only outputs of level 2 and 3 | |
| # have top-down propagation, while outputs of level 0 and level 1 have only | |
| # lateral features from the same backbone level. | |
| if fpn_top_down_levels is None: | |
| # default is to have top-down features on all levels | |
| fpn_top_down_levels = range(len(self.convs)) | |
| self.fpn_top_down_levels = list(fpn_top_down_levels) | |
| def forward(self, xs: List[torch.Tensor]): | |
| """ | |
| Performs forward pass through the Feature Pyramid Network (FPN) neck. | |
| This method processes a list of input tensors from the backbone through the FPN, applying lateral connections | |
| and top-down feature fusion. It generates output feature maps and corresponding positional encodings. | |
| Args: | |
| xs (List[torch.Tensor]): List of input tensors from the backbone, each with shape (B, C, H, W). | |
| Returns: | |
| (Tuple[List[torch.Tensor], List[torch.Tensor]]): A tuple containing: | |
| - out (List[torch.Tensor]): List of output feature maps after FPN processing, each with shape | |
| (B, d_model, H, W). | |
| - pos (List[torch.Tensor]): List of positional encodings corresponding to each output feature map. | |
| Examples: | |
| >>> fpn_neck = FpnNeck(d_model=256, backbone_channel_list=[64, 128, 256, 512]) | |
| >>> inputs = [torch.rand(1, c, 32, 32) for c in [64, 128, 256, 512]] | |
| >>> outputs, positions = fpn_neck(inputs) | |
| >>> print(len(outputs), len(positions)) | |
| 4 4 | |
| """ | |
| out = [None] * len(self.convs) | |
| pos = [None] * len(self.convs) | |
| assert len(xs) == len(self.convs) | |
| # fpn forward pass | |
| # see https://github.com/facebookresearch/detectron2/blob/main/detectron2/modeling/backbone/fpn.py | |
| prev_features = None | |
| # forward in top-down order (from low to high resolution) | |
| n = len(self.convs) - 1 | |
| for i in range(n, -1, -1): | |
| x = xs[i] | |
| lateral_features = self.convs[n - i](x) | |
| if i in self.fpn_top_down_levels and prev_features is not None: | |
| top_down_features = F.interpolate( | |
| prev_features.to(dtype=torch.float32), | |
| scale_factor=2.0, | |
| mode=self.fpn_interp_model, | |
| align_corners=(None if self.fpn_interp_model == "nearest" else False), | |
| antialias=False, | |
| ) | |
| prev_features = lateral_features + top_down_features | |
| if self.fuse_type == "avg": | |
| prev_features /= 2 | |
| else: | |
| prev_features = lateral_features | |
| x_out = prev_features | |
| out[i] = x_out | |
| pos[i] = self.position_encoding(x_out).to(x_out.dtype) | |
| return out, pos | |
| class Hiera(nn.Module): | |
| """ | |
| Hierarchical vision transformer for efficient multiscale feature extraction in image processing tasks. | |
| This class implements a Hiera model, which is a hierarchical vision transformer architecture designed for | |
| efficient multiscale feature extraction. It uses a series of transformer blocks organized into stages, | |
| with optional pooling and global attention mechanisms. | |
| Attributes: | |
| window_spec (Tuple[int, ...]): Window sizes for each stage. | |
| q_stride (Tuple[int, int]): Downsampling stride between stages. | |
| stage_ends (List[int]): Indices of the last block in each stage. | |
| q_pool_blocks (List[int]): Indices of blocks where pooling is applied. | |
| return_interm_layers (bool): Whether to return intermediate layer outputs. | |
| patch_embed (PatchEmbed): Module for patch embedding. | |
| global_att_blocks (Tuple[int, ...]): Indices of blocks with global attention. | |
| window_pos_embed_bkg_spatial_size (Tuple[int, int]): Spatial size for window positional embedding background. | |
| pos_embed (nn.Parameter): Positional embedding for the background. | |
| pos_embed_window (nn.Parameter): Positional embedding for the window. | |
| blocks (nn.ModuleList): List of MultiScaleBlock modules. | |
| channel_list (List[int]): List of output channel dimensions for each stage. | |
| Methods: | |
| _get_pos_embed: Generates positional embeddings by interpolating and combining window and background embeddings. | |
| forward: Performs the forward pass through the Hiera model. | |
| Examples: | |
| >>> model = Hiera(embed_dim=96, num_heads=1, stages=(2, 3, 16, 3)) | |
| >>> input_tensor = torch.randn(1, 3, 224, 224) | |
| >>> output_features = model(input_tensor) | |
| >>> for feat in output_features: | |
| ... print(feat.shape) | |
| """ | |
| def __init__( | |
| self, | |
| embed_dim: int = 96, # initial embed dim | |
| num_heads: int = 1, # initial number of heads | |
| drop_path_rate: float = 0.0, # stochastic depth | |
| q_pool: int = 3, # number of q_pool stages | |
| q_stride: Tuple[int, int] = (2, 2), # downsample stride bet. stages | |
| stages: Tuple[int, ...] = (2, 3, 16, 3), # blocks per stage | |
| dim_mul: float = 2.0, # dim_mul factor at stage shift | |
| head_mul: float = 2.0, # head_mul factor at stage shift | |
| window_pos_embed_bkg_spatial_size: Tuple[int, int] = (14, 14), | |
| # window size per stage, when not using global att. | |
| window_spec: Tuple[int, ...] = ( | |
| 8, | |
| 4, | |
| 14, | |
| 7, | |
| ), | |
| # global attn in these blocks | |
| global_att_blocks: Tuple[int, ...] = ( | |
| 12, | |
| 16, | |
| 20, | |
| ), | |
| return_interm_layers=True, # return feats from every stage | |
| ): | |
| """Initializes the Hiera model, configuring its hierarchical vision transformer architecture.""" | |
| super().__init__() | |
| assert len(stages) == len(window_spec) | |
| self.window_spec = window_spec | |
| depth = sum(stages) | |
| self.q_stride = q_stride | |
| self.stage_ends = [sum(stages[:i]) - 1 for i in range(1, len(stages) + 1)] | |
| assert 0 <= q_pool <= len(self.stage_ends[:-1]) | |
| self.q_pool_blocks = [x + 1 for x in self.stage_ends[:-1]][:q_pool] | |
| self.return_interm_layers = return_interm_layers | |
| self.patch_embed = PatchEmbed( | |
| embed_dim=embed_dim, | |
| kernel_size=(7, 7), | |
| stride=(4, 4), | |
| padding=(3, 3), | |
| ) | |
| # Which blocks have global att? | |
| self.global_att_blocks = global_att_blocks | |
| # Windowed positional embedding (https://arxiv.org/abs/2311.05613) | |
| self.window_pos_embed_bkg_spatial_size = window_pos_embed_bkg_spatial_size | |
| self.pos_embed = nn.Parameter(torch.zeros(1, embed_dim, *self.window_pos_embed_bkg_spatial_size)) | |
| self.pos_embed_window = nn.Parameter(torch.zeros(1, embed_dim, self.window_spec[0], self.window_spec[0])) | |
| dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule | |
| cur_stage = 1 | |
| self.blocks = nn.ModuleList() | |
| for i in range(depth): | |
| dim_out = embed_dim | |
| # lags by a block, so first block of | |
| # next stage uses an initial window size | |
| # of previous stage and final window size of current stage | |
| window_size = self.window_spec[cur_stage - 1] | |
| if self.global_att_blocks is not None: | |
| window_size = 0 if i in self.global_att_blocks else window_size | |
| if i - 1 in self.stage_ends: | |
| dim_out = int(embed_dim * dim_mul) | |
| num_heads = int(num_heads * head_mul) | |
| cur_stage += 1 | |
| block = MultiScaleBlock( | |
| dim=embed_dim, | |
| dim_out=dim_out, | |
| num_heads=num_heads, | |
| drop_path=dpr[i], | |
| q_stride=self.q_stride if i in self.q_pool_blocks else None, | |
| window_size=window_size, | |
| ) | |
| embed_dim = dim_out | |
| self.blocks.append(block) | |
| self.channel_list = ( | |
| [self.blocks[i].dim_out for i in self.stage_ends[::-1]] | |
| if return_interm_layers | |
| else [self.blocks[-1].dim_out] | |
| ) | |
| def _get_pos_embed(self, hw: Tuple[int, int]) -> torch.Tensor: | |
| """Generates positional embeddings by interpolating and combining window and background embeddings.""" | |
| h, w = hw | |
| window_embed = self.pos_embed_window | |
| pos_embed = F.interpolate(self.pos_embed, size=(h, w), mode="bicubic") | |
| pos_embed = pos_embed + window_embed.tile([x // y for x, y in zip(pos_embed.shape, window_embed.shape)]) | |
| pos_embed = pos_embed.permute(0, 2, 3, 1) | |
| return pos_embed | |
| def forward(self, x: torch.Tensor) -> List[torch.Tensor]: | |
| """Performs forward pass through Hiera model, extracting multiscale features from input images.""" | |
| x = self.patch_embed(x) | |
| # x: (B, H, W, C) | |
| # Add pos embed | |
| x = x + self._get_pos_embed(x.shape[1:3]) | |
| outputs = [] | |
| for i, blk in enumerate(self.blocks): | |
| x = blk(x) | |
| if (i == self.stage_ends[-1]) or (i in self.stage_ends and self.return_interm_layers): | |
| feats = x.permute(0, 3, 1, 2) | |
| outputs.append(feats) | |
| return outputs | |