复现论文《Distant Vehicle Detection Using Radar and Vis...

下面给你一个PyTorch实现，复现论文 Distant Vehicle Detection Using Radar and Vision 的特征级融合（concatenation 版本）结构：左支路处理三通道图像，右支路处理两通道雷达投影；在 image layer2 与 radar layer2 之后通道维拼接，再接 3 级 ResNet18 block（均 stride=2, out=256），于三种尺度分别接入 Detection head（SSD 风格，默认每格 num_anchors=6，类别数 num_classes=2）。对应关系与论文图5一致：40×16 → 20×8 → 10×4 三个特征层各自输出分类与回归（未实现 NMS/损失/数据管线）。参见论文 IV 节与图示说明。(1901.10951v2.pdf)

说明

• 输入尺寸任意（例如 3×256×640 的 RGB，2×256×640 的雷达 range/rangerate 投影），网络会按 stride 递减到 1/16、1/32、1/64 三个尺度。
• 代码只给模型结构，便于直接接你现有数据与损失函数。
• 如果想切换到 element-wise 融合（论文的另一版本），把 Concat 换成逐元素相加，并把雷达分支的最大池化去掉即可（论文 IV 节）。(1901.10951v2.pdf)

python
import torch
import torch.nn as nn
import torch.nn.functional as F


# -------------------------
# ResNet18 Basic Block
# -------------------------
class BasicBlock(nn.Module):
    expansion = 1
    def __init__(self, in_planes, planes, stride=1):
        super().__init__()
        self.conv1 = nn.Conv2d(in_planes, planes, kernel_size=3, stride=stride,
                               padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(planes)
        self.conv2 = nn.Conv2d(planes, planes, kernel_size=3, stride=1,
                               padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(planes)

        self.downsample = None
        if stride != 1 or in_planes != planes * self.expansion:
            self.downsample = nn.Sequential(
                nn.Conv2d(in_planes, planes * self.expansion,
                          kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(planes * self.expansion),
            )

    def forward(self, x):
        identity = x
        out = F.relu(self.bn1(self.conv1(x)), inplace=True)
        out = self.bn2(self.conv2(out))
        if self.downsample is not None:
            identity = self.downsample(x)
        out = F.relu(out + identity, inplace=True)
        return out


def make_layer(in_planes, planes, num_blocks, stride_first):
    """Return nn.Sequential of num_blocks BasicBlocks.
    The first block uses stride=stride_first, others stride=1.
    """
    layers = []
    strides = [stride_first] + [1] * (num_blocks - 1)
    for s in strides:
        layers.append(BasicBlock(in_planes, planes, stride=s))
        in_planes = planes * BasicBlock.expansion
    return nn.Sequential(*layers), in_planes


# -------------------------
# Detection Head (SSD-like)
# -------------------------
class DetectionHead(nn.Module):
    """Per feature map head -> classification + bbox regression.
    outputs:
      - cls_logits: [B, H*W*num_anchors, num_classes]
      - bbox_deltas: [B, H*W*num_anchors, 4]
    """
    def __init__(self, in_channels, num_anchors=6, num_classes=2):
        super().__init__()
        self.num_anchors = num_anchors
        self.num_classes = num_classes
        self.cls_conv = nn.Conv2d(in_channels, num_anchors * num_classes,
                                  kernel_size=3, padding=1)
        self.reg_conv = nn.Conv2d(in_channels, num_anchors * 4,
                                  kernel_size=3, padding=1)

    def forward(self, x):
        B, _, H, W = x.shape
        cls = self.cls_conv(x).permute(0, 2, 3, 1).contiguous()
        reg = self.reg_conv(x).permute(0, 2, 3, 1).contiguous()
        cls = cls.view(B, H * W * self.num_anchors, self.num_classes)
        reg = reg.view(B, H * W * self.num_anchors, 4)
        return cls, reg


# -------------------------
# Fusion Detection Model
# -------------------------
class FusionDetectionModel(nn.Module):
    """
    RGB branch (3ch):
        conv7x7 s2 -> bn/relu -> maxpool s2
        layer1: ResNet18 block(s) 64, stride=1
        layer2: ResNet18 block(s) 128, stride=2     -> feat @ 1/8

    Radar branch (2ch):
        conv7x7 s2 -> bn/relu -> maxpool s2
        layer2r: ResNet18 block(s) 64, stride=2      -> feat @ 1/8

    Fusion:
        concat along channels  (C=128+64=192)
        fuse3: ResNet18 block(s) 256, stride=2  -> P3 @ 1/16 (e.g., 40x16)
        fuse4: ResNet18 block(s) 256, stride=2  -> P4 @ 1/32 (20x8)
        fuse5: ResNet18 block(s) 256, stride=2  -> P5 @ 1/64 (10x4)

    Each Pi passes a DetectionHead (SSD style).
    """
    def __init__(self, num_classes=2, num_anchors=6):
        super().__init__()

        # ---- Image encoder ----
        self.img_conv = nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3, bias=False)
        self.img_bn = nn.BatchNorm2d(64)
        self.img_pool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
        self.img_layer1, c_img1 = make_layer(64, 64, num_blocks=2, stride_first=1)   # s1, 64
        self.img_layer2, c_img2 = make_layer(c_img1, 128, num_blocks=2, stride_first=2)  # s2, 128

        # ---- Radar encoder ----
        self.rad_conv = nn.Conv2d(2, 64, kernel_size=7, stride=2, padding=3, bias=False)
        self.rad_bn = nn.BatchNorm2d(64)
        self.rad_pool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
        self.rad_layer2, c_rad2 = make_layer(64, 64, num_blocks=2, stride_first=2)   # s2, 64

        # ---- Fusion ResNet blocks ----
        c_fuse_in = c_img2 + c_rad2  # concat channels: 128 + 64 = 192
        self.fuse3, c_f3 = make_layer(c_fuse_in, 256, num_blocks=2, stride_first=2)  # -> 1/16
        self.fuse4, c_f4 = make_layer(c_f3, 256, num_blocks=2, stride_first=2)      # -> 1/32
        self.fuse5, c_f5 = make_layer(c_f4, 256, num_blocks=2, stride_first=2)      # -> 1/64

        # ---- Detection heads ----
        self.head3 = DetectionHead(c_f3, num_anchors=num_anchors, num_classes=num_classes)
        self.head4 = DetectionHead(c_f4, num_anchors=num_anchors, num_classes=num_classes)
        self.head5 = DetectionHead(c_f5, num_anchors=num_anchors, num_classes=num_classes)

        self._init_weights()

    def _init_weights(self):
        for m in self.modules():
            if isinstance(m, nn.Conv2d):
                nn.init.kaiming_normal_(m.weight, mode="fan_out", nonlinearity="relu")
            elif isinstance(m, nn.BatchNorm2d):
                nn.init.constant_(m.weight, 1.)
                nn.init.constant_(m.bias, 0.)

    def forward_backbone_image(self, x):
        x = F.relu(self.img_bn(self.img_conv(x)), inplace=True)
        x = self.img_pool(x)
        x = self.img_layer1(x)
        x = self.img_layer2(x)
        return x  # 1/8

    def forward_backbone_radar(self, r):
        r = F.relu(self.rad_bn(self.rad_conv(r)), inplace=True)
        r = self.rad_pool(r)
        r = self.rad_layer2(r)
        return r  # 1/8

    def forward(self, image_input, radar_input):
        # 图像特征提取
        f_img = self.forward_backbone_image(image_input)
        # 雷达特征提取
        f_rad = self.forward_backbone_radar(radar_input)
        # 特征融合（Concat）
        f = torch.cat([f_img, f_rad], dim=1)

        # 下采样3个尺度
        p3 = self.fuse3(f)  # 1/16
        p4 = self.fuse4(p3) # 1/32
        p5 = self.fuse5(p4) # 1/64

        # 融合特征检测
        out3 = self.head3(p3)  # (cls, reg)
        out4 = self.head4(p4)
        out5 = self.head5(p5)

        # 为了与你给的接口一致，返回三个尺度的输出
        return out3, out4, out5


# -------------------------
# quick test
# -------------------------
if __name__ == "__main__":
    net = FusionDetectionModel(num_classes=2, num_anchors=6)
    img = torch.randn(2, 3, 256, 640)   # (B, 3, H, W)
    rad = torch.randn(2, 2, 256, 640)   # (B, 2, H, W)
    (cls3, reg3), (cls4, reg4), (cls5, reg5) = net(img, rad)
    for i, (c, r) in enumerate([(cls3, reg3), (cls4, reg4), (cls5, reg5)], 3):
        print(f"P{i}: cls {c.shape}, box {r.shape}")

如何对上论文结构

• 左分支：7×7 s2 → 3×3 maxpool s2 → ResNet18 block(s) 64 (s1) → ResNet18 block(s) 128 (s2)；
• 右分支：7×7 s2 → 3×3 maxpool s2 → ResNet18 block(s) 64 (s2)；
• Concat 后进入 3 个 ResNet18 block(s), s2, 256，在三个输出尺度上接入 Detection head（如图示中的三个头）。这些与论文图5网络示意完全对应；图 4 给出雷达投影到图像的 2 通道示例（range / range-rate），与我们 radar_input 的形状一致。(1901.10951v2.pdf)

需要我再给一个 element-wise 加法的版本、或把 anchors/NMS/loss 一并补齐吗？

好问题！核心点在于**“ResNet18 block”到底指什么？**

• 论文第 IV 节写的是：“We construct our network from ResNet blocks using the 18-layer variant …”——也就是采用 ResNet-18 的 BasicBlock 形式来搭网络（每个 BasicBlock = 两个 3×3 卷积 + 残差）。图 5 只是示意图，把一堆重复的 block 画成一个盒子，未逐一展开数量。(1901.10951v2.pdf)

为什么 num_blocks=2

在标准 ResNet-18 中，每个 stage 的 BasicBlock 数量是 2, 2, 2, 2。所以我在图像分支的：

python
img_layer1  # s1, 64
img_layer2  # s2, 128

都设成 num_blocks=2——这正是 ResNet-18 的每个 stage 由两个 BasicBlock 堆叠的做法（第 1 个 block 负责降采样时用 stride=2）。因此这里写成 2 是遵循 ResNet-18 的规范，而不是图里只画“一个盒子”就只放一个 block。(1901.10951v2.pdf)

图里看起来像“一个 ResNet18 模块”，你的实现对吗？

对的——我实现里每个“ResNet18 block”指的是ResNet-18 的 BasicBlock（两层 3×3），而不是“整张 ResNet-18 网络”。论文图 5 说明了在多尺度上把这些 block 输出接到 detection head 上，图里只标了“s1 / s2、通道数、特征图大小”，并没有把同一 stage 内重复的 block 个数画出来。(1901.10951v2.pdf)

如果你想跟图 5 画法更“像”

图 5 在融合后的三层（40×16、20×8、10×4）每一层都画了一个盒子。若你希望实现和图示一一对应（每个盒子=1 个 BasicBlock），可以把融合塔与雷达分支改成每层只用 1 个 BasicBlock（更轻量），而图像分支仍保持 ResNet-18 的 2 个 BasicBlock/Stage：

把我之前的几行换成：

python
# 雷达分支更贴图：只用 1 个 BasicBlock
self.rad_layer2, c_rad2 = make_layer(64, 64, num_blocks=1, stride_first=2)

# 融合塔每级 1 个 BasicBlock（贴近图 5 的每个盒子）
self.fuse3, c_f3 = make_layer(c_fuse_in, 256, num_blocks=1, stride_first=2)
self.fuse4, c_f4 = make_layer(c_f3,     256, num_blocks=1, stride_first=2)
self.fuse5, c_f5 = make_layer(c_f4,     256, num_blocks=1, stride_first=2)

这样做的影响：

• 尺寸/下采样比例不变（依旧 1/16、1/32、1/64），与图 5 标注的 40×16、20×8、10×4 对齐。
• 参数和计算量更小；精度可能略有差别，需用你的数据集验证。

小结

• num_blocks=2 的来源：ResNet-18 每个 stage 两个 BasicBlock 的标准设定；图是示意并未展开重复数。
• 你的疑问很正常：图 5 每层画“一个盒子”容易被读成“只放 1 个 block”。若想完全按图的最简实现，可把融合塔/雷达分支改为 num_blocks=1；而图像主干保持 2 更符合“18-layer variant”的描述。(1901.10951v2.pdf)

如果你要，我可以把两种版本（“标准 2 个 block/层”和“贴图 1 个 block/层”）都打包给你做 A/B 对比推理速度与精度。