File size: 11,549 Bytes
ddaad99
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
# SPDX-License-Identifier: Apache-2.0
"""The three TensorRT plugins of the deployed graph as plain torch fp32 (no mmcv, no CUDA), plus literal float64 ports
of the CUDA loops that unit-test them. Ported from the verified research reference
(``research/bevformer/scripts/bevformer_ref.py``; checks in ``research/bevformer/test_ref_ops.log``).

``AW/`` = autoware_universe ``perception/autoware_tensorrt_bevformer`` @ 9ceaccf:

- :func:`msda` == ``MultiScaleDeformableAttnTRT`` (``AW/TensorRT/plugin/multi_scale_deformable_attn/
  multiScaleDeformableAttnKernel.cu:637-707``, ``...Plugin.cpp:95-105,208-290``): softmax over levels x points INSIDE
  the op; point ``p`` uses reference ``p % Pg``; ``loc_px = ref * (W, H) + offset`` (offsets in pixels of the level);
  bilinear at ``loc_px - 0.5`` with zeros outside (== ``grid_sample(align_corners=False, padding_mode="zeros")``).
- :func:`mdcn` == ``ModulatedDeformableConv2dTRT`` (``AW/TensorRT/plugin/modulated_deformable_conv2d/
  modulatedDeformableConv2dKernel.cu:291-349``): mmcv DCNv2, offsets interleaved ``(dy_k, dx_k)`` per tap
  ``k = i * 3 + j``, sample at ``(h - 1 + i + dy, w - 1 + j + dx)``, bilinear, zeros outside, times the sigmoid mask,
  then the GEMM (+ the bias the port gets from the folded BatchNorm).
- :func:`rotate` == ``RotateTRT`` nearest (``AW/TensorRT/plugin/rotate/rotateKernel.cu:166-243``): the inverse
  affine grid of torchvision ``rotate`` about ``center`` with ``::round`` (half away from zero) and zero fill.

Importing this module imports torch (the reference runs where torch is: tt-env and the research venv); never ttnn.
"""
from __future__ import annotations

import math
from typing import Sequence, Tuple

import torch
import torch.nn.functional as F

__all__ = ["msda", "msda_naive", "mdcn", "mdcn_naive", "rotate", "round_half_away"]


def msda(value: torch.Tensor, spatial_shapes: Sequence[Tuple[int, int]], reference_points: torch.Tensor,
         sampling_offsets: torch.Tensor, attention_weights: torch.Tensor, q_chunk: int = 4096) -> torch.Tensor:
    """``MultiScaleDeformableAttnTRT`` semantics.

    value             [N, S, H, C]    (S = sum h_l * w_l)
    spatial_shapes    [(h, w), ...]   python ints (ONNX: an INT64 constant [L, 2])
    reference_points  [N, Q, 1, Pg*2] normalised (x, y); sampling point p uses reference p % Pg
    sampling_offsets  [N, Q, H, L*P*2] in PIXELS of the level, (x, y)
    attention_weights [N, Q, H, L*P]  logits: softmax over L*P inside the op
    returns           [N, Q, H, C]
    """
    N, S, H, C = value.shape
    L = len(spatial_shapes)
    Q = reference_points.shape[1]
    Pg = reference_points.shape[-1] // 2
    P = attention_weights.shape[-1] // L
    w = torch.softmax(attention_weights, dim=-1).view(N, Q, H, L, P)
    off = sampling_offsets.view(N, Q, H, L, P, 2)
    ref = reference_points.view(N, Q, 1, Pg, 2)
    ref_p = ref[:, :, :, torch.arange(P) % Pg, :]                         # [N, Q, 1, P, 2]
    out = None
    start = 0
    for lvl, (h, wd) in enumerate(spatial_shapes):
        v = value[:, start:start + h * wd].permute(0, 2, 3, 1).reshape(N * H, C, h, wd)
        start += h * wd
        norm = torch.tensor([wd, h], dtype=value.dtype)
        parts = []
        for q0 in range(0, Q, q_chunk):
            q1 = min(Q, q0 + q_chunk)
            loc = ref_p[:, q0:q1] + off[:, q0:q1, :, lvl] / norm          # [N, q, H, P, 2]
            grid = (2.0 * loc - 1.0).permute(0, 2, 1, 3, 4).reshape(N * H, q1 - q0, P, 2)
            s = F.grid_sample(v, grid, mode="bilinear", padding_mode="zeros", align_corners=False)
            wl = w[:, q0:q1, :, lvl].permute(0, 2, 1, 3).reshape(N * H, 1, q1 - q0, P)
            parts.append((s * wl).sum(-1))                                # [N*H, C, q]
        lvl_out = parts[0] if len(parts) == 1 else torch.cat(parts, dim=2)
        out = lvl_out if out is None else out + lvl_out
    return out.view(N, H, C, Q).permute(0, 3, 1, 2).contiguous()


def msda_naive(value, spatial_shapes, reference_points, sampling_offsets, attention_weights) -> torch.Tensor:
    """Literal float64 port of the CUDA kernel loop (unit tests on small sizes)."""
    value = value.double()
    reference_points = reference_points.double()
    sampling_offsets = sampling_offsets.double()
    attention_weights = attention_weights.double()
    N, S, H, C = value.shape
    L = len(spatial_shapes)
    Q = reference_points.shape[1]
    Pg = reference_points.shape[-1] // 2
    P = attention_weights.shape[-1] // L
    out = torch.zeros(N, Q, H, C, dtype=torch.float64)
    for b in range(N):
        for q in range(Q):
            for hd in range(H):
                aw = attention_weights[b, q, hd]
                mx = aw.max()
                ssum = 0.0
                acc = torch.zeros(C, dtype=torch.float64)
                widx = 0
                start = 0
                for (h, wd) in spatial_shapes:
                    vl = value[b, start:start + h * wd, hd].view(h, wd, C)
                    for p in range(P):
                        pg = p % Pg
                        lw = reference_points[b, q, 0, 2 * pg] * wd + sampling_offsets[b, q, hd, 2 * widx]
                        lh = reference_points[b, q, 0, 2 * pg + 1] * h + sampling_offsets[b, q, hd, 2 * widx + 1]
                        wt = torch.exp(aw[widx] - mx)
                        ssum = ssum + wt
                        him, wim = lh - 0.5, lw - 0.5
                        if -1 < him < h and -1 < wim < wd:
                            hl, wl = int(math.floor(him)), int(math.floor(wim))
                            hh, wh = hl + 1, wl + 1
                            dh, dw = him - hl, wim - wl
                            val = torch.zeros(C, dtype=torch.float64)
                            if hl >= 0 and wl >= 0:
                                val += (1 - dh) * (1 - dw) * vl[hl, wl]
                            if hl >= 0 and wh <= wd - 1:
                                val += (1 - dh) * dw * vl[hl, wh]
                            if hh <= h - 1 and wl >= 0:
                                val += dh * (1 - dw) * vl[hh, wl]
                            if hh <= h - 1 and wh <= wd - 1:
                                val += dh * dw * vl[hh, wh]
                            acc += val * wt
                        widx += 1
                    start += h * wd
                out[b, q, hd] = acc / ssum
    return out


def mdcn(x: torch.Tensor, offset: torch.Tensor, mask: torch.Tensor, weight: torch.Tensor,
         bias: torch.Tensor = None, stride: int = 1, padding: int = 1, dilation: int = 1) -> torch.Tensor:
    """``ModulatedDeformableConv2dTRT`` (groups = deform_groups = 1). ``offset`` [N, 2K, Ho, Wo] interleaved
    (dy_k, dx_k), ``mask`` [N, K, Ho, Wo] (post-sigmoid); nine ``grid_sample`` calls + a GEMM per tap."""
    N, C, H, W = x.shape
    Cout, Cin, kh, kw = weight.shape
    Ho = (H + 2 * padding - dilation * (kh - 1) - 1) // stride + 1
    Wo = (W + 2 * padding - dilation * (kw - 1) - 1) // stride + 1
    by = (torch.arange(Ho, dtype=x.dtype) * stride - padding).view(1, Ho, 1)
    bx = (torch.arange(Wo, dtype=x.dtype) * stride - padding).view(1, 1, Wo)
    out = None
    for i in range(kh):
        for j in range(kw):
            k = i * kw + j
            py = by + i * dilation + offset[:, 2 * k]
            px = bx + j * dilation + offset[:, 2 * k + 1]
            grid = torch.stack([(2.0 * px + 1.0) / W - 1.0, (2.0 * py + 1.0) / H - 1.0], dim=-1)
            s = F.grid_sample(x, grid, mode="bilinear", padding_mode="zeros", align_corners=False)
            s = s * mask[:, k:k + 1]
            contrib = torch.einsum("nchw,oc->nohw", s, weight[:, :, i, j])
            out = contrib if out is None else out + contrib
    if bias is not None:
        out = out + bias.view(1, -1, 1, 1)
    return out


def mdcn_naive(x, offset, mask, weight, stride: int = 1, padding: int = 1, dilation: int = 1) -> torch.Tensor:
    """Literal float64 port of ``modulated_deformable_im2col`` + GEMM (unit tests)."""
    x, offset, mask, weight = x.double(), offset.double(), mask.double(), weight.double()
    N, C, H, W = x.shape
    Cout, Cin, kh, kw = weight.shape
    Ho = (H + 2 * padding - dilation * (kh - 1) - 1) // stride + 1
    Wo = (W + 2 * padding - dilation * (kw - 1) - 1) // stride + 1
    cols = torch.zeros(N, C, kh * kw, Ho, Wo, dtype=torch.float64)
    for b in range(N):
        for ho in range(Ho):
            for wo in range(Wo):
                for i in range(kh):
                    for j in range(kw):
                        k = i * kw + j
                        him = ho * stride - padding + i * dilation + offset[b, 2 * k, ho, wo].item()
                        wim = wo * stride - padding + j * dilation + offset[b, 2 * k + 1, ho, wo].item()
                        val = torch.zeros(C, dtype=torch.float64)
                        if -1 < him < H and -1 < wim < W:
                            hl, wl = int(math.floor(him)), int(math.floor(wim))
                            hh, wh = hl + 1, wl + 1
                            dh, dw = him - hl, wim - wl
                            if hl >= 0 and wl >= 0:
                                val += (1 - dh) * (1 - dw) * x[b, :, hl, wl]
                            if hl >= 0 and wh <= W - 1:
                                val += (1 - dh) * dw * x[b, :, hl, wh]
                            if hh <= H - 1 and wl >= 0:
                                val += dh * (1 - dw) * x[b, :, hh, wl]
                            if hh <= H - 1 and wh <= W - 1:
                                val += dh * dw * x[b, :, hh, wh]
                        cols[b, :, k, ho, wo] = val * mask[b, k, ho, wo]
    return torch.einsum("nckhw,ock->nohw", cols, weight.view(Cout, Cin, kh * kw))


def round_half_away(x: torch.Tensor) -> torch.Tensor:
    """CUDA ``::round``: half away from zero."""
    return torch.sign(x) * torch.floor(torch.abs(x) + 0.5)


def rotate(img: torch.Tensor, angle_deg: torch.Tensor, center: Sequence[float], *,
           plugin_rounding: bool = True) -> torch.Tensor:
    """``RotateTRT`` (nearest) on ``img`` [C, H, W]; ``angle_deg`` (CCW, degrees, the graph's ``can_bus[17]``);
    ``center`` (x, y) in pixels. ``plugin_rounding=False`` is torchvision's ``rotate`` (training: round half to even);
    they differ only on exact ties (2 of 22,500 pixels at 0.37 degrees, research/bevformer/test_ref_ops.log)."""
    C, H, W = img.shape
    ang = -angle_deg.reshape(()) * math.pi / 180.0
    cx = center[0] - 0.5 * W
    cy = center[1] - 0.5 * H
    ca, sa = torch.cos(ang), torch.sin(ang)
    m = [ca, sa, -cx * ca - cy * sa + cx, -sa, ca, cx * sa - cy * ca + cy]
    xs = (torch.arange(W, dtype=img.dtype) - W * 0.5 + 0.5).view(1, W).expand(H, W)
    ys = (torch.arange(H, dtype=img.dtype) - H * 0.5 + 0.5).view(H, 1).expand(H, W)
    gx = (m[0] * xs + m[1] * ys + m[2]) / (0.5 * W)
    gy = (m[3] * xs + m[4] * ys + m[5]) / (0.5 * H)
    if not plugin_rounding:
        grid = torch.stack([gx, gy], -1).unsqueeze(0)
        return F.grid_sample(img.unsqueeze(0), grid, mode="nearest", padding_mode="zeros",
                             align_corners=False).squeeze(0)
    ix = ((gx + 1.0) * W - 1.0) / 2.0
    iy = ((gy + 1.0) * H - 1.0) / 2.0
    ixn = round_half_away(ix)
    iyn = round_half_away(iy)
    valid = (ixn >= 0) & (ixn <= W - 1) & (iyn >= 0) & (iyn <= H - 1)
    flat = (iyn.clamp(0, H - 1) * W + ixn.clamp(0, W - 1)).long().view(-1)
    out = img.reshape(C, H * W)[:, flat].view(C, H, W)
    return out * valid.to(img.dtype).unsqueeze(0)