Qualcomm® Intelligent Multimedia (IM) SDK

​

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

​

设置 GStreamer 和 IM SDK

1. 安装 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
   
2. 获取 python 示例，解压后创建 venv，并安装其依赖项：
   # 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：
        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. 设置环境变量（我们将在示例中使用它）：
        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 上进行调整大小和裁剪

1. 创建一个新文件 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')
   
2. 启动这个 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 摄像头测得）。
3. 现在让我们改为在 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')
   
4. 再次运行：
   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 名称，值为数据）。

1. 创建一个新文件 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')
   
2. 运行此 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：运行神经网络

​

3.1：在 Python 中运行神经网络和图像合成

1. 首先我们做一个”常规”实现：从 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))
   
2. 现在运行此应用程序：
   # 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 运行神经网络

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

1. 创建一个新文件 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))
   

1. 现在运行此应用程序：
   # 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)
   
   好！模型现在在 IM SDK 流水线内部的 NPU 上运行。如果您想获得前 5 个输出（就像 3.1 中那样），可以在 qtimltflite 元素之后对流进行 tee 分流，并将原始输出张量也发送回应用程序。

​

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
   
2. 创建一个新文件 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))
   
3. 运行此应用程序：
   # 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：将神经网络与叠加层结合

1. 创建一个新文件 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))  
   
2. 运行此应用程序：
   # 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 python3 ex1.py

​

QMMF Recorder StartCamera Failed / Failed to Open Camera

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