基于 PaddleVideo 的花滑骨骼点动作识别 2s-AGCN

配置文件

注意，2s-AGCN 是双流框架，分为节点流 joint 和骨骼流 bone，所以配置文件也有两个分别用于 joint 和 bone。

训练 AGCN 模型时的配置文件是 agcn_fsd.yaml，
详解参见博文：PaddleVideo 中 agcn_fsd.yaml 配置文件代码详解

# 使用 GPU 版本
!python3.7 main.py -c configs/recognition/agcn/agcn_fsd.yaml

现在，PaddleVideo 更新了，新加入了2s-AGCN 和 CTR-GCN，但是需要自己调整配置文件，因为没有和花滑比赛数据集 fsd-10花样滑冰数据集 完全匹配的配置。

节点流

PaddleVideo/configs/recognition/agcn2s/ 文件夹下的配置文件有：

使用 2s-AGCN 进行训练，可以尝试使用 2s-agcn_ntucs_joint_fsd.yaml 配置文件作为节点流的配置。

该配置文件是用于 NTU-CS 数据集和 FSD 数据集的联合训练的，适用于多模态数据的分类任务，与使用的原始 AGCN 配置文件 agcn_fsd.yaml 类似。

将命令行中的配置文件路径改为：

configs/recognition/2s-agcn/2s-agcn_ntucs_joint_fsd.yaml

然后对 2s-agcn_ntucs_joint_fsd.yaml 稍加修改，使其用于 FSD 数据集，（.yaml 文件名也可以修改如 agcn2s_fsd_joint.yaml）即可开始 2s-AGCN 的节点流的训练。

配置文件 2s-agcn_ntucs_joint_fsd.yaml

MODEL: #MODEL fieldframework: "RecognizerGCN" #Mandatory, indicate the type of network, associate to the 'paddlevideo/modeling/framework/' .backbone: #Mandatory, indicate the type of backbone, associate to the 'paddlevideo/modeling/backbones/' .name: "AGCN2s" #Mandatory, The name of backbone.num_point: 25num_person: 1graph: "ntu_rgb_d"graph_args:labeling_mode: "spatial"in_channels: 2head:name: "AGCN2sHead" #Mandatory, indicate the type of head, associate to the 'paddlevideo/modeling/heads'num_classes: 60  #Optional, the number of classes to be classified.in_channels: 64  #output the number of classes.M: 1  #number of people.DATASET: #DATASET fieldbatch_size: 64 #Mandatory, bacth sizenum_workers: 4 #Mandatory, the number of subprocess on each GPU.test_batch_size: 64test_num_workers: 0train:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "data/fsd10/FSD_train_data.npy" #Mandatory, train data index file pathlabel_path: "data/fsd10/FSD_train_label.npy"test:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "data/fsd10/test_A_data.npy" #Mandatory, valid data index file pathtest_mode: TruePIPELINE: #PIPELINE fieldtrain: #Mandotary, indicate the pipeline to deal with the training data, associate to the 'paddlevideo/loader/pipelines/'sample:name: "AutoPadding"window_size: 300transform: #Mandotary, image transfrom operator- SkeletonNorm:test: #Mandotary, indicate the pipeline to deal with the training data, associate to the 'paddlevideo/loader/pipelines/'sample:name: "AutoPadding"window_size: 300transform: #Mandotary, image transfrom operator- SkeletonNorm:OPTIMIZER: #OPTIMIZER fieldname: 'Momentum'momentum: 0.9learning_rate:iter_step: Truename: 'CustomWarmupAdjustDecay'step_base_lr: 0.1warmup_epochs: 5lr_decay_rate: 0.1boundaries: [ 30, 40 ]weight_decay:name: 'L2'value: 1e-4use_nesterov: TrueMETRIC:name: 'SkeletonMetric'out_file: 'submission.csv'INFERENCE:name: 'STGCN_Inference_helper'num_channels: 2window_size: 350vertex_nums: 25person_nums: 1model_name: "AGCN2s"
log_interval: 10 #Optional, the interal of logger, default:10
epochs: 50 #Mandatory, total epoch
save_interval: 10

针对 FSD 数据集，需要进行如下修改，才能与该配置文件相适配：

MODEL 字段

在 MODEL 字段中，将 num_classes 修改为 30，对应 FSD 数据集中的 30 个类别。

head:name: "AGCN2sHead"num_classes: 30in_channels: 64M: 1

这段代码定义了 AGCN2s 模型的头部，也就是最后一层网络结构，用于将经过编码器和解码器的中间表示转化为分类结果。其中：

• name: "AGCN2sHead" 表示使用 AGCN2s 模型的头部结构。
• num_classes：30 表示 FSD 数据集中总共有 30 个标签类别需要分类，因此网络的最后一层输出大小为 30。
• in_channels: 64 表示每个时间序列的输入向量的维度为 64，这个值应该根据数据集的特点进行设置，以便网络能够更好地感知时序上的不同特征。
• M: 1 表示每个时间序列中只包含一个人物的骨架数据，因为 FSD 数据集中每个样本只包含一个人物的动作数据。

其中，in_channels: 64 是一个可以调整的参数，具体取值应该根据数据集的特点和模型的架构来进行设置。

在 AGCN2s 模型中，输入的时序数据首先会经过空间时序图卷积网络（STGCN）的多层卷积、池化和归一化操作，将 25 个节点上的特征转化为一个时序上的特征表示。然后，这个时序上的特征表示会被 送到 AGCN 模块中进行更加精细的建模操作，最终输出到头部网络中进行分类。

因此，in_channels 的取值应该考虑 STGCN 和 AGCN 模块的设计，以及数据集中时序数据的性质。

例如，在使用 FSD 数据集时，可以根据实验结果确定一个相对合适的 in_channels 值，一般在 64 到 256 之间选择。

• 当 in_channels 取值较小时，模型可能难以捕捉复杂的时序特征。
• 当 in_channels 取值较大时，模型可能容易过拟合，训练时间也会增加。

因此，需要在实验中尝试不同的 in_channels 值，并根据实验结果来确定最佳的取值。

DATASET 字段

在 DATASET 字段中，将 train 和 test 中的 file_path 分别修改为 FSD 数据集中的训练集和测试集的路径。

train:format: "SkeletonDataset"file_path: "/home/aistudio/data/data104925/train_data.npy" #训练数据集路径label_path: "/home/aistudio/data/data104925/train_label.npy" #训练数据集路径
test:format: "SkeletonDataset"file_path: "/home/aistudio/data/data104924/test_A_data.npy"  #测试数据集路径test_mode: True

PIPELINE 和 INFERENCE 字段

window_size: 300 是一个可以调整的参数。

window_size 用于指定自动填充补零的窗口大小（单位为帧数），由于不同视频序列的长度可能不同，因此在训练过程中需要将视频序列填充到相同的长度，以便于模型处理。

OPTIMIZER 字段

OPTIMIZER: #OPTIMIZER fieldname: 'Momentum'momentum: 0.9learning_rate:iter_step: Truename: 'CustomWarmupAdjustDecay'step_base_lr: 0.1warmup_epochs: 5lr_decay_rate: 0.1boundaries: [ 30, 40 ]weight_decay:name: 'L2'value: 1e-4use_nesterov: True

这段代码定义了一个优化器（optimizer），用于在训练 AGCN2s 模型时更新网络参数。

具体来说：

• name: 'Momentum' 表示使用动量（momentum）优化器，即 SGD with Momentum。
• momentum: 0.9 表示设置动量系数为 0.9，这个系数表示在更新梯度时引入前一次梯度的影响程度，主要用于加速优化过程。
• learning_rate 定义了学习率（learning rate）的调整策略，包括学习率衰减和学习率热身两个过程：
  • name: 'CustomWarmupAdjustDecay' 表示使用一个自定义的学习率调整策略，即先进行学习率热身，然后根据预定的步骤调整学习率大小，并在每个阶段结束时进行学习率衰减。
  • step_base_lr: 0.1 表示初始学习率为 0.1。
  • warmup_epochs: 5 表示设置学习率热身的轮数为 5 轮，即在前 5 轮迭代中，学习率会从很小的值逐步增加到设定的初始值。
  • lr_decay_rate: 0.1 表示设置学习率的衰减率为 0.1，即在预定的迭代轮数结束时将学习率缩小到原来的 0.1 倍。
  • boundaries: [ 30, 40 ] 表示在第 30 轮和第 40 轮结束时进行学习率调整。具体来说，将学习率按照一定比例进行缩小，并在后续训练中保持调整后的大小。
• weight_decay 定义了权重衰减（weight decay）的方式，即在优化过程中对参数进行正则化以避免过拟合：
  • name: 'L2' 表示使用 L2 正则化方式对网络参数进行约束。
  • value: 1e-4 表示设置 L2 正则化系数为 0.0001。
• use_nesterov: True 表示同时采用 Nesterov 动量（Nesterov Momentum）来加速优化过程。Nesterov 动量相比于普通动量算法，可以更好地处理优化问题中的高曲率区域，从而提升优化效果。

agcn2s.py

文件路径：PaddleVideo/paddlevideo/modeling/backbones/agcn2s.py
参见博文：2s-AGCN 代码理解

import paddle
import paddle.nn as nn
import numpy as np
from ..registry import BACKBONESdef import_class(name):components = name.split('.')mod = __import__(components[0])for comp in components[1:]:mod = getattr(mod, comp)return modclass UnitTCN(nn.Layer):def __init__(self, in_channels, out_channels, kernel_size=9, stride=1):super(UnitTCN, self).__init__()pad = int((kernel_size - 1) / 2)self.conv = nn.Conv2D(in_channels,out_channels,kernel_size=(kernel_size, 1),padding=(pad, 0),stride=(stride, 1))self.bn = nn.BatchNorm2D(out_channels)self.relu = nn.ReLU()def forward(self, x):" input size : (N*M, C, T, V)"x = self.bn(self.conv(x))return xclass UnitGCN(nn.Layer):def __init__(self,in_channels,out_channels,A,coff_embedding=4,num_subset=3):super(UnitGCN, self).__init__()inter_channels = out_channels // coff_embeddingself.inter_c = inter_channelsPA = self.create_parameter(shape=A.shape, dtype='float32')self.PA = PAself.A = paddle.to_tensor(A.astype(np.float32))self.num_subset = num_subsetself.conv_a = nn.LayerList()self.conv_b = nn.LayerList()self.conv_d = nn.LayerList()for i in range(self.num_subset):self.conv_a.append(nn.Conv2D(in_channels, inter_channels, 1))self.conv_b.append(nn.Conv2D(in_channels, inter_channels, 1))self.conv_d.append(nn.Conv2D(in_channels, out_channels, 1))if in_channels != out_channels:self.down = nn.Sequential(nn.Conv2D(in_channels, out_channels, 1),nn.BatchNorm2D(out_channels))else:self.down = lambda x: xself.bn = nn.BatchNorm2D(out_channels)self.soft = nn.Softmax(-2)self.relu = nn.ReLU()def forward(self, x):N, C, T, V = x.shapeA = self.A + self.PAy = Nonefor i in range(self.num_subset):A1 = paddle.transpose(self.conv_a[i](x),perm=[0, 3, 1,2]).reshape([N, V, self.inter_c * T])A2 = self.conv_b[i](x).reshape([N, self.inter_c * T, V])A1 = self.soft(paddle.matmul(A1, A2) / A1.shape[-1])A1 = A1 + A[i]A2 = x.reshape([N, C * T, V])z = self.conv_d[i](paddle.matmul(A2, A1).reshape([N, C, T, V]))y = z + y if y is not None else zy = self.bn(y)y += self.down(x)return self.relu(y)class Block(nn.Layer):def __init__(self, in_channels, out_channels, A, stride=1, residual=True):super(Block, self).__init__()self.gcn1 = UnitGCN(in_channels, out_channels, A)self.tcn1 = UnitTCN(out_channels, out_channels, stride=stride)self.relu = nn.ReLU()if not residual:self.residual = lambda x: 0elif (in_channels == out_channels) and (stride == 1):self.residual = lambda x: xelse:self.residual = UnitTCN(in_channels,out_channels,kernel_size=1,stride=stride)def forward(self, x):x = self.tcn1(self.gcn1(x)) + self.residual(x)return self.relu(x)# This Graph structure is for the NTURGB+D dataset. If you use a custom dataset, modify num_node and the corresponding graph adjacency structure.
class Graph:def __init__(self, labeling_mode='spatial'):num_node = 25self_link = [(i, i) for i in range(num_node)]inward_ori_index = [(1, 2), (2, 21), (3, 21), (4, 3), (5, 21), (6, 5),(7, 6), (8, 7), (9, 21), (10, 9), (11, 10),(12, 11), (13, 1), (14, 13), (15, 14), (16, 15),(17, 1), (18, 17), (19, 18), (20, 19), (22, 23),(23, 8), (24, 25), (25, 12)]inward = [(i - 1, j - 1) for (i, j) in inward_ori_index]outward = [(j, i) for (i, j) in inward]neighbor = inward + outwardself.num_node = num_nodeself.self_link = self_linkself.inward = inwardself.outward = outwardself.neighbor = neighborself.A = self.get_adjacency_matrix(labeling_mode)def edge2mat(self, link, num_node):A = np.zeros((num_node, num_node))for i, j in link:A[j, i] = 1return Adef normalize_digraph(self, A):Dl = np.sum(A, 0)h, w = A.shapeDn = np.zeros((w, w))for i in range(w):if Dl[i] > 0:Dn[i, i] = Dl[i]**(-1)AD = np.dot(A, Dn)return ADdef get_spatial_graph(self, num_node, self_link, inward, outward):I = self.edge2mat(self_link, num_node)In = self.normalize_digraph(self.edge2mat(inward, num_node))Out = self.normalize_digraph(self.edge2mat(outward, num_node))A = np.stack((I, In, Out))return Adef get_adjacency_matrix(self, labeling_mode=None):if labeling_mode is None:return self.Aif labeling_mode == 'spatial':A = self.get_spatial_graph(self.num_node, self.self_link,self.inward, self.outward)else:raise ValueError()return A@BACKBONES.register()
class AGCN2s(nn.Layer):def __init__(self,num_point=25,num_person=2,graph='ntu_rgb_d',graph_args=dict(),in_channels=3):super(AGCN2s, self).__init__()if graph == 'ntu_rgb_d':self.graph = Graph(**graph_args)else:raise ValueError()A = self.graph.Aself.data_bn = nn.BatchNorm1D(num_person * in_channels * num_point)self.l1 = Block(in_channels, 64, A, residual=False)self.l2 = Block(64, 64, A)self.l3 = Block(64, 64, A)self.l4 = Block(64, 64, A)self.l5 = Block(64, 128, A, stride=2)self.l6 = Block(128, 128, A)self.l7 = Block(128, 128, A)self.l8 = Block(128, 256, A, stride=2)self.l9 = Block(256, 256, A)self.l10 = Block(256, 256, A)def forward(self, x):N, C, T, V, M = x.shapex = x.transpose([0, 4, 3, 1, 2]).reshape_([N, M * V * C, T])x = self.data_bn(x)x = x.reshape_([N, M, V, C,T]).transpose([0, 1, 3, 4,2]).reshape_([N * M, C, T, V])x = self.l1(x)x = self.l2(x)x = self.l3(x)x = self.l4(x)x = self.l5(x)x = self.l6(x)x = self.l7(x)x = self.l8(x)x = self.l9(x)x = self.l10(x)return x

graph

注意 class AGCN2s 中，邻接矩阵 $A$ 的构造是根据 graph == 'ntu_rgb_d'，

num_node = 25
self_link = [(i, i) for i in range(num_node)]
inward_ori_index = [(1, 2), (2, 21), (3, 21), (4, 3), (5, 21), (6, 5), (7, 6),(8, 7), (9, 21), (10, 9), (11, 10), (12, 11), (13, 1),(14, 13), (15, 14), (16, 15), (17, 1), (18, 17), (19, 18),(20, 19), (22, 23), (23, 8), (24, 25), (25, 12)]
inward = [(i - 1, j - 1) for (i, j) in inward_ori_index]
outward = [(j, i) for (i, j) in inward]
neighbor = inward + outward

其中，

• 变量 num_node 表示节点数量，
• self_link 是自环的节点列表，
• inward_ori_index 和 inward 是表示从其他节点指向该节点的边的列表，
• outward 是表示从该节点指向其他节点的边的列表，
• neighbor 是所有边的列表。

邻接矩阵是由 self_link, inward 和 outward 构成的一个三维数组。

$inward\_ori\_index$ 是一个变量，用于存储 每个节点与向心节点的连接对。向心节点是指 从其他节点指向该节点的节点。例如，$(1,2)$ 表示节点1与节点2相连，且节点2是向心节点。这个变量是用于创建 NTU RGB-D 数据集对应的图结构的，其中每个节点代表一个人体关节，每条边代表一个人体骨骼。
$inward\_ori\_index$ 是根据下图定义的，左图显示了 Kinetics-Skeleton 数据集的关节标签，右图显示了 NTU-RGBD 数据集的关节标签(21是中心关节)。

NTU-RGBD 数据集的关节标签(21是中心关节)，但是 fsd-10花样滑冰数据集中，8号索引关键点为人体中心。要做如下修改：

		self.num_node = 25self_link = [(i, i) for i in range(self.num_node)]inward_ori_index = [(1, 8), (0, 1), (15, 0), (17, 15), (16, 0),(18, 16), (5, 1), (6, 5), (7, 6), (2, 1), (3, 2),(4, 3), (9, 8), (10, 9), (11, 10), (24, 11),(22, 11), (23, 22), (12, 8), (13, 12), (14, 13),(21, 14), (19, 14), (20, 19)]

修改如上后，可以把 2s-agcn_ntucs_joint_fsd.yaml 另存为 2s-agcn_fsd_joint.yaml 表示用于 fsd 的节点流配置文件。

输入通道数

self.data_bn = nn.BatchNorm1D(num_person * in_channels * num_point)

原来的 NTU 数据集是 25 num_point * 3 in_channels * 1 num_person = 75 输入通道数，但是现在 FSD 数据集是 25 num_point * 2 in_channels * 1 num_person = 50 输入通道数，注意在配置文件中修改 num_person、in_channels 和 num_point 即可。

骨骼流

刚开始只改了配置文件中的 num_person、in_channels 和 num_point 参数，但是运行时候遇到了问题：

 File "/home/aistudio/work/FigureSkating/paddlevideo/modeling/backbones/agcn2s.py", line 218, in forwardx = self.data_bn(x)ValueError: (InvalidArgument) ShapeError: the shape of scale must equal to [75]But received: the shape of scale is [50][Hint: Expected scale_dim[0] == C, but received scale_dim[0]:50 != C:75.] (at /paddle/paddle/phi/infermeta/multiary.cc:593)

问题出在这一行代码 x = self.data_bn(x)，报错提示预期的维度是 [75]，但是我的输入通道数是 [50]（同上节点流中的 self.data_bn = nn.BatchNorm1D(num_person * in_channels * num_point)）。

那我就好奇了，为什么节点流的可以运行，到了骨骼流就不行了，于是我打印了相关数据，

def forward(self, x):N, C, T, V, M = x.shapeprint(N,C,T,V,M)  # 打印x = x.transpose([0, 4, 3, 1, 2]).reshape_([N, M * V * C, T])x = self.data_bn(x)...

发现，节点流的输出是：64 2 350 25 1，骨骼流的输出是：64 3 2500 25 1。

• 64 很容易理解，因为 batch_size: 64 所以每个批次都是 64 个样本。
• 节点流的从 3 2500 变成 2 350，猜测是因为在 Dataset 和 Pipeline 部分做了某些变换：去除了（x, y, conf）中的 conf 置信度，所以只保留了节点坐标 （x, y） 两个维度。2500 帧只保留 350 帧，这和 window_size: 350 保持一致。

那么，为什么骨骼流没有发生变换呢，那得去 Dataset 和 Pipeline 源码看看。

Dataset 和 Pipeline

配置文件

DATASET

DATASET: #DATASET fieldbatch_size: 64 #Mandatory, bacth sizenum_workers: 4 #Mandatory, the number of subprocess on each GPU.test_batch_size: 1test_num_workers: 0train:format: "SkeletonDataset"file_path: "/home/aistudio/work/dataset/train_data.npy"label_path: "/home/aistudio/work/dataset/train_label.npy"...

这是节点流的 DATASET 配置，骨骼流的 DATASET 配置也一模一样。

• batch_size: 64，这和输出中的 N 的大小是 64 一致，说明每个批次确实是 64 个样本。
• format: "SkeletonDataset" 说明训练数据集的格式是 SkeletonDataset。

在训练模型时，通常需要 从数据集中读取数据进行训练，不同的数据集可能有不同的格式。

PIPELINE

PIPELINE: #PIPELINE fieldtrain: sample:name: "AutoPadding"window_size: 350transform: #Mandotary, image transfrom operator- SkeletonNorm:...

这是节点流的 PIPELINE 配置，采用了 AutoPadding 样本处理器和 SkeletonNorm 数据变换。

PIPELINE: #PIPELINE fieldtrain:sample:- Iden:transform: #Mandotary, image transfrom operator- SketeonModalityTransform:joint: Falsebone: Truemotion: Falsegraph: 'fsd'...

这是骨骼流的 PIPELINE 配置，采用了 Iden 样本处理器和 SketeonModalityTransform 数据变换。

源码

skeleton.py

文件路径：/paddlevideo/loader/dataset/skeleton.py

import os.path as osp
import copy
import random
import numpy as np
import picklefrom ..registry import DATASETS
from .base import BaseDataset
from ...utils import get_loggerlogger = get_logger("paddlevideo")@DATASETS.register()
class SkeletonDataset(BaseDataset):"""Skeleton dataset for action recognition.The dataset loads skeleton feature, and apply norm operatations.Args:file_path (str): Path to the index file.pipeline(obj): Define the pipeline of data preprocessing.data_prefix (str): directory path of the data. Default: None.test_mode (bool): Whether to bulid the test dataset. Default: False."""def __init__(self, file_path, pipeline, label_path=None, test_mode=False):self.label_path = label_pathsuper().__init__(file_path, pipeline, test_mode=test_mode)def load_file(self):"""Load feature file to get skeleton information."""logger.info("Loading data, it will take some moment...")self.data = np.load(self.file_path)if self.label_path:if self.label_path.endswith('npy'):self.label = np.load(self.label_path)elif self.label_path.endswith('pkl'):with open(self.label_path, 'rb') as f:sample_name, self.label = pickle.load(f)else:logger.info("Label path not provided when test_mode={}, here just output predictions.".format(self.test_mode))logger.info("Data Loaded!")return self.data  # used for __len__def prepare_train(self, idx):"""Prepare the feature for training/valid given index. """results = dict()results['data'] = copy.deepcopy(self.data[idx])results['label'] = copy.deepcopy(self.label[idx])results = self.pipeline(results)return results['data'], results['label']def prepare_test(self, idx):"""Prepare the feature for test given index. """results = dict()results['data'] = copy.deepcopy(self.data[idx])if self.label_path:results['label'] = copy.deepcopy(self.label[idx])results = self.pipeline(results)return results['data'], results['label']else:results = self.pipeline(results)return [results['data']]

这段代码定义了一个名为 SkeletonDataset 的类，它继承了 BaseDataset 这个基类。这个类的作用是为了 实现骨架数据集的动作识别，即从3D骨架关节数据序列中识别人类的动作。

这个类有以下几个参数：

• file_path (str)：数据文件的路径。
• pipeline(obj)：定义 数据预处理的流程。
• label_path (str)：标签文件的路径。默认为 None。
• test_mode (bool)：是否构建测试数据集。默认为 False。

这个类有以下几个方法：

• __init__(self, file_path, pipeline, label_path=None, test_mode=False)：这是类的 构造函数，用于 初始化类的属性和调用基类的构造函数。
• load_file(self)：这是一个 加载数据文件 的方法，用于 获取骨架信息。

• 它会打印一条日志信息 "Data Loaded!"，然后从 file_path 中 加载数据 到 self.data 中。
• 如果提供了 label_path，它会根据文件后缀名是 npy 还是 pkl 来 加载标签 到 self.label中。
• 如果没有提供 label_path，它会打印一条日志信息，表示只输出预测结果。
• 最后它会返回 self.data 作为数据集的长度。

• prepare_train(self, idx)：这是一个 准备训练/验证数据 的方法。

• 给定索引 idx，它会创建一个字典 results，然后将 self.data[idx] 和self.label[idx] 分别复制到 results['data'] 和 results['label'] 中。
• 然后它会调用 pipeline 对 results 进行预处理，并返回 results['data'] 和 results['label'] 作为训练/验证数据。

• prepare_test(self, idx)：这是一个 准备测试数据 的方法。

• 给定索引 idx，它会创建一个字典 results，然后将 self.data[idx] 复制到 results['data'] 中。
• 如果提供了 label_path，它会将 self.label[idx] 复制到 results['label'] 中。
• 然后它会调用 pipeline 对 results 进行预处理，并返回 results['data'] 和 results['label'] 作为测试数据。
• 如果没有提供 label_path，它会只返回 [results['data']] 作为测试数据。

总结：

DATASET 中加载了文件中的数据到 self.data 和 self.label 变量中，然后根据 idx 取出训练集和验证集数据，并对其进行 results = self.pipeline(results) 操作后，作为模型训练的输入。

skeleton_pipeline.py

文件路径：/paddlevideo/loader/pipelines/skeleton_pipeline.py

AutoPadding

import os
import numpy as np
import paddle.nn.functional as F
import random
import paddle
from ..registry import PIPELINES@PIPELINES.register()
class AutoPadding(object):"""Sample or Padding frame skeleton feature.Args:window_size: int, temporal size of skeleton feature.random_pad: bool, whether do random padding when frame length < window size. Default: False."""def __init__(self, window_size, random_pad=False):self.window_size = window_sizeself.random_pad = random_paddef get_frame_num(self, data):C, T, V, M = data.shapefor i in range(T - 1, -1, -1):tmp = np.sum(data[:, i, :, :])if tmp > 0:T = i + 1breakreturn Tdef __call__(self, results):data = results['data']C, T, V, M = data.shapeT = self.get_frame_num(data)if T == self.window_size:data_pad = data[:, :self.window_size, :, :]elif T < self.window_size:begin = random.randint(0, self.window_size - T) if self.random_pad else 0data_pad = np.zeros((C, self.window_size, V, M))data_pad[:, begin:begin + T, :, :] = data[:, :T, :, :]else:if self.random_pad:index = np.random.choice(T, self.window_size, replace=False).astype('int64')else:index = np.linspace(0, T, self.window_size).astype("int64")data_pad = data[:, index, :, :]results['data'] = data_padreturn results

这段代码定义了一个名为 AutoPadding 的类，它继承了 object 这个基类。这个类的作用是为了 对骨架特征进行采样或填充，使其具有相同的时间长度。

类的参数：

• window_size: int，骨架特征的帧数。
• random_pad: bool，当帧长度小于 window_size 时，是否进行随机填充。默认为 False。

类的方法：

• __init__(self, window_size, random_pad=False)：这是类的构造函数，用于初始化类的属性。
• get_frame_num(self, data)：这是一个 获取有效帧数 的方法，给定数据 data。它会 从后往前遍历数据的时间维度，找到 第一个非零帧，然后返回其 索引加一 作为 有效帧数 T。
• __call__(self, results)：这是一个 对数据进行采样或填充 的方法。给定字典 results，它会从 results 中获取数据 data，并获取其有效帧数 T。

• 如果 T 等于 window_size，它会直接返回数据的前 window_size 帧作为 data_pad。
• 如果 T 小于 window_size，它会创建一个全零数组 data_pad，并根据 random_pad 参数决定在哪个位置开始将数据的前 T 帧复制到 data_pad 中。
• 如果 T 大于 window_size，它会根据 random_pad 参数决定从数据中随机或均匀地选择 window_size 个帧作为 data_pad。
• 最后它会将 data_pad 赋值给 results['data'] 并返回 results。

所以节点流的数据 results['data'] = data_pad 经过这个处理后，都是 350 帧。

SkeletonNorm

@PIPELINES.register()
class SkeletonNorm(object):"""Normalize skeleton feature.Args:aixs: dimensions of vertex coordinate. 2 for (x,y), 3 for (x,y,z). Default: 2."""def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']# Centralizationdata = data - data[:, :, 8:9, :]data = data[:self.axis, :, :, :]  # get (x,y) from (x,y, acc)C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results

这段代码定义了一个名为 SkeletonNorm 的类，它继承了 object 这个基类。这个类的作用是为了 对骨架特征进行归一化处理。

类的参数：

• axis: int，顶点坐标的维度。2表示 (x,y)，3表示 (x,y,z)。默认为2。
• squeeze: bool，是否将数据的最后一个维度压缩。默认为 False。

类的方法：

• __call__(self, results)：这是 对数据进行归一化 的方法。

• 给定字典 results，它会从 results 中获取数据 data，并对其进行 中心化处理，即减去第9个顶点（鼻子）的坐标。
• 然后它会根据 axis 参数选择前两个或三个维度作为顶点坐标，忽略加速度信息。
• 接着它会获取数据的形状 C, T, V, M，并根据 squeeze 参数决定是否将数据的最后一个维度压缩（当M=1时）。
• 最后它会将数据转换为 float32 类型并赋值给 results['data'] 并返回 results。
• 如果 results 中有 'label' 键，它还会将标签扩展一个维度并转换为int64类型并赋值给 results['label']。

这一句代码 data = data[:self.axis, :, :, :] # get (x,y) from (x,y, acc) 就是只取 (x, y) 坐标而 去除了置信度。

Iden

@PIPELINES.register()
class Iden(object):"""Wrapper Pipeline"""def __init__(self, label_expand=True):self.label_expand = label_expanddef __call__(self, results):data = results['data']results['data'] = data.astype('float32')if 'label' in results and self.label_expand:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results

这段代码定义了一个名为 Iden 的类，它继承了 object 这个基类。这个类的作用是为了 包装流水线处理。

类的参数：

• label_expand: bool，是否 对标签进行扩展维度。默认为 True。

类的方法：

• __call__(self, results)：这是一个 对数据进行包装 的方法。

给定字典 results，它会从 results 中获取数据 data，并将其转换为 float32 类型并赋值给 results['data']。

• 如果 results 中有 'label' 键，并且 label_expand 参数为 True，它还会 将标签扩展一个维度 并转换为 int64 类型并赋值给 results['label']。
• 最后返回 results。

对标签进行扩展维度 的目的是为了 使标签的形状与数据的形状一致，方便后续的处理。

例如，如果数据的形状是 (C, T, V)，而标签的形状是 (1)，那么对标签进行扩展维度后，标签的形状就变成了 (1, 1, 1)。这样就可以将数据和标签拼接在一起，形成一个 (C+1, T, V) 的数组。

对标签进行扩展维度的方法是使用 numpy 的 expand_dims 函数，指定要扩展的轴。例如，如果要在第0轴扩展一个维度，可以写成 np.expand_dims(label, 0)。

SketeonModalityTransform

@PIPELINES.register()
class SketeonModalityTransform(object):"""Sketeon Crop Sampler.Args:crop_model: str, crop model, support: ['center'].p_interval: list, crop lenwindow_size: int, sample windows size."""def __init__(self, bone, motion, joint=True, graph='fsd'):  # 改为 fsdself.joint = jointself.bone = boneself.motion = motionself.graph = graphif self.graph == "fsd":self.bone_pairs = ((1, 8), (0, 1), (15, 0), (17, 15), (16, 0),(18, 16), (5, 1), (6, 5), (7, 6), (2, 1), (3, 2),(4, 3), (9, 8), (10, 9), (11, 10), (24, 11),(22, 11), (23, 22), (12, 8), (13, 12), (14, 13),(21, 14), (19, 14), (20, 19))else:raise NotImplementedErrordef __call__(self, results):if self.joint:return resultsdata_numpy = results['data']if self.bone:bone_data_numpy = np.zeros_like(data_numpy)for v1, v2 in self.bone_pairs:bone_data_numpy[:, :, v1 -1] = data_numpy[:, :, v1 -1] - data_numpy[:, :, v2 - 1]data_numpy = bone_data_numpyif self.motion:data_numpy[:, :-1] = data_numpy[:, 1:] - data_numpy[:, :-1]data_numpy[:, -1] = 0results['data'] = data_numpyreturn results

这段代码定义了一个名为 SketeonModalityTransform 的类，它继承了 object 这个基类。这个类的作用是为了 对骨架特征进行不同的变换，如骨架、运动和图结构。

类的参数：

• bone: bool，是否对骨架特征进行 骨架变换，即 将每个顶点的坐标减去其连接的另一个顶点的坐标。默认为 False。
• motion: bool，是否对骨架特征进行 运动变换，即 将每个时间步的坐标减去前一个时间步的坐标。默认为 False。
• joint: bool，是否 保持原始的骨架特征不变。默认为 True。
• graph: str，选择使用的 图结构，默认为 ‘fsd’。

类的方法：

• __call__(self, results)：这是一个 对数据进行变换 的方法。

• 给定字典 results，它会从 results 中获取数据 data_numpy，并根据 bone 参数决定是否进行 骨架变换。

• 如果进行骨架变换，它会创建一个全零数组 bone_data_numpy，并根据 self.bone_pairs 中定义的 骨架连接关系，计算每个顶点与其连接顶点的差值，并赋值给 bone_data_numpy。
• 然后它会将 bone_data_numpy 赋值给 data_numpy。

• 接着它会根据 motion 参数决定是否进行 运动变换。

• 如果进行运动变换，它会将 data_numpy 中 除了最后一帧之外的每一帧减去前一帧，并将最后一帧置零。

• 最后它会将 data_numpy 赋值给 results['data'] 并返回results。

所以这里没有进行 data = data[:self.axis, :, :, :] # get (x,y) from (x,y, acc) 操作。

那这个是给 NTU-RGB 数据集适配的，所以就要去看看 NTU 数据集的节点坐标是怎样的。NTU 数据集是（x, y, z）三维的坐标，那就不能只是简单地把配置文件中的 in_channels: 3 改成 in_channels: 2，还要在 SketeonModalityTransform 类中加上 data = data[:self.axis, :, :, :] # get (x,y) from (x,y, acc)，去掉置信度后的输入才是真正的二维。

解决维度不匹配问题

至此，找到问题所在了，把上面的 SketeonModalityTransform 类按如下修改。

        data_numpy = results['data']# print(data_numpy.shape) # (3, 2500, 25, 1)data_numpy = data_numpy[:2, :, :, :]  # get (x,y) from (x,y, acc)# print(data_numpy.shape) # (2, 2500, 25, 1)if self.bone:bone_data_numpy = np.zeros_like(data_numpy)for v1, v2 in self.bone_pairs:bone_data_numpy[:, :, v1 -1] = data_numpy[:, :, v1 -1] - data_numpy[:, :, v2 - 1]data_numpy = bone_data_numpy

报错内存溢出：

ResourceExhaustedError: Out of memory error on GPU 0. 
Cannot allocate 976.562500MB memory on GPU 0,
31.607422GB memory has been allocated and available memory is only 144.500000MB.

我把 batch_size: 64 改成 batch_size: 32 还是溢出，不过需要的内存更少了，说明是有效的：

ResourceExhaustedError: Out of memory error on GPU 0. 
Cannot allocate 488.281250MB memory on GPU 0, 
31.472656GB memory has been allocated and available memory is only 282.500000MB.

改成 batch_size: 16 后终于开始训练了：

16 2 2500 25 1
[05/23 18:24:50] epoch:[  1/90 ] train step:0    loss: 6.33603 lr: 0.020000 top1: 0.06250 top5: 0.06250 batch_cost: 3.76629 sec, reader_cost: 0.36598 sec, ips: 4.24822 instance/sec, eta: 13:44:45

总之，经历了这样的过程：

N, C, T, V, M = x.shape
print(N,C,T,V,M)  
# 64 3 2500 25 1 (维度不匹配) -> 64 2 2500 25 1 (内存溢出)-> 16 2 2500 25 1(成了)

但是，运行完第一个 epoch 后，新 bug 又出现了。

File "/home/aistudio/work/FigureSkating/paddlevideo/solver/custom_lr.py", line 322, in stepself.last_epoch += 1 / self.num_iters  # update step with iters
TypeError: unsupported operand type(s) for /: 'int' and 'NoneType'

问题定位，

class CustomWarmupAdjustDecay(LRScheduler):r"""We combine warmup and stepwise-cosine which is used in slowfast model.Args:step_base_lr (float): start learning rate used in warmup stage.warmup_epochs (int): the number epochs of warmup.lr_decay_rate (float|int, optional): base learning rate decay rate.step (int): step in change learning rate.last_epoch (int, optional):  The index of last epoch. Can be set to restart training. Default: -1, means initial learning rate.verbose (bool, optional): If ``True``, prints a message to stdout for each update. Default: ``False`` .Returns:``CosineAnnealingDecay`` instance to schedule learning rate."""def __init__(self,step_base_lr,warmup_epochs,lr_decay_rate,boundaries,num_iters=None,last_epoch=-1,verbose=False):self.step_base_lr = step_base_lrself.warmup_epochs = warmup_epochsself.lr_decay_rate = lr_decay_rateself.boundaries = boundariesself.num_iters = num_iters#call step() in base class, last_lr/last_epoch/base_lr will be updatesuper(CustomWarmupAdjustDecay, self).__init__(last_epoch=last_epoch,verbose=verbose)def step(self, epoch=None):"""``step`` should be called after ``optimizer.step`` . It will update the learning rate in optimizer according to current ``epoch`` .The new learning rate will take effect on next ``optimizer.step`` .Args:epoch (int, None): specify current epoch. Default: None. Auto-increment from last_epoch=-1.Returns:None"""if epoch is None:if self.last_epoch == -1:self.last_epoch += 1else:self.last_epoch += 1 / self.num_iters  # update step with iterselse:self.last_epoch = epochself.last_lr = self.get_lr()if self.verbose:print('Epoch {}: {} set learning rate to {}.'.format(self.last_epoch, self.__class__.__name__, self.last_lr))

上述代码中，self.num_iters 初始化为 None，且后面没有赋其他值就用了 self.last_epoch += 1 / self.num_iters，才导致了上面的报错。

但是 joint 流的配置是可以跑通的，于是我仔细对比了两个流的 OPTIMIZER 的配置。

OPTIMIZER: #OPTIMIZER fieldname: 'Momentum'momentum: 0.9learning_rate:iter_step: Truename: 'CustomWarmupAdjustDecay'step_base_lr: 0.1warmup_epochs: 5lr_decay_rate: 0.1boundaries: [ 30, 40 ]weight_decay:name: 'L2'value: 1e-4use_nesterov: True

发现，原来 bone 的配置中少了 iter_step: True，加上后继续跑了。
iter_step: True 是一个配置选项，用于指定 是否在每个迭代步骤中更新学习率。

至此，骨骼流也跑通了。

结果融合

PaddleVideo 没有提供双流的结果融合，需要自行添加 ensemble.py，然后执行 !python3.7 ensemble.py 命令。

ensemble.py

import os
import re
import numpy as np
import csvdef softmax(X):m = np.max(X, axis=1, keepdims=True)exp_X = np.exp(X - m)exp_X = np.exp(X)prob = exp_X / np.sum(exp_X, axis=1, keepdims=True)return proboutput_prob = None
folder = './logits'
for logits_file in os.listdir(folder):logits = np.load(os.path.join(folder, logits_file))prob = softmax(logits)if output_prob is None:output_prob = probelse:output_prob = output_prob + prob
pred = np.argmax(output_prob, axis=1)with open('./submission_ensemble.csv', 'w') as f:writer = csv.writer(f)writer.writerow(('sample_index', 'predict_category'))for i, p in enumerate(pred):writer.writerow((i, p))

这段代码是从一个文件夹中读取多个 logits 文件，对每个 logits 文件应用 softmax 函数，得到一个概率矩阵，然后将所有概率矩阵相加，得到一个输出概率矩阵。最后，对输出概率矩阵按行取最大值的索引，作为预测类别，写入一个 csv 文件中。

具体来说：

• softmax 函数，它接受一个二维数组 $X$ 作为输入，沿着第二个维度（即每一行）计算每个元素的指数值，然后除以每一行的指数和，得到一个归一化的概率矩阵。
• 接着初始化了一个空的输出概率矩阵 output_prob。
• 接着指定了一个文件夹的路径，假设该文件夹中存放了多个 logits 文件。
• 然后遍历该文件夹中的每个 logits 文件，使用 numpy.load 函数读取文件内容，得到一个二维数组 logits，然后调用 softmax 函数对其进行处理，得到一个概率矩阵 prob。如果 output_prob 为空，则将 prob 赋值给 output_prob；否则将 prob 与 output_prob 相加，并更新 output_prob。
• 接着对输出概率矩阵按行取最大值的索引，得到一个一维数组 pred，表示预测类别。
• 最后使用 csv 模块创建一个 csv 文件，并写入表头和数据。表头包含两列：sample_index 和 predict_category。数据包含每个样本的索引和预测类别。

logits 文件是一种存储了 模型输出的未归一化概率 的文件。通常是由某种机器学习算法或框架生成的，可以用于计算 softmax 函数或交叉熵损失等操作，或者用于评估模型的性能。

test.py

文件路径：/paddlevideo/tasks/test.py。

想要得到上面的 logits 文件，还要相应修改 test.py 中的代码，让模型在测试的过程中生成 logits 文件。

import paddle
from paddlevideo.utils import get_logger
from ..loader.builder import build_dataloader, build_dataset
from ..metrics import build_metric
from ..modeling.builder import build_model
from paddlevideo.utils import loadimport numpy as np
import os
import paddle.nn.functional as Flogger = get_logger("paddlevideo")@paddle.no_grad()
def test_model(cfg, weights, parallel=True):"""Test model entryArgs:cfg (dict): configuration.weights (str): weights path to load.parallel (bool): Whether to do multi-cards testing. Default: True."""# 1. Construct model.if cfg.MODEL.backbone.get('pretrained'):cfg.MODEL.backbone.pretrained = ''  # disable pretrain model initmodel = build_model(cfg.MODEL)if parallel:model = paddle.DataParallel(model)# 2. Construct dataset and dataloader.cfg.DATASET.test.test_mode = Truedataset = build_dataset((cfg.DATASET.test, cfg.PIPELINE.test))batch_size = cfg.DATASET.get("test_batch_size", 8)places = paddle.set_device('gpu')# default num worker: 0, which means no subprocess will be creatednum_workers = cfg.DATASET.get('num_workers', 0)num_workers = cfg.DATASET.get('test_num_workers', num_workers)dataloader_setting = dict(batch_size=batch_size,num_workers=num_workers,places=places,drop_last=False,shuffle=False)data_loader = build_dataloader(dataset, **dataloader_setting)model.eval()state_dicts = load(weights)model.set_state_dict(state_dicts)# add params to metricscfg.METRIC.data_size = len(dataset)cfg.METRIC.batch_size = batch_sizeprint('{} inference start!!!'.format(cfg.model_name))Metric = build_metric(cfg.METRIC)ans = np.zeros((len(data_loader), 30))for batch_id, data in enumerate(data_loader):outputs = model(data, mode='test')ans[batch_id, :] = outputsMetric.update(batch_id, data, outputs)os.makedirs('logits', exist_ok=True)with open('logits/{}.npy'.format(cfg.model_name), 'wb') as f:np.save(f, ans)print('{} inference finished!!!'.format(cfg.model_name))Metric.accumulate()

这段代码的目的是 测试一个模型在一个数据集上的性能，具体来说：

• test 函数三个参数：cfg 是一个配置字典，包含了模型、数据集、处理流程和评估指标的相关设置；weights 是一个字符串，表示 要加载的模型权重的路径（通常是训练出的 best 模型权重）；parallel 是一个布尔值，表示是否使用多卡进行测试，默认为 True。
• 根据配置字典中的 MODEL 部分，构建一个模型对象，并根据 parallel 参数决定是否使用 paddle.DataParallel 进行多卡同步。
• 根据配置字典中的 DATASET 和 PIPELINE 部分，构建一个测试数据集和一个数据加载器。数据加载器的一些参数，如 batch_size、num_workers、places等，也可以从配置字典中获取或设置默认值。
• 将模型 设置为评估模式，不进行梯度更新。
• 使用 load 函数从 weights 路径 加载模型权重，并使用 model.set_state_dict 方法 将权重赋值给模型。
• 根据配置字典中的 METRIC 部分，构建一个评估指标对象，并将数据集的大小和批次大小作为参数传入。
• 初始化一个零矩阵 ans，用于 存储模型输出的 logits。
• 遍历数据加载器中的 每个批次的数据，将数据输入模型，得到输出 logits，并将其存入 ans 矩阵中。同时，调用评估指标对象的 update 方法，更新评估结果。
• 创建一个 logits 文件夹，并将 ans 矩阵保存为一个 npy 文件，文件名为模型名称。
• 打印一条信息，表示测试完成。
• 调用评估指标对象的 accumulate 方法，计算并打印最终的评估结果。

基于飞桨 PaddleVideo 的骨骼行为识别模型 CTR-GCN

该项目见飞桨：

PaddleVideo 的文件结构如下图：

• 其中 output 文件夹用于保存训练过程中生成的权重文件、优化器参数等 .paparams 和 .pdopt 文件，如 CTRGCN_J_fold0_0.6403_best.pdparams 和 CTRGCN_J_fold0_0.6403_best.pdopt；
• model 文件夹用于保存每个模型训练过程中的最优模型权重文件，如 model/CTRGCN_J_fold0.pdparams。
• requirements.txt 文件是要安装的依赖，每一行内容是一个要安装的依赖，其中包含了 Python 第三方库的名称和版本信息。直接执行 pip install -r requirements.txt 即可快速安装所有依赖项，并保证各依赖项的版本一致。
• run_train.sh 和 run_test.sh 分别是训练命令和测试命令的集成，因为该模型数较多，一个一个训练和测试过于繁琐。

requirements.txt 内容如下所示：

下面主要讲两个脚本文件、 configs 和 paddlevideo 文件夹。

main.py

文件路径：work/PaddleVideo/main.py

import paddle
import argparse
from paddlevideo.utils import get_config
from paddlevideo.tasks import train_model, train_model_multigrid, test_model, train_dali
from paddlevideo.utils import get_dist_info
import numpy as np
import random
import paddle.fluid as fluiddef same_seeds(seed):np.random.seed(seed)random.seed(seed)fluid.default_startup_program().random_seed = seedpaddle.seed(seed)def parse_args():parser = argparse.ArgumentParser("PaddleVideo train script")parser.add_argument('-c','--config',type=str,default='configs/example.yaml',help='config file path')parser.add_argument('-o','--override',action='append',default=[],help='config options to be overridden')parser.add_argument('--test',action='store_true',help='whether to test a model')parser.add_argument('--train_dali',action='store_true',help='whether to use dali to speed up training')parser.add_argument('--multigrid',action='store_true',help='whether to use multigrid training')parser.add_argument('-w','--weights',type=str,help='weights for finetuning or testing')parser.add_argument('--fleet',action='store_true',help='whether to use fleet run distributed training')parser.add_argument('--amp',action='store_true',help='whether to open amp training.')parser.add_argument('--validate',action='store_true',help='whether to evaluate the checkpoint during training')args = parser.parse_args()return argsdef main():same_seeds(0)args = parse_args()cfg = get_config(args.config, overrides=args.override, show=(not args.test))_, world_size = get_dist_info()parallel = world_size != 1if parallel:paddle.distributed.init_parallel_env()if args.test:test_model(cfg, weights=args.weights, parallel=parallel)elif args.train_dali:train_dali(cfg, weights=args.weights, parallel=parallel)elif args.multigrid:train_model_multigrid(cfg, world_size, validate=args.validate)else:train_model(cfg,weights=args.weights,parallel=parallel,validate=args.validate,use_fleet=args.fleet,amp=args.amp)if __name__ == '__main__':main()

通过命令行参数传入配置文件路径、权重路径等信息进行模型训练或测试。

具体实现了 test_model、train_model、train_model_multigrid、train_dali 四个视频任务训练函数。
其中

• test_model 函数用于模型测试，
• train_model 函数用于模型训练，
• train_model_multigrid 函数用于多尺度训练，
• train_dali 函数用于训练数据处理加速。

same_seeds

def same_seeds(seed):np.random.seed(seed)random.seed(seed)fluid.default_startup_program().random_seed = seedpaddle.seed(seed)

这段代码的作用是设定随机数种子，以保证实验结果的可重复性。
具体地，

• np.random.seed(seed) 设定了 numpy 库中随机数生成的种子，
• random.seed(seed) 设定了 Python 内置库中随机数生成的种子，
• fluid.default_startup_program().random_seed = seed 设定了 fluid
  框架中随机数生成的种子，
• paddle.seed(seed) 设定了 PaddlePaddle 中随机数生成的种子。

这些随机数生成器通常用于网络初始化、数据增强等场景，通过固定随机数种子，我们可以控制每一次生成的随机数序列是相同的，从而保证实验结果的可重复性。

parse_args

def parse_args():parser = argparse.ArgumentParser("PaddleVideo train script")parser.add_argument('-c','--config',type=str,default='configs/example.yaml',help='config file path')parser.add_argument('-o','--override',action='append',default=[],help='config options to be overridden')parser.add_argument('--test',action='store_true',help='whether to test a model')parser.add_argument('--train_dali',action='store_true',help='whether to use dali to speed up training')parser.add_argument('--multigrid',action='store_true',help='whether to use multigrid training')parser.add_argument('-w','--weights',type=str,help='weights for finetuning or testing')parser.add_argument('--fleet',action='store_true',help='whether to use fleet run distributed training')parser.add_argument('--amp',action='store_true',help='whether to open amp training.')parser.add_argument('--validate',action='store_true',help='whether to evaluate the checkpoint during training')args = parser.parse_args()return args

这段代码定义了一个命令行参数解析器，用于解析用户在命令行中输入的参数。

• 解析器使用 argparse 库进行构建，在 argparse.ArgumentParser 的参数中通过字符串 “PaddleVideo train script” 定义了解析器的描述信息。
• 接下来，解析器使用 add_argument 方法添加了多个命令行参数选项，可以根据用户的需求选择性地解析这些选项。
  例如，–test 参数用于指示是否进行模型测试，-c/–config 参数用于指定配置文件路径等。
• 最后，解析器调用 parse_args 方法解析出命令行参数，并将解析出的结果以一个 Namespace 对象的形式返回给主函数，由主函数根据解析得到的参数执行相应的操作。

main

def main():same_seeds(0)args = parse_args()cfg = get_config(args.config, overrides=args.override, show=(not args.test))_, world_size = get_dist_info()parallel = world_size != 1if parallel:paddle.distributed.init_parallel_env()if args.test:test_model(cfg, weights=args.weights, parallel=parallel)elif args.train_dali:train_dali(cfg, weights=args.weights, parallel=parallel)elif args.multigrid:train_model_multigrid(cfg, world_size, validate=args.validate)else:train_model(cfg,weights=args.weights,parallel=parallel,validate=args.validate,use_fleet=args.fleet,amp=args.amp)if __name__ == '__main__':main()

这段代码是主函数，程序从这里开始执行。

• 首先，调用 same_seeds(0) 函数，设定随机数种子以保证实验结果的可重复性。
• 接着，调用 parse_args() 函数解析命令行参数，并获取程序配置。根据命令行参数的不同选项，程序将执行不同的任务。

• 如果 args.test 为 True，则调用 test_model() 函数进行模型测试，同时传入相应的参数；
• 如果 args.train_dali 为 True，则调用 train_dali() 函数进行训练数据处理加速；
• 如果 args.multigrid 为 True，则调用 train_model_multigrid() 函数进行多尺度训练；
• 否则，则调用 train_model() 函数进行普通的单尺度训练。

• 最后，程序判断当前模块是否被作为脚本直接运行，如果是，则执行主函数 main()。

 _, world_size = get_dist_info()parallel = world_size != 1if parallel:paddle.distributed.init_parallel_env()

这段代码的作用是获取当前程序运行的分布式环境信息，并根据是否处于分布式环境下决定是否初始化分布式并行运行环境。

在 PaddlePaddle 中，如果使用多卡训练或分布式训练，则需要初始化分布式并行运行环境。get_dist_info() 函数用于获取当前程序运行的分布式环境信息，返回一个元组 (local_rank, world_size)，其中 local_rank 表示当前进程在本地机器中的编号，world_size 表示当前分布式环境下总共有多少个进程在运行。
接着，程序判断 world_size 是否为 1，即当前程序是否在分布式环境下运行。如果 world_size 不为 1，则表明当前程序运行在分布式环境中，需要调用 paddle.distributed.init_parallel_env() 函数初始化分布式并行运行环境。通过初始化后，后续的训练操作将可以自动使用多卡或者分布式运算。

ensemble.py

import os
import re
import numpy as np
import csvdef softmax(X):m = np.max(X, axis=1, keepdims=True)exp_X = np.exp(X - m)exp_X = np.exp(X)prob = exp_X / np.sum(exp_X, axis=1, keepdims=True)return probdef is_Mb(file_name):pattern = 'CTRGCN_Mb_fold\d+\.npy'return re.match(pattern, file_name) is not Noneoutput_prob = None
folder = './logits'
for logits_file in os.listdir(folder):logits = np.load(os.path.join(folder, logits_file))prob = softmax(logits)if is_Mb(logits_file):prob *= 0.7if output_prob is None:output_prob = probelse:output_prob = output_prob + prob
pred = np.argmax(output_prob, axis=1)with open('./submission_ensemble.csv', 'w') as f:writer = csv.writer(f)writer.writerow(('sample_index', 'predict_category'))for i, p in enumerate(pred):writer.writerow((i, p))

configs 文件夹

里面是以下7种特征的配置 .yaml 文件：

Joint（J）的配置文件

ctrgcn_fsd_J_fold0.yaml

MODEL: #MODEL fieldframework: "RecognizerGCN" #Mandatory, indicate the type of network, associate to the 'paddlevideo/modeling/framework/'.backbone: #Mandatory, indicate the type of backbone, associate to the 'paddlevideo/modeling/backbones/' .name: "CTRGCN" #Mandatory, The name of backbone.in_channels: 2head:name: "CTRGCNHead" #Mandatory, indicate the type of head, associate to the 'paddlevideo/modeling/heads'num_classes: 30 #Optional, the number of classes to be classified.ls_eps: 0.1DATASET: #DATASET fieldbatch_size: 16  #Mandatory, bacth sizenum_workers: 2  #Mandatory, the number of subprocess on each GPU.test_batch_size: 1test_num_workers: 0train:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "../dataset/train/J_fold0.npy" #Mandatory, train data index file pathlabel_path: "../dataset/train/fold0_label.npy"valid:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "../dataset/valid/J_fold0.npy" #Mandatory, train data index file pathlabel_path: "../dataset/valid/fold0_label.npy"test:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "../dataset/test/J.npy" #Mandatory, valid data index file pathtest_mode: TruePIPELINE: #PIPELINE fieldtrain: #Mandotary, indicate the pipeline to deal with the training data, associate to the 'paddlevideo/loader/pipelines/'sample:name: "UniformSampleFrames"window_size: 350transform: #Mandotary, image transfrom operator- SkeletonNorm_J:valid: #Mandotary, indicate the pipeline to deal with the training data, associate to the 'paddlevideo/loader/pipelines/'sample:name: "UniformSampleFrames"window_size: 350test_mode: Truetransform: #Mandotary, image transfrom operator- SkeletonNorm_J:test: #Mandatory, indicate the pipeline to deal with the validing data. associate to the 'paddlevideo/loader/pipelines/'sample:name: "UniformSampleFrames"window_size: 350test_mode: Truetransform: #Mandotary, image transfrom operator- SkeletonNorm_J:OPTIMIZER: #OPTIMIZER fieldname: 'Momentum'momentum: 0.9learning_rate:iter_step: Falsename: 'CustomWarmupCosineDecay'max_epoch: 90warmup_epochs: 10warmup_start_lr: 0.01cosine_base_lr: 0.1weight_decay:name: 'L2'value: 4e-4METRIC:name: 'SkeletonMetric'out_file: 'submission.csv'INFERENCE:name: 'STGCN_Inference_helper'num_channels: 5window_size: 350vertex_nums: 25person_nums: 1model_name: "CTRGCN_J_fold0"
save_interval: 10
val_interval: 1
log_interval: 20 #Optional, the interal of logger, default:10
epochs: 90 #Mandatory, total epoch

ctrgcn_fsd_J_fold1.yaml

同 J_fold0.yaml，区别在于 DATASET 中文件路径不同，修改成 fold1 的训练和测试文件路径即可，fold2、fold3、fold4 同理。

 train:format: "SkeletonDataset" file_path: "../dataset/train/J_fold1.npy" label_path: "../dataset/train/fold1_label.npy"valid:format: "SkeletonDataset" file_path: "../dataset/valid/J_fold1.npy"label_path: "../dataset/valid/fold1_label.npy"

Joint Angle（JA）的配置文件

ctrgcn_fsd_JA_fold0.yaml

MODEL: #MODEL fieldframework: "RecognizerGCN" #Mandatory, indicate the type of network, associate to the 'paddlevideo/modeling/framework/'.backbone: #Mandatory, indicate the type of backbone, associate to the 'paddlevideo/modeling/backbones/' .name: "CTRGCN" #Mandatory, The name of backbone.in_channels: 9head:name: "CTRGCNHead" #Mandatory, indicate the type of head, associate to the 'paddlevideo/modeling/heads'num_classes: 30 #Optional, the number of classes to be classified.ls_eps: 0.1DATASET: #DATASET fieldbatch_size: 16  #Mandatory, bacth sizenum_workers: 2  #Mandatory, the number of subprocess on each GPU.test_batch_size: 1test_num_workers: 0train:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "../dataset/train/JA_fold0.npy" #Mandatory, train data index file pathlabel_path: "../dataset/train/fold0_label.npy"valid:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "../dataset/valid/JA_fold0.npy" #Mandatory, train data index file pathlabel_path: "../dataset/valid/fold0_label.npy"test:format: "SkeletonDataset" #Mandatory, indicate the type of dataset, associate to the 'paddlevidel/loader/dateset'file_path: "../dataset/test/JA.npy" #Mandatory, valid data index file pathtest_mode: TruePIPELINE: #PIPELINE fieldtrain: #Mandotary, indicate the pipeline to deal with the training data, associate to the 'paddlevideo/loader/pipelines/'sample:name: "UniformSampleFrames"window_size: 350transform: #Mandotary, image transfrom operator- SkeletonNorm_JA:valid: #Mandotary, indicate the pipeline to deal with the training data, associate to the 'paddlevideo/loader/pipelines/'sample:name: "UniformSampleFrames"window_size: 350test_mode: Truetransform: #Mandotary, image transfrom operator- SkeletonNorm_JA:test: #Mandatory, indicate the pipeline to deal with the validing data. associate to the 'paddlevideo/loader/pipelines/'sample:name: "UniformSampleFrames"window_size: 350test_mode: Truetransform: #Mandotary, image transfrom operator- SkeletonNorm_JA:OPTIMIZER: #OPTIMIZER fieldname: 'Momentum'momentum: 0.9learning_rate:iter_step: Falsename: 'CustomWarmupCosineDecay'max_epoch: 90warmup_epochs: 10warmup_start_lr: 0.01cosine_base_lr: 0.1weight_decay:name: 'L2'value: 4e-4METRIC:name: 'SkeletonMetric'out_file: 'submission.csv'INFERENCE:name: 'STGCN_Inference_helper'num_channels: 5window_size: 350vertex_nums: 25person_nums: 1model_name: "CTRGCN_JA_fold0"
save_interval: 10
val_interval: 1
log_interval: 20 #Optional, the interal of logger, default:10
epochs: 90 #Mandatory, total epoch

JA 区别于 J 的在于，不同的特征，除了

• model_name 不同，
• 训练和验证数据文件路径不同，
• SkeletonNorm_J（SkeletonNorm_JA）外，

关键在于 in_channels 的不同：J 特征只有2个特征维度，而 JA 有9个。

paddlevideo 文件夹

utils 文件夹

paddlevideo/utils 文件夹中包含了一些通用的工具函数和预处理方法，用于辅助视频数据的加载、预处理和后处理等。

main.py 导入了 utils 包中的 get_config 和 get_dist_info 函数，下面会讲到。

__init__.py

from .registry import Registry
from .build_utils import build
from .config import *
from .logger import setup_logger, coloring, get_logger
from .record import AverageMeter, build_record, log_batch, log_epoch
from .dist_utils import get_dist_info, main_only
from .save_load import save, load, load_ckpt, mkdir
from .precise_bn import do_preciseBN
__all__ = ['Registry', 'build']

这段代码的作用是从 paddlevideo/utils 目录下导入一些模块或函数，并将它们添加到 paddlevideo.utils 这个包的命名空间中，方便在其他地方使用。

• 例如，from .registry import Registry 这一行就是从 registry.py 文件中导入 Registry 类，并将它添加到 paddlevideo.utils 这个包的命名空间中，也就是说，你可以通过 paddlevideo.utils.Registry 来访问这个类。
• __all__ 是一个特殊的变量，它定义了当使用 from paddlevideo.utils import * 时要导入的名称。也就是说 from paddlevideo.utils import * 命令只能导入 Registry 和 build 类，而不会导入其他的如 get_logger。
• 所以，如果想导入 get_logger 这个函数，可以使用 from paddlevideo.utils import get_logger 或者 import paddlevideo.utils 然后使用 paddlevideo.utils.get_logger。

__init__.py 文件是用来标记一个目录为 Python 包的文件。如，上述是标记 paddlevideo/utils 目录为 paddlevideo.utils 包。

• 它可以包含任意的 Python 代码，也可以为空。
• 当一个包被导入时，__init__.py 文件会被隐式地执行，它定义的对象会绑定到包的命名空间中。
• __init__.py 文件是在导入包或包中的模块时运行的。

用一个简单的例子来解释一下，假设有一个目录结构如下：

my_package/__init__.pymodule1.pymodule2.py

其中，__init__.py 文件的内容是：

print("This is my package.")
from .module1 import foo
from .module2 import bar
__all__ = ["foo", "bar"]

module1.py 文件的内容是：

print("This is module1.")
def foo():print("This is foo.")

module2.py 文件的内容是：

print("This is module2.")
def bar():print("This is bar.")

现在，如果你在 Python 解释器中输入：

>>> import my_package

你会看到输出：

This is my package.
This is module1.
This is module2.

这说明，当你导入 my_package 这个包时，它的 __init__.py 文件被隐式地执行了，它打印了一句话，并且从 module1.py 和 module2.py 文件中导入了 $foo$ 和 $bar$ 这两个函数，并将它们添加到了 my_package 这个包的命名空间中。所以，你可以直接使用 my_package.foo() 和 my_package.bar() 来调用这两个函数。

---

另外，由于 __init__.py 文件中定义了 __all__ = ["foo", "bar"] 这一行，它指定了当你使用 from my_package import * 时要导入的名称。所以，如果你在 Python 解释器中输入：

>>> from my_package import *
>>> foo()
>>> bar()

你会看到输出：

This is foo.
This is bar.

这说明，当你使用 from my_package import * 时，它只导入了 __all__ 中指定的名称，即 $foo$ 和 $bar$ 这两个函数，并将它们添加到了当前的命名空间中。所以，你可以直接使用 foo() 和 bar() 来调用这两个函数。

registry.py

class Registry(object):"""The registry that provides name -> object mapping, to support third-party users' custom modules.To register an object:.. code-block:: pythonBACKBONES = Registry('backbone')@BACKBONES.register()class ResNet:passOr:.. code-block:: pythonBACKBONES = Registry('backbone')class ResNet:passBACKBONES.register(ResNet)Usage: To build a module... code-block:: pythonbackbone_name = "ResNet"b = BACKBONES.get(backbone_name)()"""def __init__(self, name):"""Args:name (str): the name of this registry"""self._name = nameself._obj_map = {}def __contains__(self, key):return self._obj_map.get(key) is not Nonedef _do_register(self, name, obj):assert (name not in self._obj_map), "An object named '{}' was already registered in '{}' registry!".format(name, self._name)self._obj_map[name] = objdef register(self, obj=None, name=None):"""Register the given object under the the name `obj.__name__`.Can be used as either a decorator or not. See docstring of this class for usage."""if obj is None:# used as a decoratordef deco(func_or_class, name=name):if name is None:name = func_or_class.__name__self._do_register(name, func_or_class)return func_or_classreturn deco# used as a function callif name is None:name = obj.__name__self._do_register(name, obj)def get(self, name):"""Get the registry record.Args:name (str): The class name.Returns:ret: The class."""ret = self._obj_map.get(name)if ret is None:raise KeyError("No object named '{}' found in '{}' registry!".format(name, self._name))return ret

这段代码定义了一个 Registry 类，作用是用来注册一些对象，并通过名称来获取它们。这个类有以下几个方法：

• __init__(self, name)：构造方法，初始化一个空的对象映射字典 $\_obj\_map$，并记录注册器的名称 $name$。
• __contains__(self, key)：判断一个名称是否已经被注册过，如果是，返回 True，否则返回 False。
• _do_register(self, name, obj)：私有方法，用来将一个对象 $obj$ 注册到一个名称 $name$ 上，如果该名称已经被注册过，就抛出断言错误。
• register(self, obj=None, name=None)：公开方法，用来注册一个对象或者作为装饰器使用。如果传入了 $obj$ 参数，就将它注册到 $name$ 参数指定的名称上（如果没有指定 $name$ 参数，就使用 obj.__name__ 作为名称）。如果没有传入 $obj$ 参数，就返回一个装饰器函数，用来装饰一个类或者函数，并将它注册到指定的名称上。

用法如下：

   .. code-block:: pythonBACKBONES = Registry('backbone') # 创建一个名为'backbone'的注册器 BACKBONES@BACKBONES.register() # 在类 ResNet 定义前加上语法糖，那么这个类 ResNet 就被注册进了这个 BACKBONES 注册器中class ResNet:pass

Or:

   .. code-block:: pythonBACKBONES = Registry('backbone')class ResNet:passBACKBONES.register(ResNet) # BACKBONES 注册器注册这个类 ResNet

• get(self, name)：根据名称获取一个已经注册的对象，如果没有找到，就抛出 KeyError 异常。

用法如下：

   .. code-block:: pythonbackbone_name = "ResNet"b = BACKBONES.get(backbone_name)()

这个类可以用来实现一种插件机制，让不同的模块可以向注册器中添加自己的对象，并通过名称来访问它们。

build_utils.py

def build(cfg, registry, key='name'):"""Build a module from config dict.Args:cfg (dict): Config dict. It should at least contain the key.registry (XXX): The registry to search the type from.key (str): the key.Returns:obj: The constructed object."""assert isinstance(cfg, dict) and key in cfgcfg_copy = cfg.copy()obj_type = cfg_copy.pop(key)obj_cls = registry.get(obj_type)if obj_cls is None:raise KeyError('{} is not in the {} registry'.format(obj_type, registry.name))return obj_cls(**cfg_copy)

这段代码是定义了一个 $build$ 函数，它的作用是根据一个配置字典和一个注册器，构建一个模块对象。它的参数和返回值如下：

• $cfg$ ($dict$)：配置字典，它至少应该包含一个 $key$，表示要构建的模块的类型。

$cfg$ 字典可以有多个键，只要其中有一个键是 $name$，用来指定要从注册器中获取的类。其他的键和值都会作为参数传递给类的构造函数。
例如，如果想要创建一个 $Thing3$ 的实例，而 $Thing3$ 的构造函数需要三个参数，$arg1$，$arg2$ 和 $arg3$，可以使用以下代码：

cfg = {'name': 'Thing3','arg1': 5,'arg2': 6,'arg3': 7
}

那么 build(cfg, registry) 就相当于调用 Thing3(arg1=5, arg2=6, arg3=7)，并返回一个 $Thing3$ 的实例。

• $registry$ (XXX)：注册器，它是一个 Registry 类的实例，用来存储不同类型的模块类。
• $key$ ($str$)：配置字典中表示模块类型的键，默认为 ‘name’。
• $obj$：返回值，是根据配置字典和注册器中获取的模块类构造的对象。

函数的逻辑如下：

• 首先断言 $cfg$ 是一个字典，并且包含 $key$ 这个键。
• 然后复制一份 $cfg$，并从中弹出 $key$ 对应的值，赋给 $obj\_type$，表示要构建的模块类型。
• 接着从注册器中根据 $obj\_type$ 获取对应的模块类，赋给 $obj\_cls$。如果没有找到，就抛出 $KeyError$ 异常。
• 最后用剩余的 $cfg\_copy$ 作为关键字参数，调用 $obj\_cls$ 构造一个对象，并返回。

举个例子，假设有以下配置字典和注册器：

cfg = {'name': 'Thing1','arg1': 1,'arg2': 2
}registry = Registry('thing') 
registry.register('Thing1', Thing1)
registry.register('Thing2', Thing2) 

• 这段代码创建一个名为 $thing$ 的注册器，然后向注册器中注册两个类，$Thing1$ 和 $Thing2$（$Thing1$ 和 $Thing2$ 是两个自定义的类），并给它们分别指定一个字符串作为键。
• 那么调用 build(cfg, registry) 就相当于调用 Thing1(arg1=1, arg2=2)（这是因为 $cfg$ 中的 'name': 'Thing1' 指定了调用 $build$ 要创建 $Thing1$ 类），并返回一个 $Thing1$ 的实例。

注册器是一个用于存储和查找类的容器，可以根据键来获取对应的类。

• 例如，如果想要创建一个 $Thing1$ 的实例，可以使用以下代码：thing1 = registry.get('Thing1')() 或者 thing1 = registry['Thing1']()。

如果想要创建 $Thing1$ 和 $Thing2$ 的实例，可以使用两个不同的 $cfg$ 字典，分别指定 $name$ 键的值为 ${}^{\prime }Thing{1}^{\prime }$ 和 ${}^{\prime }Thing{2}^{\prime }$，然后分别调用 build(cfg, registry) 函数。例如，可以使用以下代码：

cfg1 = {'name': 'Thing1','arg1': 1,'arg2': 2
}
cfg2 = {'name': 'Thing2','arg1': 3,'arg2': 4
}
thing1 = build(cfg1, registry) # 创建 Thing1 的实例
thing2 = build(cfg2, registry) # 创建 Thing2 的实例

config.py

import os
import yaml
from paddlevideo.utils.logger import coloring, get_logger, setup_logger__all__ = ['get_config']logger = setup_logger("./", name="paddlevideo", level="INFO")class AttrDict(dict):def __getattr__(self, key):return self[key]def __setattr__(self, key, value):if key in self.__dict__:self.__dict__[key] = valueelse:self[key] = valuedef create_attr_dict(yaml_config):from ast import literal_evalfor key, value in yaml_config.items():if type(value) is dict:yaml_config[key] = value = AttrDict(value)if isinstance(value, str):try:value = literal_eval(value)except BaseException:passif isinstance(value, AttrDict):create_attr_dict(yaml_config[key])else:yaml_config[key] = valuedef parse_config(cfg_file):"""Load a config file into AttrDict"""with open(cfg_file, 'r') as fopen:yaml_config = AttrDict(yaml.load(fopen, Loader=yaml.SafeLoader))create_attr_dict(yaml_config)return yaml_configdef print_dict(d, delimiter=0):"""Recursively visualize a dict andindenting acrrording by the relationship of keys."""placeholder = "-" * 60for k, v in sorted(d.items()):if isinstance(v, dict):logger.info("{}{} : ".format(delimiter * " ", coloring(k,"HEADER")))print_dict(v, delimiter + 4)elif isinstance(v, list) and len(v) >= 1 and isinstance(v[0], dict):logger.info("{}{} : ".format(delimiter * " ",coloring(str(k), "HEADER")))for value in v:print_dict(value, delimiter + 4)else:logger.info("{}{} : {}".format(delimiter * " ",coloring(k, "HEADER"),coloring(v, "OKGREEN")))if k.isupper():logger.info(placeholder)def print_config(config):"""visualize configsArguments:config: configs"""print_dict(config)def check_config(config):"""Check config"""passdef override(dl, ks, v):"""Recursively replace dict of listArgs:dl(dict or list): dict or list to be replacedks(list): list of keysv(str): value to be replaced"""def str2num(v):try:return eval(v)except Exception:return vassert isinstance(dl, (list, dict)), ("{} should be a list or a dict")assert len(ks) > 0, ('lenght of keys should larger than 0')if isinstance(dl, list):k = str2num(ks[0])if len(ks) == 1:assert k < len(dl), ('index({}) out of range({})'.format(k, dl))dl[k] = str2num(v)else:override(dl[k], ks[1:], v)else:if len(ks) == 1:#assert ks[0] in dl, ('{} is not exist in {}'.format(ks[0], dl))if not ks[0] in dl:logger.warning('A new filed ({}) detected!'.format(ks[0], dl))dl[ks[0]] = str2num(v)else:assert ks[0] in dl, ('({}) doesn\'t exist in {}, a new dict field is invalid'.format(ks[0], dl))override(dl[ks[0]], ks[1:], v)def override_config(config, options=None):"""Recursively override the configArgs:config(dict): dict to be replacedoptions(list): list of pairs(key0.key1.idx.key2=value)such as: [epochs=20','PIPELINE.train.transform.1.ResizeImage.resize_short=300']Returns:config(dict): replaced config"""if options is not None:for opt in options:assert isinstance(opt,str), ("option({}) should be a str".format(opt))assert "=" in opt, ("option({}) should contain a =""to distinguish between key and value".format(opt))pair = opt.split('=')assert len(pair) == 2, ("there can be only a = in the option")key, value = pairkeys = key.split('.')override(config, keys, value)return configdef get_config(fname, overrides=None, show=True):"""Read config from file"""assert os.path.exists(fname), ('config file({}) is not exist'.format(fname))config = parse_config(fname)override_config(config, overrides)if show:print_config(config)check_config(config)return config

• AttrDict 类，继承自 dict 类，重写了 getattr 和 setattr 方法，使得可以用点号访问字典中的键和值，而不需要用方括号。
• create_attr_dict 函数，用于把一个普通的字典转换为 AttrDict 类型，并递归地处理字典中的子字典。这个函数还会尝试把字典中的字符串值转换为 Python 的原生类型，例如数字或布尔值。
• parse_config 函数，用于从一个 YAML 文件中读取配置信息，并返回一个 AttrDict 类型的对象。这个函数会调用 create_attr_dict 函数来处理 YAML 文件中的内容。

YAML 是一种人类可读的数据序列化语言，常用于配置文件或数据交换。Python 中有一个 PyYAML 模块，可以用来加载，解析和写入 YAML 文件。这个函数就是利用了 PyYAML 模块来读取 YAML 配置文件，并把它转换为一个方便访问的 AttrDict 对象。

• print_dict 函数，用于递归地打印一个字典的键和值，并根据键的层级关系进行缩进。这个函数还会用不同的颜色来显示键和值（通过 coloring 实现），以及用一条横线来分隔大写的键。
• print_config 函数，用于调用 print_dict 函数来可视化输出一个配置对象。

• override 这个函数的作用是可以用一个简单的方式来修改一个复杂的字典或列表中的某个值，而不需要写很多层的索引或键。在替换值过程中，还会进行一些断言和警告，检查索引是否越界，键是否存在，以及是否出现了新的字段。

这样可以提高代码的可读性和可维护性。例如，如果有一个嵌套的字典，如下：

d = {'a': {'b': {'c': 1,'d': 2},'e': 3},'f': 4 
} 

如果想要把 d[‘a’][‘b’][‘c’] 的值改为 5，可以使用 override 函数，只需要传入一个键的列表 [‘a’, ‘b’, ‘c’]，而不需要写 d[‘a’][‘b’][‘c’] = 5。例如：override(d, ['a', 'b', 'c'], 5) ，这样就可以实现同样的效果，但是更简洁和清晰。

• override_config 这个函数的作用是根据一个选项列表，递归地覆盖一个配置字典中的某些值。

这个函数接受两个参数：

• config 是要被覆盖的配置字典，
• options 是一个字符串列表，每个字符串表示一个键和值的对应关系，用等号分隔。键可以用点号连接多个子键，表示配置字典中的层级关系。

例如：

options = ['epochs=20','PIPELINE.train.transform.1.ResizeImage.resize_short=300' ]

这个函数会调用之前定义的 override 函数，把每个选项中的键和值分别传入，实现对配置字典的修改。

例如，上面的选项列表会把 config[‘epochs’] 的值改为 20，把 config[‘PIPELINE’][‘train’][‘transform’][1][‘ResizeImage’][‘resize_short’] 的值改为 300。这样就可以实现对配置字典的自定义修改。

• get_config 这个函数的意思是从一个文件中读取配置信息，并根据一些选项进行覆盖和检查。

这个函数接受三个参数：

• fname 是配置文件的路径，
• overrides 是一个选项列表，用于修改配置信息，
• show 是一个布尔值，表示是否打印配置信息。

这个函数会调用之前定义的 parse_config，override_config，print_config 和 check_config 函数，分别实现解析，覆盖，打印和检查配置信息的功能。
最后，这个函数会返回一个配置对象。

logger.py

import logging
import os
import sys
import datetimefrom paddle.distributed import ParallelEnvColor = {'RED': '\033[31m','HEADER': '\033[35m',  # deep purple'PURPLE': '\033[95m',  # purple'OKBLUE': '\033[94m','OKGREEN': '\033[92m','WARNING': '\033[93m','FAIL': '\033[91m','ENDC': '\033[0m'
}def coloring(message, color="OKGREEN"):assert color in Color.keys()if os.environ.get('COLORING', True):return Color[color] + str(message) + Color["ENDC"]else:return messagelogger_initialized = []def setup_logger(output=None, name="paddlevideo", level="INFO"):"""Initialize the paddlevideo logger and set its verbosity level to "INFO".Args:output (str): a file name or a directory to save log. If None, will not save log file.If ends with ".txt" or ".log", assumed to be a file name.Otherwise, logs will be saved to `output/log.txt`.name (str): the root module name of this loggerReturns:logging.Logger: a logger"""def time_zone(sec, fmt):real_time = datetime.datetime.now()return real_time.timetuple()logging.Formatter.converter = time_zonelogger = logging.getLogger(name)if level == "INFO":logger.setLevel(logging.INFO)elif level=="DEBUG":logger.setLevel(logging.DEBUG)logger.propagate = Falseif level == "DEBUG":plain_formatter = logging.Formatter("[%(asctime)s] %(name)s %(levelname)s: %(message)s",datefmt="%m/%d %H:%M:%S")else:plain_formatter = logging.Formatter("[%(asctime)s] %(message)s",datefmt="%m/%d %H:%M:%S")# stdout logging: master onlylocal_rank = ParallelEnv().local_rankif local_rank == 0:ch = logging.StreamHandler(stream=sys.stdout)ch.setLevel(logging.DEBUG)formatter = plain_formatterch.setFormatter(formatter)logger.addHandler(ch)# file logging: all workersif output is not None:if output.endswith(".txt") or output.endswith(".log"):filename = outputelse:filename = os.path.join(output, "log.txt")if local_rank > 0:filename = filename + ".rank{}".format(local_rank)# PathManager.mkdirs(os.path.dirname(filename))os.makedirs(os.path.dirname(filename), exist_ok=True)# fh = logging.StreamHandler(_cached_log_stream(filename)fh = logging.FileHandler(filename, mode='a')fh.setLevel(logging.DEBUG)fh.setFormatter(plain_formatter)logger.addHandler(fh)logger_initialized.append(name)return loggerdef get_logger(name, output=None):logger = logging.getLogger(name)if name in logger_initialized:return loggerreturn setup_logger(name=name, output=name)

logging 模块是 Python 标准库中提供的一个功能强大而灵活的日志系统，可以让你在程序中输出不同级别的日志信息。

• 首先导入了 logging 模块，
• $Color$ 字典，用来给不同级别的日志信息添加颜色。
• coloring 函数，用来根据颜色参数给消息添加颜色。
• logger_initialized 列表，用来存储已经初始化过的 logger 对象。

logger 对象是 logging 模块中的基本类，它提供了应用程序直接使用的接口。通过调用 logging.getLogger(name) 函数，可以获取一个 logger 对象，如果 name 相同，那么返回的是同一个 logger 对象。

• setup_logger函数，用来初始化一个名为 paddlevideo 的 logger 对象，并根据参数设置其输出级别和文件。

• 设置了 logging.Formatter.converter 属性为 time_zone 函数，用来自定义日志信息中的时间格式。
• 设置 logger 对象的日志级别为 INFO 或 DEBUG。

如果level是 DEBUG，那么日志信息中会包含时间、名称、级别和消息；如果 level 是 INFO，那么日志信息中只包含时间和消息。

• 设置 logger 对象的 propagate 属性为 False，表示不向上级 logger 传递日志信息。
• 获取当前进程的 local_rank 值，如果是0，表示是主进程，那么创建一个 StreamHandler 对象，用来将日志信息输出到标准输出流。设置该 handler 对象的级别为 DEBUG，格式为 plain_formatter，并添加到 logger 对象中。
• 如果 output 参数不为空，表示需要将日志信息保存到文件中。根据 output 参数的值，确定文件名。如果 output 以".txt"或".log"结尾，那么认为它是一个文件名；否则，将在 output 目录下创建一个"log.txt"文件。如果 local_rank 值大于0，表示是子进程，那么在文件名后面加上".rank"和 local_rank 值，以区分不同进程的日志文件。

• get_logger函数，用来获取一个指定名称的 logger 对象。
  – 如果 name 已经在 logger_initialized 列表中，表示该 logger 对象已经被初始化过，那么直接返回该 logger 对象。
  – 否则，调用 setup_logger 函数，用 name 作为参数，来初始化该 logger 对象，并返回它。

dist_utils.py

import functoolsimport paddle
import paddle.distributed as distdef get_dist_info():world_size = dist.get_world_size()rank = dist.get_rank()return rank, world_sizedef main_only(func):@functools.wraps(func)def wrapper(*args, **kwargs):rank, _ = get_dist_info()if rank == 0:return func(*args, **kwargs)return wrapper

这段代码定义了一个 main_only 函数，用来作为一个装饰器。

• 装饰器是一种设计模式，可以在不修改原函数的情况下，给原函数添加一些额外的功能。
• 装饰器本身是一个函数，它接受一个函数作为参数，并返回一个修改后的函数。

main_only 函数的作用是，只在主进程中执行被装饰的函数，其他进程则不执行。

• 使用 functools.wraps(func) 装饰器，保留被装饰函数的元信息，比如名称、文档字符串等。
• 定义一个 wrapper 函数，用来包装被装饰函数。wrapper 函数接受任意数量和类型的参数，并将它们传递给被装饰函数。
• 在 wrapper 函数中，调用 get_dist_info() 函数，获取当前进程的 rank 值和 world_size 值。rank 值表示进程在分布式环境中的编号，world_size 值表示总的进程数。
• 如果 rank 值等于0，表示是主进程，那么调用被装饰函数，并返回其结果。
• 如果 rank 值不等于0，表示是子进程，那么不调用被装饰函数，也不返回任何结果。

record.py

import paddle
from collections import OrderedDict
from .logger import get_logger, coloringlogger = get_logger("paddlevideo")__all__ = ['AverageMeter', 'build_record', 'log_batch', 'log_epoch']def build_record(cfg):framework_type = cfg.get('framework')record_list = [("loss", AverageMeter('loss', '7.5f')),("lr", AverageMeter('lr', 'f', need_avg=False)),]if 'Recognizer1D' in cfg.framework:  #TODO: required specify str in frameworkrecord_list.append(("hit_at_one", AverageMeter("hit_at_one", '.5f')))record_list.append(("perr", AverageMeter("perr", '.5f')))record_list.append(("gap", AverageMeter("gap", '.5f')))elif 'Recognizer' in cfg.framework:record_list.append(("top1", AverageMeter("top1", '.5f')))record_list.append(("top5", AverageMeter("top5", '.5f')))record_list.append(("batch_time", AverageMeter('batch_cost', '.5f')))record_list.append(("reader_time", AverageMeter('reader_cost', '.5f')))record_list = OrderedDict(record_list)return record_listclass AverageMeter(object):"""Computes and stores the average and current value"""def __init__(self, name='', fmt='f', need_avg=True):self.name = nameself.fmt = fmtself.need_avg = need_avgself.reset()def reset(self):""" reset """self.val = 0self.avg = 0self.sum = 0self.count = 0def update(self, val, n=1):""" update """if isinstance(val, paddle.Tensor):val = val.numpy()[0]self.val = valself.sum += val * nself.count += nself.avg = self.sum / self.count@propertydef total(self):return '{self.name}_sum: {self.sum:{self.fmt}}'.format(self=self)@propertydef total_minute(self):return '{self.name}_sum: {s:{self.fmt}} min'.format(s=self.sum / 60,self=self)@propertydef mean(self):return '{self.name}_avg: {self.avg:{self.fmt}}'.format(self=self) if self.need_avg else ''@propertydef value(self):return '{self.name}: {self.val:{self.fmt}}'.format(self=self)def log_batch(metric_list, batch_id, epoch_id, total_epoch, mode, ips):batch_cost = str(metric_list['batch_time'].value) + ' sec,'reader_cost = str(metric_list['reader_time'].value) + ' sec,'metric_values = []for m in metric_list:if not (m == 'batch_time' or m == 'reader_time'):metric_values.append(metric_list[m].value)metric_str = ' '.join([str(v) for v in metric_values])epoch_str = "epoch:[{:>3d}/{:<3d}]".format(epoch_id, total_epoch)step_str = "{:s} step:{:<4d}".format(mode, batch_id)logger.info("{:s} {:s} {:s} {:s} {:s} {}".format(coloring(epoch_str, "HEADER") if batch_id == 0 else epoch_str,coloring(step_str, "PURPLE"), coloring(metric_str, 'OKGREEN'),coloring(batch_cost, "OKGREEN"), coloring(reader_cost, 'OKGREEN'), ips))def log_epoch(metric_list, epoch, mode, ips):batch_cost = 'avg_' + str(metric_list['batch_time'].value) + ' sec,'reader_cost = 'avg_' + str(metric_list['reader_time'].value) + ' sec,'batch_sum = str(metric_list['batch_time'].total) + ' sec,'metric_values = []for m in metric_list:if not (m == 'batch_time' or m == 'reader_time'):metric_values.append(metric_list[m].mean)metric_str = ' '.join([str(v) for v in metric_values])end_epoch_str = "END epoch:{:<3d}".format(epoch)logger.info("{:s} {:s} {:s} {:s} {:s} {:s} {}".format(coloring(end_epoch_str, "RED"), coloring(mode, "PURPLE"),coloring(metric_str, "OKGREEN"), coloring(batch_cost, "OKGREEN"),coloring(reader_cost, "OKGREEN"), coloring(batch_sum, "OKGREEN"), ips))

• build_record 函数，用来根据配置文件中的 framework 类型，创建一个有序字典，用来记录训练或评估过程中的各种指标。

根据 framework_type 的值，判断是哪种识别器类型，并在 record_list 中添加相应的指标。

• 如果是 Recognizer1D 类型，那么添加 hit_at_one, perr, gap 等指标；
• 如果是 Recognizer 类型，那么添加 top1, top5 等指标。

最后将 record_list 转换为 OrderedDict 对象，并返回它。

• AverageMeter 类，用来计算和存储一个指标的平均值和当前值。
• log_batch 函数用来记录每个批次的训练或测试的结果。

• metric_list: 一个字典，包含了不同指标的值，比如 batch_time, reader_time, accuracy 等。
• batch_id: 一个整数，表示当前的批次编号。
• epoch_id: 一个整数，表示当前的轮次编号。
• total_epoch: 一个整数，表示总的轮次数。
• mode: 一个字符串，表示当前是训练模式还是测试模式。
• ips: 一个字符串，表示每秒处理的样本数。

log_batch 函数会将这些参数拼接成一个字符串，并使用 logging.info 方法输出到日志中。它还会使用 coloring 函数给不同的部分添加颜色，以便于区分。

• log_epoch 函数用来记录每个轮次的训练或测试的平均结果。

log_epoch 函数也会将这些参数拼接成一个字符串，并使用 logging.info 方法输出到日志中。它也会使用 coloring 函数给不同的部分添加颜色，并在轮次结束时使用红色标记。

像这样：

save_load.py

import os
import os.path as osp
import timeimport pickle
from tqdm import tqdm
import paddle
import paddle.nn.functional as F
from paddlevideo.utils import get_logger
from paddlevideo.utils import main_onlydef pretrain_vit_param_trans(model, state_dicts, num_patches, seg_num, attention_type):"""Convert ViT's pre-trained model parameters to a parameter dictionary that matches the existing model"""if 'head' + '.weight' in state_dicts:del state_dicts['head' + '.weight']if 'head' + '.bias' in state_dicts:del state_dicts['head' + '.bias']total_len = len(model.state_dict())if num_patches + 1 != state_dicts['pos_embed'].shape[1]:pos_embed = state_dicts['pos_embed']cls_pos_embed = pos_embed[0, 0, :].unsqueeze(0).unsqueeze(1)other_pos_embed = pos_embed[0, 1:, :].unsqueeze(0).unsqueeze(1).transpose((0, 1, 3, 2))new_pos_embed = F.interpolate(other_pos_embed,size=(other_pos_embed.shape[-2], num_patches),mode='nearest')new_pos_embed = new_pos_embed.squeeze(0).transpose((0, 2, 1))new_pos_embed = paddle.concat((cls_pos_embed, new_pos_embed), axis=1)state_dicts['pos_embed'] = new_pos_embedtime.sleep(0.01)if 'time_embed' in state_dicts and seg_num != state_dicts['time_embed'].shape[1]:time_embed = state_dicts['time_embed'].transpose((0, 2, 1)).unsqueeze(0)new_time_embed = F.interpolate(time_embed,size=(time_embed.shape[-2], seg_num),mode='nearest')state_dicts['time_embed'] = new_time_embed.squeeze(0).transpose((0, 2, 1))time.sleep(0.01)with tqdm(total=total_len, position=1, bar_format='{desc}', desc="Loading weights") as desc:if attention_type == 'divided_space_time':new_state_dicts = state_dicts.copy()for key in tqdm(state_dicts):if 'blocks' in key and 'attn' in key:desc.set_description("Loading %s" % key)new_key = key.replace('attn', 'temporal_attn')if not new_key in state_dicts:new_state_dicts[new_key] = state_dicts[key]else:new_state_dicts[new_key] = state_dicts[new_key]if 'blocks' in key and 'norm1' in key:desc.set_description("Loading %s" % key)new_key = key.replace('norm1', 'temporal_norm1')if not new_key in state_dicts:new_state_dicts[new_key] = state_dicts[key]else:new_state_dicts[new_key] = state_dicts[new_key]time.sleep(0.01)ret_str = "loading {:<20d} weights completed.".format(len(model.state_dict()))desc.set_description(ret_str)return new_state_dicts#XXX(shipping): maybe need load N times because of different cards have different params.
@main_only
def load_ckpt(model,weight_path,**kargs):"""1. Load pre-trained model parameters2. Extract and convert from the pre-trained model to the parameters required by the existing model3. Load the converted parameters of the existing model"""#model.set_state_dict(state_dict)if not osp.isfile(weight_path):raise IOError(f'{weight_path} is not a checkpoint file')#state_dicts = load(weight_path)logger = get_logger("paddlevideo")state_dicts = paddle.load(weight_path)if "VisionTransformer" in str(model):  # For TimeSformer casetmp = pretrain_vit_param_trans(model, state_dicts, kargs['num_patches'], kargs['seg_num'], kargs['attention_type'])else:tmp = {}total_len = len(model.state_dict())with tqdm(total=total_len, position=1, bar_format='{desc}', desc="Loading weights") as desc:for item in tqdm(model.state_dict(), total=total_len, position=0):name = itemdesc.set_description('Loading %s' % name)if name not in state_dicts: # Convert from non-parallel modelif str('backbone.' + name) in state_dicts:tmp[name] = state_dicts['backbone.' + name]else:  # Convert from parallel modeltmp[name] = state_dicts[name]time.sleep(0.01)ret_str = "loading {:<20d} weights completed.".format(len(model.state_dict()))desc.set_description(ret_str)model.set_state_dict(tmp)def mkdir(dir):if not os.path.exists(dir):# avoid error when train with multiple gpustry:os.makedirs(dir)except:pass@main_only
def save(obj, path):paddle.save(obj, path)def load(file_name):if not osp.isfile(file_name):raise IOError(f'{file_name} not exist')return paddle.load(file_name)

• 首先，代码定义了一个装饰器@main_only，它的作用是只在主进程中执行被装饰的函数，以避免多卡训练时的冲突。
• 然后，代码定义了一个函数 load_ckpt，它的作用是加载预训练模型的参数，并转换为与现有模型匹配的参数字典，然后加载到现有模型中。
• 函数 mkdir，它的作用是创建一个目录。
• save 函数用来将一个 PaddlePaddle 的对象保存到一个文件中。
• load 函数用来从一个文件中加载一个 PaddlePaddle 的对象。

precise_bn.py

import paddle
import itertoolsfrom paddlevideo.utils import get_logger
logger = get_logger("paddlevideo")
"""
Implement precise bn, which is useful for improving accuracy.
"""@paddle.no_grad()  # speed up and save CUDA memory
def do_preciseBN(model, data_loader, parallel, num_iters=200):"""Recompute and update the batch norm stats to make them more precise. Duringtraining both BN stats and the weight are changing after every iteration, sothe running average can not precisely reflect the actual stats of thecurrent model.In this function, the BN stats are recomputed with fixed weights, to makethe running average more precise. Specifically, it computes the true averageof per-batch mean/variance instead of the running average.This is useful to improve validation accuracy.Args:model: the model whose bn stats will be recomputeddata_loader: an iterator. Produce data as input to the modelnum_iters: number of iterations to compute the stats.Return:the model with precise mean and variance in bn layers."""bn_layers_list = [m for m in model.sublayers()if any((isinstance(m, bn_type)for bn_type in (paddle.nn.BatchNorm1D, paddle.nn.BatchNorm2D,paddle.nn.BatchNorm3D))) and m.training]if len(bn_layers_list) == 0:return# moving_mean=moving_mean*momentum+batch_mean*(1.−momentum)# we set momentum=0. to get the true mean and variance during forwardmomentum_actual = [bn._momentum for bn in bn_layers_list]for bn in bn_layers_list:bn._momentum = 0.running_mean = [paddle.zeros_like(bn._mean)for bn in bn_layers_list]  #pre-ignorerunning_var = [paddle.zeros_like(bn._variance) for bn in bn_layers_list]ind = -1for ind, data in enumerate(itertools.islice(data_loader, num_iters)):logger.info("doing precise BN {} / {}...".format(ind + 1, num_iters))if parallel:model._layers.train_step(data)else:model.train_step(data)for i, bn in enumerate(bn_layers_list):# Accumulates the bn stats.running_mean[i] += (bn._mean - running_mean[i]) / (ind + 1)running_var[i] += (bn._variance - running_var[i]) / (ind + 1)assert ind == num_iters - 1, ("update_bn_stats is meant to run for {} iterations, but the dataloader stops at {} iterations.".format(num_iters, ind))# Sets the precise bn stats.for i, bn in enumerate(bn_layers_list):bn._mean.set_value(running_mean[i])bn._variance.set_value(running_var[i])bn._momentum = momentum_actual[i]

这段代码是用来实现精确的批量归一化（precise batch normalization）的，这是一种提高验证精度的方法。

在训练过程中，批量归一化的统计量和权重都在每次迭代后发生变化，因此滑动平均不能准确地反映当前模型的实际统计量。使用这个函数，批量归一化的统计量是用固定的权重重新计算的，使滑动平均更加精确。具体来说，它计算每个批次的均值/方差的真实平均值，而不是滑动平均值。这对于提高验证精度是有用的。

代码的主要逻辑是：

• 首先，找出模型中所有的批量归一化层（bn_layers_list），并且把它们的动量（momentum）设为0，这样就不会使用滑动平均来计算均值和方差，而是直接使用每个批次的统计量。
• 然后，初始化两个列表（running_mean 和 running_var），用来存储每个批量归一化层的累积均值和方差。
• 接着，遍历数据集（data_loader）的前 num_iters 个批次，对每个批次，用模型进行前向传播，并且把每个批量归一化层的均值和方差累加到对应的列表中。
• 最后，把每个列表中的累积均值和方差除以 num_iters，得到更精确的均值和方差，并且更新到模型的批量归一化层中。

tasks 文件夹

tasks 文件夹的作用是存放一些用于定义和执行不同的机器学习任务的类或函数。不同的机器学习任务可能需要不同的数据集，模型，指标，训练和测试流程等，例如图像分类任务，关系分类任务，语义检索任务，智能问答任务等。tasks 文件夹中的类或函数可以根据不同的任务和数据集来构建和运行相应的模型，并在训练或测试过程中使用 metrics 文件夹中的指标来评估模型的性能。

__init__.py

from .train import train_model
from .test import test_model
from .train_dali import train_dali
from .train_multigrid import train_model_multigrid__all__ = ['train_model', 'test_model', 'train_dali', 'train_model_multigrid']

不再赘述，就是说要把 paddlevideo/tasks 文件夹当作包导入时会导入哪些函数或类的模块。__all__ 定义了当 from paddlevideo.tasks import * 时会导入的模块。

train.py

训练脚本路径：work/PaddleVideo/paddlevideo/tasks/train.py

import time
import os
import os.path as ospimport paddle
import paddle.distributed as dist
import paddle.distributed.fleet as fleet
from ..loader.builder import build_dataloader, build_dataset
from ..modeling.builder import build_model
from ..solver import build_lr, build_optimizer
from ..utils import do_preciseBN
from paddlevideo.utils import get_logger
from paddlevideo.utils import (build_record, log_batch, log_epoch, save, load,mkdir)
import numpy as np
import paddle.nn.functional as Fdef train_model(cfg,weights=None,parallel=True,validate=True,amp=False,use_fleet=False):"""Train model entryArgs:cfg (dict): configuration.weights (str): weights path for finetuning.parallel (bool): Whether multi-cards training. Default: True.validate (bool): Whether to do evaluation. Default: False."""if use_fleet:fleet.init(is_collective=True)logger = get_logger("paddlevideo")batch_size = cfg.DATASET.get('batch_size', 8)valid_batch_size = cfg.DATASET.get('valid_batch_size', batch_size)use_gradient_accumulation = cfg.get('GRADIENT_ACCUMULATION', None)if use_gradient_accumulation and dist.get_world_size() >= 1:global_batch_size = cfg.GRADIENT_ACCUMULATION.get('global_batch_size', None)num_gpus = dist.get_world_size()assert isinstance(global_batch_size, int), f"global_batch_size must be int, but got {type(global_batch_size)}"assert batch_size < global_batch_size, f"global_batch_size must bigger than batch_size"cur_global_batch_size = batch_size * num_gpus  # The number of batches calculated by all GPUs at one timeassert global_batch_size % cur_global_batch_size == 0, \f"The global batchsize must be divisible by cur_global_batch_size, but \{global_batch_size} % {cur_global_batch_size} != 0"cfg.GRADIENT_ACCUMULATION["num_iters"] = global_batch_size // cur_global_batch_size# The number of iterations required to reach the global batchsizelogger.info(f"Using gradient accumulation training strategy, "f"global_batch_size={global_batch_size}, "f"num_gpus={num_gpus}, "f"num_accumulative_iters={cfg.GRADIENT_ACCUMULATION.num_iters}")places = paddle.set_device('gpu')# default num worker: 0, which means no subprocess will be creatednum_workers = cfg.DATASET.get('num_workers', 0)valid_num_workers = cfg.DATASET.get('valid_num_workers', num_workers)model_name = cfg.model_nameoutput_dir = cfg.get("output_dir", f"./output/{model_name}")mkdir(output_dir)# 1. Construct modelmodel = build_model(cfg.MODEL)if parallel:model = paddle.DataParallel(model)if use_fleet:model = paddle.distributed_model(model)# 2. Construct dataset and dataloadertrain_dataset = build_dataset((cfg.DATASET.train, cfg.PIPELINE.train))train_dataloader_setting = dict(batch_size=batch_size,num_workers=num_workers,collate_fn_cfg=cfg.get('MIX', None),places=places)train_loader = build_dataloader(train_dataset, **train_dataloader_setting)if validate:valid_dataset = build_dataset((cfg.DATASET.valid, cfg.PIPELINE.valid))validate_dataloader_setting = dict(batch_size=valid_batch_size,num_workers=valid_num_workers,places=places,drop_last=False,shuffle=cfg.DATASET.get('shuffle_valid',False)  #NOTE: attention lstm need shuffle valid data.)valid_loader = build_dataloader(valid_dataset,**validate_dataloader_setting)# 3. Construct solver.if cfg.OPTIMIZER.learning_rate.get('iter_step'):lr = build_lr(cfg.OPTIMIZER.learning_rate, len(train_loader))else:lr = build_lr(cfg.OPTIMIZER.learning_rate, 1)optimizer = build_optimizer(cfg.OPTIMIZER,lr,parameter_list=model.parameters())if use_fleet:optimizer = fleet.distributed_optimizer(optimizer)# Resumeresume_epoch = cfg.get("resume_epoch", 0)if resume_epoch:filename = osp.join(output_dir,model_name + f"_epoch_{resume_epoch:05d}")resume_model_dict = load(filename + '.pdparams')resume_opt_dict = load(filename + '.pdopt')model.set_state_dict(resume_model_dict)optimizer.set_state_dict(resume_opt_dict)# Finetune:if weights:assert resume_epoch == 0, f"Conflict occurs when finetuning, please switch resume function off by setting resume_epoch to 0 or not indicating it."model_dict = load(weights)model.set_state_dict(model_dict)# 4. Train Model###AMP###if amp:scaler = paddle.amp.GradScaler(init_loss_scaling=2.0**16,incr_every_n_steps=2000,decr_every_n_nan_or_inf=1)best = 0.for epoch in range(0, cfg.epochs):if epoch < resume_epoch:logger.info(f"| epoch: [{epoch+1}] <= resume_epoch: [{ resume_epoch}], continue... ")continuemodel.train()record_list = build_record(cfg.MODEL)tic = time.time()for i, data in enumerate(train_loader):record_list['reader_time'].update(time.time() - tic)# 4.1 forward###AMP###if amp:with paddle.amp.auto_cast(custom_black_list={"reduce_mean"}):outputs = model(data, mode='train')avg_loss = outputs['loss']scaled = scaler.scale(avg_loss)scaled.backward()# keep prior to 2.0 designscaler.minimize(optimizer, scaled)optimizer.clear_grad()else:outputs = model(data, mode='train')# 4.2 backwardif use_gradient_accumulation and i == 0:  # Use gradient accumulation strategyoptimizer.clear_grad()avg_loss = outputs['loss']avg_loss.backward()# 4.3 minimizeif use_gradient_accumulation:  # Use gradient accumulation strategyif (i + 1) % cfg.GRADIENT_ACCUMULATION.num_iters == 0:for p in model.parameters():p.grad.set_value(p.grad / cfg.GRADIENT_ACCUMULATION.num_iters)optimizer.step()optimizer.clear_grad()else:  # Common caseoptimizer.step()optimizer.clear_grad()# log recordrecord_list['lr'].update(optimizer.get_lr(), batch_size)for name, value in outputs.items():record_list[name].update(value, batch_size)record_list['batch_time'].update(time.time() - tic)tic = time.time()if i % cfg.get("log_interval", 10) == 0:ips = "ips: {:.5f} instance/sec.".format(batch_size / record_list["batch_time"].val)log_batch(record_list, i, epoch + 1, cfg.epochs, "train", ips)# learning rate iter stepif cfg.OPTIMIZER.learning_rate.get("iter_step"):lr.step()# learning rate epoch stepif not cfg.OPTIMIZER.learning_rate.get("iter_step"):lr.step()ips = "avg_ips: {:.5f} instance/sec.".format(batch_size * record_list["batch_time"].count /record_list["batch_time"].sum)log_epoch(record_list, epoch + 1, "train", ips)def evaluate(best):model.eval()record_list = build_record(cfg.MODEL)record_list.pop('lr')tic = time.time()for i, data in enumerate(valid_loader):outputs = model(data, mode='valid')# log_recordfor name, value in outputs.items():record_list[name].update(value, batch_size)record_list['batch_time'].update(time.time() - tic)tic = time.time()if i % cfg.get("log_interval", 10) == 0:ips = "ips: {:.5f} instance/sec.".format(batch_size / record_list["batch_time"].val)log_batch(record_list, i, epoch + 1, cfg.epochs, "val", ips)ips = "avg_ips: {:.5f} instance/sec.".format(batch_size * record_list["batch_time"].count /record_list["batch_time"].sum)log_epoch(record_list, epoch + 1, "val", ips)best_flag = Falsefor top_flag in ['hit_at_one', 'top1']:if record_list.get(top_flag) and record_list[top_flag].avg > best:best = record_list[top_flag].avgbest_flag = Truereturn best, best_flag# use precise bn to improve accif cfg.get("PRECISEBN") and (epoch % cfg.PRECISEBN.preciseBN_interval== 0 or epoch == cfg.epochs - 1):do_preciseBN(model, train_loader, parallel,min(cfg.PRECISEBN.num_iters_preciseBN, len(train_loader)))# 5. Validationif validate and (epoch % cfg.get("val_interval", 1) == 0or epoch == cfg.epochs - 1):with paddle.no_grad():best, save_best_flag = evaluate(best)# save bestif save_best_flag:save(optimizer.state_dict(),osp.join(output_dir, model_name + '_' + str(int(best *10000)/10000) + "_best.pdopt"))save(model.state_dict(),osp.join(output_dir, model_name + '_' + str(int(best *10000)/10000) + "_best.pdparams"))os.makedirs('./model', exist_ok=True)save(model.state_dict(),osp.join('./model', model_name + ".pdparams"))  if model_name == "AttentionLstm":logger.info(f"Already save the best model (hit_at_one){best}")else:logger.info(f"Already save the best model (top1 acc){int(best *10000)/10000}")# 6. Save model and optimizerif epoch % cfg.get("save_interval", 1) == 0 or epoch == cfg.epochs - 1:save(optimizer.state_dict(),osp.join(output_dir,model_name + f"_epoch_{epoch+1:05d}.pdopt"))save(model.state_dict(),osp.join(output_dir,model_name + f"_epoch_{epoch+1:05d}.pdparams"))logger.info(f'training {model_name} finished')

train_model 函数主要功能是根据配置信息和权重路径，创建一个模型对象，并进行训练和评估。

有以下参数：

• cfg: 一个字典，包含了模型的配置信息。
• weights: 一个字符串，表示用于微调的权重路径。
• parallel: 一个布尔值，表示是否使用多卡训练。默认为True。
• validate: 一个布尔值，表示是否进行评估。默认为False。
• amp: 一个布尔值，表示是否使用自动混合精度训练。默认为False。
• use_fleet: 一个布尔值，表示是否使用 fleet 分布式训练。默认为False。

• 如果使用 fleet 分布式训练，需要先初始化 fleet 环境。
• 如果使用梯度累积策略，需要计算全局批量大小和累积次数。
• 函数还使用了 paddle.set_device 函数来设置设备为 GPU。
• output_dir = cfg.get("output_dir", f"./output/{model_name}")
  这行代码指定了训练后保存模型的文件夹路径。

从配置文件中获取模型的输出目录（output_dir），也就是用于保存模型参数和日志的目录。如果配置文件中没有指定，就默认为当前目录下的 output 文件夹下的模型名称对应的文件夹。

1. Construct model
   这段代码是用于构建和分布式化模型的，具体来说：

• 首先，使用 build_model 函数根据配置文件中的模型参数，创建一个模型对象。
• 然后，判断是否使用多卡训练（parallel）。如果是，就使 paddle.DataParallel 函数将模型封装为一个数据并行的对象，可以在多个 GPU 上同时训练。
• 然后，判断是否使用 fleet 分布式训练（use_fleet）。如果是，就使用 paddle.distributed_model 函数将模型转换为一个分布式的对象，可以在多个节点上进行同步或异步的训练。

---

2. Construct dataset and dataloader
   这段代码用于构建数据集和数据加载器，具体来说：

• 首先，使用 build_dataset 函数根据配置文件中的数据集参数和数据处理流程，创建一个训练集对象。
• 然后，创建一个字典 train_dataloader_setting，包含了数据加载器的相关设置，如批量大小、子进程数、混合数据的函数和设备等。
• 然后，使用 build_dataloader 函数根据训练集对象和设置，创建一个训练集的数据加载器对象。
• 然后，判断是否进行评估（validate）。如果是，就重复上述步骤，创建一个验证集对象和一个验证集的数据加载器对象。验证集的数据加载器的设置可能和训练集的不同，比如批量大小、子进程数、是否丢弃最后一个不完整的批次、是否打乱数据等。

---

3. Construct solver
   这段代码用于构建和恢复优化器，具体来说：

• 首先，使用 build_lr 函数根据配置文件中的学习率参数，创建一个学习率对象。如果配置文件中指定了迭代步长（iter_step），就根据训练集的数据加载器的长度，也就是每个轮次的迭代次数，来创建学习率对象。否则，就根据1来创建学习率对象。
• 然后，使用 build_optimizer 函数根据配置文件中的优化器参数、学习率对象和模型的参数列表，创建一个优化器对象。
• 然后，判断是否使用 fleet分布式训练（use_fleet）。如果是，就使用 fleet.distributed_optimizer 函数将优化器转换为一个分布式的对象，可以在多个节点上进行同步或异步的优化。
• 然后，从配置文件中获取恢复轮次（resume_epoch），也就是想要从哪个轮次开始继续训练。如果恢复轮次不为0，就从输出目录中加载对应轮次的模型参数和优化器状态，并设置给模型和优化器。
• 然后，判断是否有微调权重（weights），也就是想要用于初始化模型的权重路径。如果有，就断言恢复轮次为0，以避免冲突。然后，从权重路径中加载模型参数，并设置给模型。

---

4. Train Model
   这段代码用于训练模型，具体来说：

• 首先，判断是否使用自动混合精度训练（amp）。如果是，就创建一个梯度缩放器对象（scaler），用于动态调整梯度的缩放因子，以避免数值下溢。
• 然后，创建一个变量 best，用于记录最佳的评估指标。
• 然后，使用一个 for 循环，遍历所有的轮次（epoch）。每个轮次表示对整个训练集的一次遍历。
• 然后，将模型设置为训练模式（model.train()），表示模型中的一些层（如 dropout、batchnorm 等）会根据训练状态进行调整。
• 创建一个字典 record_list，用于记录一些训练过程中的信息，如读取数据的时间、计算损失的时间、计算梯度的时间等。
• 记录一个时间点 tic，用于计算读取数据的时间。
• 使用另一个 for 循环，遍历训练集的数据加载器（train_loader）。每个数据加载器返回一个批次的数据（data），包含了输入特征和标签等。
• 然后，进行前向传播（forward）。

首先，判断是否使用自动混合精度训练（amp）。如果是，就执行以下步骤：

• 使用 paddle.amp.auto_cast 函数来自动选择合适的数据类型，并用 model 函数根据数据和训练模式，得到模型的输出（outputs）。
• 从输出中获取损失值（avg_loss），并用 scaler 对象对损失值进行缩放。
• 对缩放后的损失值进行反向传播（backward）。
• 使用scaler对象对优化器进行最小化操作（minimize），并清除梯度（clear_grad）。

否则，就执行以下步骤：

• 直接用 model 函数根据数据和训练模式，得到模型的输出（outputs）。
• 从输出中获取损失值（avg_loss），并对损失值进行反向传播（backward）。
• 判断是否使用梯度累积策略（use_gradient_accumulation）。如果是，
  – 如果是第一次迭代，就清除梯度（clear_grad）。
  –如果达到了累积次数（num_iters），就对模型的所有参数的梯度除以累积次数，并执行优化器的更新步骤（step），并清除梯度（clear_grad）。
• 否则，就直接执行优化器的更新步骤（step），并清除梯度（clear_grad）。

• log record 用于记录训练过程中的日志和进行评估。
• if i % cfg.get("log_interval", 10) == 0 是否达到了日志间隔（log_interval），也就是每隔多少个批次打印一次日志。
• 函数 evaluate，用于对模型进行评估。函数接受一个参数 best，表示之前最佳的评估指标。
• 执行精确批归一化。

---

5. Validation
   注意，这一步在第4步 Train Model 中的 for epoch in range(0, cfg.epochs) 循环中。这段代码用于在每个轮次结束后，对模型进行评估和保存。具体来说：

• 首先，判断是否进行评估（validate），判断当前的轮次（epoch）是否满足评估的间隔（val_interval），或者是否是最后一个轮次。
• 使用 paddle.no_grad 函数禁用梯度计算，以节省内存和提高速度。
• 使用 evaluate 函数对模型进行评估，并返回最佳的评估指标（best）和是否需要保存最佳模型的标志（save_best_flag）。
• 使用 save 函数保存优化器和模型的状态字典（optimizer.state_dict()和
  model.state_dict()）到输出目录下，文件名分别为模型名称加上最佳评估指标加上后缀"_best.pdopt"，“_best.pdparams”。
• 使用 save 函数保存模型的参数字典（model.state_dict()）到"model"文件夹下，文件名为模型名称加上后缀".pdparams"。
• 使用 logger.info 函数打印出已经保存了最佳模型和准确率的信息。

---

6. Save model and optimizer
   注意，这一步在第4步 Train Model 中的 for epoch in range(0, cfg.epochs) 循环中。这段代码用于在每个轮次结束后，保存模型和优化器。具体来说：

判断当前的轮次（epoch）是否满足保存的间隔（save_interval），或者是否是最后一个轮次。如果是，就执行以下步骤：

• 使用 save 函数保存优化器的状态字典（optimizer.state_dict()）到输出目录下，文件名为模型名称加上当前轮次加上后缀".pdopt"。
• 使用 save 函数保存模型的参数字典（model.state_dict()）到输出目录下，文件名为模型名称加上当前轮次加上后缀".pdparams"。

test.py

测试脚本路径：work/PaddleVideo/paddlevideo/tasks/test.py

import paddle
from paddlevideo.utils import get_logger
from ..loader.builder import build_dataloader, build_dataset
from ..metrics import build_metric
from ..modeling.builder import build_model
from paddlevideo.utils import loadimport numpy as np
import os
import paddle.nn.functional as Flogger = get_logger("paddlevideo")@paddle.no_grad()
def test_model(cfg, weights, parallel=True):"""Test model entryArgs:cfg (dict): configuration.weights (str): weights path to load.parallel (bool): Whether to do multi-cards testing. Default: True."""# 1. Construct model.if cfg.MODEL.backbone.get('pretrained'):cfg.MODEL.backbone.pretrained = ''  # disable pretrain model initmodel = build_model(cfg.MODEL)if parallel:model = paddle.DataParallel(model)# 2. Construct dataset and dataloader.cfg.DATASET.test.test_mode = Truedataset = build_dataset((cfg.DATASET.test, cfg.PIPELINE.test))batch_size = cfg.DATASET.get("test_batch_size", 8)places = paddle.set_device('gpu')# default num worker: 0, which means no subprocess will be creatednum_workers = cfg.DATASET.get('num_workers', 0)num_workers = cfg.DATASET.get('test_num_workers', num_workers)dataloader_setting = dict(batch_size=batch_size,num_workers=num_workers,places=places,drop_last=False,shuffle=False)data_loader = build_dataloader(dataset, **dataloader_setting)model.eval()state_dicts = load(weights)model.set_state_dict(state_dicts)# add params to metricscfg.METRIC.data_size = len(dataset)cfg.METRIC.batch_size = batch_sizeprint('{} inference start!!!'.format(cfg.model_name))Metric = build_metric(cfg.METRIC)ans = np.zeros((len(data_loader), 30))for batch_id, data in enumerate(data_loader):outputs = model(data, mode='test')ans[batch_id, :] = outputsMetric.update(batch_id, data, outputs)os.makedirs('logits', exist_ok=True)with open('logits/{}.npy'.format(cfg.model_name), 'wb') as f:np.save(f, ans)print('{} inference finished!!!'.format(cfg.model_name))Metric.accumulate()

定义一个函数 test_model，接受三个参数：cfg 是一个配置字典，weights 是一个模型权重的路径，parallel 是一个布尔值，表示是否使用多卡测试。

函数的作用是测试模型的准确率和效率。

• 首先利用 model = build_model(cfg.MODEL)构建模型，并根据 parallel 参数判断是否需要使用多卡并行测试，如果需要，则用 paddle.DataParallel 将模型包装起来。
• 根据配置文件中的 test 设置构建数据集 dataset，并设置 batch_size、num_workers 等参数构建 dataloader，以准备测试数据。
• 调用 model.eval()进入测试模式，并用预训练权重对模型参数进行初始化。
• 构建 Metric 并用 build_metric(cfg.METRIC)初始化 metrics 指标，以便于记录模型在测试集上的各项表现。
• 对于每个 mini-batch，在模型上前向传播得到输出 outputs，同时调用 Metric.update()记录当前 mini-batch 的指标，最后将 outputs 保存在 ans 数组中。
• 将测试结果保存在 logits 文件夹下，并输出测试完成信息。

logits 是一个术语，表示神经网络最后一层的未归一化的概率，也就是说，它们的值可以是任意的实数，而不一定在 0 到 1 之间，也不一定加起来等于 1。
logits 这个词来源于 logit 函数，它是 sigmoid 函数的反函数，可以将概率值转换为对数几率值。但是在深度学习中，logits 并不一定要经过 logit 函数，而是可以直接输入到 softmax 函数或者 softmax_cross_entropy_with_logits 函数中。

metrics 文件夹

metrics 文件夹的作用是存放一些用于评估模型性能的指标的类或函数。不同的机器学习任务可能需要不同的指标来衡量模型的好坏，例如，

• 分类任务常用的指标有准确率，召回率，F1分数，ROC 曲线等，
• 而回归任务常用的指标有均方误差，均方根误差，平均绝对误差等。

metrics 文件夹中的类或函数可以根据不同的任务和数据集来定义和计算相应的指标，并在训练或测试过程中更新和累积指标的值，以便于模型的选择和优化。

__init__.py

from .registry import METRIC
from .build import build_metric
from .skeleton_metric import SkeletonMetric__all__ = ['METRIC', 'build_metric', 'SkeletonMetric'
]

skeleton_metric.py

import numpy as np
import paddle
import csv
import paddle.nn.functional as Ffrom .registry import METRIC
from .base import BaseMetric
from paddlevideo.utils import get_loggerlogger = get_logger("paddlevideo")@METRIC.register
class SkeletonMetric(BaseMetric):"""Test for Skeleton based model.note: only support batch size = 1, single card test.Args:out_file: str, file to save test results."""def __init__(self,data_size,batch_size,out_file='submission.csv',log_interval=1):"""prepare for metrics"""super().__init__(data_size, batch_size, log_interval)self.top1 = []self.top5 = []self.values = []self.out_file = out_filedef update(self, batch_id, data, outputs):"""update metrics during each iter"""if len(data) == 2:  # data with labellabels = data[1]top1 = paddle.metric.accuracy(input=outputs, label=labels, k=1)top5 = paddle.metric.accuracy(input=outputs, label=labels, k=5)if self.world_size > 1:top1 = paddle.distributed.all_reduce(top1, op=paddle.distributed.ReduceOp.SUM) / self.world_sizetop5 = paddle.distributed.all_reduce(top5, op=paddle.distributed.ReduceOp.SUM) / self.world_sizeself.top1.append(top1.numpy())self.top5.append(top5.numpy())else:  # data without label, only support batch_size=1. Used for fsd-10.# prob = F.softmax(outputs)outputs = outputs.unsqueeze(0)clas = paddle.argmax(outputs, axis=1).numpy()[0]self.values.append((batch_id, clas))# preds ensemble# if batch_id % self.log_interval == 0:#     logger.info("[TEST] Processing batch {}/{} ...".format(#         batch_id,#         self.data_size // (self.batch_size * self.world_size)))def accumulate(self):"""accumulate metrics when finished all iters."""if self.top1:  # data with labellogger.info('[TEST] finished, avg_acc1= {}, avg_acc5= {}'.format(np.mean(np.array(self.top1)), np.mean(np.array(self.top5))))else:headers = ['sample_index', 'predict_category']with open(self.out_file,'w',) as fp:writer = csv.writer(fp)writer.writerow(headers)writer.writerows(self.values)logger.info("Results saved in {} !".format(self.out_file))

这段代码是定义了一个 SkeletonMetric 类，继承自 BaseMetric 类，用于评估基于骨架的模型的性能。

• 在类的初始化方法中，接受四个参数：data_size 是数据集的大小，batch_size 是批量大小，out_file 是保存测试结果的文件名，log_interval 是打印日志的间隔。然后调用父类的初始化方法，并设置一些属性，如 top1，top5，values 和 out_file。
• 在类的 update 方法中，接受三个参数：batch_id 是批次的编号，data 是输入数据，outputs 是模型的输出。然后根据数据是否有标签来进行不同的处理。如果有标签，就计算模型的 top-1 和 top-5 准确率，并将它们添加到 top1 和 top5 列表中。如果没有标签，就对模型的输出进行 argmax 操作，得到预测的类别，并将它和批次编号添加到 values 列表中。
• 在类的 accumulate 方法中，根据数据是否有标签来进行不同的处理。如果有标签，就计算并打印模型的平均 top-1 和 top-5 准确率。如果没有标签，就将 values 列表中的结果保存到 out_file 文件中。

loader 文件夹

loader 文件夹的作用是存放一些用于加载和处理数据的类或函数。
不同的机器学习任务可能需要不同的数据格式，数据预处理，数据增强，数据采样等，例如，

• 图像分类任务需要加载图像文件，进行裁剪，旋转，归一化等操作，
• 而关系分类任务需要加载文本文件，进行分词，编码，填充等操作。

loader 文件夹中的类或函数可以根据不同的任务和数据集来创建和使用相应的数据加载器，并在训练或测试过程中提供批量的数据输入。

__init__py

from .builder import build_dataset, build_dataloader, build_batch_pipeline
from .dataset import SkeletonDataset
from .dali_loader import TSN_Dali_loader, get_input_data__all__ = ['build_dataset', 'build_dataloader', 'build_batch_pipeline', 'SkeletonDataset','TSN_Dali_loader', 'get_input_data'
]

skeleton.py

文件路径：loader/dataset/skeleton.py

import os.path as osp
import copy
import random
import numpy as np
import picklefrom ..registry import DATASETS
from .base import BaseDataset
from ...utils import get_loggerlogger = get_logger("paddlevideo")# #set random seed
# random.seed(0)
# np.random.seed(0)
# copyright (c) 2021 PaddlePaddle Authors. All Rights Reserve.
#
# 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.import os.path as osp
import copy
import random
import numpy as np
import pickle
import gcfrom ..registry import DATASETS
from .base import BaseDataset
from ...utils import get_loggerlogger = get_logger("paddlevideo")@DATASETS.register()
class KFoldSkeletonDataset(BaseDataset):"""Skeleton dataset for action recognition.The dataset loads skeleton feature, and apply norm operatations.Args:file_path (str): Path to the index file.pipeline(obj): Define the pipeline of data preprocessing.data_prefix (str): directory path of the data. Default: None.test_mode (bool): Whether to bulid the test dataset. Default: False."""def __init__(self, index_path, pipeline, file_path='/home/zhaorifa/competition/home/aistudio/work/data/train_data_padd.npy',label_path='/home/zhaorifa/competition/home/aistudio/work/data/train_label.npy', test_mode=False):self.file_path = file_pathif not test_mode:self.label_path = label_pathself.index_path = index_pathsuper().__init__(file_path, pipeline, test_mode=test_mode)def load_file(self):"""Load feature file to get skeleton information."""logger.info("Loading data, it will take some moment...")self.idx = np.load(self.index_path)tmp_data = np.load(self.file_path)self.data = tmp_data[self.idx]del tmp_datagc.collect()if self.label_path:if self.label_path.endswith('npy'):self.label = np.load(self.label_path)elif self.label_path.endswith('pkl'):with open(self.label_path, 'rb') as f:sample_name, self.label = pickle.load(f)self.label = self.label[self.idx]else:logger.info("Label path not provided when test_mode={}, here just output predictions.".format(self.test_mode))logger.info("Data Loaded!")return self.data  # used for __len__def prepare_train(self, idx):"""Prepare the feature for training/valid given index. """results = dict()results['data'] = copy.deepcopy(self.data[idx])results['label'] = copy.deepcopy(self.label[idx])results = self.pipeline(results)return results['data'], results['label']def prepare_test(self, idx):"""Prepare the feature for test given index. """results = dict()results['data'] = copy.deepcopy(self.data[idx])if self.label_path:results['label'] = copy.deepcopy(self.label[idx])results = self.pipeline(results)return results['data'], results['label']else:results = self.pipeline(results)return [results['data']]@DATASETS.register()
class SkeletonDataset(BaseDataset):"""Skeleton dataset for action recognition.The dataset loads skeleton feature, and apply norm operatations.Args:file_path (str): Path to the index file.pipeline(obj): Define the pipeline of data preprocessing.data_prefix (str): directory path of the data. Default: None.test_mode (bool): Whether to bulid the test dataset. Default: False."""def __init__(self, file_path, pipeline, label_path=None, test_mode=False):self.label_path = label_pathsuper().__init__(file_path, pipeline, test_mode=test_mode)def load_file(self):"""Load feature file to get skeleton information."""logger.info("Loading data, it will take some moment...")self.data = np.load(self.file_path)if self.label_path:if self.label_path.endswith('npy'):self.label = np.load(self.label_path)elif self.label_path.endswith('pkl'):with open(self.label_path, 'rb') as f:sample_name, self.label = pickle.load(f)else:logger.info("Label path not provided when test_mode={}, here just output predictions.".format(self.test_mode))logger.info("Data Loaded!")return self.data  # used for __len__def prepare_train(self, idx):"""Prepare the feature for training/valid given index. """results = dict()results['data'] = copy.deepcopy(self.data[idx])results['label'] = copy.deepcopy(self.label[idx])results = self.pipeline(results)return results['data'], results['label']def prepare_test(self, idx):"""Prepare the feature for test given index. """results = dict()results['data'] = copy.deepcopy(self.data[idx])if self.label_path:results['label'] = copy.deepcopy(self.label[idx])results = self.pipeline(results)return results['data'], results['label']else:results = self.pipeline(results)return [results['data']]

这段代码实现了两个基于 Skeleton 特征的动作识别数据集类：KFoldSkeletonDataset 和 SkeletonDataset。这两个类都继承自 BaseDataset 类，实现了 load_file 方法和 prepare_train、prepare_test 方法。

其中，load_file 方法用于从文件中加载数据，prepare_train 方法和 prepare_test 方法用于为模型准备训练和测试数据。

• KFoldSkeletonDataset 是一个 k 折交叉验证的数据集类，需要提供一个索引文件 index_path，一个特征文件 file_path 和一个标签文件 label_path，可以使用 pipeline 对数据进行预处理。

在 load_file 方法中，首先从指定路径的 index_path 文件中读取索引信息，然后从指定路径的 file_path 文件中读取特征信息，最后根据索引信息选取对应的特征数据，并删除临时变量以释放内存。如果提供了标签数据，则也会根据索引信息选取对应的标签数据。
prepare_train 方法和 prepare_test 方法都会返回经过 pipeline 预处理过的特征数据和标签数据（如果有的话）。

• SkeletonDataset 类与 KFoldSkeletonDataset 类相似，不同之处在于它不是 k 折交叉验证的数据集类，而是单独的一个数据集类。

skeleton_pipeline.py

文件路径：loader/pipelines/skeleton_pipeline.py

import os
import numpy as np
import math
import random
import paddle
import paddle.nn.functional as F
from ..registry import PIPELINES
"""pipeline ops for Activity Net.
"""@PIPELINES.register()
class UniformSampleFrames(object):"""Uniformly sample frames from the video.To sample an n-frame clip from the video. UniformSampleFrames basicallydivide the video into n segments of equal length and randomly sample oneframe from each segment. To make the testing results reproducible, arandom seed is set during testing, to make the sampling resultsdeterministic.Required keys are "total_frames", "start_index" , added or modified keysare "frame_inds", "window_size", "frame_interval" and "num_clips".Args:window_size (int): Frames of each sampled output clip.num_clips (int): Number of clips to be sampled. Default: 1.test_mode (bool): Store True when building test or validation dataset.Default: False.seed (int): The random seed used during test time. Default: 255."""def __init__(self, window_size, num_clips=1, test_mode=False, seed=255):self.window_size = window_sizeself.num_clips = num_clipsself.test_mode = test_modeself.seed = seeddef get_frame_num(self, data):C, T, V, M = data.shapefor i in range(T - 1, -1, -1):tmp = np.sum(data[0:2, i, :, :])if tmp != 0:T = i + 1breakreturn Tdef _get_train_clips(self, num_frames, window_size):"""Uniformly sample indices for training clips.Args:num_frames (int): The number of frames.window_size (int): The length of the clip."""assert self.num_clips == 1if num_frames < window_size:start = np.random.randint(0, num_frames)inds = np.arange(start, start + window_size)elif window_size <= num_frames < 2 * window_size:basic = np.arange(window_size)inds = np.random.choice(window_size + 1, num_frames - window_size, replace=False)offset = np.zeros(window_size + 1, dtype=np.int64)offset[inds] = 1offset = np.cumsum(offset)inds = basic + offset[:-1]else:bids = np.array([i * num_frames // window_size for i in range(window_size + 1)])bsize = np.diff(bids)bst = bids[:window_size]offset = np.random.randint(bsize)inds = bst + offsetreturn indsdef _get_test_clips(self, num_frames, window_size):"""Uniformly sample indices for testing clips.Args:num_frames (int): The number of frames.window_size (int): The length of the clip."""np.random.seed(self.seed)if num_frames < window_size:# Then we use a simple strategyif num_frames < self.num_clips:start_inds = list(range(self.num_clips))else:start_inds = [i * num_frames // self.num_clipsfor i in range(self.num_clips)]inds = np.concatenate([np.arange(i, i + window_size) for i in start_inds])elif window_size <= num_frames < window_size * 2:all_inds = []for i in range(self.num_clips):basic = np.arange(window_size)inds = np.random.choice(window_size + 1, num_frames - window_size, replace=False)offset = np.zeros(window_size + 1, dtype=np.int64)offset[inds] = 1offset = np.cumsum(offset)inds = basic + offset[:-1]all_inds.append(inds)inds = np.concatenate(all_inds)else:bids = np.array([i * num_frames // window_size for i in range(window_size + 1)])bsize = np.diff(bids)bst = bids[:window_size]all_inds = []for i in range(self.num_clips):offset = np.random.randint(bsize)all_inds.append(bst + offset)inds = np.concatenate(all_inds)return indsdef __call__(self, results):data = results['data']num_frames = self.get_frame_num(data)if self.test_mode:inds = self._get_test_clips(num_frames, self.window_size)else:inds = self._get_train_clips(num_frames, self.window_size)inds = np.mod(inds, num_frames)assert inds.shape[0] == self.window_sizedata_pad = data[:, inds, :, :]results['data'] = data_padreturn results@PIPELINES.register()
class SkeletonNorm_J(object):def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']# Centralizationdata[0:2, :, :, :] = data[0:2, :, :, :] - data[0:2, :, 8:9, :]data = data[:self.axis, :, :, :]C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class SkeletonNorm_JA(object):def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']# Centralizationdata[0:2, :, :, :] = data[0:2, :, :, :] - data[0:2, :, 8:9, :]C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class SkeletonNorm_Mb(object):def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class SkeletonNorm_JMj(object):def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']# Centralizationdata[0:2, :, :, :] = data[0:2, :, :, :] - data[0:2, :, 8:9, :]C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class SkeletonNorm_BA(object):"""Normalize skeleton feature.Args:aixs: dimensions of vertex coordinate. 2 for (x,y), 3 for (x,y,z). Default: 2."""def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class SkeletonNorm_B(object):def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']# Centralizationdata = data[:self.axis, :, :, :]C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class SkeletonNorm_JB(object):def __init__(self, axis=2, squeeze=False):self.axis = axisself.squeeze = squeezedef __call__(self, results):data = results['data']# Centralizationdata[0:2, :, :, :] = data[0:2, :, :, :] - data[0:2, :, 8:9, :]C, T, V, M = data.shapeif self.squeeze:data = data.reshape((C, T, V))  # M = 1results['data'] = data.astype('float32')if 'label' in results:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results@PIPELINES.register()
class Iden(object):"""Wrapper Pipeline"""def __init__(self, label_expand=True):self.label_expand = label_expanddef __call__(self, results):data = results['data']results['data'] = data.astype('float32')if 'label' in results and self.label_expand:label = results['label']results['label'] = np.expand_dims(label, 0).astype('int64')return results

这是一个基于 PaddlePaddle 框架的数据处理流程，其中实现了几个数据处理操作。具体来说，

• UniformSampleFrames 操作是为了从视频中均匀采样一些帧用于模型训练或测试；
• SkeletonNorm_* 操作是为了对骨骼关键点进行归一化处理，以便训练或测试需要；
• 而 Iden 则是一个包装器操作，用于将数据类型转换为 float32，并可选是否扩展标签。

这些操作通过将它们组织成一个 Pipeline 对象来串起来执行，从而最终完成数据处理。

此外定义了 SkeletonNorm_J 到 SkeletonNorm_JB 七种特征的 SkeletonNorm。

solver 文件夹

solver 文件夹的作用是存放一些用于优化和求解模型的类或函数。不同的机器学习任务可能需要不同的优化算法，损失函数，正则化项，学习率策略等，例如，

• 图像分类任务需要使用随机梯度下降，交叉熵损失，权重衰减，余弦退火等，
• 而关系分类任务需要使用自适应矩估计，对比损失，对抗训练，线性衰减等。

solver 文件夹中的类或函数可以根据不同的任务和数据集来创建和使用相应的优化器，并在训练或测试过程中更新和调整模型的参数。

__init__py

from .optimizer import build_optimizer
from .lr import build_lr

custom_lr.py

import math
from paddle.optimizer.lr import *
"""
PaddleVideo Learning Rate Schedule:
You can use paddle.optimizer.lr
or define your custom_lr in this file.
"""class CustomWarmupCosineDecay(LRScheduler):r"""We combine warmup and stepwise-cosine which is used in slowfast model.Args:warmup_start_lr (float): start learning rate used in warmup stage.warmup_epochs (int): the number epochs of warmup.cosine_base_lr (float|int, optional): base learning rate in cosine schedule.max_epoch (int): total training epochs.num_iters(int): number iterations of each epoch.last_epoch (int, optional):  The index of last epoch. Can be set to restart training. Default: -1, means initial learning rate.verbose (bool, optional): If ``True``, prints a message to stdout for each update. Default: ``False`` .Returns:``CosineAnnealingDecay`` instance to schedule learning rate."""def __init__(self,warmup_start_lr,warmup_epochs,cosine_base_lr,max_epoch,num_iters,last_epoch=-1,verbose=False):self.warmup_start_lr = warmup_start_lrself.warmup_epochs = warmup_epochsself.cosine_base_lr = cosine_base_lrself.max_epoch = max_epochself.num_iters = num_iters#call step() in base class, last_lr/last_epoch/base_lr will be updatesuper(CustomWarmupCosineDecay, self).__init__(last_epoch=last_epoch,verbose=verbose)def step(self, epoch=None):"""``step`` should be called after ``optimizer.step`` . It will update the learning rate in optimizer according to current ``epoch`` .The new learning rate will take effect on next ``optimizer.step`` .Args:epoch (int, None): specify current epoch. Default: None. Auto-increment from last_epoch=-1.Returns:None"""if epoch is None:if self.last_epoch == -1:self.last_epoch += 1else:self.last_epoch += 1 / self.num_iters  # update step with iterselse:self.last_epoch = epochself.last_lr = self.get_lr()if self.verbose:print('Epoch {}: {} set learning rate to {}.'.format(self.last_epoch, self.__class__.__name__, self.last_lr))def _lr_func_cosine(self, cur_epoch, cosine_base_lr, max_epoch):return cosine_base_lr * (math.cos(math.pi * cur_epoch / max_epoch) +1.0) * 0.5def get_lr(self):"""Define lr policy"""lr = self._lr_func_cosine(self.last_epoch, self.cosine_base_lr,self.max_epoch)lr_end = self._lr_func_cosine(self.warmup_epochs, self.cosine_base_lr,self.max_epoch)# Perform warm up.if self.last_epoch < self.warmup_epochs:lr_start = self.warmup_start_lralpha = (lr_end - lr_start) / self.warmup_epochslr = self.last_epoch * alpha + lr_startreturn lrclass CustomWarmupPiecewiseDecay(LRScheduler):r"""This op combine warmup and stepwise-cosine which is used in slowfast model.Args:warmup_start_lr (float): start learning rate used in warmup stage.warmup_epochs (int): the number epochs of warmup.step_base_lr (float|int, optional): base learning rate in step schedule.max_epoch (int): total training epochs.num_iters(int): number iterations of each epoch.last_epoch (int, optional):  The index of last epoch. Can be set to restart training. Default: -1, means initial learning rate.verbose (bool, optional): If ``True``, prints a message to stdout for each update. Default: ``False`` .Returns:``CustomWarmupPiecewiseDecay`` instance to schedule learning rate."""def __init__(self,warmup_start_lr,warmup_epochs,step_base_lr,lrs,gamma,steps,max_epoch,num_iters,last_epoch=0,verbose=False):self.warmup_start_lr = warmup_start_lrself.warmup_epochs = warmup_epochsself.step_base_lr = step_base_lrself.lrs = lrsself.gamma = gammaself.steps = stepsself.max_epoch = max_epochself.num_iters = num_itersself.last_epoch = last_epochself.last_lr = self.warmup_start_lr  # used in first iterself.verbose = verboseself._var_name = Nonedef step(self, epoch=None, rebuild=False):"""``step`` should be called after ``optimizer.step`` . It will update the learning rate in optimizer according to current ``epoch`` .The new learning rate will take effect on next ``optimizer.step`` .Args:epoch (int, None): specify current epoch. Default: None. Auto-increment from last_epoch=-1.Returns:None"""if epoch is None:if not rebuild:self.last_epoch += 1 / self.num_iters  # update step with iterselse:self.last_epoch = epochself.last_lr = self.get_lr()if self.verbose:print('Epoch {}: {} set learning rate to {}.'.format(self.last_epoch, self.__class__.__name__, self.last_lr))def _lr_func_steps_with_relative_lrs(self, cur_epoch, lrs, base_lr, steps,max_epoch):# get step indexsteps = steps + [max_epoch]for ind, step in enumerate(steps):if cur_epoch < step:breakreturn lrs[ind - 1] * base_lrdef get_lr(self):"""Define lr policy"""lr = self._lr_func_steps_with_relative_lrs(self.last_epoch,self.lrs,self.step_base_lr,self.steps,self.max_epoch,)lr_end = self._lr_func_steps_with_relative_lrs(self.warmup_epochs,self.lrs,self.step_base_lr,self.steps,self.max_epoch,)# Perform warm up.if self.last_epoch < self.warmup_epochs:lr_start = self.warmup_start_lralpha = (lr_end - lr_start) / self.warmup_epochslr = self.last_epoch * alpha + lr_startreturn lrclass CustomPiecewiseDecay(PiecewiseDecay):def __init__(self, **kargs):kargs.pop('num_iters')super().__init__(**kargs)

这段代码定义了三个学习率调度器，分别是 CustomWarmupCosineDecay、CustomWarmupPiecewiseDecay 和 CustomPiecewiseDecay。这些调度器可以在训练深度学习模型时根据不同的需求调整学习率。
其中，

• CustomWarmupCosineDecay 结合了 warmup 和 stepwise-cosine 策略，
• CustomWarmupPiecewiseDecay 结合了 warmup 和 stepwise 策略，
• 而 CustomPiecewiseDecay 则采用 piecewise 策略。

另外，在CustomPiecewiseDecay 中，它继承了 PaddlePaddle 中的 PiecewiseDecay 类，并重写了初始化函数，以便去除不必要的参数 num_iters。

modeling 文件夹

modeling 文件夹的作用是存放一些用于构建和定义模型的类或函数。不同的机器学习任务可能需要不同的模型结构，模型参数，模型层，模型激活等，例如，

• 图像分类任务需要使用卷积神经网络，全连接层，softmax层，ReLU激活等，
• 而关系分类任务需要使用循环神经网络，注意力机制，线性层，tanh激活等。

modeling 文件夹中的类或函数可以根据不同的任务和数据集来创建和使用相应的模型，并在训练或测试过程中实现模型的前向传播和反向传播。

__init__py

from .backbones import CTRGCN
from .builder import (build_backbone, build_head, build_recognizer, build_loss)
from .heads import BaseHead
from .losses import CrossEntropyLoss, MuliFocalLoss
from .framework.recognizers import BaseRecognizer
from .registry import BACKBONES, HEADS, LOSSES, RECOGNIZERS
from .weight_init import weight_init___all__ = ['BACKBONES','HEADS','RECOGNIZERS','LOSSES','build_recognizer','build_head','build_backbone','build_loss','BaseHead','BaseRecognizer','CrossEntropyLoss','MuliFocalLoss'
]

RecognizerGCN

文件路径：modeling/framework/recognizers/recognizer_gcn.py

from ...registry import RECOGNIZERS
from .base import BaseRecognizer
from paddlevideo.utils import get_loggerlogger = get_logger("paddlevideo")@RECOGNIZERS.register()
class RecognizerGCN(BaseRecognizer):"""GCN Recognizer model framework."""def forward_net(self, data):"""Define how the model is going to run, from input to output."""feature = self.backbone(data)cls_score = self.head(feature)return cls_scoredef train_step(self, data_batch):"""Training step."""data = data_batch[0]label = data_batch[1:]# call forwardcls_score = self.forward_net(data)loss_metrics = self.head.loss(cls_score, label)return loss_metricsdef val_step(self, data_batch):"""Validating setp."""data = data_batch[0]label = data_batch[1:]# call forwardcls_score = self.forward_net(data)loss_metrics = self.head.loss(cls_score, label, valid_mode=True)return loss_metricsdef test_step(self, data_batch):"""Test step."""data = data_batch[0]# call forwardcls_score = self.forward_net(data)return cls_scoredef infer_step(self, data_batch):"""Infer step."""data = data_batch[0]# call forwardcls_score = self.forward_net(data)return cls_score

这段代码定义了名为 RecognizerGCN 的类，该类继承自 BaseRecognizer 类，并使用装饰器 @RECOGNIZERS.register() 将其注册为一种视频识别模型。

在该类中，包含了模型的前向计算函数 forward_net() 和训练、验证、测试、推理等步骤。具体来说：

• forward_net 方法接收一个数据张量作为输入，将其通过 backbone 模块提取特征，然后通过 head 模块得到分类得分。
• train_step 方法接收一个数据批次，包含数据张量和标签张量，调用 forward_net 方法得到分类得分，然后调用 head 模块的 loss 方法计算损失指标。
• val_step 方法与 train_step 方法类似，但是在调用 head 模块的 loss 方法时，设置 valid_mode=True 参数，表示在验证模式下计算损失指标。
• test_step 方法接收一个数据批次，只包含数据张量，调用 forward_net 方法得到分类得分，然后返回分类得分。
• infer_step 方法定义了模型的推理步骤，与 test_step 方法基本相同，但不需要返回预测结果的置信度。

from paddlevideo.utils import get_loggerlogger = get_logger("paddlevideo")

logger = get_logger("paddlevideo") 这行代码定义了一个名为 logger 的日志对象，并使用 get_logger() 函数对其进行初始化。该函数的参数是一个字符串 “paddlevideo”，表示日志对象的名称。

在 Python 中，可以通过 logging 模块来打印日志信息。

• get_logger() 函数是 paddlevideo.utils 包中的一个工具函数，用于获取和配置一个 logger，使得我们能够在程序中输出日志信息。
• 通过 logger 对象可以调用相应的方法（例如 logger.info()），来实现在程序中打印对应级别的日志信息，便于开发者查看和排查问题。

ctrgcn.py

以下是 modeling/backbones/ctrgcn.py 中 ctrgcn.py 的代码，代码详解见：CTR-GCN 代码理解

import math
from ..registry import BACKBONES
import numpy as np
import paddle
import paddle.nn as nn
from .graph_ctrgcn import Graphdef _calculate_fan_in_and_fan_out(tensor):dimensions = tensor.ndimif dimensions < 2:raise ValueError("Fan in and fan out can not be computed for tensor with fewer than 2 dimensions")num_input_fmaps = tensor.shape[1]num_output_fmaps = tensor.shape[0]receptive_field_size = 1if tensor.ndim > 2:for s in tensor.shape[2:]:receptive_field_size *= sfan_in = num_input_fmaps * receptive_field_sizefan_out = num_output_fmaps * receptive_field_sizereturn fan_in, fan_outdef _calculate_correct_fan(tensor, mode):mode = mode.lower()valid_modes = ['fan_in', 'fan_out']if mode not in valid_modes:raise ValueError("Mode {} not supported, please use one of {}".format(mode, valid_modes))fan_in, fan_out = _calculate_fan_in_and_fan_out(tensor)return fan_in if mode == 'fan_in' else fan_outdef kaiming_normal_(tensor, a=0, mode='fan_out', nonlinearity='leaky_relu'):fan = _calculate_correct_fan(tensor, mode)gain = math.sqrt(2.0)std = gain / math.sqrt(fan)with paddle.no_grad():return nn.initializer.Normal(0.0, std)def einsum(x, A):   #'ncuv,nctv->nctu'x = x.transpose((0, 1, 3, 2))y = paddle.matmul(A, x)return ydef conv_branch_init(conv, branches):weight = conv.weightn = weight.shape[0]k1 = weight.shape[1]k2 = weight.shape[2]nn.init.normal_(weight, 0, math.sqrt(2. / (n * k1 * k2 * branches)))nn.init.constant_(conv.bias, 0)def conv_init(conv):if conv.weight is not None:kaiming_normal_(conv.weight, mode='fan_out')(conv.weight)if conv.bias is not None:nn.initializer.Constant(0)(conv.bias)def bn_init(bn, scale):nn.initializer.Constant(scale)(bn.weight)nn.initializer.Constant(0)(bn.bias)def weights_init(m):classname = m.__class__.__name__if classname.find('Conv') != -1:if hasattr(m, 'weight'):kaiming_normal_(m.weight, mode='fan_out')(m.weight)if hasattr(m, 'bias') and m.bias is not None:nn.initializer.Constant(0)(m.bias)elif classname.find('BatchNorm') != -1:if hasattr(m, 'weight') and m.weight is not None:nn.initializer.Normal(1.0, 0.02)(m.weight)if hasattr(m, 'bias') and m.bias is not None:nn.initializer.Constant(0)(m.bias)class TemporalConv(nn.Layer):def __init__(self, in_channels, out_channels, kernel_size, stride=1, dilation=1):super(TemporalConv, self).__init__()pad = (kernel_size + (kernel_size-1) * (dilation-1) - 1) // 2self.conv = nn.Conv2D(in_channels,out_channels,kernel_size=(kernel_size, 1),padding=(pad, 0),stride=(stride, 1),dilation=(dilation, 1))self.bn = nn.BatchNorm2D(out_channels)def forward(self, x):x = self.conv(x)x = self.bn(x)return xclass MultiScale_TemporalConv(nn.Layer):def __init__(self,in_channels,out_channels,kernel_size=3,stride=1,dilations=[1,2,3,4],residual=True,residual_kernel_size=1):super().__init__()assert out_channels % (len(dilations) + 2) == 0, '# out channels should be multiples of # branches'# Multiple branches of temporal convolutionself.num_branches = len(dilations) + 2branch_channels = out_channels // self.num_branchesif type(kernel_size) == list:assert len(kernel_size) == len(dilations)else:kernel_size = [kernel_size]*len(dilations)# Temporal Convolution branchesself.branches = nn.LayerList([nn.Sequential(nn.Conv2D(in_channels,branch_channels,kernel_size=1,padding=0),nn.BatchNorm2D(branch_channels),nn.ReLU(),TemporalConv(branch_channels,branch_channels,kernel_size=ks,stride=stride,dilation=dilation),)for ks, dilation in zip(kernel_size, dilations)])# Additional Max & 1x1 branchself.branches.append(nn.Sequential(nn.Conv2D(in_channels, branch_channels, kernel_size=1, padding=0),nn.BatchNorm2D(branch_channels),nn.ReLU(),nn.MaxPool2D(kernel_size=(3,1), stride=(stride,1), padding=(1,0)),nn.BatchNorm2D(branch_channels)  # 为什么还要加bn))self.branches.append(nn.Sequential(nn.Conv2D(in_channels, branch_channels, kernel_size=1, padding=0, stride=(stride,1)),nn.BatchNorm2D(branch_channels)))# Residual connectionif not residual:self.residual = lambda x: 0elif (in_channels == out_channels) and (stride == 1):self.residual = lambda x: xelse:self.residual = TemporalConv(in_channels, out_channels, kernel_size=residual_kernel_size, stride=stride)# initializeself.apply(weights_init)def forward(self, x):# Input dim: (N,C,T,V)res = self.residual(x)branch_outs = []for tempconv in self.branches:out = tempconv(x)branch_outs.append(out)out = paddle.concat(branch_outs, axis=1)out += resreturn outclass CTRGC(nn.Layer):def __init__(self, in_channels, out_channels, rel_reduction=8, mid_reduction=1):super(CTRGC, self).__init__()self.in_channels = in_channelsself.out_channels = out_channelsif in_channels <= 16:self.rel_channels = 8self.mid_channels = 16else:self.rel_channels = in_channels // rel_reductionself.mid_channels = in_channels // mid_reductionself.conv1 = nn.Conv2D(self.in_channels, self.rel_channels, kernel_size=1)self.conv2 = nn.Conv2D(self.in_channels, self.rel_channels, kernel_size=1)self.conv3 = nn.Conv2D(self.in_channels, self.out_channels, kernel_size=1)self.conv4 = nn.Conv2D(self.rel_channels, self.out_channels, kernel_size=1)self.tanh = nn.Tanh()for m in self.sublayers():if isinstance(m, nn.Conv2D):conv_init(m)elif isinstance(m, nn.BatchNorm2D):bn_init(m, 1)def forward(self, x, A=None, alpha=1):x1, x2, x3 = self.conv1(x).mean(-2), self.conv2(x).mean(-2), self.conv3(x)x1 = self.tanh(x1.unsqueeze(-1) - x2.unsqueeze(-2))x1 = self.conv4(x1) * alpha + (A.unsqueeze(0).unsqueeze(0) if A is not None else 0)  # N,C,V,Vx1 = einsum(x1, x3)return x1class unit_tcn(nn.Layer):def __init__(self, in_channels, out_channels, kernel_size=9, stride=1):super(unit_tcn, self).__init__()pad = int((kernel_size - 1) / 2)self.conv = nn.Conv2D(in_channels, out_channels, kernel_size=(kernel_size, 1), padding=(pad, 0),stride=(stride, 1))self.bn = nn.BatchNorm2D(out_channels)self.relu = nn.ReLU()conv_init(self.conv)bn_init(self.bn, 1)def forward(self, x):x = self.bn(self.conv(x))return xclass unit_gcn(nn.Layer):def __init__(self, in_channels, out_channels, A, coff_embedding=4, adaptive=True, residual=True):super(unit_gcn, self).__init__()inter_channels = out_channels // coff_embeddingself.inter_c = inter_channelsself.out_c = out_channelsself.in_c = in_channelsself.adaptive = adaptiveself.num_subset = A.shape[0]self.convs = nn.LayerList()for i in range(self.num_subset):self.convs.append(CTRGC(in_channels, out_channels))if residual:if in_channels != out_channels:self.down = nn.Sequential(nn.Conv2D(in_channels, out_channels, 1),nn.BatchNorm2D(out_channels))else:self.down = lambda x: xelse:self.down = lambda x: 0if self.adaptive:self.PA = paddle.static.create_parameter(A.shape, 'float32', default_initializer=nn.initializer.Assign(paddle.to_tensor(A.astype(np.float32), stop_gradient=False)))else:self.A = paddle.to_tensor(A.astype(np.float32), stop_gradient=True)self.alpha = paddle.static.create_parameter([1], 'float32', default_initializer=nn.initializer.Assign(paddle.to_tensor(paddle.zeros(shape=[1]), stop_gradient=False)))self.bn = nn.BatchNorm2D(out_channels)self.soft = nn.Softmax(axis=-2)self.relu = nn.ReLU()for m in self.sublayers():if isinstance(m, nn.Conv2D):conv_init(m)elif isinstance(m, nn.BatchNorm2D):bn_init(m, 1)bn_init(self.bn, 1e-6)def forward(self, x):y = Noneif self.adaptive:A = self.PAelse:A = self.Afor i in range(self.num_subset):z = self.convs[i](x, A[i], self.alpha)y = z + y if y is not None else zy = self.bn(y)y += self.down(x)y = self.relu(y)return yclass TCN_GCN_unit(nn.Layer):def __init__(self, in_channels, out_channels, A, stride=1, residual=True, adaptive=True, kernel_size=5, dilations=[1,2]):super(TCN_GCN_unit, self).__init__()self.gcn1 = unit_gcn(in_channels, out_channels, A, adaptive=adaptive)self.tcn1 = MultiScale_TemporalConv(out_channels, out_channels, kernel_size=kernel_size, stride=stride, dilations=dilations,residual=False)self.relu = nn.ReLU()if not residual:self.residual = lambda x: 0elif (in_channels == out_channels) and (stride == 1):self.residual = lambda x: xelse:self.residual = unit_tcn(in_channels, out_channels, kernel_size=1, stride=stride)def forward(self, x):y = self.relu(self.tcn1(self.gcn1(x)) + self.residual(x))return y@BACKBONES.register()
class CTRGCN(nn.Layer):def __init__(self, in_channels=2, num_class=30, num_point=25, num_person=1, drop_out=0, adaptive=True):super(CTRGCN, self).__init__()self.graph = Graph()A = self.graph.A      # 3,25,25self.num_class = num_classself.num_point = num_pointself.data_bn = nn.BatchNorm1D(num_person * in_channels * num_point)base_channel = 64self.l1 = TCN_GCN_unit(in_channels, base_channel, A, residual=False, adaptive=adaptive)self.l2 = TCN_GCN_unit(base_channel, base_channel, A, adaptive=adaptive)self.l3 = TCN_GCN_unit(base_channel, base_channel, A, adaptive=adaptive)self.l4 = TCN_GCN_unit(base_channel, base_channel, A, adaptive=adaptive)self.l5 = TCN_GCN_unit(base_channel, base_channel*2, A, stride=2, adaptive=adaptive)self.l6 = TCN_GCN_unit(base_channel*2, base_channel*2, A, adaptive=adaptive)self.l7 = TCN_GCN_unit(base_channel*2, base_channel*2, A, adaptive=adaptive)self.l8 = TCN_GCN_unit(base_channel*2, base_channel*4, A, stride=2, adaptive=adaptive)self.l9 = TCN_GCN_unit(base_channel*4, base_channel*4, A, adaptive=adaptive)self.l10 = TCN_GCN_unit(base_channel*4, base_channel*4, A, adaptive=adaptive)self.fc = nn.Linear(base_channel*4, num_class, weight_attr=nn.initializer.Normal(0, math.sqrt(2. / num_class)))bn_init(self.data_bn, 1)if drop_out:self.drop_out = nn.Dropout(drop_out)else:self.drop_out = lambda x: xdef forward(self, x):x.stop_gradient = Falseif len(x.shape) == 3:N, T, VC = x.shapex = x.reshape((N, T, self.num_point, -1))x = x.transpose((0, 3, 1, 2)).unsqueeze(-1)N, C, T, V, M = x.shapex = x.transpose((0, 4, 3, 1, 2))x = x.reshape((N, M * V * C, T))x = self.data_bn(x)x = x.reshape((N, M, V, C, T))x = x.transpose((0, 1, 3, 4, 2))x = x.reshape((N * M, C, T, V))x = self.l1(x)x = self.l2(x)x = self.l3(x)x = self.l4(x)x = self.l5(x)x = self.l6(x)x = self.l7(x)x = self.l8(x)x = self.l9(x)x = self.l10(x)# N*M,C,T,Vc_new = x.shape[1]x = x.reshape((N, M, c_new, -1))x = x.mean(3).mean(1)x = self.drop_out(x)return self.fc(x)

graph_ctrgcn.py

文件路径：modeling/backbones/graph_ctrgcn.py

import numpy as np
from . import tools_ctrgcnnum_node = 25
self_link = [(i, i) for i in range(num_node)]inward_ori_index = [(2, 1), (3, 2), (4, 3), (5, 1), (6, 5), (7, 6),(1, 8), (9, 8), (10, 9), (11, 10), (24, 11), (22, 11), (23, 22),(12, 8), (13, 12), (14, 13), (21, 14), (19, 14), (20, 19),(0, 1), (17, 15), (15, 0), (16, 0), (18, 16)]
inward = [(i, j) for (i, j) in inward_ori_index]
outward = [(j, i) for (i, j) in inward]
neighbor = inward + outwardnum_node_1 = 11
indices_1 = [8, 0, 6, 7, 3, 4, 13, 19, 10, 22, 1]
self_link_1 = [(i, i) for i in range(num_node_1)]
inward_ori_index_1 = [(1, 11), (2, 11), (3, 11), (4, 3), (5, 11), (6, 5), (7, 1), (8, 7), (9, 1), (10, 9)]
inward_1 = [(i - 1, j - 1) for (i, j) in inward_ori_index_1]
outward_1 = [(j, i) for (i, j) in inward_1]
neighbor_1 = inward_1 + outward_1num_node_2 = 5
indices_2 = [3, 5, 6, 8, 10]
self_link_2 = [(i ,i) for i in range(num_node_2)]
inward_ori_index_2 = [(0, 4), (1, 4), (2, 4), (3, 4), (0, 1), (2, 3)]
inward_2 = [(i - 1, j - 1) for (i, j) in inward_ori_index_2]
outward_2 = [(j, i) for (i, j) in inward_2]
neighbor_2 = inward_2 + outward_2class Graph:def __init__(self, labeling_mode='spatial', scale=1):self.num_node = num_nodeself.self_link = self_linkself.inward = inwardself.outward = outwardself.neighbor = neighborself.A = self.get_adjacency_matrix(labeling_mode)self.A1 = tools_ctrgcn.get_spatial_graph(num_node_1, self_link_1, inward_1, outward_1)self.A2 = tools_ctrgcn.get_spatial_graph(num_node_2, self_link_2, inward_2, outward_2)self.A_binary = tools_ctrgcn.edge2mat(neighbor, num_node)self.A_norm = tools_ctrgcn.normalize_adjacency_matrix(self.A_binary + 2*np.eye(num_node))self.A_binary_K = tools_ctrgcn.get_k_scale_graph(scale, self.A_binary)self.A_A1 = ((self.A_binary + np.eye(num_node)) / np.sum(self.A_binary + np.eye(self.A_binary.shape[0]), axis=1, keepdims=True))[indices_1]self.A1_A2 = tools_ctrgcn.edge2mat(neighbor_1, num_node_1) + np.eye(num_node_1)self.A1_A2 = (self.A1_A2 / np.sum(self.A1_A2, axis=1, keepdims=True))[indices_2]def get_adjacency_matrix(self, labeling_mode=None):if labeling_mode is None:return self.Aif labeling_mode == 'spatial':A = tools_ctrgcn.get_spatial_graph(num_node, self_link, inward, outward)else:raise ValueError()return A

tools_ctrgcn.py

文件路径：modeling/backbones/tools_ctrgcn.py

import numpy as npdef get_sgp_mat(num_in, num_out, link):A = np.zeros((num_in, num_out))for i, j in link:A[i, j] = 1A_norm = A / np.sum(A, axis=0, keepdims=True)return A_normdef edge2mat(link, num_node):A = np.zeros((num_node, num_node))for i, j in link:A[j, i] = 1return Adef get_k_scale_graph(scale, A):if scale == 1:return AAn = np.zeros_like(A)A_power = np.eye(A.shape[0])for k in range(scale):A_power = A_power @ AAn += A_powerAn[An > 0] = 1return Andef normalize_digraph(A):Dl = np.sum(A, 0)h, w = A.shapeDn = np.zeros((w, w))for i in range(w):if Dl[i] > 0:Dn[i, i] = Dl[i] ** (-1)AD = np.dot(A, Dn)return ADdef get_spatial_graph(num_node, self_link, inward, outward):I = edge2mat(self_link, num_node)In = normalize_digraph(edge2mat(inward, num_node))Out = normalize_digraph(edge2mat(outward, num_node))A = np.stack((I, In, Out))return Adef normalize_adjacency_matrix(A):node_degrees = A.sum(-1)degs_inv_sqrt = np.power(node_degrees, -0.5)norm_degs_matrix = np.eye(len(node_degrees)) * degs_inv_sqrtreturn (norm_degs_matrix @ A @ norm_degs_matrix).astype(np.float32)def k_adjacency(A, k, with_self=False, self_factor=1):assert isinstance(A, np.ndarray)I = np.eye(len(A), dtype=A.dtype)if k == 0:return IAk = np.minimum(np.linalg.matrix_power(A + I, k), 1) \- np.minimum(np.linalg.matrix_power(A + I, k - 1), 1)if with_self:Ak += (self_factor * I)return Akdef get_multiscale_spatial_graph(num_node, self_link, inward, outward):I = edge2mat(self_link, num_node)A1 = edge2mat(inward, num_node)A2 = edge2mat(outward, num_node)A3 = k_adjacency(A1, 2)A4 = k_adjacency(A2, 2)A1 = normalize_digraph(A1)A2 = normalize_digraph(A2)A3 = normalize_digraph(A3)A4 = normalize_digraph(A4)A = np.stack((I, A1, A2, A3, A4))return Adef get_uniform_graph(num_node, self_link, neighbor):A = normalize_digraph(edge2mat(neighbor + self_link, num_node))return A

ctrgcn_head.py

以下是 modeling/heads/ctrgcn_head.py 路径下 ctrgcn_head.py 的源代码：

import paddle
import paddle.nn as nnfrom .base import BaseHead
from ..registry import HEADS
from ..weight_init import weight_init_@HEADS.register()
class CTRGCNHead(BaseHead):"""Head for ST-GCN model.Args:in_channels: int, input feature channels. Default: 256.num_classes: int, number classes. Default: 10."""def __init__(self, in_channels=256, num_classes=10, **kwargs):super().__init__(num_classes, in_channels, **kwargs)def forward(self, x):"""Define how the head is going to run."""return x

cross_entropy_loss.py

文件路径：modeling/losses/cross_entropy_loss.py

import paddle
import paddle.nn.functional as Ffrom ..registry import LOSSES
from .base import BaseWeightedLoss@LOSSES.register()
class CrossEntropyLoss(BaseWeightedLoss):"""Cross Entropy Loss."""def _forward(self, score, labels, **kwargs):"""Forward function.Args:score (paddle.Tensor): The class score.labels (paddle.Tensor): The ground truth labels.kwargs: Any keyword argument to be used to calculateCrossEntropy loss.Returns:loss (paddle.Tensor): The returned CrossEntropy loss."""loss = F.cross_entropy(score, labels, **kwargs)return loss