快速入门指南

sudo tee /etc/configs/config_detection.json << EOF
{
  "file-path": "$HOME/media/$SRC_VIDEO_NAME",
  "ml-framework": "tflite",
  "yolo-model-type": "yolox",
  "model": "$HOME/models/$MODEL_NAME",
  "labels": "$HOME/labels/$LABELS_NAME",
  "threshold": 40,
  "runtime": "dsp",
  "output-type": "waylandsink"
}
EOF

cat > obj_det.py << 'PYEOF'
#!/usr/bin/env python3
import os, signal, gi
gi.require_version("Gst", "1.0")
gi.require_version("GLib", "2.0")
from gi.repository import Gst, GLib

SAMPLES = os.environ.get("QIMSDK_SAMPLES", "/etc")
MODEL   = f"{SAMPLES}/models/{os.environ.get('MODEL_NAME',    'yolo_x_w8a8.tflite')}"
LABELS  = f"{SAMPLES}/labels/{os.environ.get('LABELS_NAME',   'yolov8.json')}"
VIDEO   = f"{SAMPLES}/media/{os.environ.get('SRC_VIDEO_NAME', 'video.mp4')}"


def make(pipeline, factory, **props):
    el = Gst.ElementFactory.make(factory)
    for k, v in props.items():
        el.set_property(k.replace("_", "-"), v)
    pipeline.add(el)
    return el


def on_demux_pad(demux, pad, next_el):
    if "video" in pad.get_current_caps().to_string():
        pad.link(next_el.get_static_pad("sink"))


def build_pipeline():
    p = Gst.Pipeline.new()

    src      = make(p, "filesrc",    location=VIDEO)
    demux    = make(p, "qtdemux")
    parse    = make(p, "h264parse")
    decoder  = make(p, "v4l2h264dec", capture_io_mode=4, output_io_mode=4)
    q0       = make(p, "queue")
    tee      = make(p, "tee")

    q1       = make(p, "queue")
    pre_proc = make(p, "qtimlvconverter")
    q2       = make(p, "queue")
    infer    = make(p, "qtimltflite",
                   model=MODEL, delegate="external",
                   external_delegate_path="libQnnTFLiteDelegate.so")
    infer.set_property("external-delegate-options",
                       Gst.Structure.new_from_string(
                           "QNNExternalDelegate,backend_type=htp,log_level=(string)1"))
    q3       = make(p, "queue")
    post_proc= make(p, "qtimlpostprocess",
                   module="yolov8", labels=LABELS,
                   settings='{"confidence": 51.0}')
    q4       = make(p, "queue")

    mux      = make(p, "qtimetamux")
    q5       = make(p, "queue")
    overlay  = make(p, "qtivoverlay")
    q6       = make(p, "queue")
    sink     = make(p, "waylandsink", fullscreen=True, sync=False)
    q7       = make(p, "queue")

    src.link(demux)
    demux.connect("pad-added", on_demux_pad, parse)
    parse.link(decoder)
    decoder.link_filtered(q0, Gst.Caps.from_string("video/x-raw,format=NV12"))
    q0.link(tee)

    tee.request_pad_simple("src_%u").link(q1.get_static_pad("sink"))
    for a, b in [(q1, pre_proc), (pre_proc, q2), (q2, infer),
                 (infer, q3), (q3, post_proc)]:
        a.link(b)
    post_proc.link_filtered(q4, Gst.Caps.from_string("text/x-raw"))
    q4.link(mux)

    tee.request_pad_simple("src_%u").link(q7.get_static_pad("sink"))
    q7.link(mux)

    for a, b in [(mux, q5), (q5, overlay), (overlay, q6), (q6, sink)]:
        a.link(b)

    return p


Gst.init(None)
loop     = GLib.MainLoop()
pipeline = build_pipeline()

def on_message(bus, msg):
    if msg.type == Gst.MessageType.ERROR:
        print("Error:", msg.parse_error()[0].message)
    if msg.type in (Gst.MessageType.EOS, Gst.MessageType.ERROR):
        loop.quit()

pipeline.get_bus().add_watch(GLib.PRIORITY_DEFAULT, lambda b, m: (on_message(b, m), True)[1])
GLib.unix_signal_add(GLib.PRIORITY_HIGH, signal.SIGINT, lambda: loop.quit() or GLib.SOURCE_CONTINUE)

pipeline.set_state(Gst.State.PLAYING)
loop.run()
pipeline.set_state(Gst.State.NULL)
PYEOF