Skip to content

KLV

A KLV stream is a sequence of Key-Length-Value encoded messages Using gstreamer we can send and receive klv data in SMPTE 336M standard

SMPTE 336M

The standard require 16 byte Universal Label (UL) as the key. more

Demo

Send and Receive klv using gstreamer python binding

sender
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import time

Gst.init()

pipeline_str = "appsrc name=klv_source caps=meta/x-klv,parsed=true ! rtpklvpay ! udpsink host=127.0.0.1 port=5002"
pipeline = Gst.parse_launch(pipeline_str)

appsrc = pipeline.get_by_name("klv_source")

def push_klv_data():
    # Create dummy KLV data
    klv_data = bytes([
        0x06, 0x0E, 0x2B, 0x34, 0x01, 0x01, 0x01, 0x01, 0x0E, 0x01, 0x03, 0x01, 0x01, 0x00, 0x00, 0x00,  # Key
        0x02,  # Length
        0x01, 0x02   # Value
    ])

    buf = Gst.Buffer.new_allocate(None, len(klv_data), None)
    buf.fill(0, klv_data)
    appsrc.emit("push-buffer", buf)
    return True

# Set up the main loop
loop = GLib.MainLoop()

# Start the pipeline
pipeline.set_state(Gst.State.PLAYING)

# Push KLV data periodically
GLib.timeout_add_seconds(1, push_klv_data)

try:
    loop.run()
except KeyboardInterrupt:
    pass

# Clean up
pipeline.set_state(Gst.State.NULL)
receiver
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# klv_receiver.py
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib

Gst.init(None)

def on_new_sample(sink):
    sample = sink.emit("pull-sample")
    buf = sample.get_buffer()
    success, map_info = buf.map(Gst.MapFlags.READ)
    if success:
        data = map_info.data
        print("Received KLV:", list(data))
        buf.unmap(map_info)
    return Gst.FlowReturn.OK

pipeline = Gst.parse_launch("""
udpsrc port=5002 caps="application/x-rtp, media=application, encoding-name=SMPTE336M, payload=96" !
rtpklvdepay ! appsink name=sink emit-signals=true sync=true
""")

sink = pipeline.get_by_name("sink")
sink.connect("new-sample", on_new_sample)

pipeline.set_state(Gst.State.PLAYING)
loop = GLib.MainLoop()
loop.run()