YOLO 源代码分析(三) YOLOv3 主干网络

Posted by feizaipp on June 25, 2020

我的博客

1. 概述

       ultralytics 版 YOLOv3 的主干网络是由 Darknet53 加上 SPP 结构再加上残差网络构成的。这个版本的主干网络是通过解析配置文件来加载的,由于网络整体结构比较简单,不过多介绍,本文主要介绍如果解析配置文件。

2. 配置文件格式

       主干网络的配置文件再 my_yolov3.cfg 配置文件中,在搭建自己的网络时只需要根据网络的预测类别进行相关参数的修改即可。如果使用 VOC 数据集,数据集的类别是 20 ,每个预测器预测 (20 + 1 + 4)=25 个值,分别为 20 个类别和背景,再加上边界框回归参数, feature map 上的每一个点预测 3 个 anchor 也就是每个点预测 75 个值。

2.1. 类型

       net 该类型定义网络的超参数。

       convolutional 该类型定义网络卷积神经网络。

       shortcut 残差网络,各个元素对应相加。

       maxpool 最大池化层。

       route 残差网络,在 channel 维度上进行拼接。

       yolo 网络预测层。

       upsample 上采样层。

2.2. 实现

       首先看 Darknet 网络的实现,构造函数实现如下:

  • parse_model_cfg: 解析配置文件,以字典的形式缓存配置
  • create_modules: 根据配置文件构建网络,下面重点介绍
  • self.yolo_layers: 获取网络中的 yolo 层,一共有 3 个 yolo 层
    class Darknet(nn.Module):
      def __init__(self, cfg, img_size=(416, 416), verbose=False):
          super(Darknet, self).__init__()
          self.input_size = [img_size] * 2 if isinstance(img_size, int) else img_size
          self.module_defs = parse_model_cfg(cfg)
          self.module_list, self.routs = create_modules(self.module_defs, img_size)
          self.yolo_layers = get_yolo_layers(self)
          self.info(verbose) if not ONNX_EXPORT else None  # print model description
    

           让我们看一下网络中的各个模块是如何搭建起来的,在 create_modules 函数中,根据配置文的缓存数据进行搭建网络:

  • 从缓存中删除列表的第一项,该项为网络超参数,未使用,超参数在 hyp.yaml 中
  • output_filters: 该变量记录每一层的输入通道数,初始值为 3 ,输入的是 RGB 彩色图像
  • module_list: 该变量存储网络的每一个模块
  • routs: 改变量统计哪些特征层的输出会被后续的层使用到(可能是特征融合,也可能是拼接)
  • yolo_index: 存储 yolo 层的索引,一共有 3 个 yolo 层
  • 接下来是遍历 modules_defs 中的每一项,存储的是一个字典
  • convolutional: 卷积层,添加 Conv2d 模块,添加 activation 激活函数,添加 BatchNorm2d ,如果不存在 BN 层,意味着该层为 yolo 层,将层号添加到 routs 变量中
  • maxpool: 最大池化层
  • upsample: 上采样层
  • route: 这一层代表残差网络,在 channel 维度上进行拼接。 layers 表示与哪些层进行拼接, layers 是负值,表示与 output_filters 倒数第 layers 层进行拼接, layers 为正值,表示与 output_filters 正数第 layers+1 层为与之拼接的层进行拼接; filters 求和就是经过拼接后网络输出的维度,添加到 output_filter 中; routs 记录层号,这些层的输出以后会用到,当 l < 0 时, 向前跳过 l 层,为拼接的层,当 l > 0 时, 则第 l 层为拼接的层
  • shortcut: 这一层表示特征融合,对应元素相加。融合后网络的维度, shortcut 是对应元素相加,维度不变。
  • yolo: 网络有 3 个,每个 yolo 层的缩放比例分别为 [32, 16, 8] ,下面对于 bias_ 的设置我目前还每看懂是什么意思。
  • 最后返回 module_list 和 routs_binary , routs_binary 记录哪些层需要融合或者拼接
    def create_modules(modules_defs: list, img_size):
      img_size = [img_size] * 2 if isinstance(img_size, int) else img_size
      modules_defs.pop(0)  # cfg training hyperparams (unused)
      output_filters = [3]  # input channels
      module_list = nn.ModuleList()
      routs = []  # list of layers which rout to deeper layers
      yolo_index = -1
    
      for i, mdef in enumerate(modules_defs):
          modules = nn.Sequential()
    
          if mdef["type"] == "convolutional":
              bn = mdef["batch_normalize"]  # 1 or 0 / use or not
              filters = mdef["filters"]
              k = mdef["size"]  # kernel size
              stride = mdef["stride"] if "stride" in mdef else (mdef['stride_y'], mdef["stride_x"])
              if isinstance(k, int):
                  modules.add_module("Conv2d", nn.Conv2d(in_channels=output_filters[-1],
                                                         out_channels=filters,
                                                         kernel_size=k,
                                                         stride=stride,
                                                         padding=k // 2 if mdef["pad"] else 0,
                                                         bias=not bn))
              else:
                  raise TypeError("conv2d filter size must be int type.")
    
              if bn:
                  modules.add_module("BatchNorm2d", nn.BatchNorm2d(filters))
              else:
                  routs.append(i)  # detection output (goes into yolo layer)
    
              if mdef["activation"] == "leaky":
                  modules.add_module("activation", nn.LeakyReLU(0.1, inplace=True))
              else:
                  pass
    
          elif mdef["type"] == "BatchNorm2d":
              pass
    
          elif mdef["type"] == "maxpool":
              k = mdef["size"]  # kernel size
              stride = mdef["stride"]
              modules = nn.MaxPool2d(kernel_size=k, stride=stride, padding=(k - 1) // 2)
    
          elif mdef["type"] == "upsample":
              if ONNX_EXPORT:  # explicitly state size, avoid scale_factor
                  g = (yolo_index + 1) * 2 / 32  # gain
                  modules = nn.Upsample(size=tuple(int(x * g) for x in img_size))
              else:
                  modules = nn.Upsample(scale_factor=mdef["stride"])
    
          elif mdef["type"] == "route":  # [-2],  [-1,-3,-5,-6], [-1, 61]
              layers = mdef["layers"]
              filters = sum([output_filters[l + 1 if l > 0 else l] for l in layers])
              routs.extend([i + l if l < 0 else l for l in layers])
              modules = FeatureConcat(layers=layers)
    
          elif mdef["type"] == "shortcut":
              layers = mdef["from"]
              filters = output_filters[-1]
              routs.append(i + layers[0])
              modules = WeightedFeatureFusion(layers=layers, weight="weights_type" in mdef)
    
          elif mdef["type"] == "yolo":
              yolo_index += 1  # 记录是第几个yolo_layer [0, 1, 2]
              stride = [32, 16, 8]  # 预测特征层对应原图的缩放比例
    
              modules = YOLOLayer(anchors=mdef["anchors"][mdef["mask"]],  # anchor list
                                  nc=mdef["classes"],  # number of classes
                                  img_size=img_size,
                                  stride=stride[yolo_index])
    
              # Initialize preceding Conv2d() bias (https://arxiv.org/pdf/1708.02002.pdf section 3.3)
              try:
                  j = -1
                  bias_ = module_list[j][0].bias  # shape(255,) 索引0对应Sequential中的Conv2d
                  bias = bias_.view(modules.na, -1)  # shape(3, 85)
                  bias[:, 4] += -4.5  # obj
                  bias[:, 5:] += math.log(0.6 / (modules.nc - 0.99))  # cls (sigmoid(p) = 1/nc)
                  module_list[j][0].bias = torch.nn.Parameter(bias_, requires_grad=bias_.requires_grad)
              except Exception as e:
                  print('WARNING: smart bias initialization failure.', e)
          else:
              print("Warning: Unrecognized Layer Type: " + mdef["type"])
    
          # Register module list and number of output filters
          module_list.append(modules)
          output_filters.append(filters)
    
      routs_binary = [False] * len(modules_defs)
      for i in routs:
          routs_binary[i] = True
      return module_list, routs_binary
    

       接下来介绍 route 层的实现,该层实现在 FeatureConcat 类中:

  • 将多个特征层在 channel 维度上拼接在一起
    class FeatureConcat(nn.Module):
      def __init__(self, layers):
          super(FeatureConcat, self).__init__()
          self.layers = layers  # layer indices
          self.multiple = len(layers) > 1  # multiple layers flag
    
      def forward(self, x, outputs):
          return torch.cat([outputs[i] for i in self.layers], 1) if self.multiple else outputs[self.layers[0]]
    

       特征融合的实现在 WeightedFeatureFusion 类中。

  • 默认 weight=False ,所以前向传播就是直接各个层的输出特征相加
    class WeightedFeatureFusion(nn.Module):  # weighted sum of 2 or more layers https://arxiv.org/abs/1911.09070
      def __init__(self, layers, weight=False):
          super(WeightedFeatureFusion, self).__init__()
          self.layers = layers  # layer indices
          self.weight = weight  # apply weights boolean
          self.n = len(layers) + 1  # number of layers 融合的特征矩阵个数
          if weight:
              self.w = nn.Parameter(torch.zeros(self.n), requires_grad=True)  # layer weights
    
      def forward(self, x, outputs):
          # Weights
          if self.weight:
              w = torch.sigmoid(self.w) * (2 / self.n)  # sigmoid weights (0-1)
              x = x * w[0]
    
          # Fusion
          nx = x.shape[1]  # input channels
          for i in range(self.n - 1):
              a = outputs[self.layers[i]] * w[i + 1] if self.weight else outputs[self.layers[i]]  # feature to add
              na = a.shape[1]  # feature channels
    
              if nx == na:  # same shape 如果channel相同,直接相加
                  x = x + a
              elif nx > na:  # slice input 如果channel不同,将channel多的特征矩阵砍掉部分channel保证相加的channel一致
                  x[:, :na] = x[:, :na] + a  # or a = nn.ZeroPad2d((0, 0, 0, 0, 0, dc))(a); x = x + a
              else:  # slice feature
                  x = x + a[:, :nx]
    
          return x
    

       预测器实现在 YOLOLayer 类中,构造函数实现如下:

  • 传入当前预测特征层的 anchor 的大小,类别个数, img_size ,下采样倍数
  • self.anchor_vec: 将 anchors 大小缩放到 grid 尺度
  • self.anchor_wh: 各个元素的定义为 [batch_size, na, grid_h, grid_w, wh] ,值为 1 的维度对应的值不是固定值,后续操作可根据 broadcast 广播机制自动扩充
    class YOLOLayer(nn.Module):
      def __init__(self, anchors, nc, img_size, stride):
          super(YOLOLayer, self).__init__()
          self.anchors = torch.Tensor(anchors)
          self.stride = stride
          self.na = len(anchors)  # number of anchors (3)
          self.nc = nc  # number of classes (80)
          self.no = nc + 5  # number of outputs (85: x, y, w, h, obj, cls1, ...)
          self.nx, self.ny, self.ng = 0, 0, (0, 0)  # initialize number of x, y gridpoints
          self.anchor_vec = self.anchors / self.stride
          self.anchor_wh = self.anchor_vec.view(1, self.na, 1, 1, 2)
          self.grid = None
    
          if ONNX_EXPORT:
              self.training = False
              self.create_grids((img_size[1] // stride, img_size[0] // stride))  # number x, y grid points
    

           接下来看 forward 函数,首先是 create_grids 函数,该函数主要用于生成网格,且在推理时才使用。

  • self.ng: 生成网格大小
  • 使用 torch.meshgrid 函数在特征图上获得网格点的坐标
  • self.grid: 将网格点坐标堆叠在一起,将坐标 reshape 到 (batch_size, na, grid_h, grid_w, wh) 维度,值为 1 的可根据广播机制进行扩充
    def create_grids(self, ng=(13, 13), device="cpu"):
      self.nx, self.ny = ng
      self.ng = torch.tensor(ng, dtype=torch.float)
    
      # build xy offsets 构建每个cell处的anchor的xy偏移量(在feature map上的)
      if not self.training:  # 训练模式不需要回归到最终预测boxes
          yv, xv = torch.meshgrid([torch.arange(self.ny, device=device),
                                      torch.arange(self.nx, device=device)])
          # batch_size, na, grid_h, grid_w, wh
          self.grid = torch.stack((xv, yv), 2).view((1, 1, self.ny, self.nx, 2)).float()
    
      if self.anchor_vec.device != device:
          self.anchor_vec = self.anchor_vec.to(device)
          self.anchor_wh = self.anchor_wh.to(device)
    

           下面看 forward 函数的实现:

  • 对输入进行维度的调整 [bs, anchor, grid, grid, xywh + obj + classes]
  • 如果时训练模式,直接返回调整维度后的值,进行损失计算
  • 如果时验证模式,将网络输出值处理,首先将 xy 转换成在 feature map 上的 xy 坐标,将 wh 转换成在 feature map 上的 wh 值,将 xywh 四个值映射到原图上,计算网络输出的置信度。最后将预测值 reshape 到 [bs, -1, self.no] 格式。 ``` def forward(self, p): if ONNX_EXPORT: bs = 1 # batch size else: bs, _, ny, nx = p.shape # batch_size, predict_param(255), grid(13), grid(13) if (self.nx, self.ny) != (nx, ny) or self.grid is None: # fix no grid bug self.create_grids((nx, ny), p.device)

    p = p.view(bs, self.na, self.no, self.ny, self.nx).permute(0, 1, 3, 4, 2).contiguous() # prediction

    if self.training: return p elif ONNX_EXPORT: # Avoid broadcasting for ANE operations m = self.na * self.nx * self.ny # 3* ng = 1. / self.ng.repeat(m, 1) grid = self.grid.repeat(1, self.na, 1, 1, 1).view(m, 2) anchor_wh = self.anchor_wh.repeat(1, 1, self.nx, self.ny, 1).view(m, 2) * ng

      p = p.view(m, self.no)
      # xy = torch.sigmoid(p[:, 0:2]) + grid  # x, y
      # wh = torch.exp(p[:, 2:4]) * anchor_wh  # width, height
      # p_cls = torch.sigmoid(p[:, 4:5]) if self.nc == 1 else \
      #     torch.sigmoid(p[:, 5:self.no]) * torch.sigmoid(p[:, 4:5])  # conf
      p[:, :2] = (torch.sigmoid(p[:, 0:2]) + grid) * ng  # x, y
      p[:, 2:4] = torch.exp(p[:, 2:4]) * anchor_wh  # width, height
      p[:, 4:] = torch.sigmoid(p[:, 4:])
      p[:, 5:] = p[:, 5:self.no] * p[:, 4:5]
      return p   else:  # inference
      # [bs, anchor, grid, grid, xywh + obj + classes]
      io = p.clone()  # inference output
      io[..., :2] = torch.sigmoid(io[..., :2]) + self.grid
      io[..., 2:4] = torch.exp(io[..., 2:4]) * self.anchor_wh
      io[..., :4] *= self.stride
      torch.sigmoid_(io[..., 4:])
      return io.view(bs, -1, self.no), p  # view [1, 3, 13, 13, 85] as [1, 507, 85]
    
&#160; &#160; &#160; &#160;Darknet 网络中的子模块就介绍完了,现在我们回到 Darknet 实现上。在 Darknet 构造函数中调用了获取 yolo 层的接口,实现如下:
* yolo 层共有 3 个,分别是 [89, 101, 113]

def get_yolo_layers(self): return [i for i, m in enumerate(self.module_list) if m.class.name == ‘YOLOLayer’]


&#160; &#160; &#160; &#160;下面分析一下 Darknet 的前向传播函数, forward 函数直接调用了 forward_once 函数。
* yolo_out, out: yolo_out 收集每个 yolo_layer 层的输出; out 收集每个模块的输出。
* 前向传播过程中要记录需要网络的输出,以供 WeightedFeatureFusion 和 FeatureConcat 两个层使用。
* 如果是训练模式,返回 yolo 层的输出, shape 为 [bs, anchor, grid, grid, xywh + obj + classes]
* 如果是验证模式,返回 yolo 层的输出, shape 为 [bs, -1, self.no] 和 [bs, anchor, grid, grid, xywh + obj + classes] ,最后要将 [bs, -1, self.no] 在维度 1 上进行堆叠。

def forward(self, x, verbose=False): return self.forward_once(x, verbose=verbose)

def forward_once(self, x, verbose=False): yolo_out, out = [], []

for i, module in enumerate(self.module_list):
    name = module.__class__.__name__
    if name in ["WeightedFeatureFusion", "FeatureConcat"]:  # sum, concat
        x = module(x, out)  # WeightedFeatureFusion(), FeatureConcat()
    elif name == "YOLOLayer":
        yolo_out.append(module(x))
    else:  # run module directly, i.e. mtype = 'convolutional', 'upsample', 'maxpool', 'batchnorm2d' etc.
        x = module(x)

    out.append(x if self.routs[i] else [])

if self.training:  # train
    return yolo_out
elif ONNX_EXPORT:  # export
    p = torch.cat(yolo_out, dim=0)

    return p
else:  # inference or test
    x, p = zip(*yolo_out)  # inference output, training output
    x = torch.cat(x, 1)  # cat yolo outputs

    return x, p ``` &#160; &#160; &#160; &#160;至此, YOLOv3 主干网络的源码分析就结束了。