RidgeRun Metadata/Use Cases/Cases

From RidgeRun Developer Wiki

Follow Us On Twitter LinkedIn Email Share this page


NVIDIA partner logo NXP partner logo






Use Case: Surveillance or Reconnaissance Drones

In defense, security, and monitoring applications, drones (UAVs) equipped with cameras are commonly used to capture real-time video. However, the value of this video increases significantly when it is enriched with metadata that provides contextual information such as GPS coordinates, altitude, camera orientation, precise timestamps, and more.

One of the most widely adopted standards for transmitting this type of information is MISB (Motion Imagery Standards Board). MISB defines how to structure and transport metadata relevant to visual intelligence. This metadata can be encapsulated and transmitted alongside the video using the MPEG Transport Stream (MPEG-TS) container, which supports multiplexing of video, audio, and metadata into a single synchronized stream.

Applicable RidgeRun Products

MPEG-TS

By leveraging RidgeRun's MPEG-TS metadata framework, developers can easily integrate synchronized, real-time metadata into video pipelines used in UAV applications. The solution simplifies the injection and extraction process using the metasrc and metasink elements, and supports both asynchronous and synchronous metadata conforming to standards like MISB ST 1402.

This enables a powerful and standards-compliant method of enriching video streams with mission-critical data—ideal for drones engaged in surveillance, reconnaissance, or tactical operations.

Code Example

Inject and extract Drone metadata (MPEG‑TS + RidgeRun)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import time
from gi.repository import Gst, GLib

# Initialize GStreamer
Gst.init(None)

# Pipeline:
# - Inject KLV (synchronous stream_type=21) with metasrc
# - Mux to MPEG-TS (mux)
# - Tee the TS: one branch to file, the other to tsdemux → metasink (extract)
pipeline_desc = (
    'metasrc name=meta ! meta/x-klv,stream_type=21 ! '
    'mpegtsmux name=mux ! tee name=t '
    'videotestsrc is-live=true pattern=18 '
    '! video/x-raw,width=640,height=480,framerate=30/1 '
    '! queue ! x264enc ! mux. '
    't. ! queue ! filesink location=./metadata.ts '
    't. ! queue ! tsdemux name=demux demux.private_1 ! metasink name=msink'
)

pipeline = Gst.parse_launch(pipeline_desc)
metasrc = pipeline.get_by_name("meta")
msink = pipeline.get_by_name("msink")

# Simulate one KLV-like payload
metadata = {
    "lat": 9.9281,
    "lon": -84.0907,
    "alt": 105.2,
    "speed": 8.0,
    "timestamp": time.time()
}
payload = json.dumps(metadata).encode("utf-8")

# Inject 
metasrc.set_property("metadata-binary", payload)

# Print extracted metadata from metasink using pad probe
def print_extracted(pad, info):
    buf = info.get_buffer()
    if not buf:
        return Gst.PadProbeReturn.OK
    ok, map_info = buf.map(Gst.MapFlags.READ)
    if ok:
        try:
            print(json.dumps(json.loads(map_info.data.decode("utf-8")), indent=2))
        except Exception:
            print(f"[metasink] extracted {len(map_info.data)} bytes")
        buf.unmap(map_info)
    return Gst.PadProbeReturn.OK

msink.get_static_pad("sink").add_probe(Gst.PadProbeType.BUFFER, print_extracted)

# Run a few seconds
pipeline.set_state(Gst.State.PLAYING)
print("Recording and extracting metadata.")
time.sleep(5)
pipeline.set_state(Gst.State.NULL)
print("Done. File saved")

Use Case: Video Analysis with Bounding Boxes (SEI)

In computer vision applications such as object detection and video analytics, models often output bounding boxes, class labels, and confidence scores per frame. To transport this metadata without modifying the video frames and without using a parallel data stream, a widely adopted method is to embed the metadata in SEI (Supplemental Enhancement Information) .

Applicable RidgeRun Products SEI

The GstSEIMetadata plugin provides a robust solution for inserting and extracting metadata in H.264/H.265 video streams, enabling the synchronization of additional information—such as bounding boxes—directly with each video frame. Through its seiinject element, bounding boxes can be embedded as metadata using properties, binary data, or GstMeta structures. The seiextract element then retrieves this metadata from the bitstream and reattaches it as GstMeta, making it available for downstream pipeline elements or external applications. Additionally, seimetatimestamp inserts precise timestamps as metadata, helping maintain temporal correlation between detected objects and their corresponding frames. This capability is especially valuable in computer vision and real-time analytics systems, where ensuring that bounding boxes remain tightly synchronized with the video—even through encoding and transmission—is critical.

Code Example

Inject + Extract SEI Metadata

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import time
from gi.repository import Gst, GLib

Gst.init(None)

PIPELINE = (
    'videotestsrc is-live=true pattern=ball ! video/x-raw,width=640,height=480,framerate=30/1 ! '
    'x264enc tune=zerolatency key-int-max=30 ! '
    'seiinject name=sei ! h264parse ! seiextract ! fakesink name=sink'
)

def make_metadata_payload():
    return json.dumps({
        "frame_ts_ms": int(time.time() * 1000),
        "img_size": [640, 480],
        "objects": [
            {"id": 1, "cls": "person", "score": 0.92, "bbox_xywh": [100, 150, 60, 120]},
            {"id": 2, "cls": "car",    "score": 0.85, "bbox_xywh": [300, 220, 80, 50]}
        ]
    }, separators=(",", ":")).encode("utf-8")

def extract_metadata_probe(pad, info):
    buffer = info.get_buffer()
    if buffer and buffer.map(Gst.MapFlags.READ)[0]:
        success, map_info = buffer.map(Gst.MapFlags.READ)
        if success:
            try:
                print(json.dumps(json.loads(map_info.data.decode("utf-8")), indent=2))
            except:
                pass
            buffer.unmap(map_info)
    return Gst.PadProbeReturn.OK

def main():
    pipeline = Gst.parse_launch(PIPELINE)
    sei = pipeline.get_by_name("sei")
    sink_pad = pipeline.get_by_name("sink").get_static_pad("sink")
    sink_pad.add_probe(Gst.PadProbeType.BUFFER, extract_metadata_probe)

    pipeline.set_state(Gst.State.PLAYING)

    def inject():
        sei.set_property("metadata-binary", make_metadata_payload())
        return True

    GLib.timeout_add(100, inject)

    try:
        GLib.MainLoop().run()
    except KeyboardInterrupt:
        pass
    finally:
        pipeline.set_state(Gst.State.NULL)

if __name__ == "__main__":
    main()