Skip to content

H.26X SEI

H.264 SEI (Supplemental Enhancement Information) is optional metadata carried inside an H.264 video bitstream. SEI messages do not change the decoded pixels, but they let encoders, decoders, or downstream applications attach extra information to frames.

Common uses include timestamps, camera or sensor metadata, HDR/display information, captions, stream identifiers, and custom application data. In a GStreamer pipeline, SEI is useful when metadata must stay synchronized with the video frames and travel together with the encoded stream.

h264_sei_pipe.py

h264_sei_pipe.py demonstrates a small app bridge: one GStreamer pipeline encodes raw video to Annex B H.264 access units, an appsink gives Python each encoded frame, Python inserts a user_data_unregistered SEI NAL unit, and an appsrc pushes the modified H.264 frame to an RTP/UDP pipeline.

The SEI is injected before the first VCL NAL unit, so metadata is carried in the same access unit as the frame without changing the decoded image.

1
2
3
4
5
6
7
access_unit = buffer_to_bytes(buffer)
payload = build_sei_payload(buffer)
output_data = insert_sei_before_first_vcl(access_unit, payload)

output_buffer = Gst.Buffer.new_wrapped(output_data)
clone_timing_and_flags(buffer, output_buffer)
appsrc.emit("push-buffer", output_buffer)

VCL Video Coding Layer

1
[SPS/PPS/AUD metadata] [SEI metadata] [actual encoded frame data]

The payload is wrapped as user_data_unregistered SEI, identified by a UUID, then inserted into the H.264 access unit before the actual video slice. The UUID acts like an identifier for your custom metadata format. Many tools or apps can insert SEI data into the same H.264 stream, so the receiver needs a way to know which SEI messages belong to your application.

1
SEI payload = [16-byte UUID][your JSON/message bytes]

Receiver

h264_sei_receiver.py receives the RTP stream, depayloads it back to Annex B H.264 access units, and attaches a pad probe before the decoder:

1
udpsrc ! rtpjitterbuffer ! rtph264depay ! h264parse ! identity name=sei_tap ! avdec_h264

The probe sees each encoded access unit while the SEI NAL units are still present. It maps the Gst.Buffer, scans Annex B NAL units for SEI NAL type 6, parses user_data_unregistered payloads, keeps only messages whose first 16 bytes match the application UUID, and then decodes the remaining bytes as the application payload:

1
2
3
4
5
access_unit = buffer_to_bytes(buffer)
payloads = extract_user_data_unregistered(access_unit)

for payload in payloads:
    decoded = json.loads(payload.decode("utf-8"))

This keeps metadata extraction outside the decoder and does not alter the video frames. The cost is still on the streaming path: every probed buffer is mapped and scanned, and every matching SEI payload is decoded. Keep the probe work small, avoid blocking I/O or heavy parsing in the callback, and move expensive processing to another thread or queue if the receiver must sustain high framerate or high-bitrate streams.

H.265 / HEVC version

The same idea is implemented in h265_sei.py, h265_sei_pipe.py, and h265_sei_receiver.py. The GStreamer shape is the same, but the codec elements change to HEVC:

1
2
x265enc ! h265parse ! rtph265pay
udpsrc ! rtph265depay ! h265parse ! identity name=sei_tap ! avdec_h265

The parser differences are in the NAL-unit header. H.265 NAL units have a two-byte header, the prefix SEI NAL type is 39, and VCL slice NAL types are 0..31. The SEI payload itself still uses user_data_unregistered:

1
HEVC prefix SEI = [NAL type 39 header][payload type 5][size][16-byte UUID][payload bytes]

Run h265_sei_receiver.py first, then h265_sei_pipe.py. The H.265 example uses UDP port 5002 so it can run beside the H.264 example on port 5000.

Annex B

Annex B is a common byte-stream format for H.264 and H.265 video. In Annex B, each encoded chunk, called a NAL unit, is separated by a start code:

1
00 00 01
Annex B stream looks roughly like:
1
2
3
4
5
[start code][SPS]
[start code][PPS]
[start code][SEI]
[start code][video slice]
[start code][video slice]

Demos:

h264

Helper 
  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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
#!/usr/bin/env python3

from __future__ import annotations

from collections.abc import Iterator


USER_DATA_UNREGISTERED = 5
H264_SEI_NAL_TYPE = 6
BT_SEI_UUID = b"BTGSTSEI01234567"


def make_user_data_unregistered_sei(payload: bytes, uuid: bytes = BT_SEI_UUID) -> bytes:
    if len(uuid) != 16:
        raise ValueError("H.264 user_data_unregistered SEI UUID must be 16 bytes")

    sei_payload = uuid + payload
    rbsp = (
        _encode_sei_value(USER_DATA_UNREGISTERED)
        + _encode_sei_value(len(sei_payload))
        + sei_payload
        + b"\x80"
    )

    return b"\x00\x00\x00\x01" + bytes([H264_SEI_NAL_TYPE]) + _escape_rbsp(rbsp)


def insert_sei_before_first_vcl(
    access_unit: bytes,
    payload: bytes,
    uuid: bytes = BT_SEI_UUID,
) -> bytes:
    sei_nal = make_user_data_unregistered_sei(payload, uuid)
    insert_at = _first_vcl_start(access_unit)

    if insert_at is None:
        return sei_nal + access_unit

    return access_unit[:insert_at] + sei_nal + access_unit[insert_at:]


def extract_user_data_unregistered(
    access_unit: bytes,
    uuid: bytes = BT_SEI_UUID,
) -> list[bytes]:
    payloads = []

    for _start, nal_header, nal_end in iter_annexb_nalus(access_unit):
        if access_unit[nal_header] & 0x1F != H264_SEI_NAL_TYPE:
            continue

        rbsp = _unescape_ebsp(access_unit[nal_header + 1 : nal_end])
        payloads.extend(_extract_matching_sei_payloads(rbsp, uuid))

    return payloads


def iter_annexb_nalus(data: bytes) -> Iterator[tuple[int, int, int]]:
    offset = 0

    while True:
        start = _find_start_code(data, offset)
        if start is None:
            return

        nal_header = start + _start_code_size(data, start)
        if nal_header >= len(data):
            return

        next_start = _find_start_code(data, nal_header + 1)
        nal_end = next_start if next_start is not None else len(data)
        yield start, nal_header, nal_end

        if next_start is None:
            return
        offset = next_start


def _first_vcl_start(data: bytes) -> int | None:
    for start, nal_header, _nal_end in iter_annexb_nalus(data):
        nal_type = data[nal_header] & 0x1F
        if 1 <= nal_type <= 5:
            return start
    return None


def _find_start_code(data: bytes, offset: int) -> int | None:
    three_byte = data.find(b"\x00\x00\x01", offset)
    four_byte = data.find(b"\x00\x00\x00\x01", offset)

    if three_byte == -1 and four_byte == -1:
        return None
    if three_byte == -1:
        return four_byte
    if four_byte == -1:
        return three_byte
    return min(three_byte, four_byte)


def _start_code_size(data: bytes, start: int) -> int:
    if data[start : start + 4] == b"\x00\x00\x00\x01":
        return 4
    return 3


def _encode_sei_value(value: int) -> bytes:
    chunks = bytearray()
    while value >= 255:
        chunks.append(255)
        value -= 255
    chunks.append(value)
    return bytes(chunks)


def _escape_rbsp(rbsp: bytes) -> bytes:
    escaped = bytearray()
    zero_count = 0

    for byte in rbsp:
        if zero_count >= 2 and byte <= 3:
            escaped.append(3)
            zero_count = 0

        escaped.append(byte)
        zero_count = zero_count + 1 if byte == 0 else 0

    return bytes(escaped)


def _unescape_ebsp(ebsp: bytes) -> bytes:
    rbsp = bytearray()
    zero_count = 0
    index = 0

    while index < len(ebsp):
        byte = ebsp[index]
        if zero_count >= 2 and byte == 3:
            index += 1
            zero_count = 0
            continue

        rbsp.append(byte)
        zero_count = zero_count + 1 if byte == 0 else 0
        index += 1

    return bytes(rbsp)


def _extract_matching_sei_payloads(rbsp: bytes, uuid: bytes) -> list[bytes]:
    payloads = []
    offset = 0

    while offset + 1 < len(rbsp):
        if rbsp[offset] == 0x80:
            break

        payload_type, offset = _read_sei_value(rbsp, offset)
        payload_size, offset = _read_sei_value(rbsp, offset)
        end = offset + payload_size
        if end > len(rbsp):
            break

        payload = rbsp[offset:end]
        if (
            payload_type == USER_DATA_UNREGISTERED
            and len(payload) >= 16
            and payload[:16] == uuid
        ):
            payloads.append(payload[16:])

        offset = end

    return payloads


def _read_sei_value(rbsp: bytes, offset: int) -> tuple[int, int]:
    value = 0

    while offset < len(rbsp):
        byte = rbsp[offset]
        offset += 1
        value += byte
        if byte != 255:
            return value, offset

    return value, offset
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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
#!/usr/bin/env python3

import json
import time
from itertools import count

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h264_sei import insert_sei_before_first_vcl

Gst.init(None)

sei_counter = count()

def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def clone_timing_and_flags(source, target):
    target.pts = source.pts
    target.dts = source.dts
    target.duration = source.duration
    target.offset = source.offset
    target.offset_end = source.offset_end
    target.set_flags(source.get_flags())


def build_sei_payload(buffer):
    counter = next(sei_counter)
    return json.dumps(
        {
            "counter": counter,
            "unix_time": time.time(),
            "pts": int(buffer.pts),
            "message": "bt-gst h264 sei",
        },
        separators=(",", ":"),
    ).encode("utf-8")


def on_encoded_sample(appsink, appsrc):
    sample = appsink.emit("pull-sample")
    if sample is None:
        return Gst.FlowReturn.EOS

    buffer = sample.get_buffer()
    if buffer is None:
        return Gst.FlowReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.FlowReturn.OK

    payload = build_sei_payload(buffer)
    output_data = insert_sei_before_first_vcl(access_unit, payload)
    output_buffer = Gst.Buffer.new_wrapped(output_data)
    clone_timing_and_flags(buffer, output_buffer)

    print(f"send H264 SEI app-bridge: pts={buffer.pts}, bytes={len(payload)}")
    return appsrc.emit("push-buffer", output_buffer)


source_pipeline = Gst.parse_launch(
    "videotestsrc is-live=true pattern=ball ! "
    "video/x-raw,width=640,height=480,framerate=1/1 ! "
    "x264enc tune=zerolatency speed-preset=ultrafast key-int-max=30 "
    "byte-stream=true aud=true ! "
    "h264parse config-interval=-1 ! "
    "video/x-h264,stream-format=byte-stream,alignment=au ! "
    "appsink name=encoded_sink emit-signals=true sync=false max-buffers=1 drop=true"
)

rtp_pipeline = Gst.parse_launch(
    "appsrc name=encoded_src is-live=true format=time do-timestamp=false "
    'caps="video/x-h264,stream-format=byte-stream,alignment=au" ! '
    "rtph264pay pt=96 config-interval=1 ! "
    "udpsink host=127.0.0.1 port=5000 sync=false async=false"
)

appsink = source_pipeline.get_by_name("encoded_sink")
appsrc = rtp_pipeline.get_by_name("encoded_src")
appsink.connect("new-sample", on_encoded_sample, appsrc)

loop = GLib.MainLoop()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


for pipeline in (source_pipeline, rtp_pipeline):
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", on_message, loop)

rtp_pipeline.set_state(Gst.State.PLAYING)
source_pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    source_pipeline.set_state(Gst.State.NULL)
    rtp_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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/env python3

import json

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h264_sei import extract_user_data_unregistered

Gst.init(None)


def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def sei_probe(pad, info, user_data):
    buffer = info.get_buffer()
    if buffer is None:
        return Gst.PadProbeReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.PadProbeReturn.OK

    payloads = extract_user_data_unregistered(access_unit)
    if not payloads:
        print(f"recv H264: pts={buffer.pts}, no bt sei")
        return Gst.PadProbeReturn.OK

    for payload in payloads:
        try:
            decoded = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            decoded = payload.hex()

        print(f"recv H264 SEI: pts={buffer.pts}, payload={decoded}")

    return Gst.PadProbeReturn.OK


pipeline = Gst.parse_launch(
    'udpsrc port=5000 '
    'caps="application/x-rtp,media=video,encoding-name=H264,payload=96,clock-rate=90000" ! '
    "rtpjitterbuffer latency=100 ! "
    "rtph264depay ! "
    "h264parse config-interval=-1 ! "
    "video/x-h264,stream-format=byte-stream,alignment=au ! "
    "identity name=sei_tap ! "
    "avdec_h264 ! "
    "videoconvert ! "
    "autovideosink sync=true"
)

tap = pipeline.get_by_name("sei_tap")
pad = tap.get_static_pad("sink")
pad.add_probe(Gst.PadProbeType.BUFFER, sei_probe, None)

loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus.connect("message", on_message, loop)

pipeline.set_state(Gst.State.PLAYING)

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

h265

Helper 
  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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
#!/usr/bin/env python3

from __future__ import annotations

from collections.abc import Iterator


USER_DATA_UNREGISTERED = 5
H265_PREFIX_SEI_NAL_TYPE = 39
BT_SEI_UUID = b"BTGSTSEI01234567"


def make_user_data_unregistered_sei(payload: bytes, uuid: bytes = BT_SEI_UUID) -> bytes:
    if len(uuid) != 16:
        raise ValueError("H.265 user_data_unregistered SEI UUID must be 16 bytes")

    sei_payload = uuid + payload
    rbsp = (
        _encode_sei_value(USER_DATA_UNREGISTERED)
        + _encode_sei_value(len(sei_payload))
        + sei_payload
        + b"\x80"
    )

    nal_header = bytes([(H265_PREFIX_SEI_NAL_TYPE << 1) & 0x7E, 0x01])
    return b"\x00\x00\x00\x01" + nal_header + _escape_rbsp(rbsp)


def insert_sei_before_first_vcl(
    access_unit: bytes,
    payload: bytes,
    uuid: bytes = BT_SEI_UUID,
) -> bytes:
    sei_nal = make_user_data_unregistered_sei(payload, uuid)
    insert_at = _first_vcl_start(access_unit)

    if insert_at is None:
        return sei_nal + access_unit

    return access_unit[:insert_at] + sei_nal + access_unit[insert_at:]


def extract_user_data_unregistered(
    access_unit: bytes,
    uuid: bytes = BT_SEI_UUID,
) -> list[bytes]:
    payloads = []

    for _start, nal_header, nal_end in iter_annexb_nalus(access_unit):
        if nal_header + 2 > nal_end:
            continue
        if _nal_type(access_unit[nal_header]) != H265_PREFIX_SEI_NAL_TYPE:
            continue

        rbsp = _unescape_ebsp(access_unit[nal_header + 2 : nal_end])
        payloads.extend(_extract_matching_sei_payloads(rbsp, uuid))

    return payloads


def iter_annexb_nalus(data: bytes) -> Iterator[tuple[int, int, int]]:
    offset = 0

    while True:
        start = _find_start_code(data, offset)
        if start is None:
            return

        nal_header = start + _start_code_size(data, start)
        if nal_header >= len(data):
            return

        next_start = _find_start_code(data, nal_header + 2)
        nal_end = next_start if next_start is not None else len(data)
        yield start, nal_header, nal_end

        if next_start is None:
            return
        offset = next_start


def _first_vcl_start(data: bytes) -> int | None:
    for start, nal_header, nal_end in iter_annexb_nalus(data):
        if nal_header + 2 > nal_end:
            continue
        if 0 <= _nal_type(data[nal_header]) <= 31:
            return start
    return None


def _nal_type(first_header_byte: int) -> int:
    return (first_header_byte >> 1) & 0x3F


def _find_start_code(data: bytes, offset: int) -> int | None:
    three_byte = data.find(b"\x00\x00\x01", offset)
    four_byte = data.find(b"\x00\x00\x00\x01", offset)

    if three_byte == -1 and four_byte == -1:
        return None
    if three_byte == -1:
        return four_byte
    if four_byte == -1:
        return three_byte
    return min(three_byte, four_byte)


def _start_code_size(data: bytes, start: int) -> int:
    if data[start : start + 4] == b"\x00\x00\x00\x01":
        return 4
    return 3


def _encode_sei_value(value: int) -> bytes:
    chunks = bytearray()
    while value >= 255:
        chunks.append(255)
        value -= 255
    chunks.append(value)
    return bytes(chunks)


def _escape_rbsp(rbsp: bytes) -> bytes:
    escaped = bytearray()
    zero_count = 0

    for byte in rbsp:
        if zero_count >= 2 and byte <= 3:
            escaped.append(3)
            zero_count = 0

        escaped.append(byte)
        zero_count = zero_count + 1 if byte == 0 else 0

    return bytes(escaped)


def _unescape_ebsp(ebsp: bytes) -> bytes:
    rbsp = bytearray()
    zero_count = 0
    index = 0

    while index < len(ebsp):
        byte = ebsp[index]
        if zero_count >= 2 and byte == 3:
            index += 1
            zero_count = 0
            continue

        rbsp.append(byte)
        zero_count = zero_count + 1 if byte == 0 else 0
        index += 1

    return bytes(rbsp)


def _extract_matching_sei_payloads(rbsp: bytes, uuid: bytes) -> list[bytes]:
    payloads = []
    offset = 0

    while offset + 1 < len(rbsp):
        if rbsp[offset] == 0x80:
            break

        payload_type, offset = _read_sei_value(rbsp, offset)
        payload_size, offset = _read_sei_value(rbsp, offset)
        end = offset + payload_size
        if end > len(rbsp):
            break

        payload = rbsp[offset:end]
        if (
            payload_type == USER_DATA_UNREGISTERED
            and len(payload) >= 16
            and payload[:16] == uuid
        ):
            payloads.append(payload[16:])

        offset = end

    return payloads


def _read_sei_value(rbsp: bytes, offset: int) -> tuple[int, int]:
    value = 0

    while offset < len(rbsp):
        byte = rbsp[offset]
        offset += 1
        value += byte
        if byte != 255:
            return value, offset

    return value, offset
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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
#!/usr/bin/env python3

import json
import time
from itertools import count

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h265_sei import insert_sei_before_first_vcl

Gst.init(None)

sei_counter = count()


def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def clone_timing_and_flags(source, target):
    target.pts = source.pts
    target.dts = source.dts
    target.duration = source.duration
    target.offset = source.offset
    target.offset_end = source.offset_end
    target.set_flags(source.get_flags())


def build_sei_payload(buffer):
    counter = next(sei_counter)
    return json.dumps(
        {
            "counter": counter,
            "unix_time": time.time(),
            "pts": int(buffer.pts),
            "message": "bt-gst h265 sei",
        },
        separators=(",", ":"),
    ).encode("utf-8")


def on_encoded_sample(appsink, appsrc):
    sample = appsink.emit("pull-sample")
    if sample is None:
        return Gst.FlowReturn.EOS

    buffer = sample.get_buffer()
    if buffer is None:
        return Gst.FlowReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.FlowReturn.OK

    payload = build_sei_payload(buffer)
    output_data = insert_sei_before_first_vcl(access_unit, payload)
    output_buffer = Gst.Buffer.new_wrapped(output_data)
    clone_timing_and_flags(buffer, output_buffer)

    print(f"send H265 SEI app-bridge: pts={buffer.pts}, bytes={len(payload)}")
    return appsrc.emit("push-buffer", output_buffer)


source_pipeline = Gst.parse_launch(
    "videotestsrc is-live=true pattern=ball ! "
    "video/x-raw,width=640,height=480,framerate=1/1 ! "
    "x265enc tune=zerolatency key-int-max=30 ! "
    "h265parse config-interval=-1 ! "
    "video/x-h265,stream-format=byte-stream,alignment=au ! "
    "appsink name=encoded_sink emit-signals=true sync=false max-buffers=1 drop=true"
)

rtp_pipeline = Gst.parse_launch(
    "appsrc name=encoded_src is-live=true format=time do-timestamp=false "
    'caps="video/x-h265,stream-format=byte-stream,alignment=au" ! '
    "rtph265pay pt=96 config-interval=1 ! "
    "udpsink host=127.0.0.1 port=5002 sync=false async=false"
)

appsink = source_pipeline.get_by_name("encoded_sink")
appsrc = rtp_pipeline.get_by_name("encoded_src")
appsink.connect("new-sample", on_encoded_sample, appsrc)

loop = GLib.MainLoop()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


for pipeline in (source_pipeline, rtp_pipeline):
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", on_message, loop)

rtp_pipeline.set_state(Gst.State.PLAYING)
source_pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    source_pipeline.set_state(Gst.State.NULL)
    rtp_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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/env python3

import json

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h265_sei import extract_user_data_unregistered

Gst.init(None)


def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def sei_probe(pad, info, user_data):
    buffer = info.get_buffer()
    if buffer is None:
        return Gst.PadProbeReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.PadProbeReturn.OK

    payloads = extract_user_data_unregistered(access_unit)
    if not payloads:
        print(f"recv H265: pts={buffer.pts}, no bt sei")
        return Gst.PadProbeReturn.OK

    for payload in payloads:
        try:
            decoded = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            decoded = payload.hex()

        print(f"recv H265 SEI: pts={buffer.pts}, payload={decoded}")

    return Gst.PadProbeReturn.OK


pipeline = Gst.parse_launch(
    'udpsrc port=5002 '
    'caps="application/x-rtp,media=video,encoding-name=H265,payload=96,clock-rate=90000" ! '
    "rtpjitterbuffer latency=100 ! "
    "rtph265depay ! "
    "h265parse config-interval=-1 ! "
    "video/x-h265,stream-format=byte-stream,alignment=au ! "
    "identity name=sei_tap ! "
    "avdec_h265 ! "
    "videoconvert ! "
    "autovideosink sync=true"
)

tap = pipeline.get_by_name("sei_tap")
pad = tap.get_static_pad("sink")
pad.add_probe(Gst.PadProbeType.BUFFER, sei_probe, None)

loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus.connect("message", on_message, loop)

pipeline.set_state(Gst.State.PLAYING)

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

nvidia h265

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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
#!/usr/bin/env python3

import json
import time
from itertools import count

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h265_sei import insert_sei_before_first_vcl

Gst.init(None)

sei_counter = count()


def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def clone_timing_and_flags(source, target):
    target.pts = source.pts
    target.dts = source.dts
    target.duration = source.duration
    target.offset = source.offset
    target.offset_end = source.offset_end
    target.set_flags(source.get_flags())


def build_sei_payload(buffer):
    counter = next(sei_counter)
    return json.dumps(
        {
            "counter": counter,
            "unix_time": time.time(),
            "pts": int(buffer.pts),
            "message": "bt-gst nvidia h265 sei",
        },
        separators=(",", ":"),
    ).encode("utf-8")


def on_encoded_sample(appsink, appsrc):
    sample = appsink.emit("pull-sample")
    if sample is None:
        return Gst.FlowReturn.EOS

    buffer = sample.get_buffer()
    if buffer is None:
        return Gst.FlowReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.FlowReturn.OK

    payload = build_sei_payload(buffer)
    output_data = insert_sei_before_first_vcl(access_unit, payload)
    output_buffer = Gst.Buffer.new_wrapped(output_data)
    clone_timing_and_flags(buffer, output_buffer)

    print(f"send NVIDIA H265 SEI app-bridge: pts={buffer.pts}, bytes={len(payload)}")
    return appsrc.emit("push-buffer", output_buffer)


source_pipeline = Gst.parse_launch(
    "videotestsrc is-live=true pattern=ball ! "
    "video/x-raw,width=640,height=480,framerate=30/1 ! "
    "videoconvert ! "
    "nvh265enc zerolatency=true gop-size=30 bitrate=4000 ! "
    "h265parse config-interval=-1 ! "
    "video/x-h265,stream-format=byte-stream,alignment=au ! "
    "appsink name=encoded_sink emit-signals=true sync=false max-buffers=1 drop=true"
)

rtp_pipeline = Gst.parse_launch(
    "appsrc name=encoded_src is-live=true format=time do-timestamp=false "
    'caps="video/x-h265,stream-format=byte-stream,alignment=au" ! '
    "rtph265pay pt=96 config-interval=1 ! "
    "udpsink host=127.0.0.1 port=5004 sync=false async=false"
)

appsink = source_pipeline.get_by_name("encoded_sink")
appsrc = rtp_pipeline.get_by_name("encoded_src")
appsink.connect("new-sample", on_encoded_sample, appsrc)

loop = GLib.MainLoop()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


for pipeline in (source_pipeline, rtp_pipeline):
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", on_message, loop)

rtp_pipeline.set_state(Gst.State.PLAYING)
source_pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    source_pipeline.set_state(Gst.State.NULL)
    rtp_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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/env python3

import json

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h265_sei import extract_user_data_unregistered

Gst.init(None)


def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def sei_probe(pad, info, user_data):
    buffer = info.get_buffer()
    if buffer is None:
        return Gst.PadProbeReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.PadProbeReturn.OK

    payloads = extract_user_data_unregistered(access_unit)
    if not payloads:
        print(f"recv NVIDIA H265: pts={buffer.pts}, no bt sei")
        return Gst.PadProbeReturn.OK

    for payload in payloads:
        try:
            decoded = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            decoded = payload.hex()

        print(f"recv NVIDIA H265 SEI: pts={buffer.pts}, payload={decoded}")

    return Gst.PadProbeReturn.OK


pipeline = Gst.parse_launch(
    'udpsrc port=5004 '
    'caps="application/x-rtp,media=video,encoding-name=H265,payload=96,clock-rate=90000" ! '
    "rtpjitterbuffer latency=100 ! "
    "rtph265depay ! "
    "h265parse config-interval=-1 ! "
    "video/x-h265,stream-format=byte-stream,alignment=au ! "
    "identity name=sei_tap ! "
    "nvh265dec ! "
    "videoconvert ! "
    "autovideosink sync=true"
)

tap = pipeline.get_by_name("sei_tap")
pad = tap.get_static_pad("sink")
pad.add_probe(Gst.PadProbeType.BUFFER, sei_probe, None)

loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus.connect("message", on_message, loop)

pipeline.set_state(Gst.State.PLAYING)

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

Demo using plugin

Plugin 
  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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
import json
import sys
import time
from pathlib import Path

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")
from gi.repository import GObject, Gst, GstBase  # noqa: E402

EXAMPLES_PATH = Path(__file__).resolve().parents[2] / "bt_gst" / "examples"
if str(EXAMPLES_PATH) not in sys.path:
    sys.path.insert(0, str(EXAMPLES_PATH))

from h264_sei import insert_sei_before_first_vcl  # noqa: E402

Gst.init(None)

H264_CAPS = "video/x-h264,stream-format=byte-stream,alignment=au"


def buffer_to_bytes(buffer: Gst.Buffer) -> bytes | None:
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def clone_timing_and_flags(source: Gst.Buffer, target: Gst.Buffer) -> None:
    target.pts = source.pts
    target.dts = source.dts
    target.duration = source.duration
    target.offset = source.offset
    target.offset_end = source.offset_end
    target.set_flags(source.get_flags())


def build_sei_payload(buffer: Gst.Buffer) -> bytes:
    return json.dumps(
        {
            "unix_time": time.time(),
            "pts": int(buffer.pts),
            "message": "bt-gst h264 sei",
        },
        separators=(",", ":"),
    ).encode("utf-8")


class BtH264Sei(GstBase.BaseTransform):
    __gstmetadata__ = (
        "BT H264 SEI",
        "Filter/Video",
        "Injects user_data_unregistered SEI into H.264 access units",
        "bt_ws",
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.from_string(H264_CAPS),
        ),
        Gst.PadTemplate.new(
            "sink",
            Gst.PadDirection.SINK,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.from_string(H264_CAPS),
        ),
    )

    def __init__(self) -> None:
        super().__init__()
        self.set_in_place(False)
        self.set_passthrough(False)

    def do_prepare_output_buffer(
        self,
        input_buffer: Gst.Buffer,
    ) -> tuple[Gst.FlowReturn, Gst.Buffer | None]:
        access_unit = buffer_to_bytes(input_buffer)
        if access_unit is None:
            return Gst.FlowReturn.ERROR, None

        payload = build_sei_payload(input_buffer)
        output_data = insert_sei_before_first_vcl(access_unit, payload)
        output_buffer = Gst.Buffer.new_wrapped(output_data)
        clone_timing_and_flags(input_buffer, output_buffer)

        print(f"send H264 SEI: pts={input_buffer.pts}, bytes={len(payload)}")
        return Gst.FlowReturn.OK, output_buffer

    def do_transform(
        self,
        input_buffer: Gst.Buffer,
        output_buffer: Gst.Buffer,
    ) -> Gst.FlowReturn:
        return Gst.FlowReturn.OK


GObject.type_register(BtH264Sei)
if Gst.ElementFactory.find("bt_h264_sei") is None:
    Gst.Element.register(None, "bt_h264_sei", Gst.Rank.NONE, BtH264Sei)
__gstelementfactory__ = ("bt_h264_sei", Gst.Rank.NONE, BtH264Sei)
Sender use the plugin 
 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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#!/usr/bin/env python3

import importlib.util
import os
from pathlib import Path

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst


def configure_plugin_path() -> None:
    plugin_path = str(Path(__file__).resolve().parents[2] / "plugins")
    existing_path = os.environ.get("GST_PLUGIN_PATH")
    if not existing_path:
        os.environ["GST_PLUGIN_PATH"] = plugin_path
        return

    paths = existing_path.split(os.pathsep)
    if plugin_path not in paths:
        os.environ["GST_PLUGIN_PATH"] = os.pathsep.join([plugin_path, *paths])


def register_h264_sei_element() -> None:
    plugin_path = (
        Path(__file__).resolve().parents[2]
        / "plugins"
        / "python"
        / "gstbt_h264_sei.py"
    )
    spec = importlib.util.spec_from_file_location("gstbt_h264_sei", plugin_path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Could not load {plugin_path}")

    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)


configure_plugin_path()
Gst.init(None)
register_h264_sei_element()

pipeline = Gst.parse_launch(
    "videotestsrc is-live=true pattern=ball ! "
    "video/x-raw,width=640,height=480,framerate=1/1 ! "
    "x264enc tune=zerolatency speed-preset=ultrafast key-int-max=30 "
    "byte-stream=true aud=true ! "
    "h264parse config-interval=-1 ! "`
    "video/x-h264,stream-format=byte-stream,alignment=au ! "
    "bt_h264_sei ! "
    "rtph264pay pt=96 config-interval=1 ! "
    "udpsink host=127.0.0.1 port=5000 sync=false async=false"
)

loop = GLib.MainLoop()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message", on_message, loop)

pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    pipeline.set_state(Gst.State.NULL)
Plugin 
 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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/env python3

import json

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

from h264_sei import extract_user_data_unregistered

Gst.init(None)


def buffer_to_bytes(buffer):
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if not success:
        return None

    try:
        return bytes(map_info.data)
    finally:
        buffer.unmap(map_info)


def sei_probe(pad, info, user_data):
    buffer = info.get_buffer()
    if buffer is None:
        return Gst.PadProbeReturn.OK

    access_unit = buffer_to_bytes(buffer)
    if access_unit is None:
        return Gst.PadProbeReturn.OK

    payloads = extract_user_data_unregistered(access_unit)
    if not payloads:
        print(f"recv H264: pts={buffer.pts}, no bt sei")
        return Gst.PadProbeReturn.OK

    for payload in payloads:
        try:
            decoded = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            decoded = payload.hex()

        print(f"recv H264 SEI: pts={buffer.pts}, payload={decoded}")

    return Gst.PadProbeReturn.OK


pipeline = Gst.parse_launch(
    'udpsrc port=5000 '
    'caps="application/x-rtp,media=video,encoding-name=H264,payload=96,clock-rate=90000" ! '
    "rtpjitterbuffer latency=100 ! "
    "rtph264depay ! "
    "h264parse config-interval=-1 ! "
    "video/x-h264,stream-format=byte-stream,alignment=au ! "
    "identity name=sei_tap ! "
    "avdec_h264 ! "
    "videoconvert ! "
    "autovideosink sync=true"
)

tap = pipeline.get_by_name("sei_tap")
pad = tap.get_static_pad("sink")
pad.add_probe(Gst.PadProbeType.BUFFER, sei_probe, None)

loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus.connect("message", on_message, loop)

pipeline.set_state(Gst.State.PLAYING)

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

do_prepare_output_buffer

BtH264Sei is a non-in-place GstBase.BaseTransform, so it does not edit the incoming encoded buffer directly. do_prepare_output_buffer is where the plugin builds the replacement output buffer for each H.264 access unit:

1
2
3
4
5
access_unit = buffer_to_bytes(input_buffer)
payload = build_sei_payload(input_buffer)
output_data = insert_sei_before_first_vcl(access_unit, payload)
output_buffer = Gst.Buffer.new_wrapped(output_data)
clone_timing_and_flags(input_buffer, output_buffer)

The method maps the input Gst.Buffer to bytes, creates the JSON metadata payload, inserts it as an SEI NAL unit before the first VCL NAL unit, then wraps the modified bytes in a new Gst.Buffer. Because this creates a fresh buffer, the original PTS, DTS, duration, offsets, and buffer flags are copied to keep the encoded frame synchronized with the rest of the pipeline.

If the input buffer cannot be mapped, the method returns Gst.FlowReturn.ERROR. Otherwise it returns Gst.FlowReturn.OK with the prepared output buffer. The do_transform method can then be a no-op because the output buffer has already been fully produced by do_prepare_output_buffer.