# Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import absolute_import from __future__ import division from __future__ import print_function import numpy as np import paddle import paddle.nn as nn import paddle.nn.functional as F from ppdet.core.workspace import register from ppdet.modeling.rbox_utils import box2corners, check_points_in_polys, paddle_gather __all__ = ['FCOSRAssigner'] EPS = 1e-9 @register class FCOSRAssigner(nn.Layer): """ FCOSR Assigner, refer to https://arxiv.org/abs/2111.10780 for details 1. compute normalized gaussian distribution score and refined gaussian distribution score 2. refer to ellipse center sampling, sample points whose normalized gaussian distribution score is greater than threshold 3. refer to multi-level sampling, assign ground truth to feature map which follows two conditions. i). first, the ratio between the short edge of the target and the stride of the feature map is less than 2. ii). second, the long edge of minimum bounding rectangle of the target is larger than the acceptance range of feature map 4. refer to fuzzy sample label assignment, the points satisfying 2 and 3 will be assigned to the ground truth according to gaussian distribution score """ __shared__ = ['num_classes'] def __init__(self, num_classes=80, factor=12, threshold=0.23, boundary=[[-1, 128], [128, 320], [320, 10000]], score_type='iou'): super(FCOSRAssigner, self).__init__() self.num_classes = num_classes self.factor = factor self.threshold = threshold self.boundary = [ paddle.to_tensor( l, dtype=paddle.float32).reshape([1, 1, 2]) for l in boundary ] self.score_type = score_type def get_gaussian_distribution_score(self, points, gt_rboxes, gt_polys): # projecting points to coordinate system defined by each rbox # [B, N, 4, 2] -> 4 * [B, N, 1, 2] a, b, c, d = gt_polys.split(4, axis=2) # [1, L, 2] -> [1, 1, L, 2] points = points.unsqueeze(0) ab = b - a ad = d - a # [B, N, 5] -> [B, N, 2], [B, N, 2], [B, N, 1] xy, wh, angle = gt_rboxes.split([2, 2, 1], axis=-1) # [B, N, 2] -> [B, N, 1, 2] xy = xy.unsqueeze(2) # vector of points to center [B, N, L, 2] vec = points - xy # = |ab| * |vec| * cos(theta) [B, N, L] vec_dot_ab = paddle.sum(vec * ab, axis=-1) # = |ad| * |vec| * cos(theta) [B, N, L] vec_dot_ad = paddle.sum(vec * ad, axis=-1) # norm_ab [B, N, L] norm_ab = paddle.sum(ab * ab, axis=-1).sqrt() # norm_ad [B, N, L] norm_ad = paddle.sum(ad * ad, axis=-1).sqrt() # min(h, w), [B, N, 1] min_edge = paddle.min(wh, axis=-1, keepdim=True) # delta_x, delta_y [B, N, L] delta_x = vec_dot_ab.pow(2) / (norm_ab.pow(3) * min_edge + EPS) delta_y = vec_dot_ad.pow(2) / (norm_ad.pow(3) * min_edge + EPS) # score [B, N, L] norm_score = paddle.exp(-0.5 * self.factor * (delta_x + delta_y)) # simplified calculation sigma = min_edge / self.factor refined_score = norm_score / (2 * np.pi * sigma + EPS) return norm_score, refined_score def get_rotated_inside_mask(self, points, gt_polys, scores): inside_mask = check_points_in_polys(points, gt_polys) center_mask = scores >= self.threshold return (inside_mask & center_mask).cast(paddle.float32) def get_inside_range_mask(self, points, gt_bboxes, gt_rboxes, stride_tensor, regress_range): # [1, L, 2] -> [1, 1, L, 2] points = points.unsqueeze(0) # [B, n, 4] -> [B, n, 1, 4] x1y1, x2y2 = gt_bboxes.unsqueeze(2).split(2, axis=-1) # [B, n, L, 2] lt = points - x1y1 rb = x2y2 - points # [B, n, L, 4] ltrb = paddle.concat([lt, rb], axis=-1) # [B, n, L, 4] -> [B, n, L] inside_mask = paddle.min(ltrb, axis=-1) > EPS # regress_range [1, L, 2] -> [1, 1, L, 2] regress_range = regress_range.unsqueeze(0) # stride_tensor [1, L, 1] -> [1, 1, L] stride_tensor = stride_tensor.transpose((0, 2, 1)) # fcos range # [B, n, L, 4] -> [B, n, L] ltrb_max = paddle.max(ltrb, axis=-1) # [1, 1, L, 2] -> [1, 1, L] low, high = regress_range[..., 0], regress_range[..., 1] # [B, n, L] regress_mask = (ltrb_max >= low) & (ltrb_max <= high) # mask for rotated # [B, n, 1] min_edge = paddle.min(gt_rboxes[..., 2:4], axis=-1, keepdim=True) # [B, n , L] rotated_mask = ((min_edge / stride_tensor) < 2.0) & (ltrb_max > high) mask = inside_mask & (regress_mask | rotated_mask) return mask.cast(paddle.float32) @paddle.no_grad() def forward(self, anchor_points, stride_tensor, num_anchors_list, gt_labels, gt_bboxes, gt_rboxes, pad_gt_mask, bg_index, pred_rboxes=None): r""" Args: anchor_points (Tensor, float32): pre-defined anchor points, shape(1, L, 2), "x, y" format stride_tensor (Tensor, float32): stride tensor, shape (1, L, 1) num_anchors_list (List): num of anchors in each level gt_labels (Tensor, int64|int32): Label of gt_bboxes, shape(B, n, 1) gt_bboxes (Tensor, float32): Ground truth bboxes, shape(B, n, 4) gt_rboxes (Tensor, float32): Ground truth bboxes, shape(B, n, 5) pad_gt_mask (Tensor, float32): 1 means bbox, 0 means no bbox, shape(B, n, 1) bg_index (int): background index pred_rboxes (Tensor, float32, optional): predicted bounding boxes, shape(B, L, 5) Returns: assigned_labels (Tensor): (B, L) assigned_rboxes (Tensor): (B, L, 5) assigned_scores (Tensor): (B, L, C), if pred_rboxes is not None, then output ious """ _, num_anchors, _ = anchor_points.shape batch_size, num_max_boxes, _ = gt_rboxes.shape if num_max_boxes == 0: assigned_labels = paddle.full( [batch_size, num_anchors], bg_index, dtype=gt_labels.dtype) assigned_rboxes = paddle.zeros([batch_size, num_anchors, 5]) assigned_scores = paddle.zeros( [batch_size, num_anchors, self.num_classes]) return assigned_labels, assigned_rboxes, assigned_scores # get normalized gaussian distribution score and refined distribution score gt_polys = box2corners(gt_rboxes) score, refined_score = self.get_gaussian_distribution_score( anchor_points, gt_rboxes, gt_polys) inside_mask = self.get_rotated_inside_mask(anchor_points, gt_polys, score) regress_ranges = [] for num, bound in zip(num_anchors_list, self.boundary): regress_ranges.append(bound.tile((1, num, 1))) regress_ranges = paddle.concat(regress_ranges, axis=1) regress_mask = self.get_inside_range_mask( anchor_points, gt_bboxes, gt_rboxes, stride_tensor, regress_ranges) # [B, n, L] mask_positive = inside_mask * regress_mask * pad_gt_mask refined_score = refined_score * mask_positive - (1. - mask_positive) argmax_refined_score = refined_score.argmax(axis=-2) max_refined_score = refined_score.max(axis=-2) assigned_gt_index = argmax_refined_score # assigned target batch_ind = paddle.arange( end=batch_size, dtype=gt_labels.dtype).unsqueeze(-1) assigned_gt_index = assigned_gt_index + batch_ind * num_max_boxes assigned_labels = paddle.gather( gt_labels.flatten(), assigned_gt_index.flatten(), axis=0) assigned_labels = assigned_labels.reshape([batch_size, num_anchors]) assigned_labels = paddle.where( max_refined_score > 0, assigned_labels, paddle.full_like(assigned_labels, bg_index)) assigned_rboxes = paddle.gather( gt_rboxes.reshape([-1, 5]), assigned_gt_index.flatten(), axis=0) assigned_rboxes = assigned_rboxes.reshape([batch_size, num_anchors, 5]) assigned_scores = F.one_hot(assigned_labels, self.num_classes + 1) ind = list(range(self.num_classes + 1)) ind.remove(bg_index) assigned_scores = paddle.index_select( assigned_scores, paddle.to_tensor(ind), axis=-1) if self.score_type == 'gaussian': selected_scores = paddle_gather( score, 1, argmax_refined_score.unsqueeze(-2)).squeeze(-2) assigned_scores = assigned_scores * selected_scores.unsqueeze(-1) elif self.score_type == 'iou': assert pred_rboxes is not None, 'If score type is iou, pred_rboxes should not be None' from ext_op import matched_rbox_iou b, l = pred_rboxes.shape[:2] iou_score = matched_rbox_iou( pred_rboxes.reshape((-1, 5)), assigned_rboxes.reshape( (-1, 5))).reshape((b, l, 1)) assigned_scores = assigned_scores * iou_score return assigned_labels, assigned_rboxes, assigned_scores