|
|
|
|
|
import numpy as np |
|
from typing import Dict, List, Optional |
|
import fvcore.nn.weight_init as weight_init |
|
import torch |
|
import torch.nn as nn |
|
from torch.nn import functional as F |
|
|
|
from detectron2.layers import Conv2d, ShapeSpec, get_norm |
|
from detectron2.modeling import ROI_HEADS_REGISTRY, StandardROIHeads |
|
from detectron2.modeling.poolers import ROIPooler |
|
from detectron2.modeling.roi_heads import select_foreground_proposals |
|
from detectron2.structures import ImageList, Instances |
|
|
|
from .. import ( |
|
build_densepose_data_filter, |
|
build_densepose_embedder, |
|
build_densepose_head, |
|
build_densepose_losses, |
|
build_densepose_predictor, |
|
densepose_inference, |
|
) |
|
|
|
|
|
class Decoder(nn.Module): |
|
""" |
|
A semantic segmentation head described in detail in the Panoptic Feature Pyramid Networks paper |
|
(https://arxiv.org/abs/1901.02446). It takes FPN features as input and merges information from |
|
all levels of the FPN into single output. |
|
""" |
|
|
|
def __init__(self, cfg, input_shape: Dict[str, ShapeSpec], in_features): |
|
super(Decoder, self).__init__() |
|
|
|
|
|
self.in_features = in_features |
|
feature_strides = {k: v.stride for k, v in input_shape.items()} |
|
feature_channels = {k: v.channels for k, v in input_shape.items()} |
|
num_classes = cfg.MODEL.ROI_DENSEPOSE_HEAD.DECODER_NUM_CLASSES |
|
conv_dims = cfg.MODEL.ROI_DENSEPOSE_HEAD.DECODER_CONV_DIMS |
|
self.common_stride = cfg.MODEL.ROI_DENSEPOSE_HEAD.DECODER_COMMON_STRIDE |
|
norm = cfg.MODEL.ROI_DENSEPOSE_HEAD.DECODER_NORM |
|
|
|
|
|
self.scale_heads = [] |
|
for in_feature in self.in_features: |
|
head_ops = [] |
|
head_length = max( |
|
1, int(np.log2(feature_strides[in_feature]) - np.log2(self.common_stride)) |
|
) |
|
for k in range(head_length): |
|
conv = Conv2d( |
|
feature_channels[in_feature] if k == 0 else conv_dims, |
|
conv_dims, |
|
kernel_size=3, |
|
stride=1, |
|
padding=1, |
|
bias=not norm, |
|
norm=get_norm(norm, conv_dims), |
|
activation=F.relu, |
|
) |
|
weight_init.c2_msra_fill(conv) |
|
head_ops.append(conv) |
|
if feature_strides[in_feature] != self.common_stride: |
|
head_ops.append( |
|
nn.Upsample(scale_factor=2, mode="bilinear", align_corners=False) |
|
) |
|
self.scale_heads.append(nn.Sequential(*head_ops)) |
|
self.add_module(in_feature, self.scale_heads[-1]) |
|
self.predictor = Conv2d(conv_dims, num_classes, kernel_size=1, stride=1, padding=0) |
|
weight_init.c2_msra_fill(self.predictor) |
|
|
|
def forward(self, features: List[torch.Tensor]): |
|
for i, _ in enumerate(self.in_features): |
|
if i == 0: |
|
x = self.scale_heads[i](features[i]) |
|
else: |
|
x = x + self.scale_heads[i](features[i]) |
|
x = self.predictor(x) |
|
return x |
|
|
|
|
|
@ROI_HEADS_REGISTRY.register() |
|
class DensePoseROIHeads(StandardROIHeads): |
|
""" |
|
A Standard ROIHeads which contains an addition of DensePose head. |
|
""" |
|
|
|
def __init__(self, cfg, input_shape): |
|
super().__init__(cfg, input_shape) |
|
self._init_densepose_head(cfg, input_shape) |
|
|
|
def _init_densepose_head(self, cfg, input_shape): |
|
|
|
self.densepose_on = cfg.MODEL.DENSEPOSE_ON |
|
if not self.densepose_on: |
|
return |
|
self.densepose_data_filter = build_densepose_data_filter(cfg) |
|
dp_pooler_resolution = cfg.MODEL.ROI_DENSEPOSE_HEAD.POOLER_RESOLUTION |
|
dp_pooler_sampling_ratio = cfg.MODEL.ROI_DENSEPOSE_HEAD.POOLER_SAMPLING_RATIO |
|
dp_pooler_type = cfg.MODEL.ROI_DENSEPOSE_HEAD.POOLER_TYPE |
|
self.use_decoder = cfg.MODEL.ROI_DENSEPOSE_HEAD.DECODER_ON |
|
|
|
if self.use_decoder: |
|
dp_pooler_scales = (1.0 / input_shape[self.in_features[0]].stride,) |
|
else: |
|
dp_pooler_scales = tuple(1.0 / input_shape[k].stride for k in self.in_features) |
|
in_channels = [input_shape[f].channels for f in self.in_features][0] |
|
|
|
if self.use_decoder: |
|
self.decoder = Decoder(cfg, input_shape, self.in_features) |
|
|
|
self.densepose_pooler = ROIPooler( |
|
output_size=dp_pooler_resolution, |
|
scales=dp_pooler_scales, |
|
sampling_ratio=dp_pooler_sampling_ratio, |
|
pooler_type=dp_pooler_type, |
|
) |
|
self.densepose_head = build_densepose_head(cfg, in_channels) |
|
self.densepose_predictor = build_densepose_predictor( |
|
cfg, self.densepose_head.n_out_channels |
|
) |
|
self.densepose_losses = build_densepose_losses(cfg) |
|
self.embedder = build_densepose_embedder(cfg) |
|
|
|
def _forward_densepose(self, features: Dict[str, torch.Tensor], instances: List[Instances]): |
|
""" |
|
Forward logic of the densepose prediction branch. |
|
|
|
Args: |
|
features (dict[str, Tensor]): input data as a mapping from feature |
|
map name to tensor. Axis 0 represents the number of images `N` in |
|
the input data; axes 1-3 are channels, height, and width, which may |
|
vary between feature maps (e.g., if a feature pyramid is used). |
|
instances (list[Instances]): length `N` list of `Instances`. The i-th |
|
`Instances` contains instances for the i-th input image, |
|
In training, they can be the proposals. |
|
In inference, they can be the predicted boxes. |
|
|
|
Returns: |
|
In training, a dict of losses. |
|
In inference, update `instances` with new fields "densepose" and return it. |
|
""" |
|
if not self.densepose_on: |
|
return {} if self.training else instances |
|
|
|
features_list = [features[f] for f in self.in_features] |
|
if self.training: |
|
proposals, _ = select_foreground_proposals(instances, self.num_classes) |
|
features_list, proposals = self.densepose_data_filter(features_list, proposals) |
|
if len(proposals) > 0: |
|
proposal_boxes = [x.proposal_boxes for x in proposals] |
|
|
|
if self.use_decoder: |
|
features_list = [self.decoder(features_list)] |
|
|
|
features_dp = self.densepose_pooler(features_list, proposal_boxes) |
|
densepose_head_outputs = self.densepose_head(features_dp) |
|
densepose_predictor_outputs = self.densepose_predictor(densepose_head_outputs) |
|
densepose_loss_dict = self.densepose_losses( |
|
proposals, densepose_predictor_outputs, embedder=self.embedder |
|
) |
|
return densepose_loss_dict |
|
else: |
|
pred_boxes = [x.pred_boxes for x in instances] |
|
|
|
if self.use_decoder: |
|
features_list = [self.decoder(features_list)] |
|
|
|
features_dp = self.densepose_pooler(features_list, pred_boxes) |
|
if len(features_dp) > 0: |
|
densepose_head_outputs = self.densepose_head(features_dp) |
|
densepose_predictor_outputs = self.densepose_predictor(densepose_head_outputs) |
|
else: |
|
densepose_predictor_outputs = None |
|
|
|
densepose_inference(densepose_predictor_outputs, instances) |
|
return instances |
|
|
|
def forward( |
|
self, |
|
images: ImageList, |
|
features: Dict[str, torch.Tensor], |
|
proposals: List[Instances], |
|
targets: Optional[List[Instances]] = None, |
|
): |
|
instances, losses = super().forward(images, features, proposals, targets) |
|
del targets, images |
|
|
|
if self.training: |
|
losses.update(self._forward_densepose(features, instances)) |
|
return instances, losses |
|
|
|
def forward_with_given_boxes( |
|
self, features: Dict[str, torch.Tensor], instances: List[Instances] |
|
): |
|
""" |
|
Use the given boxes in `instances` to produce other (non-box) per-ROI outputs. |
|
|
|
This is useful for downstream tasks where a box is known, but need to obtain |
|
other attributes (outputs of other heads). |
|
Test-time augmentation also uses this. |
|
|
|
Args: |
|
features: same as in `forward()` |
|
instances (list[Instances]): instances to predict other outputs. Expect the keys |
|
"pred_boxes" and "pred_classes" to exist. |
|
|
|
Returns: |
|
instances (list[Instances]): |
|
the same `Instances` objects, with extra |
|
fields such as `pred_masks` or `pred_keypoints`. |
|
""" |
|
|
|
instances = super().forward_with_given_boxes(features, instances) |
|
instances = self._forward_densepose(features, instances) |
|
return instances |
|
|
