Qualcomm® 智能多媒体 (IM) SDK

Qualcomm® 智能多媒体 (IM) SDK 是一组 GStreamer 插件,可让您在 Dragonwing 开发板的 GPU 上运行计算机视觉操作;并可以创建完全运行在 GPU 和 NPU 上的 AI 流水线,而无需回到 CPU(零拷贝)。综合来看,这使您能够获得比自己用 OpenCV + TFLite 等方式实现 AI 计算机视觉流水线高得多的吞吐率。

那么…… GStreamer 流水线是什么?

IM SDK 构建在 GStreamer 之上。GStreamer 是一个多媒体框架,可让您描述视频或音频的处理流水线,并负责按顺序运行每个步骤。在”普通 Python”中,您可能会编写 OpenCV 代码:从摄像头抓取一帧,对其进行缩放和裁剪,调用推理函数,在结果上绘制边界框,然后再输出或显示该帧——除非您显式接入 GPU/NPU API,否则每个步骤都在 CPU 上运行。使用 GStreamer + IM SDK,您只需在一个流水线字符串中声明同样的处理序列,框架就会替您将帧依次流经整个处理链。 IM SDK 在 Qualcomm 硬件上带来的能力是让这些步骤可以被透明地加速:缩放/裁剪和绘制边界框可以在 GPU 上运行,推理可以在 NPU 上运行,而整条操作链(例如裁剪 → 缩放 → 神经网络推理)可以在完全不回到 CPU 的情况下执行(零拷贝)。在您的应用程序中,您只需配置流水线;底层框架会处理逐帧调度、同步和加速器卸载。 IM SDK 提供了使这一切成为可能的特殊 GStreamer 插件。例如,qtivtransform 将颜色转换、裁剪和缩放卸载到 GPU,而 qtimltflite 负责在 NPU 上进行推理。这样,您用标准 GStreamer 编写的同样高层的流水线,现在几乎可以完全运行在专用加速器上,以极小的 CPU 负载实现实时吞吐。

设置 GStreamer 和 IM SDK

好了,让我们开始使用 IM SDK 构建一些应用程序。

安装 GStreamer、IM SDK 以及本示例中需要的一些额外依赖。在开发板上打开终端,或通过 SSH 会话连接到开发板,然后运行:

if [ ! -f /etc/apt/sources.list.d/ubuntu-qcom-iot-ubuntu-qcom-ppa-noble.list ]; then
    sudo apt-add-repository -y ppa:ubuntu-qcom-iot/qcom-ppa
fi

# Install GStreamer / IM SDK
sudo apt update
sudo apt install -y gstreamer1.0-tools gstreamer1.0-plugins-good gstreamer1.0-plugins-base gstreamer1.0-plugins-base-apps gstreamer1.0-plugins-qcom-good gstreamer1.0-qcom-sample-apps

# Install Python bindings for GStreamer, and some build dependencies
sudo apt install -y v4l-utils libcairo2-dev pkg-config python3-dev libgirepository1.0-dev gir1.2-gstreamer-1.0

获取 Python 示例,创建一个 venv,并安装其依赖:

# Create a new venv
python3 -m venv .venv
source .venv/bin/activate

# Install Python dependencies
pip3 install -r requirements.txt

您需要一个摄像头(内置摄像头,如 RB3 Gen 2 Vision Kit 上的那种,或 USB 网络摄像头)。

如果要使用 USB 网络摄像头:

找出设备 ID:

v4l2-ctl --list-devices
# msm_vidc_media (platform:aa00000.video-codec):
#         /dev/media0
#
# msm_vidc_decoder (platform:msm_vidc_bus):
#         /dev/video32
#         /dev/video33
#
# C922 Pro Stream Webcam (usb-0000:01:00.0-2):
#         /dev/video2     <-- So /dev/video2
#         /dev/video3
#         /dev/media3

设置环境变量(我们将在示例中使用它):

export IMSDK_VIDEO_SOURCE="v4l2src device=/dev/video2"

如果您使用的是 RB3 Gen 2 Vision Kit,并希望使用内置摄像头:
export IMSDK_VIDEO_SOURCE="qtiqmmfsrc name=camsrc camera=0"

示例 1:在 GPU 与 CPU 上进行缩放和裁剪

让我们展示一下在 GPU 上处理相比 CPU 能快多少。如果您有一个需要 224x224 RGB 输入的神经网络,则需要对数据进行预处理:首先从网络摄像头抓取帧(例如原生分辨率为 1980x1080),然后裁剪为 1/1 宽高比(例如裁剪为 1080x1080),再缩放到所需分辨率(224x224),最后从像素创建 Numpy 数组。

创建一个新文件 ex1.py,并添加:

from gst_helper import gst_grouped_frames, atomic_save_image, timing_marks_to_str
import time, argparse

parser = argparse.ArgumentParser(description='GStreamer -> Python RGB frames')
parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
args, unknown = parser.parse_known_args()

PIPELINE = (
    # Video source
    f"{args.video_source} ! "
    # Properties for the video source
    "video/x-raw,width=1920,height=1080 ! "
    # An identity element so we can track when a new frame is ready (so we can calc. processing time)
    "identity name=frame_ready_webcam silent=false ! "
    # Crop to square
    "videoconvert ! aspectratiocrop aspect-ratio=1/1 ! "
    # Scale to 224x224 and RGB
    "videoscale ! video/x-raw,format=RGB,width=224,height=224 ! "
    # Event when the crop/scale are done
    "identity name=transform_done silent=false ! "
    # Send out the resulting frame to an appsink (where we can pick it up from Python)
    "queue max-size-buffers=2 leaky=downstream ! "
    "appsink name=frame drop=true sync=false max-buffers=1 emit-signals=true"
)

for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
    print(f"Frame ready")
    print('    Data:', end='')
    for key in list(frames_by_sink):
        print(f' name={key} {frames_by_sink[key].shape}', end='')
    print('')
    print('    Timings:', timing_marks_to_str(marks))

    # Save image to disk, frames_by_sink has all the
    frame = frames_by_sink['frame']
    atomic_save_image(frame=frame, path='out/gstreamer.png')

让我们启动这个 Python 脚本。该流水线运行在 CPU 上(使用原生 GStreamer 组件):

python3 ex1.py --video-source "$IMSDK_VIDEO_SOURCE"

# Frame ready
#     Data: name=frame (224, 224, 3)
#     Timings: frame_ready_webcam->transform_done: 22.16ms, transform_done->pipeline_finished: 2.14ms (total 24.31ms)
# Frame ready
#     Data: name=frame (224, 224, 3)
#     Timings: frame_ready_webcam->transform_done: 22.21ms, transform_done->pipeline_finished: 1.25ms (total 23.46ms)

可以看到缩放/裁剪耗时 22ms(在 IQ9 上使用 USB 摄像头测得)。

现在让我们改为在 GPU 上运行…… 将:

    # Crop to square
    "videoconvert ! aspectratiocrop aspect-ratio=1/1 ! "
    # Scale to 224x224 and RGB
    "videoscale ! video/x-raw,format=RGB,width=224,height=224 ! "

替换为:

    # Crop (square), the crop syntax is ('<X, Y, WIDTH, HEIGHT >').
    # So here we use 1920x1080 input, then center crop to 1080x1080 ((1920-1080)/2 = 420 = x crop)
    f'qtivtransform crop="<420, 0, 1080, 1080>" ! '
    # then resize to 224x224
    "video/x-raw,format=RGB,width=224,height=224 ! "

以下是完整文件 ex1_imsdk.py:

from gst_helper import gst_grouped_frames, atomic_save_image, timing_marks_to_str
import time, argparse

parser = argparse.ArgumentParser(description='GStreamer -> Python RGB frames')
parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
args, unknown = parser.parse_known_args()

PIPELINE = (
    # Video source
    f"{args.video_source} ! "
    # Properties for the video source
    "video/x-raw,width=1920,height=1080 ! "
    # An identity element so we can track when a new frame is ready (so we can calc. processing time)
    "identity name=frame_ready_webcam silent=false ! "
    # Crop (square), the crop syntax is ('<X, Y, WIDTH, HEIGHT >').
    # So here we use 1920x1080 input, then center crop to 1080x1080 ((1920-1080)/2 = 420 = x crop)
    f'qtivtransform crop="<420, 0, 1080, 1080>" ! '
    # then resize to 224x224
    "video/x-raw,format=RGB,width=224,height=224 ! "
    # Event when the crop/scale are done
    "identity name=transform_done silent=false ! "
    # Send out the resulting frame to an appsink (where we can pick it up from Python)
    "queue max-size-buffers=2 leaky=downstream ! "
    "appsink name=frame drop=true sync=false max-buffers=1 emit-signals=true"
)

for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
    print(f"Frame ready")
    print('    Data:', end='')
    for key in list(frames_by_sink):
       print(f' name={key} {frames_by_sink[key].shape}', end='')
    print('')
    print('    Timings:', timing_marks_to_str(marks))

    # Save image to disk, frames_by_sink has all the
    frame = frames_by_sink['frame']
    atomic_save_image(frame=frame, path='out/gstreamer.png')

再次运行:

python3 ex1.py --video-source "$IMSDK_VIDEO_SOURCE"

# Frame ready
#     Data: name=frame (224, 224, 3)
#     Timings: frame_ready_webcam->transform_done: 6.55ms, transform_done->pipeline_finished: 0.78ms (total 7.33ms)
# Frame ready
#     Data: name=frame (224, 224, 3)
#     Timings: frame_ready_webcam->transform_done: 6.60ms, transform_done->pipeline_finished: 0.78ms (total 7.38ms)

🚀 只用两行代码,您就将裁剪/缩放操作从约 22ms 加速到了约 6ms!

示例 2:流的分支(Tee)与多路输出

在上面的流水线中,您已经看到了几个与您自己的代码交互时相关的元素:

Identity 元素(例如 identity name=frame_ready_webcam silent=false)。可用于调试流水线中的时序。它们被触发时的时间戳会被保存,并在流水线结束时通过 marks 元素返回(键值对,键为 identity 名称,值为时间戳)。
Appsink 元素(例如 appsink name=frame)。用于将数据从 GStreamer 流水线发送到您的应用程序。这里 appsink 之前的元素是 video/x-raw,format=RGB,width=224,height=224——因此我们会向 Python 发送一个 224x224 的 RGB 数组。您可以在 frames_by_sink 元素中接收它们(键值对,键为 appsink 名称,值为数据)。

每条流水线可以有多个 appsink。例如,您可能还想获取原始的 1920x1080 图像。在这种情况下,您可以在 identity name=frame_ready_webcam 之后将流水线拆分为两部分:一部分发送到新的 appsink,另一部分经过缩放/裁剪流水线。

创建一个新文件 ex2.py 并添加:

from gst_helper import gst_grouped_frames, atomic_save_image, timing_marks_to_str
import time, argparse

parser = argparse.ArgumentParser(description='GStreamer -> Python RGB frames')
parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
args, unknown = parser.parse_known_args()

PIPELINE = (
    # Video source
    f"{args.video_source} ! "
    # Properties for the video source
    "video/x-raw,width=1920,height=1080 ! "
    # An identity element so we can track when a new frame is ready (so we can calc. processing time)
    "identity name=frame_ready_webcam silent=false ! "

    # Split the stream
    "tee name=t "

    # Branch A) convert to RGB and send to original appsink
        "t. ! queue max-size-buffers=1 leaky=downstream ! "
        "qtivtransform ! video/x-raw,format=RGB ! "
        "appsink name=original drop=true sync=false max-buffers=1 emit-signals=true "

    # Branch B) resize/crop to 224x224 -> send to another appsink
        "t. ! queue max-size-buffers=1 leaky=downstream ! "
        # Crop (square), the crop syntax is ('<X, Y, WIDTH, HEIGHT >').
        # So here we use 1920x1080 input, then center crop to 1080x1080 ((1920-1080)/2 = 420 = x crop)
        f'qtivtransform crop="<420, 0, 1080, 1080>" ! '
        # then resize to 224x224
        "video/x-raw,format=RGB,width=224,height=224 ! "
        # Event when the crop/scale are done
        "identity name=transform_done silent=false ! "
        # Send out the resulting frame to an appsink (where we can pick it up from Python)
        "queue max-size-buffers=2 leaky=downstream ! "
        "appsink name=frame drop=true sync=false max-buffers=1 emit-signals=true "
)

for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
    print(f"Frame ready")
    print('    Data:', end='')
    for key in list(frames_by_sink):
        print(f' name={key} {frames_by_sink[key].shape}', end='')
    print('')
    print('    Timings:', timing_marks_to_str(marks))

    # Save image to disk
    frame = frames_by_sink['frame']
    atomic_save_image(frame=frame, path='out/imsdk.png')
    original = frames_by_sink['original']
    atomic_save_image(frame=original, path='out/imsdk_original.png')

运行这段 Python 代码:

python3 ex2.py --video-source "$IMSDK_VIDEO_SOURCE"

# Frame ready
#      Data: name=frame (224, 224, 3) name=original (1080, 1920, 3)
#      Timings: frame_ready_webcam->transform_done: 5.42ms, transform_done->pipeline_finished: 4.22ms (total 9.64ms)
# Frame ready
#      Data: name=frame (224, 224, 3) name=original (1080, 1920, 3)
#      Timings: frame_ready_webcam->transform_done: 5.51ms, transform_done->pipeline_finished: 4.41ms (total 9.92ms)

(out/ 目录中保存了最近处理的帧,包括原始分辨率和缩放后的分辨率)

很好!这样一条流水线就有了两个输出。现在您已经知道如何在单条流水线中构建更复杂的应用程序了。

示例 3:运行神经网络

现在我们已经能以正确的分辨率从网络摄像头获取图像流,让我们加入一个神经网络。

以下工作流仅适用于 Ubuntu Server 版本,不适用于 Ubuntu Desktop 操作系统。

3.1:在 Python 中进行神经网络推理与图像合成

首先我们做一个”常规”实现:从 IM SDK 流水线获取缩放后的帧,然后使用 LiteRT 运行模型(在 NPU 上)。之后我们会将最高置信度的预测绘制到图像上并写入磁盘。创建一个新文件 ex3_from_python.py 并添加:

from gst_helper import gst_grouped_frames, atomic_save_pillow_image, timing_marks_to_str, download_file_if_needed, softmax
import time, argparse, numpy as np
from ai_edge_litert.interpreter import Interpreter, load_delegate
from PIL import ImageDraw, Image

parser = argparse.ArgumentParser(description='GStreamer -> SqueezeNet')
parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
args, unknown = parser.parse_known_args()

MODEL_PATH = download_file_if_needed('models/squeezenet1_1-squeezenet-1.1-w8a8.tflite', 'https://cdn.edgeimpulse.com/qc-ai-docs/models/squeezenet1_1-squeezenet-1.1-w8a8.tflite')
LABELS_PATH = download_file_if_needed('models/SqueezeNet-1.1_labels.txt', 'https://cdn.edgeimpulse.com/qc-ai-docs/models/SqueezeNet-1.1_labels.txt')

# Parse labels
with open(LABELS_PATH, 'r') as f:
    labels = [line for line in f.read().splitlines() if line.strip()]

# Load TFLite model and allocate tensors, note: this is a 224x224 model with uint8 input!
# If your models are different, then you'll need to update the pipeline below.
interpreter = Interpreter(
    model_path=MODEL_PATH,
    experimental_delegates=[load_delegate("libQnnTFLiteDelegate.so", options={"backend_type": "htp"})]     # Use NPU
)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

PIPELINE = (
    # Video source
    f"{args.video_source} ! "
    # Properties for the video source
    "video/x-raw,width=1920,height=1080 ! "
    # An identity element so we can track when a new frame is ready (so we can calc. processing time)
    "identity name=frame_ready_webcam silent=false ! "
    # Crop (square), the crop syntax is ('<X, Y, WIDTH, HEIGHT >').
    # So here we use 1920x1080 input, then center crop to 1080x1080 ((1920-1080)/2 = 420 = x crop)
    f'qtivtransform crop="<420, 0, 1080, 1080>" ! '
    # then resize to 224x224
    "video/x-raw,format=RGB,width=224,height=224 ! "
    # Event when the crop/scale are done
    "identity name=transform_done silent=false ! "
    # Send out the resulting frame to an appsink (where we can pick it up from Python)
    "queue max-size-buffers=2 leaky=downstream ! "
    "appsink name=frame drop=true sync=false max-buffers=1 emit-signals=true "
)

for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
    print(f"Frame ready")
    print('    Data:', end='')
    for key in list(frames_by_sink):
        print(f' name={key} {frames_by_sink[key].shape}', end='')
    print('')

    # Begin inference timer
    inference_start = time.perf_counter()

    # Set tensor with the image received in "frames_by_sink['frame']", add batch dim, and run inference
    interpreter.set_tensor(input_details[0]['index'], frames_by_sink['frame'].reshape((1, 224, 224, 3)))
    interpreter.invoke()

    # Get prediction (dequantized)
    q_output = interpreter.get_tensor(output_details[0]['index'])
    scale, zero_point = output_details[0]['quantization']
    f_output = (q_output.astype(np.float32) - zero_point) * scale

    # Image classification models in AI Hub miss a Softmax() layer at the end of the model, so add it manually
    scores = softmax(f_output[0])

    # End inference timer
    inference_end = time.perf_counter()

    # Add an extra mark, so we have timing info for the complete pipeline
    marks['inference_done'] = list(marks.items())[-1][1] + (inference_end - inference_start)

    # Print top-5 predictions
    top_k = scores.argsort()[-5:][::-1]
    print(f"    Top-5 predictions:")
    for i in top_k:
        print(f"        Class {labels[i]}: score={scores[i]}")

    # Image composition timer
    image_composition_start = time.perf_counter()

    # Add the top 5 scores to the image, and save image to disk (for debug purposes)
    frame = frames_by_sink['frame']
    img = Image.fromarray(frame)
    img_draw = ImageDraw.Draw(img)
    img_draw.text((10, 10), f"{labels[top_k[0]]} ({scores[top_k[0]]:.2f})", fill="black")
    atomic_save_pillow_image(img=img, path='out/imsdk_with_prediction.png')

    image_composition_end = time.perf_counter()

    # Add an extra mark, so we have timing info for the complete pipeline
    marks['image_composition_end'] = list(marks.items())[-1][1] + (image_composition_end - image_composition_start)

    print('    Timings:', timing_marks_to_str(marks))

现在运行此应用程序:

# We use '| grep -v "<W>"' to filter out some warnings - you can omit it if you want.
python3 ex3_from_python.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "<W>"

# Frame ready
#     Data: name=frame (224, 224, 3)
# Top-5 predictions:
#     Class laptop: score=0.4951600134372711
#     Class notebook: score=0.33943524956703186
#     Class computer keyboard: score=0.07495530694723129
#     Class space bar: score=0.062059469521045685
#     Class typewriter keyboard: score=0.007778045255690813
# Timings: frame_ready_webcam->transform_done: 6.81ms, transform_done->pipeline_finished: 0.77ms, pipeline_finished->inference_done: 1.19ms, inference_done->image_composition_end: 37.85ms (total 46.61ms)

结果相当不错,但让我们看看能否做得更好……

3.2:使用 IM SDK 运行神经网络

让我们把神经网络推理移到 IM SDK 中。这需要通过三个插件来实现:

qtimlvconverter——将帧转换为输入张量。
qtimltflite——运行神经网络(LiteRT 格式)。如果您通过 appsink 发送这些结果,您会得到与之前完全相同的张量(只是不再需要经过 CPU 来调用推理引擎)。
一个类似 qtimlpostprocess 的元素,用于解析输出。这里该插件适用于输出形状为 (1, n) 的图像分类用例(如我们使用的 SqueezeNet 模型)。该插件的输出可以是文本(预测结果),也可以是叠加层(绘制到原始图像上)。

注意: 此元素要求特定的标签格式(见下文)。

创建一个新文件 ex3_nn_imsdk.py 并添加:

 from gst_helper import gst_grouped_frames, atomic_save_pillow_image, timing_marks_to_str, download_file_if_needed
 import argparse, numpy as np
 from PIL import Image

 parser = argparse.ArgumentParser(description='GStreamer -> SqueezeNet')
 parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
 args, unknown = parser.parse_known_args()

 MODEL_PATH = download_file_if_needed('models/squeezenet1_1-squeezenet-1.1-w8a8.tflite', 'https://cdn.edgeimpulse.com/qc-ai-docs/models/squeezenet1_1-squeezenet-1.1-w8a8.tflite')
 LABELS_PATH = download_file_if_needed('models/SqueezeNet-1.1_labels.txt', 'https://cdn.edgeimpulse.com/qc-ai-docs/models/SqueezeNet-1.1_labels.txt')

 # Parse labels
 with open(LABELS_PATH, 'r') as f:
     labels = [line for line in f.read().splitlines() if line.strip()]

 PIPELINE = (
     # Video source
     f"{args.video_source} ! "
     # Properties for the video source
     "video/x-raw,width=1920,height=1080 ! "
     # An identity element so we can track when a new frame is ready (so we can calc. processing time)
     "identity name=frame_ready_webcam silent=false ! "
     'qtivtransform ! '
     # NV12 for tightly packed buffer
     "video/x-raw,format=NV12 ! "
     # Mark after transform
     "identity name=transform_done silent=false ! "

     # turn into right format (UINT8 data type) and add batch dimension
     'qtimlvconverter ! neural-network/tensors,type=UINT8,dimensions=<<1,224,224,3>> ! '
     # Event when conversion is done
     "identity name=conversion_done silent=false ! "
     # run inference (using the QNN delegates to run on NPU)
     f'qtimltflite delegate=external external-delegate-path=libQnnTFLiteDelegate.so external-delegate-options="QNNExternalDelegate,backend_type=htp;" model="{MODEL_PATH}" ! '
     # Event when inference is done
     "identity name=inference_done silent=false ! "

     # Post-process (Mobilenet-style softmax).
     f'qtimlpostprocess name=postproc module=mobilenet-softmax '
     f'labels="{LABELS_PATH}" results=5 settings="{{\\"confidence\\": 10.0}}" ! '
     "text/x-raw,format=utf8 ! "
     # Send to application
     "queue max-size-buffers=2 leaky=downstream ! "
     'appsink name=qtimlpostprocess_text drop=true sync=false max-buffers=1 emit-signals=true '
 )

 for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
     print("Frame ready")
     print(' Data:', end='')
     for key in list(frames_by_sink):
         print(f' name={key} {frames_by_sink[key].shape} ({frames_by_sink[key].dtype})', end='')
     print('')

     # Grab the qtimlpostprocess_text (utf8 text) with predictions from IM SDK
     cls_text = frames_by_sink['qtimlpostprocess_text'].tobytes().decode('utf-8')
     print(' qtimlpostprocess_text:', cls_text)
     print(' Timings:', timing_marks_to_str(marks))

NV12: 这里我们(在 qtivtransform 之后)从 RGB 格式切换到了 NV12 格式,因为 qtimltflite 需要紧密打包的缓冲区——而 RGB 输出使用了行跨距(row-stride)填充。这类问题可能非常难以调试。在命令前添加 GST_DEBUG=3(例如 GST_DEBUG=3 python3 ex3_nn_imsdk.py),并将流水线和错误信息提供给 ChatGPT 之类的 LLM,有时可以帮助您排查问题。 module=mobilenet-softmax: qtimlpostprocess 将其用于输出为 FLOAT32 1×N logits 向量的分类模型。它应用 Softmax 将 logits 归一化为概率。

现在运行此应用程序:

# We use '| grep -v "<W>"' to filter out some warnings - you can omit it if you want.
python3 ex3_nn_imsdk.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "<W>"

# Frame ready
#     Data: name=qtimlpostprocess_text (892,) (uint8)
#     qtimlpostprocess_text: { (structure)"ImageClassification\,\ labels\=\(structure\)\<\ \"laptop\\\,\\\ id\\\=\\\(uint\\\)256\\\,\\\ confidence\\\=\\\(double\\\)45.182029724121094\\\,\\\ color\\\=\\\(uint\\\)3211364863\\\;\"\,\ \"notebook\\\,\\\ id\\\=\\\(uint\\\)256\\\,\\\ confidence\\\=\\\(double\\\)17.578998565673828\\\,\\\ color\\\=\\\(uint\\\)2015954943\\\;\"\,\ \"computer.keyboard\\\,\\\ id\\\=\\\(uint\\\)256\\\,\\\ confidence\\\=\\\(double\\\)2.6610369682312012\\\,\\\ color\\\=\\\(uint\\\)2861764863\\\;\"\,\ \"monitor\\\,\\\ id\\\=\\\(uint\\\)256\\\,\\\ confidence\\\=\\\(double\\\)2.2032134532928467\\\,\\\ color\\\=\\\(uint\\\)1091632127\\\;\"\,\ \"desktop.computer\\\,\\\ id\\\=\\\(uint\\\)256\\\,\\\ confidence\\\=\\\(double\\\)2.2032134532928467\\\,\\\ color\\\=\\\(uint\\\)3377702399\\\;\"\ \>\,\ timestamp\=\(guint64\)6025656009\,\ sequence-index\=\(uint\)1\,\ sequence-num-entries\=\(uint\)1\;" }
#     Timings: frame_ready_webcam->transform_done: 7.29ms, transform_done->conversion_done: 0.83ms, conversion_done->inference_done: 1.25ms, inference_done->postproc_done: 0.45ms (total 9.82ms)

OK!模型现在在 IM SDK 流水线内部的 NPU 上运行。如果您更希望获取前 5 名的输出(就像 3.1 中那样),可以在 qtimltflite 元素之后对流进行分支(tee),并将原始输出张量同时发送回应用程序。

3.3:叠加层

为了复现 3.1 中的输出,我们还需要绘制一个叠加层。首先用一张静态叠加图像来演示。

下载一张半透明图像(来源):

mkdir -p images
wget -O images/imsdk-transparent-static.png https://cdn.edgeimpulse.com/qc-ai-docs/example-images/imsdk-transparent-static.png

创建一个新文件 ex3_overlay.py 并添加:

from gst_helper import gst_grouped_frames, atomic_save_image, timing_marks_to_str, download_file_if_needed, softmax
import time, argparse, numpy as np
from ai_edge_litert.interpreter import Interpreter, load_delegate
from PIL import ImageDraw, Image

parser = argparse.ArgumentParser(description='GStreamer -> SqueezeNet')
parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
args, unknown = parser.parse_known_args()

if args.video_source.strip() == '':
    raise Exception('--video-source is empty, did you not set the IMSDK_VIDEO_SOURCE env variable? E.g.:\n' +
    '    export IMSDK_VIDEO_SOURCE="v4l2src device=/dev/video2"')

# Source: https://commons.wikimedia.org/wiki/File:Arrow_png_image.png
OVERLAY_IMAGE = download_file_if_needed('images/imsdk-transparent-static.png', 'https://cdn.edgeimpulse.com/qc-ai-docs/example-images/imsdk-transparent-static.png')
OVERLAY_WIDTH = 128
OVERLAY_HEIGHT = 96

PIPELINE = (
    # Part 1: Create a qtivcomposer with two sinks (we'll write webcam to sink 0, overlay to sink 1)
    "qtivcomposer name=comp sink_0::zorder=0 "
        # Sink 1 (the overlay) will be at x=10, y=10; and sized 128x96
        f"sink_1::zorder=1 sink_1::alpha=1.0 sink_1::position=<10,10> sink_1::dimensions=<{OVERLAY_WIDTH},{OVERLAY_HEIGHT}> ! "
    "videoconvert ! "
    "video/x-raw,format=RGBA,width=224,height=224 ! "
    # Write frames to appsink
    "appsink name=overlay_raw drop=true sync=false max-buffers=1 emit-signals=true "

    # Part 2: Grab image from webcam and write the composer
        # Video source
        f"{args.video_source} ! "
        # Properties for the video source
        "video/x-raw,width=1920,height=1080 ! "
        # An identity element so we can track when a new frame is ready (so we can calc. processing time)
        "identity name=frame_ready_webcam silent=false ! "
        # Crop (square), the crop syntax is ('<X, Y, WIDTH, HEIGHT >').
        # So here we use 1920x1080 input, then center crop to 1080x1080 ((1920-1080)/2 = 420 = x crop)
        f'qtivtransform crop="<420, 0, 1080, 1080>" ! '
        # then resize to 224x224
        "video/x-raw,width=224,height=224,format=NV12 ! "
        # Event when the crop/scale are done
        "identity name=transform_done silent=false ! "
        # Write to sink 0 on the composer
        "comp.sink_0 "

    # Part 3: Load overlay from disk and write to composer (sink 1)
        # Image (statically from disk)
        f'filesrc location="{OVERLAY_IMAGE}" ! '
        # Decode PNG
        "pngdec ! "
        # Turn into a video (scaled to 128x96, RGBA format so we keep transparency, requires a framerate)
        "imagefreeze ! "
        "videoscale ! "
        "videoconvert ! "
        f"video/x-raw,format=RGBA,width={OVERLAY_WIDTH},height={OVERLAY_HEIGHT},framerate=30/1 ! "
        # Write to sink 1 on the composer
        "comp.sink_1 "
)

for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
    print(f"Frame ready")
    print('    Data:', end='')
    for key in list(frames_by_sink):
        print(f' name={key} {frames_by_sink[key].shape} ({frames_by_sink[key].dtype})', end='')
    print('')

    # Save image to disk
    save_image_start = time.perf_counter()
    frame = frames_by_sink['overlay_raw']
    atomic_save_image(frame=frame, path='out/webcam_with_overlay.png')
    save_image_end = time.perf_counter()

    # Add an extra mark, so we have timing info for the complete pipeline
    marks['save_image_end'] = list(marks.items())[-1][1] + (save_image_end - save_image_start)

    print('    Timings:', timing_marks_to_str(marks))

运行此应用程序:

# We use '| grep -v "<W>"' to filter out some warnings - you can omit it if you want.
python3 ex3_overlay.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "<W>"

# Frame ready
#     Data: name=overlay_raw (224, 224, 4) (uint8)
#     Timings: frame_ready_webcam->transform_done: 1.03ms, transform_done->pipeline_finished: 3.28ms, pipeline_finished->save_image_end: 31.28ms (total 35.59ms)

3.4:将神经网络与叠加层结合

您现在已经了解了如何在 IM SDK 流水线中运行神经网络,也了解了如何绘制叠加层。让我们将两者合并到一条流水线中,把预测结果叠加到图像上——整个过程完全不经过 CPU。

创建一个新文件 ex3_from_imsdk.py 并添加:

 from gst_helper import gst_grouped_frames, timing_marks_to_str, download_file_if_needed
 import argparse, os

 parser = argparse.ArgumentParser(description='GStreamer -> SqueezeNet')
 parser.add_argument('--video-source', type=str, required=True, help='GStreamer video source (e.g. "v4l2src device=/dev/video2" or "qtiqmmfsrc name=camsrc camera=0")')
 args, unknown = parser.parse_known_args()

 if args.video_source.strip() == '':
     raise Exception('--video-source is empty, did you not set the IMSDK_VIDEO_SOURCE env variable? E.g.:\n' +
     '    export IMSDK_VIDEO_SOURCE="v4l2src device=/dev/video2"')

 MODEL_PATH = download_file_if_needed('models/squeezenet1_1-squeezenet-1.1-w8a8.tflite', 'https://cdn.edgeimpulse.com/qc-ai-docs/models/squeezenet1_1-squeezenet-1.1-w8a8.tflite')
 LABELS_PATH = download_file_if_needed('models/SqueezeNet-1.1_labels.txt', 'https://cdn.edgeimpulse.com/qc-ai-docs/models/SqueezeNet-1.1_labels.txt')

 PIPELINE = (
     # Part 1: Create a qtivcomposer with two sinks (we'll write webcam to sink 0, overlay to sink 1)
     "qtivcomposer name=comp sink_0::zorder=0 "
     "sink_1::zorder=1 sink_1::alpha=1.0 ! "
     "video/x-raw,format=NV12,width=1920,height=1080 ! "
     "v4l2h264enc capture-io-mode=4 output-io-mode=4 ! "
     "queue ! h264parse ! mp4mux ! "
     "filesink location=output/out.mp4 "

     # Video source
     f"{args.video_source} ! "
     # Properties for the video source
     "video/x-raw,width=1920,height=1080 ! "
     # An identity element so we can track when a new frame is ready (so we can calc. processing time)
     "identity name=frame_ready_webcam silent=false ! "
     "qtivtransform ! "
     "video/x-raw,format=NV12 ! "
     "identity name=transform_done silent=false ! "
     "tee name=v "
     "v. ! queue max-size-buffers=1 leaky=downstream ! "
     "comp.sink_0 "

     # Part 3: NN path ? postprocess overlay ? comp.sink_1 + nn_overlay appsink
     "v. ! queue max-size-buffers=1 leaky=downstream ! "
     # (1) Input of qtimlvconverter
     "identity name=converter_in silent=false ! "
     "qtimlvconverter ! neural-network/tensors,type=UINT8,dimensions=<<1,224,224,3>> ! "
     # (2) Output of qtimlvconverter
     "identity name=converter_out silent=false ! "
     # qtimltflite (inference on HTP via QNN delegate)
     f'qtimltflite delegate=external external-delegate-path=libQnnTFLiteDelegate.so '
     f'external-delegate-options="QNNExternalDelegate,backend_type=htp;" model="{MODEL_PATH}" ! '
     # (3) Output of qtimltflite
     "identity name=inference_done silent=false ! "
     # qtimlpostprocess (mobilenet-softmax): dequant + softmax + overlay render
     f'qtimlpostprocess name=postproc module=mobilenet-softmax labels="{LABELS_PATH}" '
     'results=1 settings="{\\"confidence\\": 10.0}" ! '
     # (4) Output of qtimlpostprocess
     "identity name=postproc_done silent=false ! "
     # Overlay frame (BGRA) that feeds both composer and the NN appsink
     "video/x-raw,format=BGRA,width=224,height=224 ! "
     "tee name=nn_t "

     # Branch A (to composer)
     "nn_t. ! queue max-size-buffers=1 leaky=downstream ! "
     "comp.sink_1 "

     # Branch B (to appsink on the NN branch)
     "nn_t. ! queue max-size-buffers=1 leaky=downstream ! "
     "appsink name=nn_overlay drop=true sync=false max-buffers=1 emit-signals=true "
 )

 os.makedirs('output', exist_ok=True)

 for frames_by_sink, marks in gst_grouped_frames(PIPELINE):
     # Consume/inspect the NN-branch overlay (BGRA 224x224) so groups progress
     if 'nn_overlay' in frames_by_sink:
         nn_overlay = frames_by_sink['nn_overlay']  # ndarray (224, 224, 4), dtype=uint8
         print(f"[appsink:nn_overlay] frame {nn_overlay.shape} {nn_overlay.dtype}")

     # Print timing markers (now includes converter_* / inference_done / postproc_done / composer_out)
     print('Timings:', timing_marks_to_str(marks))  

运行此应用程序:

# We use '| grep -v "<W>"' to filter out some warnings - you can omit it if you want.
python3 ex3_from_imsdk.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "<W>"

# Frame ready
#     Timings: frame_ready_webcam->transform_done: 6.42ms, transform_done->converter_in: 0.40ms, converter_in->converter_out: 1.09ms, converter_out->inference_done: 1.29ms, inference_done->postproc_done: 0.63ms (total 9.84ms)

太棒了!现在整条流水线都运行在 IM SDK 中。输出文件位于 out/imsdk-webcam-nn-overlay.mp4。

故障排除

流水线没有任何输出

如果您没有看到任何输出,请添加 GST_DEBUG=3 以查看更详细的调试信息。

GST_DEBUG=3 python3 ex1.py

QMMF Recorder StartCamera Failed / Failed to Open Camera

如果您看到如下所示的 QMMF 错误:

0:00:00.058915726  7329     0x1faf28a0 ERROR             qtiqmmfsrc qmmf_source_context.cc:1426:gst_qmmf_context_open: QMMF Recorder StartCamera Failed!
0:00:00.058955986  7329     0x1faf28a0 WARN              qtiqmmfsrc qmmf_source.c:1206:qmmfsrc_change_state:<camsrc> error: Failed to Open Camera!

可以通过运行以下命令释放摄像头:

sudo killall cam-server

Introduction

Architecture

Key Documents

Technologies

System

Release Notes

那么…… GStreamer 流水线是什么?

设置 GStreamer 和 IM SDK

示例 1:在 GPU 与 CPU 上进行缩放和裁剪

示例 2:流的分支(Tee)与多路输出

示例 3:运行神经网络

3.1:在 Python 中进行神经网络推理与图像合成

3.2:使用 IM SDK 运行神经网络

3.3:叠加层

3.4:将神经网络与叠加层结合

故障排除

流水线没有任何输出

QMMF Recorder StartCamera Failed / Failed to Open Camera

​那么…… GStreamer 流水线是什么?

​设置 GStreamer 和 IM SDK

​示例 1:在 GPU 与 CPU 上进行缩放和裁剪

​示例 2:流的分支(Tee)与多路输出

​示例 3:运行神经网络

​3.1:在 Python 中进行神经网络推理与图像合成

​3.2:使用 IM SDK 运行神经网络

​3.3:叠加层

​3.4:将神经网络与叠加层结合

​故障排除

​流水线没有任何输出

​QMMF Recorder StartCamera Failed / Failed to Open Camera

那么…… GStreamer 流水线是什么?

设置 GStreamer 和 IM SDK

示例 1:在 GPU 与 CPU 上进行缩放和裁剪

示例 2:流的分支(Tee)与多路输出

示例 3:运行神经网络

3.1:在 Python 中进行神经网络推理与图像合成

3.2:使用 IM SDK 运行神经网络

3.3:叠加层

3.4:将神经网络与叠加层结合

故障排除

流水线没有任何输出

QMMF Recorder StartCamera Failed / Failed to Open Camera