GStreamer In-Band Metadata for MPEG Transport Stream - Examples - Python
| GStreamer In-Band Metadata for MPEG Transport Stream |
|---|
|
| MPEG TS Metadata Basics |
| Getting Started |
| User Guide |
| Examples |
| Performance |
| FAQ |
| Contact Us |
 | Evaluation binary should be installed first to run any of the example programs in this guide. |
Write Metadata to TS
The following Python script generates an MPEG-TS file containing an H.264 video stream together with dynamic KLV metadata. The metadata is generated during execution and injected into the transport stream at a configurable frame interval using the metasrc element.
The script builds the complete GStreamer pipeline using the Python GStreamer bindings (PyGObject) and records the resulting transport stream into a TS file.
#!/usr/bin/env python3
"""Create an MPEG-TS file with video and dynamic metadata from Python."""
import argparse
import json
import signal
import sys
import time
import gi
gi.require_version("Gst", "1.0")
from gi.repository import GLib, Gst
PLUGIN_HINT = (
"Build/install gst-plugin-meta, or run from the source tree with "
'GST_PLUGIN_PATH="$PWD/src/.libs". Use the same GStreamer runtime that '
"was used to build the plugin."
)
def require_factories(*names):
missing = [name for name in names if Gst.ElementFactory.find(name) is None]
if missing:
raise SystemExit(
f"Missing GStreamer element(s): {', '.join(missing)}. {PLUGIN_HINT}"
)
def make_element(factory, name):
element = Gst.ElementFactory.make(factory, name)
if element is None:
raise RuntimeError(f"Could not create GStreamer element {factory!r}")
return element
def link_many(*elements):
for upstream, downstream in zip(elements, elements[1:]):
if not upstream.link(downstream):
raise RuntimeError(
f"Could not link {upstream.get_name()} to {downstream.get_name()}"
)
def format_clock_time(value):
if value == Gst.CLOCK_TIME_NONE:
return None
return int(value)
def on_bus_message(bus, message, loop):
if message.type == Gst.MessageType.ERROR:
error, debug = message.parse_error()
print(f"ERROR from {message.src.get_name()}: {error}", file=sys.stderr)
if debug:
print(debug, file=sys.stderr)
loop.quit()
elif message.type == Gst.MessageType.EOS:
loop.quit()
return True
def build_pipeline(output, every_n_frames):
pipeline = Gst.Pipeline.new("dynamic-metadata-ts-writer")
metadata_src = make_element("metasrc", "metadata_src")
metadata_caps = make_element("capsfilter", "metadata_caps")
metadata_queue = make_element("queue", "metadata_queue")
video_src = make_element("videotestsrc", "video_src")
raw_caps = make_element("capsfilter", "raw_video_caps")
overlay = make_element("timeoverlay", "time_overlay")
convert = make_element("videoconvert", "video_convert")
encoder = make_element("x264enc", "h264_encoder")
parser = make_element("h264parse", "h264_parser")
h264_caps = make_element("capsfilter", "h264_caps")
video_queue = make_element("queue", "video_queue")
mux = make_element("mpegtsmux", "mux")
sink = make_element("filesink", "ts_file")
metadata_src.set_property("metadata", '{"event":"pipeline-start"}')
metadata_caps.set_property("caps", Gst.Caps.from_string("meta/x-klv,parsed=true"))
video_src.set_property("is-live", True)
raw_caps.set_property(
"caps",
Gst.Caps.from_string("video/x-raw,width=640,height=360,framerate=30/1"),
)
encoder.set_property("tune", "zerolatency")
encoder.set_property("speed-preset", "ultrafast")
encoder.set_property("key-int-max", 30)
parser.set_property("config-interval", -1)
h264_caps.set_property(
"caps",
Gst.Caps.from_string("video/x-h264,stream-format=byte-stream,alignment=au"),
)
sink.set_property("location", output)
sink.set_property("sync", False)
sink.set_property("async", False)
elements = (
metadata_src,
metadata_caps,
metadata_queue,
video_src,
raw_caps,
overlay,
convert,
encoder,
parser,
h264_caps,
video_queue,
mux,
sink,
)
for element in elements:
pipeline.add(element)
link_many(metadata_src, metadata_caps, metadata_queue)
if not metadata_queue.link(mux):
raise RuntimeError("Could not link metadata stream to mpegtsmux")
link_many(
video_src,
raw_caps,
overlay,
convert,
encoder,
parser,
h264_caps,
video_queue,
)
if not video_queue.link(mux):
raise RuntimeError("Could not link video stream to mpegtsmux")
if not mux.link(sink):
raise RuntimeError("Could not link mpegtsmux to filesink")
frame_state = {"count": 0}
def push_metadata_for_video_buffer(pad, info):
buffer = info.get_buffer()
if buffer is None:
return Gst.PadProbeReturn.OK
frame_state["count"] += 1
frame_number = frame_state["count"]
if frame_number % every_n_frames != 0:
return Gst.PadProbeReturn.OK
payload = {
"source": "python-dynamic-metadata-ts",
"frame": frame_number,
"pts_ns": format_clock_time(buffer.pts),
"unix_time_ns": time.time_ns(),
}
data = json.dumps(payload, separators=(",", ":"))
metadata_src.set_property("metadata", data)
return Gst.PadProbeReturn.OK
video_src.get_static_pad("src").add_probe(
Gst.PadProbeType.BUFFER, push_metadata_for_video_buffer
)
return pipeline
def main():
parser = argparse.ArgumentParser(
description="Write an MPEG-TS file with dynamic metadata from Python."
)
parser.add_argument("--output", default="metadata.ts", help="Output TS file")
parser.add_argument(
"--seconds", type=int, default=8, help="Recording duration in seconds"
)
parser.add_argument(
"--every-n-frames",
type=int,
default=30,
help="Push one metadata buffer for every N video frames",
)
args = parser.parse_args()
if args.every_n_frames <= 0:
raise SystemExit("--every-n-frames must be greater than zero")
Gst.init(None)
require_factories(
"metasrc",
"capsfilter",
"queue",
"videotestsrc",
"timeoverlay",
"videoconvert",
"x264enc",
"h264parse",
"mpegtsmux",
"filesink",
)
loop = GLib.MainLoop()
pipeline = build_pipeline(args.output, args.every_n_frames)
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message", on_bus_message, loop)
def stop_pipeline(*unused):
pipeline.send_event(Gst.Event.new_eos())
return False
signal.signal(signal.SIGINT, stop_pipeline)
result = pipeline.set_state(Gst.State.PLAYING)
if result == Gst.StateChangeReturn.FAILURE:
raise RuntimeError("Could not set pipeline to PLAYING")
GLib.timeout_add_seconds(args.seconds, stop_pipeline)
try:
loop.run()
finally:
pipeline.set_state(Gst.State.NULL)
bus.remove_signal_watch()
print(f"Wrote MPEG-TS with dynamic metadata to {args.output}")
if __name__ == "__main__":
main()
Once generated the script execute it as follows.
python3 write_dynamic_metadata_ts.py --output metadata.ts
Read Metadata
The following Python script generates an MPEG-TS file containing an H.264 video stream together with dynamic KLV metadata. The metadata is generated during execution and injected into the transport stream at a configurable frame interval using the metasrc element.
The script builds the complete GStreamer pipeline using the Python GStreamer bindings (PyGObject) and records the resulting transport stream into a TS file.
import argparse
import string
import sys
import gi
gi.require_version("Gst", "1.0")
from gi.repository import GLib, Gst
PLUGIN_HINT = (
"Build/install gst-plugin-meta, or run from the source tree with "
'GST_PLUGIN_PATH="$PWD/src/.libs". Use the same GStreamer runtime that '
"was used to build the plugin."
)
def require_factories(*names):
missing = [name for name in names if Gst.ElementFactory.find(name) is None]
if missing:
raise SystemExit(
f"Missing GStreamer element(s): {', '.join(missing)}. {PLUGIN_HINT}"
)
def make_element(factory, name):
element = Gst.ElementFactory.make(factory, name)
if element is None:
raise RuntimeError(f"Could not create GStreamer element {factory!r}")
return element
def link_many(*elements):
for upstream, downstream in zip(elements, elements[1:]):
if not upstream.link(downstream):
raise RuntimeError(
f"Could not link {upstream.get_name()} to {downstream.get_name()}"
)
def clock_to_string(value):
if value == Gst.CLOCK_TIME_NONE:
return "none"
seconds, ns = divmod(int(value), Gst.SECOND)
return f"{seconds}.{ns:09d}s"
def printable(data):
allowed = set(string.printable.encode("ascii"))
return "".join(
chr(byte) if byte in allowed and byte not in (0x0B, 0x0C) else "."
for byte in data
)
def on_bus_message(bus, message, loop):
if message.type == Gst.MessageType.ERROR:
error, debug = message.parse_error()
print(f"ERROR from {message.src.get_name()}: {error}", file=sys.stderr)
if debug:
print(debug, file=sys.stderr)
loop.quit()
elif message.type == Gst.MessageType.EOS:
loop.quit()
return True
def make_appsink(on_sample):
appsink = make_element("appsink", "metadata_sink")
appsink.set_property("emit-signals", True)
appsink.set_property("sync", False)
appsink.connect("new-sample", on_sample)
return appsink
def build_raw_pipeline(input_file, appsink):
pipeline = Gst.Pipeline.new("raw-metadata-reader")
src = make_element("filesrc", "input")
capsfilter = make_element("capsfilter", "metadata_caps")
src.set_property("location", input_file)
capsfilter.set_property("caps", Gst.Caps.from_string("meta/x-klv,parsed=true"))
for element in (src, capsfilter, appsink):
pipeline.add(element)
link_many(src, capsfilter, appsink)
return pipeline
def build_ts_pipeline(input_file, appsink, parse_misb):
pipeline = Gst.Pipeline.new("ts-metadata-reader")
src = make_element("filesrc", "input")
demux = make_element("tsdemux", "demux")
queue = make_element("queue", "metadata_queue")
src.set_property("location", input_file)
elements = [src, demux, queue]
parser = None
if parse_misb:
parser = make_element("misbparser", "misb_parser")
parser.set_property("dump", False)
elements.append(parser)
elements.append(appsink)
for element in elements:
pipeline.add(element)
if not src.link(demux):
raise RuntimeError("Could not link filesrc to tsdemux")
if parser is None:
link_many(queue, appsink)
else:
link_many(queue, parser, appsink)
def on_pad_added(demux, pad):
caps = pad.get_current_caps() or pad.query_caps(None)
caps_name = caps.to_string() if caps is not None else ""
if not caps_name.startswith("meta/x-klv"):
return
sink_pad = queue.get_static_pad("sink")
if sink_pad.is_linked():
return
result = pad.link(sink_pad)
if result != Gst.PadLinkReturn.OK:
print(f"Could not link metadata pad: {result.value_nick}", file=sys.stderr)
demux.connect("pad-added", on_pad_added)
return pipeline
def main():
parser = argparse.ArgumentParser(description="Read metadata buffers with appsink.")
parser.add_argument("--input", required=True, help="Input raw KLV or MPEG-TS file")
parser.add_argument(
"--raw",
action="store_true",
help="Treat input as a raw metadata file instead of MPEG-TS",
)
parser.add_argument(
"--parse-misb",
action="store_true",
help="Run TS metadata through misbparser before appsink",
)
args = parser.parse_args()
if args.raw and args.parse_misb:
raise SystemExit("--parse-misb is only supported for MPEG-TS input")
Gst.init(None)
required = ["filesrc", "appsink"]
if args.raw:
required.append("capsfilter")
else:
required.extend(["tsdemux", "queue"])
if args.parse_misb:
required.append("misbparser")
require_factories(*required)
loop = GLib.MainLoop()
state = {"count": 0}
def on_sample(appsink):
sample = appsink.emit("pull-sample")
if sample is None:
return Gst.FlowReturn.ERROR
buffer = sample.get_buffer()
success, info = buffer.map(Gst.MapFlags.READ)
if not success:
return Gst.FlowReturn.ERROR
try:
data = bytes(info.data)
finally:
buffer.unmap(info)
state["count"] += 1
print(
f"{state['count']:04d} pts={clock_to_string(buffer.pts)} "
f"size={len(data)} ascii={printable(data)} hex={data.hex(' ')}"
)
return Gst.FlowReturn.OK
appsink = make_appsink(on_sample)
if args.raw:
pipeline = build_raw_pipeline(args.input, appsink)
else:
pipeline = build_ts_pipeline(args.input, appsink, args.parse_misb)
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message", on_bus_message, loop)
pipeline.set_state(Gst.State.PLAYING)
try:
loop.run()
finally:
pipeline.set_state(Gst.State.NULL)
bus.remove_signal_watch()
print(f"Read {state['count']} metadata buffer(s)")
if __name__ == "__main__":
main()
Execute the script as follows.
python3 read_metadata.py --input metadata.ts
The output will look as follows
Setting pipeline to PAUSED ... Pipeline is PREROLLING ... Pipeline is PREROLLED ... Setting pipeline to PLAYING ... 00000000 (0x7e0ae4004b88): 70 79 74 68 6f 6e 2d 70 65 72 69 6f 64 69 63 20 python-periodic 00000010 (0x7e0ae4004b98): 32 30 32 36 2d 30 36 2d 32 39 54 31 37 3a 31 33 2026-06-29T17:13 00000020 (0x7e0ae4004ba8): 3a 31 37 2d 30 36 30 30 00 70 79 74 68 6f 6e 2d :17-0600.python- 00000030 (0x7e0ae4004bb8): 70 65 72 69 6f 64 69 63 20 32 30 32 36 2d 30 36 periodic 2026-06 00000040 (0x7e0ae4004bc8): 2d 32 39 54 31 37 3a 31 33 3a 31 38 2d 30 36 30 -29T17:13:18-060 Redistribute latency... 00000050 (0x7e0ae4004bd8): 30 00 70 79 74 68 6f 6e 2d 70 65 72 69 6f 64 69 0.python-periodi 00000060 (0x7e0ae4004be8): 63 20 32 30 32 36 2d 30 36 2d 32 39 54 31 37 3a c 2026-06-29T17: 00000070 (0x7e0ae4004bf8): 31 33 3a 31 39 2d 30 36 30 30 00 70 79 74 68 6f 13:19-0600.pytho 00000080 (0x7e0ae4004c08): 6e 2d 70 65 72 69 6f 64 69 63 20 32 30 32 36 2d n-periodic 2026- 00000090 (0x7e0ae4004c18): 30 36 2d 32 39 54 31 37 3a 31 33 3a 32 30 2d 30 06-29T17:13:20-0 New clock: GstSystemClock 000000a0 (0x7e0ae4004c28): 36 30 30 00 70 79 74 68 6f 6e 2d 70 65 72 69 6f 600.python-perio 000000b0 (0x7e0ae4004c38): 64 69 63 20 32 30 32 36 2d 30 36 2d 32 39 54 31 dic 2026-06-29T1 000000c0 (0x7e0ae4004c48): 37 3a 31 33 3a 32 31 2d 30 36 30 30 00 70 79 74 7:13:21-0600.pyt 000000d0 (0x7e0ae4004c58): 68 6f 6e 2d 70 65 72 69 6f 64 69 63 20 32 30 32 hon-periodic 202 000000e0 (0x7e0ae4004c68): 36 2d 30 36 2d 32 39 54 31 37 3a 31 33 3a 32 32 6-06-29T17:13:22 000000f0 (0x7e0ae4004c78): 2d 30 36 30 30 00 -0600. Got EOS from element "pipeline0". EOS received - stopping pipeline... Execution ended after 0:00:00.000538260 Setting pipeline to NULL ...