CustomRegion 工具介绍#

章节目标#

明确 CustomRegion 训练脚本和推理脚本分别需要实现哪些 Python 接口。
明确如何在训练脚本中读取类别、用户参数、二进制参数、图像、视图和标注。
明确如何在推理脚本中读取模型、处理输入图像和视图，并返回区域检测结果。
掌握进度回调、设备分配和 BinaryPacks 模型传递的基本写法。
了解脚本 print 输出的去向，以及如何在 C++ 集成程序中为脚本输出设置自定义重定向 sink。

适用场景#

CustomRegion 用于接入用户自定义的区域检测算法。用户只需要提供训练脚本和推理脚本，就可以把自己的 Python 算法接入 VisionFlow 的数据、标注、训练、模型保存和推理流程。

典型场景包括：

用 PyTorch、TensorFlow、ONNXRuntime、OpenCV 等库实现自定义检测算法。
快速验证新的区域检测思路。
复用已有第三方模型，并把输出转换成 VisionFlow 的区域结果。

Note

CustomRegion 脚本中调用的 visionflow Python 接口大多直接对应 VisionFlow C++ 接口。阅读本项目 C++ 算法示例时，通常可以将 visionflow:: 替换为 vflow.，将 :: 命名空间分隔符替换为 . 来理解 Python 调用。例如 visionflow::props::PolygonRegionList 对应 vflow.props.PolygonRegionList。

整体流程#

用户实现 CustomRegion 算法时，通常只需要关心以下流程：

训练脚本
  读取类别、用户参数、训练二进制参数
  读取图像、视图、标注
  训练自定义模型
  返回 BinaryPacks

       |
       v

推理脚本
  读取 BinaryPacks 模型
  读取当前图像、视图、图像信息
  执行自定义推理
  返回 PolygonRegionList

最小可运行接入步骤#

第一次接入时，建议先按下面步骤跑通空训练和空推理，再替换成真实算法：

准备 VisionFlow 运行环境中的 Python 依赖。脚本会在 VisionFlow 的 Python 环境中执行，因此 PyTorch、ONNXRuntime、OpenCV 等第三方库需要安装到同一个环境里。
在 CustomRegion 的训练参数中填写训练脚本。脚本必须包含 CustomConfigurator 类，并保证 execute 返回 vflow.param.BinaryPacks。
在 CustomRegion 的推理参数中填写推理脚本。脚本必须包含 CustomOperator 类，并保证 execute 返回 vflow.props.PolygonRegionList。
配置类别列表。PolygonRegion.name 应与这里配置的类别名一致，背景类不需要写入类别列表。
按需配置普通参数和训练二进制输入参数。普通参数通过 VFLOW_USER_VARS 读取；预训练模型、模板数据等二进制输入通过 svc.input_parameters()[2] 读取。
标注训练视图和区域标注，启动训练。训练脚本返回的 BinaryPacks 会作为训练产物保存，并传给后续推理脚本。
运行推理并检查输出区域。如果没有结果，先确认推理视图是否为空、模型 key 是否存在、阈值是否过高，以及返回 polygon 是否在原图坐标系下。

最小空脚本#

先用下面两段脚本确认 CustomRegion 工具、类别配置和脚本运行环境能正常工作。它们不会训练真实模型，也不会输出区域结果；跑通后再逐步加入参数读取、训练数据遍历和推理逻辑。

训练脚本：

from typing import Callable, List

import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

class CustomConfigurator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        svc: vflow.confs.IService,
    ) -> None:
        pass

    def execute(self, svc: vflow.confs.IService) -> vflow.param.BinaryPacks:
        return vflow.param.BinaryPacks()

推理脚本：

from typing import Callable, List

import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

class CustomOperator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        model: vflow.param.BinaryPacks,
    ) -> None:
        pass

    def execute(
        self,
        image: vflow.props.Image,
        views: vflow.props.ViewList,
        img_info: vflow.props.RawImageInfo,
    ) -> vflow.props.PolygonRegionList:
        return vflow.props.PolygonRegionList()

接入前核心概念#

后续代码块中的 type hint 用于说明接口对象类型，也便于 IDE 补全和静态检查；VisionFlow 调用脚本时仍以类名、方法名、参数数量和参数顺序为准。

脚本生命周期#

训练脚本和推理脚本的生命周期不同，建议按下面方式组织代码：

CustomConfigurator.__init__：每次启动训练任务时调用一次。适合读取类别、用户参数、训练二进制输入、申请设备和初始化训练对象。
CustomConfigurator.execute：每次训练任务调用一次。适合遍历训练集、执行训练、上报进度并返回模型 BinaryPacks。
CustomOperator.__init__：推理参数和模型加载时调用一次。适合读取推理参数、申请设备、反序列化模型、构建 ONNXRuntime session 或 PyTorch 模型。
CustomOperator.execute：每次推理一个样本时调用。CustomOperator 对象会被复用，因此模型、类别名、阈值等只读状态应放在 __init__ 中；当前样本的中间结果应放在 execute 的局部变量中。

脚本可以保存成员状态，但不要依赖样本调用顺序。若在 CustomOperator 中维护可变缓存、计数器或复用输入输出缓冲区，需要保证每次 execute 前后状态一致；如果运行环境并发调用同一个对象，还需要自行保证线程安全。

Python 运行环境#

CustomRegion 脚本运行在 VisionFlow 内嵌的 Python 环境中。训练脚本和推理脚本使用同一套 VisionFlow Python 绑定和第三方依赖环境；具体 Python 版本随 VisionFlow 发布包而定，调试时可以在脚本中打印：

import sys
print("python:", sys.version)

使用第三方库时，应将依赖安装到 VisionFlow 实际使用的 Python 环境中，并分别确认训练脚本和推理脚本都可以 import 成功。print 输出会进入 VisionFlow 脚本日志，具体去向由集成 VisionFlow 的宿主程序配置（详见脚本 print 输出与重定向）；脚本语法错误、运行时异常和用户变量类型转换错误会作为任务错误呈现，并带有 Python 异常信息。

脚本内容、普通参数、二进制输入数据和训练输出模型都应由 VisionFlow 工程维护。训练前需要传入的预训练模型、模板数据、字典数据等，应通过训练二进制输入参数 BinaryPacks 注入；训练后需要保存给推理使用的数据，应通过 execute 返回的 BinaryPacks 写回工程。脚本与 VisionFlow 之间的数据交换应通过 VFLOW_USER_VARS、svc.input_parameters()、svc.property_sets() 和返回值完成。

视图是什么#

view 表示原图上的一个待训练或待推理区域。它包含两个关键信息：视图子图的尺寸，以及从原图坐标到视图子图坐标的变换矩阵。通过 vflow.img.transform(raw_image, view).image() 可以得到该视图对应的子图。

一张样本图可以有多个 view。view 可能来自前序工具的定位或检测结果，也可能来自人工标注或项目中已有的 ROI 配置。训练时，脚本通常只使用带有 vflow.kTrain 标记的 view 生成训练样本；推理时，一般对传入的全部 views 执行推理。

如果传入的 views 为空，可以根据算法约定返回空结果，也可以在推理逻辑中按整图推理。关键是训练和推理对 view 的处理策略保持一致。

核心对象速查#

对象	常见用法	返回或行为
`views_set`	`for sample_id, views in views_set`	遍历样本 ID 和对应视图集合；`views` 可能为 `None`。
`views`	`views.tagged_views(vflow.kTrain)`、`for view_id, view in views`	返回指定集合标记下的视图集合，或遍历全部视图；集合可能为空。
`view`	`view.size()`、`view.transform_matrix()`、`view.view_on_raw_image()`	读取视图尺寸、原图到视图子图的变换矩阵、视图在原图上的四边形范围。
`image_set`	`image_set.at(sample_id)`	返回图像属性对象；图像缺失时可能为 `None`。
`image_prop`	`image_prop.image()`	返回 `vflow.img.Image`。
`vflow.img.Image`	`image.to_numpy()`、`image.width()`、`image.height()`、`image.channels()`	转成 numpy 数组，或读取宽、高、通道数。
`label_set`	`label_set.at(sample_id)`	返回整图标注 `vflow.props.PolygonRegionList`；未标注时可能为 `None` 或空列表。
`PolygonRegionList`	`for region_id, region in labels`、`labels.size()`、`result.add(region)`	遍历、计数或追加区域。
`PolygonRegion`	`region.polygon()`、`region.name()`、`region.score()`	读取区域多边形、类别名和置信度；构造结果时使用 `set_polygon`、`set_name`、`set_score`。
`BinaryPacks`	`contains(key)`、`get(key).to_bytes()`、`insert(key, vflow.Buffer.FromBytes(data))`	按 key 读写模型二进制数据。

训练脚本接口#

训练脚本必须定义 CustomConfigurator 类。类名、方法名、参数数量和参数顺序都不能修改。

from typing import Callable, List

import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

class CustomConfigurator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        svc: vflow.confs.IService,
    ) -> None:
        pass

    def execute(self, svc: vflow.confs.IService) -> vflow.param.BinaryPacks:
        model = vflow.param.BinaryPacks()
        return model

__init__: 训练对象初始化时调用一次。适合读取用户参数、申请设备、初始化模型结构、读取二进制输入参数。
execute: 训练任务执行时调用。适合遍历训练数据、执行训练、上报进度、保存模型。必须返回 vflow.param.BinaryPacks，该返回值就是后续推理脚本收到的模型。
device_allocator: 设备分配函数，类型可写为 Callable[[int, int], List[int]]，调用方式为 device_allocator(require_num, require_memory_size) -> List[int]。返回 GPU 设备 ID 列表；使用 CPU 时通常返回 [-1]。
svc: 训练服务对象。训练脚本通过它读取输入参数、训练数据、上下文并上报进度。

读取训练参数#

用户变量#

脚本执行前，VisionFlow 会将用户配置的普通参数注入到全局字典 VFLOW_USER_VARS。普通参数在工程数据中以 type 和 value 保存，其中 value 是字符串；脚本执行前会由 VisionFlow 按 type 转成 Python 对象：

epochs = int(VFLOW_USER_VARS.get("Epoch", 5))
learning_rate = float(VFLOW_USER_VARS.get("LearningRate", 0.001))
threshold = float(VFLOW_USER_VARS.get("ScoreThreshold", 0.5))
use_aug = VFLOW_USER_VARS.get("UseAugmentation", False)

常见类型对应关系：

参数类型	Python 类型	说明
`string`	`str`	字符串。
`number`	`float`	数值。需要整数时使用 `int(...)`。
`bool`	`bool`	布尔值。字符串 `"true"`、`"1"` 转为 `True`；`"false"`、`"0"` 转为 `False`。

Warning

不要使用 bool(VFLOW_USER_VARS.get("UseAugmentation", False)) 解析布尔参数。如果拿到的是字符串，Python 中 bool("False") 会得到 True。正常由 VisionFlow 注入的 bool 类型参数可以直接读取；如果脚本还需要兼容手工调试时传入的字符串，可以使用下面的安全写法：

def read_bool(name: str, default: bool = False) -> bool:
    value = VFLOW_USER_VARS.get(name, default)
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in ("true", "1", "yes", "y")
    return bool(value)

use_aug = read_bool("UseAugmentation", False)

类别和二进制参数#

训练脚本通过 svc.input_parameters() 读取类别和二进制输入参数。返回值顺序固定：

label_classes, _, binary_args = svc.input_parameters()

class_names = list(label_classes.get())

if binary_args.contains("pretrained_model"):
    pretrained_bytes = binary_args.get("pretrained_model").to_bytes()

label_classes: 类型为 vflow.param.LabelClasses，用于读取用户配置的类别名。
binary_args: 类型为 vflow.param.BinaryPacks，用于读取训练前由 VisionFlow 工程传入的二进制数据，例如预训练模型、模板数据、字典数据等。

读取训练数据#

训练脚本通过 svc.property_sets() 读取训练数据。返回值顺序固定：

image_set, views_set, img_info_set, label_set = svc.property_sets()

这些 PropertySet 按 sample_id 对齐，但不代表每个样本上的属性对象都一定存在。实际工程中，图像数据丢失、视图未生成、样本未标注等情况都可能让 image_set.at(sample_id)、views_set 中的 views、label_set.at(sample_id) 等返回 None。脚本读取后应先检查再使用；未标注样本通常可以按空标注处理，图像或视图缺失的样本通常应跳过。

各对象含义如下：

对象	含义
`image_set`	原图集合。通过 `image_set.at(sample_id).image()` 读取原图；如果对应图像缺失，`image_set.at(sample_id)` 可能为 `None`。
`views_set`	视图集合。每张图可以有多个视图，视图带有训练集、测试集或未知集合标记；未生成视图时，对应 `views` 可能为 `None` 或空列表。
`img_info_set`	原图信息集合，例如图像尺寸等；图像信息缺失时，对应对象也可能为 `None`。
`label_set`	区域标注集合，类型为 `vflow.props.PolygonRegionList`；未标注样本可能返回 `None`，也可能返回空列表。

推荐以 views_set 为主循环，再用同一个 sample_id 到其他集合读取对应数据。读取后先处理空对象，再进入图像裁剪和标注转换：

for sample_id, views in views_set:
    if views is None:
        continue

    train_views = views.tagged_views(vflow.kTrain)
    if train_views.empty():
        continue

    image_prop = image_set.at(sample_id)
    if image_prop is None:
        continue

    raw_image = image_prop.image()
    labels = label_set.at(sample_id)
    if labels is None:
        labels = vflow.props.PolygonRegionList()

    for view_id, view in train_views:
        sub_image = vflow.img.transform(raw_image, view).image()
        image_np = sub_image.to_numpy()

训练、测试和未知集合#

视图可带有集合标记：

vflow.kTrain：训练集。
vflow.kTest：测试集。
vflow.kUnknown：未知集合。

训练脚本通常只使用 vflow.kTrain：

train_views = views.tagged_views(vflow.kTrain)

如果算法需要在训练过程中评估测试集，也可以读取 vflow.kTest：

test_views = views.tagged_views(vflow.kTest)

图像、视图与坐标系#

读取原图前先确认 image_set.at(sample_id) 不是 None，再通过 image_set.at(sample_id).image() 得到 raw_image。 CustomRegion 中常见对象的坐标系如下：

对象	坐标系	说明
`raw_image`	原图坐标系	左上角通常为 `(0, 0)`。
`view`	定义在原图上的视图区域	`view.view_on_raw_image()` 或 `view.bounding_ring()` 表示该视图在原图上的四边形范围。
`sub_image`	视图子图坐标系	由 `vflow.img.transform(raw_image, view).image()` 得到，子图左上角为 `(0, 0)`。
`label_set.at(sample_id)`	原图坐标系	区域标注是整张原图上的标注，不会自动按当前 `view` 筛选或转换。

将原图裁剪成视图区域中的子图：

sub_image = vflow.img.transform(raw_image, view).image()
image_np = sub_image.to_numpy()

to_numpy() 返回的 numpy 数组可以直接交给 OpenCV、PyTorch 等库处理。

图像数组格式#

vflow.img.Image.to_numpy() 返回 3 维 numpy 数组，形状为 (height, width, channels)，即 HWC 排布。灰度图的形状通常是 (H, W, 1)，不是 (H, W)。数组 dtype 与 VisionFlow 图像深度一致，常见为 uint8，也可能是 uint16 或 float32。

to_numpy() 不会自动做模型预处理。接入 OpenCV、PyTorch 或 ONNXRuntime 时，需要按模型要求显式处理：

image_np = sub_image.to_numpy()
print("image:", image_np.shape, image_np.dtype)

# 示例：HWC uint8 -> NCHW float32
model_input = image_np.astype(np.float32) / 255.0
model_input = np.transpose(model_input, (2, 0, 1))
model_input = np.expand_dims(model_input, axis=0)

如果模型要求 RGB、BGR、灰度或固定通道数，需要在脚本中自行转换。不要假设 to_numpy() 已经完成颜色空间转换、归一化、resize、padding 或 CHW 转换。

如果真实检测算法以视图子图为训练样本，通常需要先筛选与该 view 相交的标注，再把标注从原图坐标系转换到视图子图坐标系：

def labels_in_view(
    labels,
    view: vflow.View,
    min_intersection_area: float = 1.0,
) -> list:
    labels_on_view = []
    if labels is None:
        return labels_on_view

    for region_id, region in labels:
        polygon_on_raw = region.polygon()
        clipped_on_raw = vflow.geometry.intersection(
            view.view_on_raw_image(),
            polygon_on_raw,
        )
        if clipped_on_raw.area() < min_intersection_area:
            continue

        polygon_on_view = vflow.geometry.transform(
            clipped_on_raw,
            view.transform_matrix(),
        )
        labels_on_view.append({
            "name": region.name(),
            "polygon": polygon_on_view,
        })
    return labels_on_view

view.transform_matrix() 用于从原图坐标转换到视图子图坐标。反过来，如果算法在视图子图坐标系下输出多边形，推理返回前应转换回原图坐标系：

raw_transform = vflow.geometry.get_inverse_transform(view.transform_matrix())
polygon_on_raw_image = vflow.geometry.transform(polygon_in_view, raw_transform)

标注数据#

label_set.at(sample_id) 返回 vflow.props.PolygonRegionList。可以遍历其中每个区域：

labels = label_set.at(sample_id)
if labels is None:
    labels = vflow.props.PolygonRegionList()

for region_id, region in labels:
    label_name = region.name()
    polygon = region.polygon()

label_set.at(sample_id) 是当前样本的整图标注集合。遍历某个 train_view 时，不应直接用 labels.size() > 0 判断该 view 是否为正样本；一张图有目标，不代表每个 view 都包含目标。真实检测算法应按 view 与标注 polygon 的相交关系生成该 view 对应的训练标签，并按需要把相交区域转换到视图子图坐标系。

对于区域检测算法，空标注通常可以作为无目标负样本。只接受正样本的算法，应显式跳过当前 view 内无有效目标的样本。

进度回调#

训练过程中使用 svc.on_progress(total, progress, message) 上报进度，并接收控制信号。

total: 总进度步数。不知道总步数时可以传 0。
progress: 当前进度。推荐保持 0 <= progress <= total。
message: 进度信息字符串。推荐传 JSON 字符串，并至少包含 Message 字段。常用字段包括 Message、Epoch、Step、Loss、LearningRate、Metric 等。字段名不是强制协议，但保持稳定有利于前端展示和问题排查。
返回值: vflow.TaskCtrlSignal。如果不是 vflow.TaskCtrlSignal.kContinue，脚本应尽快停止训练。

示例：

import json
import visionflow as vflow

ctrl_signal = svc.on_progress(
    total_steps,
    current_step,
    json.dumps({
        "Message": "Training...",
        "Loss": float(loss),
    }),
)
if ctrl_signal != vflow.TaskCtrlSignal.kContinue:
    return self.export_model()

训练中断时，如果已经有可用模型，可以返回当前模型；如果模型尚不可用，应抛出异常并让任务失败，不要返回结构不完整的 BinaryPacks。脚本中的 print 会被 VisionFlow 重定向到脚本日志输出（详见脚本 print 输出与重定向）；Python 语法错误、运行时异常和用户变量类型转换失败会作为任务错误呈现，并包含脚本行号或异常信息。

保存模型 BinaryPacks#

训练脚本必须返回 vflow.param.BinaryPacks。BinaryPacks 是按 key 存放二进制数据的容器。

常用接口：

Python 调用	作用
`packs.insert(key, buffer)`	写入二进制数据。相同 key 会覆盖旧数据。
`packs.contains(key)`	判断 key 是否存在。
`packs.get(key)`	读取 key 对应的 `vflow.Buffer`。
`packs.keys()`	返回全部 key。

vflow.Buffer 与 Python bytes 互转：

packs = vflow.param.BinaryPacks()
packs.insert("model_info", vflow.Buffer.FromBytes(b'{"version": 1}'))

model_info_bytes = packs.get("model_info").to_bytes()

Note

vflow.Buffer.FromBytes(data) 默认会复制一份数据，返回的 Buffer 持有自己的内存，因此可以放心写进 BinaryPacks、返回给 C++，或交给 svc.output_parameters(...)。

它还有一个零拷贝模式 vflow.Buffer.FromBytes(data, copy=False)，返回的 Buffer 只是借用 data 的内存，有效期仅限于这个 Python 对象本身。这种 Buffer 不能跨出当前 Python 作用域——一旦写进 BinaryPacks 或返回给 C++，脚本结束后 C++ 侧读到的就是已经释放的内存，轻则模型数据损坏，重则进程段错误崩溃。

# 正确：默认复制，Buffer 持有自己的内存
packs.insert("torch_model", vflow.Buffer.FromBytes(io_buffer.getvalue()))

# 错误：copy=False 只借用内存，存进 packs 后留下悬垂指针
packs.insert("torch_model",
             vflow.Buffer.FromBytes(io_buffer.getvalue(), copy=False))

只有当数据在当前函数内读完即弃、并且确实需要避免大块复制时，才使用 copy=False。

推荐至少保存：

模型权重或推理模型数据，例如 model。
模型元信息，例如 model_info，可保存版本、类别名、输入尺寸等。

工程约定建议：

key 使用稳定的英文名称，例如 torch_model、onnx_model、label_map、model_info。不要在 key 中放运行时间或随机数。
model_info 建议使用 JSON 保存，至少包含 format_version、class_names、input_size、score_threshold、created_by 等字段，方便后续兼容升级。
多段模型数据可以拆分为多个 key，例如 model.onnx、preprocess.json、label_map.json；推理脚本加载时逐个检查。
缺少必要 key 时应立即抛出明确异常，避免推理阶段返回空结果掩盖模型配置错误。
大模型会随工程数据保存和加载。真实项目中应尽量保存推理所需的最小模型数据，例如 ONNX、TorchScript 或压缩后的权重；如果必须保存多个大块二进制数据，需要评估工程体积、加载时间和版本兼容成本。

示例：

import json

MODEL_KEY = "torch_model"
INFO_KEY = "model_info"

packs = vflow.param.BinaryPacks()
packs.insert(MODEL_KEY, vflow.Buffer.FromBytes(model_bytes))
packs.insert(INFO_KEY, vflow.Buffer.FromBytes(json.dumps({
    "format_version": 1,
    "class_names": class_names,
    "input_size": [640, 640],
    "score_threshold": 0.5,
    "created_by": "custom_region_torch_example",
}).encode("utf-8")))

if not packs.contains(MODEL_KEY):
    raise RuntimeError(f"Missing required model key: {MODEL_KEY}")

推理脚本接口#

推理脚本必须定义 CustomOperator 类。类名、方法名、参数数量和参数顺序都不能修改。

from typing import Callable, List

import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

class CustomOperator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        model: vflow.param.BinaryPacks,
    ) -> None:
        pass

    def execute(
        self,
        image: vflow.props.Image,
        views: vflow.props.ViewList,
        img_info: vflow.props.RawImageInfo,
    ) -> vflow.props.PolygonRegionList:
        return vflow.props.PolygonRegionList()

__init__: 推理对象初始化时调用一次。适合读取用户参数、申请设备、从 model 中加载权重。
model: 训练脚本返回的 vflow.param.BinaryPacks。
execute: 每次推理时调用。输入分别是当前图像、当前视图列表和当前图像信息。必须返回 vflow.props.PolygonRegionList。

加载模型#

推理脚本只通过 CustomOperator.__init__(device_allocator, model) 接收训练产物。model 是训练脚本 execute 返回并由 VisionFlow 工程保存的 vflow.param.BinaryPacks。推理脚本应在初始化阶段检查必要 key，完成模型反序列化和只读状态初始化：

import json
from typing import Callable, List

import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

class CustomOperator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        model: vflow.param.BinaryPacks,
    ) -> None:
        if not model.contains("model_info"):
            raise RuntimeError("Missing required model key: model_info")
        if not model.contains("onnx_model"):
            raise RuntimeError("Missing required model key: onnx_model")

        self.model_info = json.loads(
            model.get("model_info").to_bytes().decode("utf-8")
        )
        self.model_bytes = model.get("onnx_model").to_bytes()
        self.class_names = self.model_info["class_names"]
        self.score_threshold = float(VFLOW_USER_VARS.get("ScoreThreshold", 0.5))

如果模型缺少必要 key，应直接抛出明确异常。不要在模型不完整时返回空结果，否则排查时很难区分“没有检出目标”和“模型没有正确加载”。

读取推理输入#

推理脚本的 execute 会收到三个参数：

参数	含义
`image`	当前样本原图属性。通过 `image.image()` 读取原图。
`views`	当前样本视图列表。
`img_info`	当前样本图像信息。

image、views、img_info 都来自 VisionFlow 当前工程数据。图像数据缺失或视图未生成时，对应对象可能为 None 或空集合；推理脚本应先检查，再进入模型推理。

遍历 views 并执行模型#

推理时通常对传入的全部 views 执行推理，不应默认只推理 kTest 或其他某一种集合标记下的视图。只有算法自身明确要求忽略部分视图时，才由脚本主动筛选。

如果 views 为空，脚本可以按算法约定直接返回空结果；如果算法希望没有视图时按整图推理，可以构造覆盖整张图的 view：

def full_image_view(raw_image: vflow.img.Image) -> vflow.View:
    return vflow.View(
        vflow.geometry.Matrix3f(1),
        vflow.geometry.Size2f(raw_image.width(), raw_image.height()),
    )

def execute(
    self,
    image: vflow.props.Image,
    views: vflow.props.ViewList,
    img_info: vflow.props.RawImageInfo,
) -> vflow.props.PolygonRegionList:
    result = vflow.props.PolygonRegionList()
    if image is None:
        return result

    raw_image = image.image()
    if views is None or views.empty():
        candidate_views = [("__full_image__", full_image_view(raw_image))]
    else:
        candidate_views = list(views)

    for view_id, view in candidate_views:
        sub_image = vflow.img.transform(raw_image, view).image()
        image_np = sub_image.to_numpy()
        model_input = preprocess(image_np)
        detections = run_model(model_input)
        # 后续步骤：阈值过滤、NMS、坐标转换、构造 PolygonRegion

    return result

这里的 sub_image 是视图子图坐标系下的图像。resize、归一化、padding、HWC/CHW 转换、RGB/BGR 转换等预处理都应由脚本按模型要求显式完成。

将模型输出转为 PolygonRegion#

检测模型通常输出框、类别和分数。脚本需要先按阈值过滤，再按需要做 NMS，最后把每个有效检测转成 vflow.PolygonRegion。

def box_to_polygon(
    x1: float,
    y1: float,
    x2: float,
    y2: float,
) -> vflow.geometry.Polygon2f:
    polygon = vflow.geometry.Polygon2f()
    polygon.outer = vflow.geometry.Ring2f([
        vflow.geometry.Point2f(float(x1), float(y1)),
        vflow.geometry.Point2f(float(x2), float(y1)),
        vflow.geometry.Point2f(float(x2), float(y2)),
        vflow.geometry.Point2f(float(x1), float(y2)),
    ])
    return polygon

region = vflow.PolygonRegion()
region.set_polygon(polygon_on_raw)
region.set_name(class_name)
region.set_score(float(score))
result.add(region)

如果模型已经输出 polygon，也应确认点按轮廓顺序排列，不能把矩形四点或多边形顶点乱序加入 Ring2f。

视图坐标转原图坐标#

模型通常在 sub_image 上推理，因此模型输出的 box 或 polygon 多数位于视图子图坐标系。返回给 CustomRegion 前，必须转换回原图坐标系：

raw_transform = vflow.geometry.get_inverse_transform(view.transform_matrix())
polygon_on_raw = vflow.geometry.transform(polygon_on_view, raw_transform)

如果模型输出的是 resize 或 padding 后的网络输入坐标，需要先还原到 sub_image 坐标系，再通过上面的逆变换回到原图坐标系。

返回结果约束#

推理脚本必须返回 vflow.props.PolygonRegionList。每个区域通常需要设置：

polygon：区域多边形，通常应是原图坐标系。
name：类别名，应与训练类别一致。
score：置信度，可选，推荐使用 0.0 到 1.0。

输出约定：

polygon 顶点应按轮廓顺序排列，顺时针或逆时针均可，但同一个区域内不要乱序；不要返回自交多边形。
Ring2f 使用点序列描述闭合区域，通常不需要把首点重复作为最后一个点。
坐标可以是浮点数，但应与图像像素坐标保持一致。
polygon 推荐裁剪或约束在原图范围内。越界区域可能影响显示、过滤和评估。
name 必须使用类别列表中的名称。类别名不匹配会影响前端筛选、统计和评估，应视为脚本错误处理。
矩形框应按四点 polygon 返回。若模型输出的是 x1, y1, x2, y2，需要构造左上、右上、右下、左下四个点。
多类别算法按每个检测结果分别设置 name，并加入同一个 PolygonRegionList。
如果同一目标可能从多个 view 重复检出，脚本应在返回前按算法需要执行 NMS 或其他去重逻辑。

示例：

polygon = vflow.geometry.Polygon2f()
polygon.outer = vflow.geometry.Ring2f([
    vflow.geometry.Point2f(0, 0),
    vflow.geometry.Point2f(100, 0),
    vflow.geometry.Point2f(100, 100),
    vflow.geometry.Point2f(0, 100),
])

region = vflow.PolygonRegion()
region.set_polygon(polygon)
region.set_name("defect")
region.set_score(0.98)

result = vflow.props.PolygonRegionList()
result.add(region)

设备分配#

训练脚本和推理脚本都会收到 device_allocator。它用于申请 GPU，也可以在没有 GPU 时回退到 CPU。

MB = 1024 * 1024
gpu_ids = device_allocator(1, 512 * MB)
device = (
    f"cuda:{gpu_ids[0]}"
    if gpu_ids and gpu_ids[0] >= 0
    else "cpu"
)

参数含义：

第一个参数：需要的 GPU 数量。传 0 表示由系统自动选择。
第二个参数：每个 GPU 需要的最小空闲显存，单位是字节。

脚本 print 输出与重定向#

VisionFlow 提供了 print 重定向机制：集成 VisionFlow 的 C++ 宿主程序可以注册一个自定义 sink，把脚本输出接入自己的日志系统或任务界面。未安装重定向时，脚本中的 print 输出进入宿主进程的标准输出；Runtime 推理侧默认会安装到 VisionFlow 日志系统的重定向。

重定向原理#

sink 的接口是 visionflow::IScriptStdSink，只包含两个方法：

visionflow::IScriptStdSink::write()：接收一段脚本输出文本；
visionflow::IScriptStdSink::flush()：刷新输出缓冲。

write 和 flush 与 Python file-like 对象的协议一致，因此 sink 对象可以直接通过 VisionFlow 的 Python 绑定注入脚本环境，充当 print 的 file 参数。

CustomRegion 训练器初始化脚本环境时，会通过 visionflow::confs::IService::common_configs() 读取服务上的公共配置。如果其中的 script_sink 字段非空，VisionFlow 会：

把该 sink 对象注入脚本的全局命名空间；
把脚本环境中的内置 print 替换为一个包装函数，其 file 参数默认指向注入的 sink 对象，flush 参数默认为 True。

此后脚本中每次调用 print(...)，Python 会把各个参数、分隔符和行尾分多次传入 write，最后调用一次 flush。因此 sink 实现通常在 write 中累积文本，在 flush 中将完整消息写入目标日志。

如果 script_sink 为空，则不安装重定向， print 输出进入宿主进程标准输出。训练侧的 script_sink 由数据服务适配器提供；推理侧的 script_sink 由 Runtime 根据 visionflow::runtime::StrategyOptions::redirect_python_script_print_to_log 和 visionflow::runtime::StrategyOptions::oper_sink 生成。

通过数据服务适配器设置重定向 sink#

训练流程中（见设置参数及训练模型）， visionflow::adapt 返回的数据服务适配器继承自 visionflow::confs::IService。可以在数据服务上调用 visionflow::confs::IService::set_script_sink() 设置训练脚本的 print 输出去向。

#include "visionflow/core_base/iscript_sink.hpp"

// 自定义 sink：write 中累积文本，flush 中输出完整消息
class MyScriptSink : public visionflow::IScriptStdSink {
public:
  void write(const char *message) override { buffer_ += message; }

  void flush() override {
    my_logger_write(buffer_); // 写入自己的日志系统或界面
    buffer_.clear();
  }

private:
  std::string buffer_;
};

visionflow::runtime::StrategyOptions strategy;

// 创建 CustomRegion 训练器的 ConfigureRuntime
auto trainer = project->create_config_runtime(
    {custom_region_id, visionflow::CustomRegion::trainer}, strategy);

 // 为训练执行器创建数据服务
 auto trainer_server = visionflow::adapt(project.get(), trainer);

 // 在 initialize 之前设置 print 重定向 sink
 // 第二个参数为 false 时，不额外写入 VisionFlow 日志系统
 trainer_server.set_script_sink(std::make_shared<MyScriptSink>(), false);

 trainer.initialize(trainer_server);
 trainer.execute(trainer_server);

如果希望同一份 print 输出同时写入 VisionFlow 日志系统和自定义 sink，可以传入自定义 sink，并保持参数 redirect_python_script_print_to_log 为 true：

auto trainer_server = visionflow::adapt(project.get(), trainer);

trainer_server.set_script_sink(std::make_shared<MyScriptSink>(), true);

trainer.initialize(trainer_server);
trainer.execute(trainer_server);

两个参数的组合行为如下：

`redirect_python_script_print_to_log`	`script_sink`	训练脚本 `print` 输出去向
`true` （默认）	`nullptr`	VisionFlow 日志系统。
`true` （默认）	非空	同一份输出分发给两个目标：先进入 VisionFlow 日志系统，同时进入自定义 sink。
`false`	`nullptr`	不安装重定向，进入宿主进程标准输出。
`false`	非空	仅进入自定义 sink。

使用时注意：

sink 必须在 trainer.initialize(...) 之前设置。print 重定向是在训练器初始化脚本环境时安装的，初始化之后修改 script_sink 不会生效。
重定向只接管未显式指定 file 的 print。脚本中显式使用 sys.stdout.write(...)、 print(..., file=sys.stdout) 或 print(..., file=None) 的输出不会进入 sink。
write 和 flush 在脚本执行期间被同步调用。实现中不要执行耗时阻塞操作，也不要抛出异常，否则异常会变成脚本中 print 调用处的 Python 异常。
sink 以 std::shared_ptr 形式被脚本环境持有；sink 内部引用的外部对象（日志器、界面句柄等）需要保证在训练结束前有效。
visionflow::confs::IService::set_script_sink() 会覆盖当前服务上的 script_sink 配置；如果需要同时写入多个目标，应在一次调用中传入自定义 sink 并开启日志重定向。
visionflow::confs::IService::common_configs() 是配置器实现读写该配置的接口。

推理脚本的 print 重定向#

推理脚本使用相同的重定向机制，但 sink 不通过数据服务适配器设置，而是在创建 Runtime 时通过创建运行时的策略的两个选项配置：

visionflow::runtime::StrategyOptions::redirect_python_script_print_to_log: 是否把脚本 print 输出重定向到 VisionFlow 日志系统。默认为 true，对 Runtime 中所有执行 Python 脚本的算子生效。
visionflow::runtime::StrategyOptions::oper_sink: 为指定算子设置自定义 sink，类型为 std::map<ToolNodeId, std::shared_ptr<IScriptStdSink>>。key 是算子节点的 ToolNodeId （工具 ID 加节点名，CustomRegion 推理算子的节点名为 infer），value 是自定义 sink 对象；未列入 map 的算子不受该选项影响。

两个选项可以独立配置，组合行为如下：

`redirect_python_script_print_to_log`	`oper_sink` 中包含该算子	该算子脚本 `print` 输出去向
`true` （默认）	否	VisionFlow 日志系统。
`true` （默认）	是	同一份输出分发给两个目标：先进入 VisionFlow 日志系统，同时进入自定义 sink。
`false`	否	不安装重定向，进入宿主进程标准输出。
`false`	是	仅进入自定义 sink。

示例：只把 CustomRegion 推理脚本的输出接入自己的 sink，不写 VisionFlow 日志：

visionflow::runtime::AllTools strategy;

// 关闭内置的日志重定向（默认开启）
strategy.options.redirect_python_script_print_to_log = false;

// 为 CustomRegion 的推理算子设置自定义 sink，MyScriptSink 实现见上一节
strategy.options.oper_sink[{custom_region_id, visionflow::CustomRegion::infer}] =
    std::make_shared<MyScriptSink>();

auto runtime = project->create_runtime(strategy);

使用时注意：

与训练侧不同，推理侧 redirect_python_script_print_to_log 默认开启，因此推理脚本的 print 默认就会进入 VisionFlow 日志，而不是宿主进程标准输出。
这两个选项对 Runtime 中所有执行 Python 脚本的算子生效，不限于 CustomRegion；如果流程中还有其他脚本类算子（例如 Integration、BlobAnalysis ），可以通过 oper_sink 按节点分别设置。
自定义 sink 的实现要求与训练侧一致。

完整示例：用 PyTorch 训练一个极简区域检测器#

下面示例只演示关键方法和数据流，不代表真实检测算法的建模方式：

训练时读取训练视图和标注。
将每个视图子图转为一个亮度均值特征。
用 PyTorch 训练一个只有一层线性层的二分类模型。
将模型和类别名保存到 BinaryPacks。
推理时读取模型，对每个视图判断是否有目标；若有目标，则返回覆盖该视图的多边形。

Note

这是用于说明接口的玩具示例。真实检测算法应使用 labels_in_view 这类逻辑，按当前 view 与标注的相交关系生成样本标签和检测框，而不是只根据整图是否存在标注来决定当前 view 是否为正样本。

训练脚本#

import io
import json
from typing import Callable, List

import numpy as np
import torch
import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

def has_target_in_view(
    labels,
    view: vflow.View,
    min_intersection_area: float = 1.0,
) -> bool:
    if labels is None:
        return False

    for region_id, region in labels:
        clipped = vflow.geometry.intersection(
            view.view_on_raw_image(),
            region.polygon(),
        )
        if clipped.area() >= min_intersection_area:
            return True
    return False

class CustomConfigurator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        svc: vflow.confs.IService,
    ) -> None:
        label_classes, _, _ = svc.input_parameters()
        class_names = list(label_classes.get())
        if not class_names:
            raise RuntimeError("CustomRegion requires at least one label class.")
        self.class_name = class_names[0]

        gpu_ids = device_allocator(1, 256 * 1024 * 1024)
        self.device = torch.device(
            f"cuda:{gpu_ids[0]}" if gpu_ids and gpu_ids[0] >= 0 else "cpu"
        )
        self.epochs = int(VFLOW_USER_VARS.get("Epoch", 5))
        self.lr = float(VFLOW_USER_VARS.get("LearningRate", 0.1))

    def _feature(self, image: vflow.img.Image) -> float:
        return float(np.mean(image.to_numpy()) / 255.0)

    def execute(self, svc: vflow.confs.IService) -> vflow.param.BinaryPacks:
        image_set, views_set, _, label_set = svc.property_sets()

        xs = []
        ys = []
        for sample_id, views in views_set:
            if views is None:
                continue
            train_views = views.tagged_views(vflow.kTrain)
            if train_views.empty():
                continue

            image_prop = image_set.at(sample_id)
            if image_prop is None:
                continue

            raw_image = image_prop.image()
            labels = label_set.at(sample_id)
            if labels is None:
                labels = vflow.props.PolygonRegionList()

            for view_id, view in train_views:
                sub_image = vflow.img.transform(raw_image, view).image()
                xs.append([self._feature(sub_image)])
                ys.append(1 if has_target_in_view(labels, view) else 0)

        if not xs:
            raise RuntimeError("No training views found.")

        model = torch.nn.Linear(1, 2).to(self.device)
        optimizer = torch.optim.SGD(model.parameters(), lr=self.lr)
        loss_fn = torch.nn.CrossEntropyLoss()

        x = torch.tensor(xs, dtype=torch.float32, device=self.device)
        y = torch.tensor(ys, dtype=torch.long, device=self.device)

        for epoch in range(self.epochs):
            optimizer.zero_grad()
            loss = loss_fn(model(x), y)
            loss.backward()
            optimizer.step()

            ctrl = svc.on_progress(
                self.epochs,
                epoch + 1,
                json.dumps({
                    "Message": "Training...",
                    "Loss": float(loss.item()),
                }),
            )
            if ctrl != vflow.TaskCtrlSignal.kContinue:
                break

        buffer = io.BytesIO()
        torch.save({
            "state_dict": model.state_dict(),
            "class_name": self.class_name,
        }, buffer)

        packs = vflow.param.BinaryPacks()
        packs.insert("torch_model", vflow.Buffer.FromBytes(buffer.getvalue()))
        packs.insert("model_info", vflow.Buffer.FromBytes(json.dumps({
            "format_version": 1,
            "class_names": [self.class_name],
            "feature": "mean_gray",
        }).encode("utf-8")))
        return packs

推理脚本#

import io
from typing import Callable, List

import numpy as np
import torch
import visionflow as vflow

DeviceAllocator = Callable[[int, int], List[int]]

def view_rect_on_raw_image(view: vflow.View) -> vflow.geometry.Polygon2f:
    size = view.size()
    polygon = vflow.geometry.Polygon2f()
    polygon.outer = vflow.geometry.Ring2f([
        vflow.geometry.Point2f(0.0, 0.0),
        vflow.geometry.Point2f(float(size.w), 0.0),
        vflow.geometry.Point2f(float(size.w), float(size.h)),
        vflow.geometry.Point2f(0.0, float(size.h)),
    ])
    raw_transform = vflow.geometry.get_inverse_transform(view.transform_matrix())
    return vflow.geometry.transform(polygon, raw_transform)

class CustomOperator:
    def __init__(
        self,
        device_allocator: DeviceAllocator,
        model: vflow.param.BinaryPacks,
    ) -> None:
        gpu_ids = device_allocator(1, 256 * 1024 * 1024)
        self.device = torch.device(
            f"cuda:{gpu_ids[0]}" if gpu_ids and gpu_ids[0] >= 0 else "cpu"
        )

        if not model.contains("torch_model"):
            raise RuntimeError("Missing required model key: torch_model")

        data = torch.load(
            io.BytesIO(model.get("torch_model").to_bytes()),
            map_location=self.device,
        )
        self.model = torch.nn.Linear(1, 2).to(self.device)
        self.model.load_state_dict(data["state_dict"])
        self.model.eval()
        self.class_name = data["class_name"]
        self.score_threshold = float(VFLOW_USER_VARS.get("ScoreThreshold", 0.5))

    def _feature(self, image: vflow.img.Image) -> float:
        return float(np.mean(image.to_numpy()) / 255.0)

    def execute(
        self,
        image: vflow.props.Image,
        views: vflow.props.ViewList,
        img_info: vflow.props.RawImageInfo,
    ) -> vflow.props.PolygonRegionList:
        result = vflow.props.PolygonRegionList()
        raw_image = image.image()

        for view_id, view in views:
            sub_image = vflow.img.transform(raw_image, view).image()
            x = torch.tensor(
                [[self._feature(sub_image)]],
                dtype=torch.float32,
                device=self.device,
            )

            with torch.no_grad():
                prob = torch.softmax(self.model(x), dim=1)[0, 1].item()

            if prob < self.score_threshold:
                continue

            region = vflow.PolygonRegion()
            region.set_polygon(view_rect_on_raw_image(view))
            region.set_name(self.class_name)
            region.set_score(float(prob))
            result.add(region)

        return result

真实算法替换点#

把示例迁移到真实算法时，通常只需要替换三处：

CustomConfigurator.__init__：读取用户参数、申请设备、初始化训练对象。
CustomConfigurator.execute：把 sub_image.to_numpy() 和当前 view 内的标注转成训练样本，训练后将模型序列化到 BinaryPacks。
CustomOperator.execute：把推理输出转换成 vflow.props.PolygonRegionList，并确保输出 polygon 已经回到原图坐标系。

真实检测模型的替换流程通常类似下面这样：

# 训练阶段
sub_image = vflow.img.transform(raw_image, train_view).image()
image_np = sub_image.to_numpy()
labels_on_view = labels_in_view(labels, train_view)
train_sample = build_train_sample(image_np, labels_on_view)

# 推理阶段
sub_image = vflow.img.transform(raw_image, view).image()
model_input, restore_info = preprocess_for_model(sub_image.to_numpy())
boxes, class_ids, scores = run_detector(model_input)
boxes = restore_boxes_to_view(boxes, restore_info)
boxes, class_ids, scores = nms(boxes, class_ids, scores)

for box, class_id, score in zip(boxes, class_ids, scores):
    polygon_on_view = box_to_polygon(*box)
    polygon_on_raw = vflow.geometry.transform(
        polygon_on_view,
        vflow.geometry.get_inverse_transform(view.transform_matrix()),
    )
    # 构造 PolygonRegion，并加入返回结果

替换真实检测算法时，重点检查下面几项：

训练样本是否只使用 vflow.kTrain 视图。
每个 view 的正负样本标签是否按标注与 view 的相交关系生成。
传给模型的标注坐标是否已经从原图坐标转换到视图子图坐标。
推理结果是否从视图子图坐标转换回原图坐标。
模型包中是否包含推理所需的权重、类别名和版本信息。

常见问题#

配置与环境#

依赖库导入失败怎么办？: 脚本在 VisionFlow 的 Python 环境中运行。需要把第三方库安装到同一个环境，并确认训练和推理两份脚本都能 import 成功。
普通参数读取类型不对怎么办？: 检查参数类型是否配置为 string、number 或 bool。number 会注入为 float，需要整数时再使用 int(...)；bool 正常会注入为 Python bool，不要用 bool("False") 这类写法。
GPU 申请失败怎么办？: 检查 device_allocator 返回值。返回空列表或 [-1] 时应回退 CPU，或者抛出明确错误提示显存需求和实际设备情况。不要直接假设 gpu_ids[0] 一定存在。

训练数据#

为什么训练样本数量为 0？: 优先检查是否存在 vflow.kTrain 视图、views_set 是否为空、image_set.at(sample_id) 是否为 None，以及脚本是否错误过滤了全部 view。训练脚本应在 not xs 或无有效样本时抛出明确异常。
未标注样本怎么处理？: label_set.at(sample_id) 可能为 None 或空列表。区域检测算法通常可以把当前 view 作为无目标负样本；只接受正样本的算法，应显式跳过当前 view 内无有效目标的样本。

模型传递#

训练二进制输入参数和训练输出模型有什么区别？: 训练二进制输入参数是训练前由 VisionFlow 工程传入脚本的数据，例如预训练模型。execute 返回的 BinaryPacks 是训练后的模型，会传给推理脚本。
模型 key 不存在怎么办？: 在 CustomOperator.__init__ 中使用 model.contains(key) 显式检查，并抛出 RuntimeError("Missing required model key: ...")。不要让 model.get(key) 的底层异常成为唯一提示。

推理输出#

推理脚本能否访问 svc？: 不能。推理脚本只能通过 CustomOperator.__init__(device_allocator, model) 读取模型和用户变量，通过 execute(image, views, img_info) 读取当前推理数据。
为什么推理没有任何结果？: 先检查推理视图是否为空、阈值是否过高、模型 key 是否正确、类别名是否匹配、模型输出坐标是否被误当作原图坐标。可以临时打印 view 数量、模型分数和返回区域面积定位问题。
类别名不匹配怎么办？: PolygonRegion.name 应使用类别列表中的名称。类别名不匹配会影响前端筛选、统计和评估，建议在构造结果前检查 class_name in self.class_names，发现不匹配时抛出明确异常。

坐标与视图#

返回的多边形坐标应该属于哪个坐标系？: 通常应返回原图坐标系下的区域。如果算法在裁剪后的视图子图上输出结果，需要根据视图变换矩阵转换回原图坐标系。
返回区域明显错位怎么办？: 多数情况是坐标系转换方向反了。训练时，原图标注转视图子图坐标使用 view.transform_matrix()；推理时，视图子图输出转原图坐标使用 vflow.geometry.get_inverse_transform(view.transform_matrix())。
views 为空怎么办？: 先确认上游流程是否应该产生 view。如果算法约定无 view 不推理，可以返回空 PolygonRegionList；如果算法约定无 view 时整图推理，可以在 execute 中构造覆盖整张原图的 view。两种方式都可以，但训练和推理策略要一致。

调试排错#

最小调试模板怎么写？

首次接入时建议先打印关键对象数量和模型 key，确认数据流正确后再排查算法逻辑：

print("vars:", VFLOW_USER_VARS)

label_classes, _, binary_args = svc.input_parameters()
print("classes:", list(label_classes.get()))
print("has pretrained:", binary_args.contains("pretrained_model"))

image_set, views_set, _, label_set = svc.property_sets()
sample_count = 0
view_count = 0
label_count = 0
for sample_id, views in views_set:
    sample_count += 1
    if views is not None:
        for view_id, view in views:
            view_count += 1
    labels = label_set.at(sample_id)
    if labels is not None:
        label_count += labels.size()
print("samples:", sample_count, "views:", view_count, "labels:", label_count)