> ## Documentation Index > Fetch the complete documentation index at: https://dragonwingdocs.qualcomm.com/llms.txt > Use this file to discover all available pages before exploring further. # Qualcomm® Intelligent Multimedia (IM) SDK > 使用在 Dragonwing 设备的 GPU 和 NPU 上运行的 GStreamer 插件构建零拷贝 AI 视觉流水线。 Qualcomm® Intelligent Multimedia (IM) SDK 是一组 GStreamer 插件，可让您在 Dragonwing 开发板的 GPU 上运行计算机视觉操作；并且可以创建完全在 GPU 和 NPU 上运行、无需回到 CPU 的 AI 流水线（零拷贝）。这些特性结合在一起，使您可以实现比自己用 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 构建一些应用程序。 1. 安装 GStreamer、IM SDK 以及本示例中需要的一些额外依赖项。打开开发板上的终端，或与开发板建立 SSH 会话，然后运行： ```shell theme={null} 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 ``` 2. 获取 [python 示例](/Ubuntu/python-examples/)，解压后创建 venv，并安装其依赖项： ```shell theme={null} # Download python_examples.zip from the examples page, then extract it. sudo apt install -y unzip mkdir -p ~/imsdk-python-examples cd ~/imsdk-python-examples unzip ~/Downloads/python_examples.zip # Create a new venv python3 -m venv .venv --system-site-packages source .venv/bin/activate # Install Python dependencies pip3 install -r requirements.txt ``` 3. 您需要一个摄像头（内置摄像头，如 RB3 Gen 2 Vision Kit 上的，或 USB 网络摄像头）。 * 如果您想使用 USB 网络摄像头： 1. 找出设备 ID： ```shell theme={null} 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 ``` 2. 设置环境变量（我们将在示例中使用它）： ```text theme={null} export IMSDK_VIDEO_SOURCE="v4l2src device=/dev/video2" ``` * 如果您使用的是 RB3 Gen 2 Vision Kit 并想使用内置摄像头： ```text theme={null} export IMSDK_VIDEO_SOURCE="qtiqmmfsrc name=camsrc camera=0" ``` ## 示例 1：在 GPU 与 CPU 上进行调整大小和裁剪 让我们展示在 GPU 上处理相比 CPU 能快多少。如果您有一个需要 224x224 RGB 输入的神经网络，则需要预处理数据：首先，从网络摄像头抓取帧（例如原生分辨率为 1980x1080），然后裁剪为 1/1 宽高比（例如裁剪为 1080x1080），再调整为所需分辨率（224x224），最后由像素创建一个 Numpy 数组。 1. 创建一个新文件 `ex1.py`，并添加： ```python theme={null} 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') ``` 2. 启动这个 python 脚本。此流水线在 CPU 上运行（使用原生 GStreamer 组件）： ```shell theme={null} 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 摄像头测得）。 3. 现在让我们改为在 GPU 上运行……将： ```shell theme={null} # Crop to square "videoconvert ! aspectratiocrop aspect-ratio=1/1 ! " # Scale to 224x224 and RGB "videoscale ! video/x-raw,format=RGB,width=224,height=224 ! " ``` 替换为： ```text theme={null} # Crop (square), the crop syntax is (''). # 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`： ```python theme={null} 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 (''). # 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') ``` 4. 再次运行： ```text theme={null} 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，另一部分通过调整大小/裁剪流水线。 1. 创建一个新文件 `ex2.py` 并添加： ```python theme={null} 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 (''). # 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') ``` 2. 运行此 python 脚本： ```text theme={null} 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 中运行神经网络和图像合成 1. 首先我们做一个"常规"实现：从 IM SDK 流水线获取调整大小后的帧，然后使用 [LiteRT](/zh/ai-workflows/lite-rt) 运行模型（在 NPU 上）。之后，我们将在图像上绘制最高置信度的预测并写入磁盘。创建一个新文件 `ex3_from_python.py` 并添加： ```python theme={null} 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 (''). # 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)) ``` 2. 现在运行此应用程序： ```text theme={null} # We use '| grep -v ""' to filter out some warnings - you can omit it if you want. python3 ex3_from_python.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "" # 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 模型）。该插件输出文本（包含预测结果）或叠加层（用于绘制到原始图像上）。 **注意：** 该元素有特定的标签格式（见下文）。 1. 创建一个新文件 `ex3_nn_imsdk.py` 并添加： ```python theme={null} 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 归一化为概率。 1. 现在运行此应用程序： ```text theme={null} # We use '| grep -v ""' to filter out some warnings - you can omit it if you want. python3 ex3_nn_imsdk.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "" # 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) ``` 好！模型现在在 IM SDK 流水线*内部*的 NPU 上运行。如果您想获得前 5 个输出（就像 3.1 中那样），可以在 `qtimltflite` 元素之后对流进行 tee 分流，并将原始输出张量也发送回应用程序。 ### 3.3：叠加层 为了模拟 3.1 中的输出，我们还需要绘制一个叠加层。先用一个静态叠加图像来演示。 1. 下载一张半透明图像（[来源](https://commons.wikimedia.org/wiki/File:PNG_transparency_demonstration_2.png)）： ```shell theme={null} mkdir -p images wget -O images/imsdk-transparent-static.png https://cdn.edgeimpulse.com/qc-ai-docs/example-images/imsdk-transparent-static.png ``` 2. 创建一个新文件 `ex3_overlay.py` 并添加： ```python theme={null} 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 (''). # 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)) ``` 3. 运行此应用程序： ```text theme={null} # We use '| grep -v ""' to filter out some warnings - you can omit it if you want. python3 ex3_overlay.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "" # 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。 1. 创建一个新文件 `ex3_from_imsdk.py` 并添加： ```python theme={null} 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)) ``` 2. 运行此应用程序： ```shell theme={null} # We use '| grep -v ""' to filter out some warnings - you can omit it if you want. python3 ex3_from_imsdk.py --video-source "$IMSDK_VIDEO_SOURCE" | grep -v "" # 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` 以查看更详细的调试信息。 ```text theme={null} GST_DEBUG=3 python3 ex1.py ``` ### QMMF Recorder StartCamera Failed / Failed to Open Camera 如果您看到类似以下的 QMMF 错误： ```text theme={null} 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: error: Failed to Open Camera! ``` 您可以通过运行以下命令释放摄像头： ```text theme={null} sudo killall cam-server ```