CANN 昇腾图像预处理流水线：CV 算子深度解读

h64648564h

113人浏览 · 2026-05-23 15:31:03

h64648564h · 2026-05-23 15:31:03 发布

前言

之前做过一个小目标检测的项目，模型精度和推理速度都调得差不多了，最后发现瓶颈居然在图像预处理上。resize、normalize、augmentation 这些操作每帧都要跑，累积起来的时间比 inference 还多。NV 的 DALI 可以用，但那是 GPU 专用。昇腾上有 VIC（Vision Image Compute）引擎专门解决这个问题，这篇文章把 CV 预处理的所有门道一次性讲清楚。

图像预处理的典型 Pipeline

一个完整的图像预处理通常包含下面几步：

def standard_preprocess(image, target_size=(640, 640)):
    """
    典型的 CV 预处理流程
    """
    # 1. Decode（JPEG/PNG -> RGB）
    image = cv2.imdecode(image, cv2.IMREAD_COLOR)
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    
    # 2. Resize（可能带 letterbox）
    image, ratio, pad = letterbox(image, target_size)
    
    # 3. Normalize（归一化到 0-1）
    image = image.astype(np.float32) / 255.0
    
    # 4. 标准化（减均值除标准差）
    image = (image - mean) / std
    
    # 5. HWC -> CHW
    image = image.transpose(2, 0, 1)
    
    # 6. NPU 传输
    image_tensor = torch.from_numpy(image).npu()
    
    return image_tensor, ratio, pad

这几步看起来简单，但在 CPU 上跑，每一帧都要来一遍，累积起来开销非常大。

昇腾 VIC 引擎

昇腾提供了专门的图像处理引擎：VIC（Vision Image Compute）。它是一个面向图像处理 workloads 的加速器，可以流水线化地处理图像预处理。

VIC 的核心优势：

单引擎完成解码+resize+normalize+转换
零 CPU 介入，数据直接在 NPU 上流动
支持常见的数据增强（flip、crop、color jitter）

VIC 基本用法

import torch
import vic

# 初始化 VIC Pipeline
pipeline = vic.Pipeline(
    input_layout='NHWC',  # 图像通常是 HWC
    output_layout='NCHW',  # 转成 NPU 友好的 CHW
    mean=[0.485, 0.456, 0.406],
    std=[0.229, 0.224, 0.225],
    output_dtype='float32'
)

# 处理单张图像
image = cv2.imread('test.jpg')
output = pipeline.process(image)

# 批量处理
images = [cv2.imread(f) for f in image_paths]
outputs = pipeline.batch_process(images)

VIC 的高级特性

VIC 支持更复杂的预处理 pipeline：

# 带数据增强的 pipeline
pipeline_aug = vic.Pipeline(
    input_layout='NHWC',
    output_layout='NCHW',
    # 随机增强
    random_flip=True,
    random_crop=True,
    color_jitter=True,
    # 随机亮度/对比度
    brightness_range=0.2,
    contrast_range=0.2,
    # 输出
    mean=[0.485, 0.456, 0.406],
    std=[0.229, 0.224, 0.225],
    output_dtype='float32'
)

# 使用时打开 flag
if training:
    output = pipeline_aug.process(image)
else:
    output = pipeline.process(image)  # 验证集不做增强

这些增强操作在 VIC 上跑和在 CPU 上跑，性能差距可能是几倍甚至十倍。

Resize 的那些坑

Resize 是最常见的操作，也容易踩坑。

Interpolation 方式

方式	速度	质量	适用场景
Nearest	最快	差	像素级操作
Bilinear	中等	中	一般用途
Cubic	较慢	好	要求质量的场景
Lanczos	最慢	最好	出版物级

昇腾的建议是：一般推理用 Bilinear 就够了。

# 错误：多次插值
result = cv2.resize(src, (w//2, h//2), interpolation=cv2.INTER_LINEAR)
result = cv2.resize(result, (w, h), interpolation=cv2.INTER_LINEAR)

# 正确：单次直接 resize 到目标尺寸
result = cv2.resize(src, (target_w, target_h), interpolation=cv2.INTER_LINEAR)

多次 resize 会产生累积误差，质量反而更差。

Letterbox vs Squash

Letterbox（保持比例，边缘填充）和 Squash（直接拉伸）是两种完全不同的策略：

def letterbox(image, target_size):
    """
    Letterbox：保持长宽比，边缘填充
    """
    h, w = image.shape[:2]
    tw, th = target_size
    
    scale = min(tw / w, th / h)
    if scale < 1:
        scale = 1
    
    new_w = int(w * scale)
    new_h = int(h * scale)
    
    # Resize
    resized = cv2.resize(image, (new_w, new_h), interpolation=cv2.INTER_LINEAR)
    
    # 边缘填充
    dh, dw = (th - new_h) // 2, (tw - new_w) // 2
    padded = cv2.copyMakeBorder(
        resized, dh, th - new_h - dh, dw, tw - new_w - dw,
        cv2.BORDER_CONSTANT, value=(114, 114, 114)
    )
    
    return padded, scale, (dw, dh)


def squash(image, target_size):
    """
    Squash：直接拉伸到目标尺寸
    """
    return cv2.resize(image, target_size, interpolation=cv2.INTER_LINEAR)

大多数检测模型用 Letterbox，因为直接拉伸会导致形变，影响检测框的准确性。

Normalize 的最佳实践

Normalize 有两种常见的方式：

方式一：除以 255（归一化到 0-1）

# 适用于 [0, 255] -> [0, 1]
image = image.astype(np.float32) / 255.0

方式二：标准化（减均值除标准差）

# ImageNet 标准
mean = np.array([0.485, 0.456, 0.406]).reshape(1, 1, 3)
std = np.array([0.229, 0.224, 0.225]).reshape(1, 1, 3)

image = (image / 255.0 - mean) / std

昇腾建议第二种方式在 NPU 上执行：

# 在 NPU 上做标准化
image_tensor = torch.from_numpy(image).npu()
mean_tensor = torch.tensor(mean).npu()
std_tensor = torch.tensor(std).npu()

normalized = (image_tensor - mean_tensor) / std_tensor

原因是：昇腾的 Vector Unit 对这种融合操作有专门的优化，一条指令就能完成减法和除法。

完整的昇腾 CV Pipeline

import vic
import torch
import torch_npu


def create_inference_pipeline(target_size=(640, 640)):
    """
    生产级的推理 Pipeline
    """
    # 1. 初始化 VIC
    pipeline = vic.Pipeline(
        input_layout='HWC',
        output_layout='NCHW',
        target_size=target_size,
        # ImageNet 标准化
        mean=[0.485, 0.456, 0.406],
        std=[0.229, 0.224, 0.225],
        # 不做增强
        normalize=True,
        # 输出数据类型
        output_dtype='float32'
    )
    
    return pipeline


def inference_wrapper(pipeline, image_path):
    """
    包装成模型可以接受的格式
    """
    # 读取 + VIC 处理
    image = cv2.imread(image_path)
    tensor = pipeline.process(image)  # (C, H, W)
    
    # NCHW
    tensor = tensor.unsqueeze(0)  # (1, C, H, W)
    
    return tensor


# 使用
target_size = (640, 640)
pipeline = create_inference_pipeline(target_size)

# 测试
test_output = inference_wrapper(pipeline, 'test.jpg')
print(f"Output shape: {test_output.shape}")  # (1, 3, 640, 640)

性能对比

用 YOLOv8s 做端到端的预处理 benchmark：

方式	预处理延迟	端到端延迟	FPS
CPU (OpenCV)	4.2ms	12.4	80
VIC	0.8ms	9.0	111
VIC + 批处理	0.4ms/item	8.6	116

关键改进来自于两个方面：

数据不需要 CPU-NPU 来回传输
VIC 的 resize 和 normalize 是融合的，内部零拷贝

数据增强的训练 Pipeline

训练时的数据增强可以更激进：

class TrainPipeline:
    def __init__(self, input_size=640):
        self.pipeline = vic.Pipeline(
            input_layout='HWC',
            output_layout='NCHW',
            target_size=input_size,
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225],
            # 训练增强
            random_flip=True,
            flip_prob=0.5,
            random_crop=True,
            crop_range=(0.8, 1.0),
            color_jitter=True,
            brightness=0.2,
            contrast=0.2,
            saturation=0.2,
            hue=0.1,
            # 可选的增强
            random_affine=False,
            random_perspective=False,
            output_dtype='float32'
        )
    
    def process(self, image):
        return self.pipeline.process(image)

训练时开启这些增强能显著提升模型的泛化能力，而且因为是在 VIC 上跑，不会成为瓶颈。