| |
| |
| |
| |
| |
|
|
| import gi |
|
|
| gi.require_version("Gst", "1.0") |
| gi.require_version("GstAnalytics", "1.0") |
| from gi.repository import Gst, GLib, GstAnalytics |
|
|
| Gst.init([]) |
|
|
| INPUT_VIDEO = "test_video.mp4" |
| VEHICLE_LABELS = {"car", "motorcycle", "bus", "truck"} |
|
|
| |
| MODEL = "yolo26n" |
|
|
| |
| |
| pipeline_str = ( |
| f"filesrc location={INPUT_VIDEO} ! decodebin3 ! " |
| "videoconvert ! " |
| f"gvadetect model={MODEL}_openvino_model/{MODEL}.xml " |
| "device=GPU " |
| "threshold=0.4 ! queue ! " |
| "gvawatermark displ-cfg=show-roi=car:motorcycle:bus:truck ! " |
| "videoconvert ! video/x-raw,format=I420 ! " |
| "openh264enc ! h264parse ! " |
| "mp4mux ! filesink name=sink location=output_dlstreamer.mp4" |
| ) |
| pipeline = Gst.parse_launch(pipeline_str) |
|
|
|
|
| def on_buffer(pad, info): |
| buf = info.get_buffer() |
| rmeta = GstAnalytics.buffer_get_analytics_relation_meta(buf) |
| if rmeta is None: |
| return Gst.PadProbeReturn.OK |
| vehicles = [] |
| idx = 1 |
| while True: |
| ok, od = rmeta.get_od_mtd(idx) |
| if not ok: |
| break |
| label = GLib.quark_to_string(od.get_obj_type()) |
| if label in VEHICLE_LABELS: |
| _, x, y, w, h, conf = od.get_location() |
| vehicles.append((label, int(x), int(y))) |
| idx += 1 |
| if vehicles: |
| print(f"Vehicle count: {len(vehicles)}", flush=True) |
| for label, x, y in vehicles: |
| print(f" {label} at ({x},{y})", flush=True) |
| return Gst.PadProbeReturn.OK |
|
|
|
|
| sink = pipeline.get_by_name("sink") |
| sink_pad = sink.get_static_pad("sink") |
| sink_pad.add_probe(Gst.PadProbeType.BUFFER, on_buffer) |
|
|
| pipeline.set_state(Gst.State.PLAYING) |
| bus = pipeline.get_bus() |
| bus.timed_pop_filtered( |
| Gst.CLOCK_TIME_NONE, |
| Gst.MessageType.EOS | Gst.MessageType.ERROR, |
| ) |
| pipeline.set_state(Gst.State.NULL) |
|
|
