在 NPU 上运行深度估计 - Qualcomm Dragonwing Documentation

本页是 qrb_ros_samples 目录中 sample_depth_estimation 的手工实现版本。您无需安装打包好的示例，而是亲自构建这个 ROS 2 节点——订阅摄像头话题，通过 Qualcomm QNN delegate 在 Hexagon HTP NPU 上运行量化的 TFLite 模型，并发布彩色化的深度图像以及原始逆深度图。目标是端到端地展示各个部分如何组合在一起，以便您可以为任何模型构建自己的节点。

为什么要自己构建而不是使用示例？ 示例目录是一个很好的起点，但您最终会遇到它未覆盖的模型或管线形态。本页演示了与示例相同的模式——QNN delegate 加载、预处理、推理、后处理、发布——并接入标准的 sensor_msgs / cv_bridge 和来自 Qualcomm AI Hub 的模型。只要看过一次，您就可以将 MiDaS v2 换成任何 AI Hub 模型并复用相同的脚手架。本页和 sample_depth_estimation 都以同一个 Hexagon HTP NPU 为目标——只是本页把每一处连接都展示了出来。

各阶段的运行位置

阶段	运行位置	说明
摄像头采集	ISP	摄像头硬件模块。
颜色转换（YUYV → BGR）	CPU	在 `v4l2_camera` 内部完成。可通过 IM SDK 的 GStreamer 插件卸载到 GPU。
预处理（缩放、归一化）	CPU	`midas_tflite.py` 中的 `cv2.resize` + NumPy。
推理	NPU（Hexagon HTP）	通过 QNN TFLite delegate（`libQnnTFLiteDelegate.so`，后端为 `htp`）。这是 Qualcomm 的差异化优势。
后处理（彩色化）	CPU	`cv2.applyColorMap`。
发布	CPU	`rclpy` + `cv_bridge`。

与普通 ROS 2 TFLite 节点的区别

普通 TFLite 在 CPU 上运行（最多用到 OpenCL GPU）。 Hexagon HTP NPU 只能通过 Qualcomm QNN delegate 或 QNN SDK 访问，而这正是本管线所加载的。
每个节点边界都是一次 memcpy。 cv_bridge + v4l2_camera 在每一跳都会分配并复制完整帧。对于硬件到硬件的管线（摄像头 ISP → NPU），这一复制是可以避免的——参见 qrb_ros_transport 了解 DMA‑buf fd 传递。
NVIDIA Isaac ROS / Intel OpenVINO 软件包面向不同的芯片（NVIDIA GPU / Intel VPU），无法在 Qualcomm 硬件上运行。

前提条件： 开始之前请完成软件设置工作流程。本页使用发布 /image_raw 的 USB 或 ISP 摄像头，而不是特定的机器人平台。

此工作流程中的前处理和后处理——图像解码、缩放、颜色转换和可视化——在 CPU 上运行。如需 GPU 加速的前后处理，请使用 IM SDK 中的 GStreamer 插件。

安装前提条件

安装 Python TFLite 运行时，验证 QNN TFLite delegate，并安装摄像头驱动。TFLite 运行时

pip install ai-edge-litert

如果您的平台上没有 ai-edge-litert，可以回退到 tflite-runtime：

pip install tflite-runtime

QNN TFLite delegate — 确认共享库存在：

ls /usr/lib/libQnnTFLiteDelegate.so

如果该文件缺失，请先为您的设备安装 Qualcomm AI SDK 或 QNN 运行时软件包，然后再继续。该库必须位于 /usr/lib/libQnnTFLiteDelegate.so（节点加载的默认路径）。摄像头驱动

sudo apt install -y ros-jazzy-v4l2-camera

获取模型

Qualcomm AI Hub 上的所有模型在下载之前，均已针对您的特定目标设备进行编译和验证。

访问 https://aihub.qualcomm.com/models/midas。
选择 Export 并选择您的目标设备——IQ8 选择 IQ‑8275 EVK，IQ9 选择 IQ‑9075 EVK。
选择 TFLite 作为运行时，并选择 INT8 量化（w8a8）。
下载导出的 .tflite 文件——其名称为 midas-midas-v2-w8a8.tflite。

下载后，请确认文件名以 .tflite 结尾。如果 AI Hub 只为您的设备显示 ONNX 导出选项，则该组合不支持 TFLite——请参阅 AI 工作流程章节，了解如何改在 NPU 上运行 ONNX 模型。

搭建软件包脚手架

ros2 pkg create 会生成所有样板文件——package.xml、setup.cfg、setup.py、ament 资源标记文件和 __init__.py。

运行这些命令之前请进入您的工作空间根目录。其余步骤中的所有路径均相对于工作空间根目录。

软件包名称必须是 ros2 pkg create 之后的第一个位置参数——将其放在 create 之后、任何标志之前。如果将其放在 --dependencies 之后，CLI 会将其视为另一个依赖项，命令会失败且不会创建任何软件包。

cd src
ros2 pkg create midas_depth_ros \
  --build-type ament_python \
  --dependencies rclpy sensor_msgs std_msgs cv_bridge

创建其余目录并将模型移动到位：

mkdir -p midas_depth_ros/launch
mkdir -p midas_depth_ros/config
mkdir -p midas_depth_ros/models

mv ~/Downloads/midas-midas-v2-w8a8.tflite midas_depth_ros/models/

更新 package.xml

ros2 pkg create 已经添加了 ROS 依赖项。在 <package> 块中追加以下两个 Python 系统依赖项：

<exec_depend>python3-numpy</exec_depend>
<exec_depend>python3-opencv</exec_depend>

添加这些依赖项后，重新运行 rosdep install 以解析它们：

cd ..
rosdep install --from-paths src --ignore-src -r -y

替换 setup.py

生成的 setup.py 需要更新 data_files，以便安装 launch 文件、配置和模型。

from setuptools import setup
from glob import glob

package_name = 'midas_depth_ros'

setup(
    name=package_name,
    version='0.1.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages', ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        ('share/' + package_name + '/launch', glob('launch/*.py')),
        ('share/' + package_name + '/config', glob('config/*.yaml')),
        ('share/' + package_name + '/models', glob('models/*.tflite')),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='maintainer',
    maintainer_email='you@example.com',
    description='MiDaS monocular depth estimation (TFLite) for ROS 2 on the Hexagon HTP NPU.',
    entry_points={
        'console_scripts': [
            'midas_depth_node = midas_depth_ros.midas_depth_node:main',
        ],
    },
)

编写源文件

该软件包有两个源文件：一个 TFLite 封装类，负责 delegate 加载、预处理、推理和可视化；以及一个 ROS 2 节点，负责连接摄像头订阅、运行推理并发布结果。

midas_depth_ros/midas_tflite.py

负责 QNN delegate 加载、输入预处理、推理和深度彩色化的 TFLite 封装：

"""MiDaS v2 (w8a8) TFLite runner for monocular depth estimation.

Input  : [1, H, W, 3] uint8 or float32 (auto-detected from interpreter)
Output : [1, H, W] or [1, H, W, 1] depth (inverse-depth, higher = closer)
"""
import os
import numpy as np
import cv2

try:
    from tflite_runtime.interpreter import Interpreter, load_delegate
except ImportError:
    try:
        from tensorflow.lite.python.interpreter import Interpreter  # type: ignore
        try:
            from tensorflow.lite.python.interpreter import load_delegate  # type: ignore
        except ImportError:
            load_delegate = None
    except ImportError:
        from ai_edge_litert.interpreter import Interpreter  # type: ignore
        try:
            from ai_edge_litert.interpreter import load_delegate  # type: ignore
        except ImportError:
            load_delegate = None


class MidasTFLite:
    def __init__(self, model_path,
                 use_qnn_delegate=False, qnn_delegate_path=None, qnn_backend='htp'):
        if not os.path.isfile(model_path):
            raise FileNotFoundError(model_path)

        delegates = []
        self.delegate_active = False
        self.delegate_error = None
        if use_qnn_delegate:
            if load_delegate is None:
                self.delegate_error = 'load_delegate unavailable in this TFLite runtime'
            elif not qnn_delegate_path or not os.path.isfile(qnn_delegate_path):
                self.delegate_error = f'QNN delegate .so not found at {qnn_delegate_path}'
            else:
                try:
                    opts = {'backend_type': qnn_backend}
                    delegates = [load_delegate(qnn_delegate_path, options=opts)]
                    self.delegate_active = True
                except Exception as e:
                    self.delegate_error = f'load_delegate failed: {e}'

        self.interp = Interpreter(model_path=model_path,
                                  experimental_delegates=delegates or None)
        self.interp.allocate_tensors()
        self.inp = self.interp.get_input_details()[0]
        self.out = self.interp.get_output_details()[0]

        shape = self.inp['shape']  # [1, H, W, 3]
        self.in_h = int(shape[1])
        self.in_w = int(shape[2])

    @staticmethod
    def _dequant(arr, detail):
        q = detail.get('quantization', (0.0, 0))
        scale, zp = q[0], q[1]
        if scale and scale > 0:
            return (arr.astype(np.float32) - zp) * scale
        return arr.astype(np.float32)

    def infer(self, bgr):
        """Run inference on a BGR image; return a float32 HxW inverse-depth map
        in the ORIGINAL image resolution."""
        h, w = bgr.shape[:2]
        rgb = cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
        resized = cv2.resize(rgb, (self.in_w, self.in_h),
                             interpolation=cv2.INTER_CUBIC)

        dtype = self.inp['dtype']
        if dtype == np.uint8:
            x = resized.astype(np.uint8)
        else:
            x = resized.astype(np.float32) / 255.0
            mean = np.array([0.485, 0.456, 0.406], dtype=np.float32)
            std  = np.array([0.229, 0.224, 0.225], dtype=np.float32)
            x = (x - mean) / std

        x = np.expand_dims(x, 0).astype(dtype)

        self.interp.set_tensor(self.inp['index'], x)
        self.interp.invoke()
        raw = self.interp.get_tensor(self.out['index'])
        depth = self._dequant(raw, self.out)

        depth = np.squeeze(depth)  # -> (H, W)
        if depth.ndim != 2:
            depth = depth.reshape(self.in_h, self.in_w)

        return cv2.resize(depth, (w, h), interpolation=cv2.INTER_CUBIC)

    @staticmethod
    def colorize(depth, colormap=cv2.COLORMAP_INFERNO):
        """Normalize to 0..255 and apply a colormap -> BGR uint8 for display."""
        d = depth.astype(np.float32)
        dmin, dmax = float(np.min(d)), float(np.max(d))
        if dmax - dmin < 1e-6:
            norm = np.zeros_like(d, dtype=np.uint8)
        else:
            norm = ((d - dmin) / (dmax - dmin) * 255.0).astype(np.uint8)
        return cv2.applyColorMap(norm, colormap)

midas_depth_ros/midas_depth_node.py

将摄像头订阅、推理和话题发布连接在一起的 ROS 2 节点：

"""ROS 2 node: MiDaS monocular depth estimation from a mono RGB camera."""
import os
import time

import numpy as np
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from cv_bridge import CvBridge
from ament_index_python.packages import get_package_share_directory

from .midas_tflite import MidasTFLite


class MidasDepthNode(Node):
    def __init__(self):
        super().__init__('midas_depth_node')
        self._declare_params()
        self.bridge = CvBridge()

        model_path = self.get_parameter('model_path').value
        if not model_path:
            model_path = os.path.join(
                get_package_share_directory('midas_depth_ros'),
                'models', 'midas-midas-v2-w8a8.tflite')

        self.midas = MidasTFLite(
            model_path=model_path,
            use_qnn_delegate=bool(self.get_parameter('use_qnn_delegate').value),
            qnn_delegate_path=str(self.get_parameter('qnn_delegate_path').value),
            qnn_backend=str(self.get_parameter('qnn_backend').value),
        )
        self.get_logger().info(
            f'MiDaS loaded: {model_path}  input={self.midas.in_w}x{self.midas.in_h}')
        want_npu = bool(self.get_parameter('use_qnn_delegate').value)
        if want_npu and self.midas.delegate_active:
            self.get_logger().info(
                f"✅ QNN delegate ACTIVE on '{self.get_parameter('qnn_backend').value}' "
                f"backend — inference runs on the NPU.")
        elif want_npu:
            self.get_logger().error(
                f'❌ QNN delegate requested but NOT active — running on CPU. '
                f'Reason: {self.midas.delegate_error}')
        else:
            self.get_logger().warn('QNN delegate disabled — running on CPU.')

        qos = 10
        self.sub = self.create_subscription(
            Image, self.get_parameter('image_topic').value, self.on_image, qos)

        self.pub_vis = self.create_publisher(
            Image, self.get_parameter('depth_image_topic').value, qos)
        self.pub_raw = self.create_publisher(
            Image, self.get_parameter('depth_raw_topic').value, qos)

        self._perf_interval = float(self.get_parameter('perf_log_interval_sec').value)
        self._perf_reset()

    def _declare_params(self):
        defaults = {
            'image_topic':           '/image_raw',
            'depth_image_topic':     '/midas/depth_image',
            'depth_raw_topic':       '/midas/depth',
            'model_path':            '',
            'use_qnn_delegate':      False,
            'qnn_delegate_path':     '/usr/lib/libQnnTFLiteDelegate.so',
            'qnn_backend':           'htp',
            'colormap':              'inferno',
            'perf_log_interval_sec': 2.0,
        }
        for name, value in defaults.items():
            self.declare_parameter(name, value)

    _COLORMAPS = {
        'inferno': 14, 'magma': 13, 'viridis': 16, 'plasma': 15,
        'jet': 2, 'turbo': 20, 'hot': 11, 'bone': 1,
    }

    def _perf_reset(self):
        self._perf_window_start = time.monotonic()
        self._perf_frames       = 0
        self._perf_total_ms     = 0.0
        self._perf_max_ms       = 0.0

    def _perf_maybe_log(self, frame_ms: float):
        self._perf_frames   += 1
        self._perf_total_ms += frame_ms
        if frame_ms > self._perf_max_ms:
            self._perf_max_ms = frame_ms
        elapsed = time.monotonic() - self._perf_window_start
        if elapsed < self._perf_interval:
            return
        fps    = self._perf_frames / elapsed if elapsed > 0 else 0.0
        avg_ms = self._perf_total_ms / self._perf_frames if self._perf_frames else 0.0
        self.get_logger().info(
            f'[perf] {fps:5.2f} Hz  avg {avg_ms:6.2f} ms  max {self._perf_max_ms:6.2f} ms  '
            f'(window {self._perf_frames} frames / {elapsed:.1f}s)')
        self._perf_reset()

    def on_image(self, msg: Image):
        t0 = time.monotonic()
        try:
            bgr = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')
        except Exception as e:
            self.get_logger().error(f'cv_bridge failed: {e}')
            return

        depth = self.midas.infer(bgr)

        cmap_name = str(self.get_parameter('colormap').value).lower()
        cmap      = self._COLORMAPS.get(cmap_name, 14)
        vis       = MidasTFLite.colorize(depth, colormap=cmap)

        vis_msg        = self.bridge.cv2_to_imgmsg(vis, encoding='bgr8')
        vis_msg.header = msg.header
        self.pub_vis.publish(vis_msg)

        raw_msg        = self.bridge.cv2_to_imgmsg(depth.astype(np.float32), encoding='32FC1')
        raw_msg.header = msg.header
        self.pub_raw.publish(raw_msg)

        self._perf_maybe_log((time.monotonic() - t0) * 1000.0)


def main():
    rclpy.init()
    node = MidasDepthNode()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

编写 launch 和配置文件

from launch import LaunchDescription
from launch_ros.actions import Node
from ament_index_python.packages import get_package_share_directory
import os


def generate_launch_description():
    pkg_share = get_package_share_directory('midas_depth_ros')
    params    = os.path.join(pkg_share, 'config', 'params.yaml')

    return LaunchDescription([
        Node(
            package='midas_depth_ros',
            executable='midas_depth_node',
            name='midas_depth_node',
            output='screen',
            parameters=[params],
        )
    ])

midas_depth_node:
  ros__parameters:
    image_topic: /image_raw
    depth_image_topic: /midas/depth_image
    depth_raw_topic: /midas/depth
    model_path: ''
    use_qnn_delegate: true
    qnn_delegate_path: /usr/lib/libQnnTFLiteDelegate.so
    qnn_backend: htp
    colormap: inferno
    perf_log_interval_sec: 2.0

关键参数：

参数	选项	说明
`use_qnn_delegate`	`true` / `false`	`true` 在 HTP NPU 上运行；`false` 回退到 CPU
`qnn_backend`	`htp`、`gpu`、`cpu`	`htp` 面向 Hexagon NPU
`colormap`	`inferno`、`magma`、`viridis`、`plasma`、`jet`、`turbo`、`hot`、`bone`	应用于发布的深度可视化的色彩映射

构建

colcon build --packages-select midas_depth_ros --symlink-install
source install/setup.bash

运行

如果摄像头尚未运行，请先启动它：

ros2 run v4l2_camera v4l2_camera_node \
  --ros-args -p video_device:=/dev/video0 -p pixel_format:=YUYV \
  -p image_size:=[640,480] -p camera_frame_id:=camera_link

然后启动推理节点：

ros2 launch midas_depth_ros midas_depth.launch.py

启动时，节点会记录 NPU delegate 是否加载成功：

✅ QNN delegate ACTIVE on 'htp' backend — inference runs on the NPU.

如果 delegate 加载失败，则回退到 CPU 并记录原因：

❌ QNN delegate requested but NOT active — running on CPU. Reason: ...

话题

方向	话题	类型	说明
sub	`/image_raw`	`sensor_msgs/Image` bgr8	来自 `v4l2_camera` 的摄像头输入
pub	`/midas/depth_image`	`sensor_msgs/Image` bgr8	用于 RViz 的彩色化逆深度
pub	`/midas/depth`	`sensor_msgs/Image` 32FC1	原始逆深度（值越大表示越近）

在 RViz 中可视化

添加一个 Image 显示项并将话题设置为 /midas/depth_image。使用默认的 inferno 色彩映射时，彩色化输出会将更近的物体映射为更亮的值。

后续步骤

将此脚手架适配到其他模型。 将步骤 2 中的 MiDaS 导出模型换成 Qualcomm AI Hub 上的任何 TFLite 模型，并在 midas_tflite.py 中调整预处理。delegate 加载、话题连接以及 launch/config 文件均可原样沿用。
想避免摄像头与此节点之间的逐帧 CPU 复制？ 参见 qrb_ros_transport 了解零拷贝 DMA-buf 传递。
更喜欢此管线的打包版本？ qrb_ros_samples 中的 sample_depth_estimation 提供了预先连接好的相同管线——当您想在不自行构建节点的情况下运行深度估计时可以使用它。

​各阶段的运行位置

​与普通 ROS 2 TFLite 节点的区别

​话题

​在 RViz 中可视化

​后续步骤

各阶段的运行位置

与普通 ROS 2 TFLite 节点的区别

话题

在 RViz 中可视化

后续步骤