在 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/配置文件均可原样复用。
想避免摄像头与此节点之间每帧的 CPU 复制？ 参见 qrb_ros_transport，了解零拷贝 DMA-buf 传递。
更喜欢此流水线的打包版本？ qrb_ros_samples 中的 sample_depth_estimation 提供了预先连接好的相同流水线——当您想运行深度估计而无需自己构建节点时可以使用它。

​每个阶段在哪里运行

​这与标准 ROS 2 TFLite 节点有何不同

​话题

​在 RViz 中可视化

​后续步骤

每个阶段在哪里运行

这与标准 ROS 2 TFLite 节点有何不同

话题

在 RViz 中可视化

后续步骤