> ## Documentation Index
> Fetch the complete documentation index at: https://dragonwingdocs-staging.qualcomm.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 在 NPU 上运行 LiteRT 模型

> 使用 QNN 委托和 Python,在 Qualcomm Dragonwing 设备的 NPU 上运行目标检测应用。

使用 [LiteRT](https://ai.google.dev/edge/litert) 运行时,在 Qualcomm® Dragonwing™ 设备的 NPU 上运行量化 LiteRT 模型。

## 前提条件

运行应用之前,请完成以下步骤以连接目标设备、下载所需工件并设置 Python 环境。

1. 在设备上启用 Wi-Fi 和 SSH。

   设备需要互联网连接才能下载运行示例应用所需的工件。如果已配置 SSH 和 Wi-Fi,请跳过此步骤。

   按照[设置 SSH 连接](https://dragonwingdocs.qualcomm.com/Technologies/Wi-Fi/setup-wi-fi#set-up-uart/ssh)在设备上启用 Wi-Fi 和 SSH。

2. 在目标设备上,下载 `download_artifacts.sh` 脚本,设置可执行权限,
   然后运行该脚本,将模型和标签文件下载到设备。

   ```shell theme={null}
   cd /tmp/
   ```

   ```shell theme={null}
   curl -L -O https://raw.githubusercontent.com/qualcomm/sample-apps-for-qualcomm-linux/refs/heads/main/qualcomm-linux/scripts/download_artifacts_2.x.sh && chmod +x download_artifacts_2.x.sh && ./download_artifacts_2.x.sh
   ```

3. 在目标设备上设置 Python 环境,以安装 LiteRT 运行时和其他依赖项。

   ```shell theme={null}
   python3 -m venv --system-site-packages python-venv
   ```

   激活 Python 环境:

   ```shell theme={null}
   source python-venv/bin/activate
   ```

   安装 LiteRT 运行时、Pillow 和 OpenCV 软件包:

   ```shell theme={null}
   pip3 install ai-edge-litert==2.1.5 Pillow opencv-python==4.10.0.84
   ```

## 运行目标检测应用

以下 Python 应用使用量化的 YOLOX LiteRT 模型,对视频文件实时执行目标检测。它将带标注的帧输出到文件或 Wayland 显示器,并通过 QNN LiteRT 委托利用硬件加速,针对边缘 AI 场景进行了优化。

1. 创建并进入 `${HOME}/Downloads/qimsdk_samples/apps/` 目录。

   ```shell theme={null}
   mkdir -p ${HOME}/Downloads/qimsdk_samples/apps/ && cd ${HOME}/Downloads/qimsdk_samples/apps/
   ```

2. 下载 `object_detection_2.x.py` 文件。

   ```shell theme={null}
   curl -L https://raw.githubusercontent.com/qualcomm/sample-apps-for-qualcomm-linux/refs/heads/main/qualcomm-linux/applications/LiteRT/object_detection_2.x.py -o object_detection.py
   ```

   有关代码说明,请参阅下文的[使用 OpenCV 和 LiteRT 进行目标检测](#create-object_detection-py)。

3. 运行应用:

   <Tabs>
     <Tab title="输出到文件">
       ```shell theme={null}
       python3 object_detection.py --output file
       ```

       将输出视频下载到主机以查看结果:

       ```shell theme={null}
       scp root@<IP_ADDRESS_OF_TARGET_DEVICE>:/tmp/output_object_detection.mp4 .
       ```

       当系统提示输入密码时,请输入 `oelinux123`。
     </Tab>

     <Tab title="输出到显示器">
       在目标设备的终端中,运行目标检测应用:

       ```shell theme={null}
       python3 object_detection.py --output wayland
       ```
     </Tab>
   </Tabs>

<h2 id="create-object_detection-py">
  使用 OpenCV 和 LiteRT 进行目标检测
</h2>

要创建与上一节所述目标检测应用类似的应用,请按如下方式创建 `object_detection.py` 文件:

1. 在 `${HOME}/Downloads/qimdk_samples/apps/` 文件夹中,创建 `object_detection.py` 文件。

2. 将以下代码添加到您的 `object_detection.py` 文件中。

   <Note>
     以下代码中的后处理与 AI Hub 中的目标检测模型兼容。
     对于自定义模型,请更新后处理逻辑以匹配模型的输出格式和要求。
   </Note>

   a. 导入所需的软件包:

   ```python theme={null}
   #!/usr/bin/env python3
   import os
   import cv2
   import numpy as np
   import argparse
   import ai_edge_litert.interpreter as tflite
   import gi
   gi.require_version('Gst', '1.0')
   from gi.repository import Gst
   ```

   b. 处理输出参数:

   ```python theme={null}
   parser = argparse.ArgumentParser(description="Run object detection and output to file or Wayland.")
   parser.add_argument("--output", choices=["file", "wayland"], default="file",
                     help="Choose output mode: 'file' (default) or 'wayland'")
   args = parser.parse_args()
   ```

   c. 初始化并配置模型参数:

   ```python theme={null}
   ARTIFACT_DIR = os.path.join(
      os.path.expanduser("~"), "Downloads", "qimsdk_samples")
   MODEL_PATH = f"${ARTIFACT_DIR}/models/yolox_quantized.tflite"  # YOLOX quantized model
   LABEL_PATH = f"${ARTIFACT_DIR}/labels/coco_labels.txt"
   VIDEO_IN = f"${ARTIFACT_DIR}/media/video.mp4"
   VIDEO_OUT = "/tmp/output_object_detection.mp4"
   DELEGATE_PATH = "libQnnTFLiteDelegate.so"

   FRAME_W, FRAME_H = 1600, 900
   FPS_OUT = 30
   CONF_THRES = 0.25
   NMS_IOU_THRES = 0.50
   BOX_SCALE = 3.2108588218688965
   BOX_ZP = 31.0
   SCORE_SCALE = 0.0038042240776121616
   ```

   d. 加载模型并设置 LiteRT 委托:

   ```python theme={null}
   delegate_options = {'backend_type': 'htp'}
   delegate = tflite.load_delegate(DELEGATE_PATH, delegate_options)
   interpreter = tflite.Interpreter(model_path=MODEL_PATH, experimental_delegates=[delegate])
   interpreter.allocate_tensors()

   in_det = interpreter.get_input_details()
   out_det = interpreter.get_output_details()
   in_h, in_w = in_det[0]["shape"][1:3]

   labels = [l.strip() for l in open(LABEL_PATH)]
   ```

   e. 设置视频捕获和预处理:

   ```python theme={null}
   try:
      cap = cv2.VideoCapture(VIDEO_IN)
      sx, sy = FRAME_W / in_w, FRAME_H / in_h
      frame_rs = np.empty((FRAME_H, FRAME_W, 3), np.uint8)
      input_tensor = np.empty((1, in_h, in_w, 3), np.uint8)
   ```

   f. 创建 GStreamer 管线,将帧流式传输到 Wayland 显示器:

   ```python theme={null}
      if args.output == "file":
         fourcc = cv2.VideoWriter_fourcc(*"mp4v")
         out_writer = cv2.VideoWriter(VIDEO_OUT, fourcc, FPS_OUT, (FRAME_W, FRAME_H))
      else:
         Gst.init(None)
         pipeline = Gst.parse_launch(
            'appsrc name=src is-live=true block=true format=time caps=video/x-raw,format=BGR,width=1600,height=900,framerate=30/1 ! videoconvert ! waylandsink'
         )
         appsrc = pipeline.get_by_name('src')
         pipeline.set_state(Gst.State.PLAYING)

      frame_cnt = 0
   ```

   g. 初始化主循环以读取视频、对每一帧运行推理,并在输出上绘制边界框:

   ```python theme={null}
      # -------------------- Main Loop --------------------
      while True:
         ok, frame = cap.read()
         if not ok:
            break
         frame_cnt += 1

         cv2.resize(frame, (FRAME_W, FRAME_H), dst=frame_rs)
         cv2.resize(frame_rs, (in_w, in_h), dst=input_tensor[0])

         interpreter.set_tensor(in_det[0]['index'], input_tensor)
         interpreter.invoke()

         boxes_q = interpreter.get_tensor(out_det[0]['index'])[0]
         scores_q = interpreter.get_tensor(out_det[1]['index'])[0]
         classes_q = interpreter.get_tensor(out_det[2]['index'])[0]

         boxes = BOX_SCALE * (boxes_q.astype(np.float32) - BOX_ZP)
         scores = SCORE_SCALE * scores_q.astype(np.float32)
         classes = classes_q.astype(np.int32)

         mask = scores >= CONF_THRES
         if np.any(mask):
            boxes_f = boxes[mask]
            scores_f = scores[mask]
            classes_f = classes[mask]

            x1, y1, x2, y2 = boxes_f.T
            boxes_cv2 = np.column_stack((x1, y1, x2 - x1, y2 - y1))

            idx_cv2 = cv2.dnn.NMSBoxes(
                  bboxes=boxes_cv2.tolist(),
                  scores=scores_f.tolist(),
                  score_threshold=CONF_THRES,
                  nms_threshold=NMS_IOU_THRES
            )

            if len(idx_cv2):
                  idx = idx_cv2.flatten()
                  sel_boxes = boxes_f[idx]
                  sel_scores = scores_f[idx]
                  sel_classes = classes_f[idx]

                  sel_boxes[:, [0, 2]] *= sx
                  sel_boxes[:, [1, 3]] *= sy
                  sel_boxes = sel_boxes.astype(np.int32)

                  sel_boxes[:, [0, 2]] = np.clip(sel_boxes[:, [0, 2]], 0, FRAME_W - 1)
                  sel_boxes[:, [1, 3]] = np.clip(sel_boxes[:, [1, 3]], 0, FRAME_H - 1)

                  for (x1i, y1i, x2i, y2i), sc, cl in zip(sel_boxes, sel_scores, sel_classes):
                     cv2.rectangle(frame_rs, (x1i, y1i), (x2i, y2i), (0, 255, 0), 2)
                     lab = labels[cl] if cl < len(labels) else str(cl)
                     cv2.putText(frame_rs, f"{lab} {sc:.2f}", (x1i, max(10, y1i - 5)),
                                 cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)

         if args.output == "file":
            out_writer.write(frame_rs)
         else:
            data = frame_rs.tobytes()
            buf = Gst.Buffer.new_allocate(None, len(data), None)
            buf.fill(0, data)
            buf.duration = Gst.util_uint64_scale_int(1, Gst.SECOND, FPS_OUT)
            timestamp = cap.get(cv2.CAP_PROP_POS_MSEC) * Gst.MSECOND
            buf.pts = buf.dts = int(timestamp)
            appsrc.emit('push-buffer', buf)
   ```

   h. 释放管线并在完成时通知用户:

   ```python theme={null}
   except KeyboardInterrupt:
      print(f"\nCtrl+C received. Cleaning up after {frame_cnt} frames...")

   finally:
      # -------------------- Finish --------------------
      cap.release()
      if args.output == "file":
         out_writer.release()
         print(f"Done – processed video saved to {VIDEO_OUT}")
      else:
         appsrc.emit('end-of-stream')
         pipeline.set_state(Gst.State.NULL)
         print("Done – video streamed to Wayland sink")
   ```