该项目见飞桨：

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 文件夹

__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()

MODEL: framework: "RecognizerGCN" backbone: name: "CTRGCN" in_channels: 9

配置文件 – MODEL

接下来是由配置文件中的配置引发的各种源码。

MODEL: framework: "RecognizerGCN" backbone: name: "CTRGCN" in_channels: 2head:name: "CTRGCNHead" num_classes: 30 ls_eps: 0.1

RecognizerGCN

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

以下是 PaddleVideo 中 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

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

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

CTRGCNHead

以下是 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