人类演示数据视觉预处理与质量清洗
更新时间：2026-07-21
概述
用于在进入昂贵的 GPU 感知之前，对海量原始视频做质量清洗，剔除低质片段，降低下游算力浪费。
- 输入：原始演示视频（含相机内参与畸变系数）
- 输出：去畸变 + 场景切分后，通过清晰度/运动量/时长过滤的高质量片段清单（JSONL 格式，供 P2 消费）
- 业务价值：GPU 感知成本高，前置 CPU 清洗可显著提升闭环整体吞吐与数据质量
处理流程
Plain Text
1原始视频 → S1 去畸变 → S2 按场景切分 → S3 元数据提取 → S4 清晰度评分 → S5 运动评分 → S6 质量过滤 → 高质量片段清单算子映射
| 步骤 | 算子 | 说明 |
|---|---|---|
| S1 | VideoUndistort |
按相机内参+畸变系数做镜头去畸变，输出校正视频（可选，no_undistort=true 跳过） |
| S2 | VideoSplitByScene |
ContentDetector 按画面内容突变将视频切成多段 clip |
| S3 | VideoExtractMetadata |
ffprobe 提取时长/分辨率/fps 等元信息 |
| S4 | VideoSharpness |
Laplacian 方差清晰度评分（数值越大越清晰） |
| S5 | VideoMotionScore |
Farneback 光流运动量评分（standardized 0~1） |
| S6 | Filter | 按清晰度/运动/时长阈值过滤，保留高质量片段 |
数据流向
Plain Text
1input_file (manifest.json)
2 → S1 → tmp_dir/s1_undistort.jsonl
3 → S2 → tmp_dir/s2_split_scene.jsonl
4 → S3 → tmp_dir/s3_metadata.jsonl
5 → S4 → tmp_dir/s4_sharpness.jsonl
6 → S5 → tmp_dir/s5_motion.jsonl
7 → S6 → output_file (clips.jsonl)前置条件
| 资源 | 说明 |
|---|---|
| AIHC 队列 | 需要有可用的计算队列（queue_id） |
| PFS 存储 | 用于存放中间数据和最终输出（pfs_id） |
| BOS Bucket | 用于挂载算子脚本和输入数据，默认 aihc-rdw-bj/aihc-daft |
| 镜像 | ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.1-cu121-py3.11-ubuntu22.04 |
全局参数说明
| 参数名 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
image |
string | 是 | ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.1-cu121-py3.11-ubuntu22.04 |
镜像地址（含 tag），不填则使用默认值 |
queue_id |
string | 是 | — | AIHC 队列 ID |
priority |
string | 否 | high |
任务优先级 |
pfs_id |
string | 是 | — | PFS 存储 ID |
pfs_source_path |
string | 否 | / |
PFS sourcePath |
pfs_mount_path |
string | 否 | /mnt/pfs |
PFS 挂载点（中间数据与最终输出存储） |
bos_source_path |
string | 否 | aihc-rdw-bj/aihc-daft |
BOS sourcePath（脚本+输入数据统一 bucket） |
bos_mount_path |
string | 否 | /opt/aihc |
BOS 挂载点 |
input_file |
string | 是 | /opt/aihc/data/embodied_pipelines/input_data/p1/manifest.json |
输入文件绝对路径（BOS 上） |
output_file |
string | 是 | — | 最终输出文件路径（PFS 上） |
tmp_dir |
string | 是 | — | 中间数据目录（6 个 step 共用，存于 PFS） |
no_undistort |
string | 否 | false |
是否跳过去畸变（true/false） |
head_cpu |
int | 否 | 4 |
Head 节点 CPU 数 |
head_memory |
int | 否 | 8 |
Head 节点内存（GB） |
shared_memory |
int | 否 | 8 |
sharedMemory 大小（GB） |
worker_replicas |
int | 否 | 1 |
Worker 副本数 |
worker_cpu |
int | 否 | 6 |
Worker 节点 CPU 数 |
worker_memory |
int | 否 | 12 |
Worker 节点内存（GB） |
worker_gpu |
int | 否 | 0 |
Worker 节点 GPU 数 |
步骤详解
S1 去畸变（Undistort）
按相机内参和畸变系数对视频帧做镜头校正，消除桶形/枕形畸变，还原真实空间几何。支持跳过（设置 no_undistort=true）。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
camera_model |
string | pinhole |
相机模型类型（pinhole / fisheye） |
- 输入列：
video,intrinsics,distortion - 输出列：新增
undistorted（去畸变后的视频路径）
S2 按场景切分（Split Scene）
使用 ContentDetector 检测画面内容突变点，将长视频切分为独立场景片段。切分后按行 explode，每行对应一个 clip。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
detector |
string | ContentDetector |
场景检测算法 |
threshold |
string | 27.0 |
检测灵敏度阈值（越小越敏感） |
min_scene_len |
string | 15 |
最小场景帧数（低于此长度的片段将被合并） |
- 输入列：
undistorted（或跳过去畸变时为video） - 输出列：新增
clip（单个片段路径）
S3 元数据提取（Metadata）
通过 ffprobe 提取视频元信息，包括时长、分辨率、帧率、编码格式等。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| （无独有参数） | — | — | — |
- 输入列：
clip - 输出列：新增
meta（含duration,video_width,video_height,video_fps等）
S4 清晰度评分（Sharpness）
对视频按固定帧率采样，计算每帧的 Laplacian 方差作为清晰度指标，输出统计值（mean/median/max/min/std）。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
method |
string | laplacian |
清晰度评估算法 |
fps |
string | 2.0 |
采样帧率（每秒采几帧用于评分） |
max_frames |
string | 60 |
最大采样帧数 |
- 输入列：
clip - 输出列：新增
clarity（含mean,median,max,min,std）
S5 运动评分（Motion）
基于 Farneback 光流算法计算视频运动量，输出标准化评分（0~1），用于识别静止/低运动片段。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
optical_flow_algorithm |
string | farneback |
光流算法 |
use_cuda |
string | false |
是否使用 CUDA 加速（true/false） |
- 输入列：
clip - 输出列：新增
motion（含standardized_score,motion_pattern,mean_score等）
S6 质量过滤（Filter）
根据清晰度、运动量、时长三个维度的阈值过滤片段，仅保留高质量视频。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
clarity_min |
string | 20.0 |
清晰度最低阈值（clarity.mean 低于此值的片段被剔除） |
motion_min |
string | 0.05 |
运动量最低阈值（motion.standardized_score 低于此值的片段被剔除） |
min_duration |
string | 0.5 |
最短时长（秒），低于此值的片段被剔除 |
- 输入列：
clip,meta,clarity,motion - 输出列：
clip,duration,clarity,motion（最终产物）
完整工作流 YAML 示例
以下为 P1 视觉清洗管线的完整工作流模板（6 步串行 DAG）：
YAML
1version: v1
2kind: PipelineTemplate
3
4inputs:
5- name: image
6 type: string
7 hint: 镜像地址（含 tag）
8 defaultValue: ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.1-cu121-py3.11-ubuntu22.04
9- name: queue_id # [必填] 替换为实际的队列 ID
10 type: string
11 hint: AIHC 队列 ID
12 defaultValue: <YOUR_QUEUE_ID>
13- name: priority
14 type: string
15 hint: 任务优先级
16 defaultValue: high
17- name: pfs_id # [必填] 替换为实际的 PFS 存储 ID
18 type: string
19 hint: PFS 存储 ID
20 defaultValue: <YOUR_PFS_ID>
21- name: pfs_source_path
22 type: string
23 defaultValue: /
24- name: pfs_mount_path
25 type: string
26 hint: PFS 挂载点（中间数据与最终输出存储）
27 defaultValue: /mnt/pfs
28- name: bos_source_path
29 type: string
30 hint: BOS sourcePath（脚本+输入数据统一 bucket）
31 defaultValue: aihc-rdw-bj/aihc-daft
32- name: bos_mount_path
33 type: string
34 hint: BOS 挂载点
35 defaultValue: /opt/aihc
36- name: tmp_dir # [必填] 替换为实际的中间数据目录
37 type: string
38 hint: 中间数据目录（6 个 step 共用，存于 PFS）
39 defaultValue: /mnt/pfs/<YOUR_PATH>/tmp_data/p1
40- name: input_file # [必填] 输入 manifest 文件路径
41 type: string
42 hint: 输入文件绝对路径（BOS 上）
43 defaultValue: /opt/aihc/data/embodied_pipelines/input_data/p1/manifest.json
44- name: output_file # [必填] 最终输出路径
45 type: string
46 hint: 最终输出文件路径（PFS 上）
47 defaultValue: /mnt/pfs/<YOUR_PATH>/output_data/p1/clips.jsonl
48- name: no_undistort
49 type: string
50 hint: 是否跳过去畸变 (true/false)
51 defaultValue: "false"
52- name: head_cpu
53 type: int
54 defaultValue: 4
55- name: head_memory
56 type: int
57 defaultValue: 8
58- name: shared_memory
59 type: int
60 defaultValue: 8
61- name: worker_replicas
62 type: int
63 defaultValue: 1
64- name: worker_cpu
65 type: int
66 defaultValue: 6
67- name: worker_memory
68 type: int
69 defaultValue: 12
70- name: worker_gpu
71 type: int
72 defaultValue: 0
73
74taskTemplates:
75# ---------- S1: 去畸变 ----------
76- name: p1-1-undistort
77 type: CustomTask
78 inputs:
79 - name: camera_model
80 type: string
81 defaultValue: pinhole
82 spec:
83 queue: '{{pipeline.parameters.queue_id}}'
84 priority: '{{pipeline.parameters.priority}}'
85 jobType: RayJob
86 command: >-
87 if [ "{{pipeline.parameters.no_undistort}}" = "true" ]; then
88 echo "[P1-S1] skip undistort"; exit 0;
89 fi;
90 DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p1_1_undistort.py
91 --input {{pipeline.parameters.input_file}}
92 --output {{pipeline.parameters.tmp_dir}}/s1_undistort.jsonl
93 --tmp-dir {{pipeline.parameters.tmp_dir}}
94 --camera-model {{inputs.parameters.camera_model}}
95 jobSpec:
96 Head:
97 replicas: 1
98 restartPolicy: Never
99 image: '{{pipeline.parameters.image}}'
100 resources:
101 - name: cpu
102 quantity: '{{pipeline.parameters.head_cpu}}'
103 - name: memory
104 quantity: '{{pipeline.parameters.head_memory}}'
105 - name: sharedMemory
106 quantity: '{{pipeline.parameters.shared_memory}}'
107 Worker:
108 replicas: '{{pipeline.parameters.worker_replicas}}'
109 restartPolicy: Never
110 image: '{{pipeline.parameters.image}}'
111 resources:
112 - name: cpu
113 quantity: '{{pipeline.parameters.worker_cpu}}'
114 - name: memory
115 quantity: '{{pipeline.parameters.worker_memory}}'
116 - name: sharedMemory
117 quantity: '{{pipeline.parameters.shared_memory}}'
118 datasources:
119 - type: bos
120 sourcePath: '{{pipeline.parameters.bos_source_path}}'
121 mountPath: '{{pipeline.parameters.bos_mount_path}}'
122 - type: pfs
123 sourcePath: '{{pipeline.parameters.pfs_source_path}}'
124 mountPath: '{{pipeline.parameters.pfs_mount_path}}'
125 name: '{{pipeline.parameters.pfs_id}}'
126
127# ---------- S2: 按场景切分 ----------
128- name: p1-2-split-scene
129 type: CustomTask
130 inputs:
131 - name: detector
132 type: string
133 defaultValue: ContentDetector
134 - name: threshold
135 type: string
136 defaultValue: "27.0"
137 - name: min_scene_len
138 type: string
139 defaultValue: "15"
140 spec:
141 queue: '{{pipeline.parameters.queue_id}}'
142 priority: '{{pipeline.parameters.priority}}'
143 jobType: RayJob
144 command: >-
145 if [ "{{pipeline.parameters.no_undistort}}" = "true" ]; then
146 INPUT_FILE="{{pipeline.parameters.input_file}}";
147 VIDEO_COL="video";
148 else
149 INPUT_FILE="{{pipeline.parameters.tmp_dir}}/s1_undistort.jsonl";
150 VIDEO_COL="undistorted";
151 fi;
152 DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p1_2_split_scene.py
153 --input "$INPUT_FILE"
154 --output {{pipeline.parameters.tmp_dir}}/s2_split_scene.jsonl
155 --tmp-dir {{pipeline.parameters.tmp_dir}}
156 --video-col "$VIDEO_COL"
157 --detector {{inputs.parameters.detector}}
158 --threshold {{inputs.parameters.threshold}}
159 --min-scene-len {{inputs.parameters.min_scene_len}}
160 # jobSpec 同上，省略...
161
162# ---------- S3: 提取元数据 ----------
163- name: p1-3-metadata
164 type: CustomTask
165 spec:
166 queue: '{{pipeline.parameters.queue_id}}'
167 priority: '{{pipeline.parameters.priority}}'
168 jobType: RayJob
169 command: >-
170 DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p1_3_metadata.py
171 --input {{pipeline.parameters.tmp_dir}}/s2_split_scene.jsonl
172 --output {{pipeline.parameters.tmp_dir}}/s3_metadata.jsonl
173 --tmp-dir {{pipeline.parameters.tmp_dir}}
174
175# ---------- S4: 清晰度评分 ----------
176- name: p1-4-sharpness
177 type: CustomTask
178 inputs:
179 - name: method
180 type: string
181 defaultValue: laplacian
182 - name: fps
183 type: string
184 defaultValue: "2.0"
185 - name: max_frames
186 type: string
187 defaultValue: "60"
188 spec:
189 queue: '{{pipeline.parameters.queue_id}}'
190 priority: '{{pipeline.parameters.priority}}'
191 jobType: RayJob
192 command: >-
193 DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p1_4_sharpness.py
194 --input {{pipeline.parameters.tmp_dir}}/s3_metadata.jsonl
195 --output {{pipeline.parameters.tmp_dir}}/s4_sharpness.jsonl
196 --tmp-dir {{pipeline.parameters.tmp_dir}}
197 --method {{inputs.parameters.method}}
198 --fps {{inputs.parameters.fps}}
199 --max-frames {{inputs.parameters.max_frames}}
200
201# ---------- S5: 运动评分 ----------
202- name: p1-5-motion
203 type: CustomTask
204 inputs:
205 - name: optical_flow_algorithm
206 type: string
207 defaultValue: farneback
208 - name: use_cuda
209 type: string
210 defaultValue: "false"
211 spec:
212 queue: '{{pipeline.parameters.queue_id}}'
213 priority: '{{pipeline.parameters.priority}}'
214 jobType: RayJob
215 command: >-
216 DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p1_5_motion.py
217 --input {{pipeline.parameters.tmp_dir}}/s4_sharpness.jsonl
218 --output {{pipeline.parameters.tmp_dir}}/s5_motion.jsonl
219 --tmp-dir {{pipeline.parameters.tmp_dir}}
220 --optical-flow-algorithm {{inputs.parameters.optical_flow_algorithm}}
221 $([ "{{inputs.parameters.use_cuda}}" = "true" ] && echo "--use-cuda" || echo "")
222
223# ---------- S6: 质量过滤 ----------
224- name: p1-6-filter
225 type: CustomTask
226 inputs:
227 - name: clarity_min
228 type: string
229 defaultValue: "20.0"
230 - name: motion_min
231 type: string
232 defaultValue: "0.05"
233 - name: min_duration
234 type: string
235 defaultValue: "0.5"
236 spec:
237 queue: '{{pipeline.parameters.queue_id}}'
238 priority: '{{pipeline.parameters.priority}}'
239 jobType: RayJob
240 command: >-
241 DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p1_6_filter.py
242 --input {{pipeline.parameters.tmp_dir}}/s5_motion.jsonl
243 --output {{pipeline.parameters.output_file}}
244 --tmp-dir {{pipeline.parameters.tmp_dir}}
245 --clarity-min {{inputs.parameters.clarity_min}}
246 --motion-min {{inputs.parameters.motion_min}}
247 --min-duration {{inputs.parameters.min_duration}}
248
249# ============ 任务编排（串行 DAG） ============
250tasks:
251- name: step1-undistort
252 taskTemplateName: p1-1-undistort
253 inputs:
254 - name: camera_model
255 value: 'pinhole'
256
257- name: step2-split-scene
258 taskTemplateName: p1-2-split-scene
259 dependencies:
260 - step1-undistort
261 inputs:
262 - name: detector
263 value: 'ContentDetector'
264 - name: threshold
265 value: '27.0'
266 - name: min_scene_len
267 value: '15'
268
269- name: step3-metadata
270 taskTemplateName: p1-3-metadata
271 dependencies:
272 - step2-split-scene
273
274- name: step4-sharpness
275 taskTemplateName: p1-4-sharpness
276 dependencies:
277 - step3-metadata
278 inputs:
279 - name: method
280 value: 'laplacian'
281 - name: fps
282 value: '2.0'
283 - name: max_frames
284 value: '60'
285
286- name: step5-motion
287 taskTemplateName: p1-5-motion
288 dependencies:
289 - step4-sharpness
290 inputs:
291 - name: optical_flow_algorithm
292 value: 'farneback'
293 - name: use_cuda
294 value: 'false'
295
296- name: step6-filter
297 taskTemplateName: p1-6-filter
298 dependencies:
299 - step5-motion
300 inputs:
301 - name: clarity_min
302 value: '20.0'
303 - name: motion_min
304 value: '0.05'
305 - name: min_duration
306 value: '0.5'上述 YAML 中 S2-S6 的
jobSpec和datasources结构与 S1 相同，实际使用时请参考完整模板文件。
任务拓扑
Plain Text
1+-------------+ +---------------+ +------------+ +-------------+ +-----------+ +----------+
2| S1 Undistort| --> | S2 SplitScene | --> | S3 Metadata| --> | S4 Sharpness| --> | S5 Motion | --> | S6 Filter|
3+-------------+ +---------------+ +------------+ +-------------+ +-----------+ +----------+6 个步骤严格串行执行，每步通过 dependencies 声明对前一步的依赖。
单步独立执行
每个步骤也提供独立的工作流模板，可以单独提交运行。适用于调试单个算子或从中间步骤恢复执行。
例如单独运行 S4 清晰度评分：
YAML
1version: v1
2kind: PipelineTemplate
3
4inputs:
5- name: image
6 type: string
7 defaultValue: ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.1-cu121-py3.11-ubuntu22.04
8
9taskTemplates:
10- name: p1-4-sharpness
11 type: CustomTask
12 inputs:
13 - name: queue_id
14 type: string
15 - name: pfs_id
16 type: string
17 - name: input_file
18 type: string
19 - name: output_file
20 type: string
21 - name: tmp_dir
22 type: string
23 - name: method
24 type: string
25 defaultValue: laplacian
26 - name: fps
27 type: string
28 defaultValue: "2.0"
29 - name: max_frames
30 type: string
31 defaultValue: "60"
32 spec:
33 queue: '{{inputs.parameters.queue_id}}'
34 priority: high
35 jobType: RayJob
36 command: >-
37 DAFT_RUNNER=ray python3 /opt/aihc/workflow/p1_4_sharpness.py
38 --input {{inputs.parameters.input_file}}
39 --output {{inputs.parameters.output_file}}
40 --tmp-dir {{inputs.parameters.tmp_dir}}
41 --method {{inputs.parameters.method}}
42 --fps {{inputs.parameters.fps}}
43 --max-frames {{inputs.parameters.max_frames}}
44 # jobSpec / datasources 配置同完整模板
45
46tasks:
47- name: p1-sharpness-task
48 taskTemplateName: p1-4-sharpness
49 inputs:
50 - name: queue_id
51 value: '<YOUR_QUEUE_ID>' # 替换为实际队列 ID
52 - name: pfs_id
53 value: '<YOUR_PFS_ID>' # 替换为实际 PFS ID
54 - name: input_file
55 value: '/mnt/pfs/<YOUR_PATH>/tmp_data/p1/s3_metadata.jsonl'
56 - name: output_file
57 value: '/mnt/pfs/<YOUR_PATH>/tmp_data/p1/s4_sharpness.jsonl'
58 - name: tmp_dir
59 value: '/mnt/pfs/<YOUR_PATH>/tmp_data/p1'使用建议
- 首次使用建议设置
no_undistort=true快速验证全链路是否跑通 clarity_min、motion_min、min_duration三个过滤阈值可根据实际数据质量灵活调整- Worker 资源配置中
worker_gpu=0，P1 全程为 CPU 计算，无需 GPU 资源 - 如数据量大，可增加
worker_replicas实现 Ray 分布式加速
评价此篇文章