MPEG TS and metadata channels

MPEG-TS in brief

MPEG Transport Stream (MPEG-TS) is a container format designed for streaming audio, video, and metadata over unreliable networks or broadcast links. The stream is split into small fixed-size packets, usually 188 bytes, where each packet carries a packet identifier (PID). A receiver uses these PIDs and the program tables in the stream to find the elementary streams that belong to a program, such as a video track, audio track, or metadata track.

MPEG-TS is commonly used for live video because it can be decoded progressively and can recover from packet loss better than formats that depend on a complete file structure.

Simple video-only UDP pipeline

The following pipeline generates a test video pattern, encodes it as H.264, muxes it into MPEG-TS, and streams it over UDP:

gst-launch-1.0 -e \
  videotestsrc is-live=true ! \
  video/x-raw,width=1280,height=720,framerate=30/1 ! \
  x264enc tune=zerolatency bitrate=2000 speed-preset=ultrafast key-int-max=30 bframes=0 ! \
  h264parse config-interval=-1 ! \
  mpegtsmux alignment=7 ! \
  udpsink host=127.0.0.1 port=5000 sync=false async=false

mpegtsmux alignment=7

mpegtsmux alignment=7 tells GStreamer how many MPEG-TS packets to group into each output buffer. MPEG-TS packets are normally 188 bytes each.

7 * 188 = 1316 bytes

This is commonly used for UDP streaming because 1316 bytes fits well inside a typical Ethernet MTU of 1500 bytes after IP/UDP headers are added. It helps avoid IP fragmentation.

h264parse config-interval=-1

tells h264parse to insert the H.264 configuration data into the stream with every IDR keyframe. For UDP/live streaming, -1 is useful because a receiver can join the stream later and still get the required decoder configuration at the next keyframe. Without SPS/PPS appearing in-band often enough, a late receiver may fail to decode until it somehow receives that config data.

IDR: Instantaneous Decoder Refresh frame. A keyframe that can be decoded without needing earlier frames. After an IDR, the decoder can start fresh.

This sends an MPEG-TS stream with only one video elementary stream and no audio or metadata channels.

Run this receiver pipeline in another terminal to receive the MPEG-TS stream, extract the H.264 video, decode it, and display it:

gst-launch-1.0 \
  udpsrc port=5000 caps="video/mpegts, systemstream=(boolean)true, packetsize=(int)188" ! \
  queue max-size-buffers=0 max-size-bytes=0 max-size-time=100000000 leaky=downstream ! \
  tsdemux latency=0 ! \
  queue max-size-buffers=2 max-size-bytes=0 max-size-time=0 leaky=downstream ! \
  h264parse ! \
  avdec_h264 ! \
  videoconvert ! \
  queue max-size-buffers=2 max-size-bytes=0 max-size-time=0 leaky=downstream ! \
  autovideosink sync=false

This version is tuned for low latency. tsdemux latency=0 removes the default demux smoothing delay, leaky queues prevent old frames from building up, and autovideosink sync=false displays frames as soon as they are decoded. If the machine cannot decode fast enough, frames may be dropped instead of increasing the visible delay.

• queue before tsdemux: absorb UDP jitter (absorb means “take in and smooth out.”)
• queue after tsdemux: separate demux from decode
• h264parse: clean/prepare H.264 for decoder
• queue before sink: prevent display slowness from causing latency buildup

---

Metadata-only UDP pipeline

MPEG-TS can carry a metadata elementary stream without video or audio. In GStreamer, mpegtsmux accepts this kind of metadata on a meta/x-klv sink pad. For a real interoperable system, wrap the JSON in valid KLV. For a simple lab pipeline, the JSON bytes can be sent on that private metadata channel.

Run this sender pipeline to stream only JSON metadata over UDP:

while true; do
  printf '{"timestamp":%s,"source":"demo","value":42}\n' "$(date +%s)"
  sleep 1
done | gst-launch-1.0 -e \
  fdsrc fd=0 do-timestamp=true ! \
  queue ! \
  meta/x-klv,parsed=true ! \
  mpegtsmux alignment=7 ! \
  udpsink host=127.0.0.1 port=5000 sync=false async=false

Run this receiver pipeline in another terminal to receive the MPEG-TS stream and print the metadata channel to stdout:

gst-launch-1.0 \
  udpsrc port=5000 caps="video/mpegts, systemstream=(boolean)true, packetsize=(int)188" ! \
  queue max-size-buffers=0 max-size-bytes=0 max-size-time=100000000 leaky=downstream ! \
  tsdemux latency=0 ! \
  queue ! "meta/x-klv" ! identity silent=false dump=true ! fakesink sync=false

The final branch handles only the demuxed metadata stream:

• queue decouples tsdemux from the metadata consumer so parsing and printing metadata does not block demuxing.
• "meta/x-klv" filters the branch to KLV metadata buffers from the transport stream.
• identity silent=false dump=true logs each buffer and dumps its bytes, which makes the JSON payload visible in the terminal.
• fakesink sync=false discards the buffers after inspection and does not wait on the pipeline clock.

---

Demo:

Sender

json_sender.py creates a live MPEG-TS sender with two synchronized inputs. For every timer tick it generates one OpenCV video frame and one UTF-8 JSON payload, gives both buffers the same PTS/DTS/duration, muxes them with mpegtsmux, and sends the transport stream to udp://HOST:PORT.

flowchart LR
    timer["GLib timer<br/>fps interval"] --> push["push_frame_and_json()"]

    push --> frame["make_counter_frame()<br/>BGR video bytes"]
    frame --> videosrc["appsrc videosrc<br/>video/x-raw, BGR"]
    videosrc --> vqueue1["queue<br/>leaky downstream"]
    vqueue1 --> tee["tee name=video_tee"]

    tee --> vqueue2["queue"]
    vqueue2 --> convert1["videoconvert"]
    convert1 --> enc["x264enc<br/>zerolatency, ultrafast"]
    enc --> parse["h264parse<br/>config-interval=1"]
    parse --> vqueue3["queue"]
    vqueue3 --> mux["mpegtsmux<br/>alignment=7"]

    tee --> previewq["queue"]
    previewq --> convert2["videoconvert"]
    convert2 --> preview["autovideosink<br/>local preview"]

    push --> json["make_json_payload()<br/>UTF-8 JSON bytes"]
    json --> jsonsrc["appsrc jsonsrc<br/>meta/x-klv, parsed=true"]
    jsonsrc --> mqueue["queue<br/>leaky downstream"]
    mqueue --> mux

    mux --> outq["queue"]
    outq --> udp["udpsink<br/>sync=false async=false"]

Sender code

#!/usr/bin/env python3
"""Send H.264 video plus private JSON metadata in MPEG-TS over UDP.

The metadata branch uses `meta/x-klv` caps because `mpegtsmux` supports that
stream type. The payload bytes are UTF-8 JSON for this private sender/receiver
pair, not standards-compliant KLV.
"""

from __future__ import annotations

import argparse
import json
import signal
import sys
import time
from typing import Optional

import gi
import cv2
import numpy as np

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

from gi.repository import GLib, Gst


Gst.init(None)


def make_json_payload(counter: int, created_unix_ns: int, sender_pts_ms: float) -> bytes:
    payload = {
        "counter": counter,
        "created_unix_ns": created_unix_ns,
        "sender_pts_ms": sender_pts_ms,
        "message": "hello from json metadata stream",
    }
    return json.dumps(payload, separators=(",", ":")).encode("utf-8")


def make_counter_frame(counter: int, pts_ms: float, width: int, height: int) -> bytes:
    frame = np.zeros((height, width, 3), dtype=np.uint8)
    frame[:, :] = (24, 24, 24)

    grid_color = (55, 55, 55)
    for x in range(0, width, 80):
        cv2.line(frame, (x, 0), (x, height), grid_color, 1)
    for y in range(0, height, 80):
        cv2.line(frame, (0, y), (width, y), grid_color, 1)

    marker_size = max(40, min(width, height) // 10)
    usable_width = max(1, width - marker_size - 40)
    x = 20 + ((counter * 13) % usable_width)
    y = height // 2 - marker_size // 2
    cv2.rectangle(frame, (x, y), (x + marker_size, y + marker_size), (0, 180, 255), -1)
    cv2.rectangle(frame, (x, y), (x + marker_size, y + marker_size), (255, 255, 255), 3)

    cv2.putText(
        frame,
        f"TX JSON #{counter}",
        (40, 90),
        cv2.FONT_HERSHEY_SIMPLEX,
        2.0,
        (255, 255, 255),
        4,
        cv2.LINE_AA,
    )
    cv2.putText(
        frame,
        f"PTS {pts_ms:.2f} ms",
        (40, 150),
        cv2.FONT_HERSHEY_SIMPLEX,
        1.2,
        (80, 220, 255),
        3,
        cv2.LINE_AA,
    )
    cv2.putText(
        frame,
        f"frame {counter}",
        (40, height - 50),
        cv2.FONT_HERSHEY_SIMPLEX,
        1.1,
        (180, 255, 120),
        3,
        cv2.LINE_AA,
    )
    return frame.tobytes()


class H264JsonUdpSender:
    def __init__(
        self,
        host: str,
        port: int,
        width: int,
        height: int,
        fps: int,
        bitrate_kbps: int,
    ) -> None:
        self.host = host
        self.port = port
        self.width = width
        self.height = height
        self.fps = fps
        self.bitrate_kbps = bitrate_kbps

        self.counter = 0
        self.loop: Optional[GLib.MainLoop] = None

        pipeline_desc = f"""
    mpegtsmux name=mux alignment=7
        ! queue
        ! udpsink host={host} port={port} sync=false async=false

    appsrc name=videosrc
           is-live=true
           format=time
           do-timestamp=false
           block=false
           caps=video/x-raw,format=BGR,width={width},height={height},framerate={fps}/1
        ! queue leaky=downstream max-size-buffers=3 max-size-time=0 max-size-bytes=0
        ! tee name=video_tee

    video_tee.
        ! queue leaky=downstream max-size-buffers=3 max-size-time=0 max-size-bytes=0
        ! videoconvert
        ! x264enc tune=zerolatency
                  speed-preset=ultrafast
                  bitrate={bitrate_kbps}
                  key-int-max={fps}
                  bframes=0
                  byte-stream=true
        ! h264parse config-interval=1
        ! queue max-size-buffers=0 max-size-time=0 max-size-bytes=0
        ! mux.

    video_tee.
        ! queue leaky=downstream max-size-buffers=3 max-size-time=0 max-size-bytes=0
        ! videoconvert
        ! autovideosink sync=false

    appsrc name=jsonsrc
           is-live=true
           format=time
           do-timestamp=false
           block=false
           caps=meta/x-klv,parsed=true
        ! queue leaky=downstream max-size-buffers=2 max-size-time=0 max-size-bytes=0
        ! mux.
"""

        self.pipeline = Gst.parse_launch(pipeline_desc)

        self.videosrc = self.pipeline.get_by_name("videosrc")
        if self.videosrc is None:
            raise RuntimeError("Could not find appsrc named videosrc")

        self.jsonsrc = self.pipeline.get_by_name("jsonsrc")
        if self.jsonsrc is None:
            raise RuntimeError("Could not find appsrc named jsonsrc")

        self.videosrc.set_property("is-live", True)
        self.videosrc.set_property("format", Gst.Format.TIME)
        self.videosrc.set_property("do-timestamp", False)
        self.videosrc.set_property("block", False)
        self.videosrc.set_property("max-bytes", width * height * 3 * 3)
        self.videosrc.set_property(
            "caps",
            Gst.Caps.from_string(
                f"video/x-raw,format=BGR,width={width},height={height},framerate={fps}/1"
            ),
        )

        self.jsonsrc.set_property("is-live", True)
        self.jsonsrc.set_property("format", Gst.Format.TIME)
        self.jsonsrc.set_property("do-timestamp", False)
        self.jsonsrc.set_property("block", False)
        self.jsonsrc.set_property("max-bytes", 4096)
        self.jsonsrc.set_property(
            "caps",
            Gst.Caps.from_string("meta/x-klv,parsed=true"),
        )

        bus = self.pipeline.get_bus()
        bus.add_signal_watch()
        bus.connect("message", self.on_bus_message)

    def on_bus_message(self, bus: Gst.Bus, message: Gst.Message) -> None:
        msg_type = message.type

        if msg_type == Gst.MessageType.ERROR:
            err, debug = message.parse_error()
            print(f"[ERROR] {err}", file=sys.stderr)
            if debug:
                print(f"[DEBUG] {debug}", file=sys.stderr)
            self.stop()

        elif msg_type == Gst.MessageType.WARNING:
            err, debug = message.parse_warning()
            print(f"[WARNING] {err}", file=sys.stderr)
            if debug:
                print(f"[DEBUG] {debug}", file=sys.stderr)

        elif msg_type == Gst.MessageType.EOS:
            print("[INFO] EOS")
            self.stop()

        elif msg_type == Gst.MessageType.STATE_CHANGED:
            if message.src == self.pipeline:
                old, new, _pending = message.parse_state_changed()
                print(f"[INFO] Pipeline state: {old.value_nick} -> {new.value_nick}")

    def start_frame_timer(self) -> bool:
        interval_ms = max(1, int(1000 / self.fps))
        print(f"[INFO] Starting frame generator: {self.fps} FPS, interval={interval_ms} ms")
        GLib.timeout_add(interval_ms, self.push_frame_and_json)
        return False

    def push_frame_and_json(self) -> bool:
        frame_duration = Gst.SECOND // self.fps
        frame_pts = self.counter * frame_duration
        pts_ms = frame_pts / Gst.MSECOND

        frame = make_counter_frame(
            self.counter,
            pts_ms,
            width=self.width,
            height=self.height,
        )
        video_buf = Gst.Buffer.new_allocate(None, len(frame), None)
        video_buf.fill(0, frame)
        video_buf.pts = frame_pts
        video_buf.dts = frame_pts
        video_buf.duration = frame_duration

        video_ret = self.videosrc.emit("push-buffer", video_buf)
        if video_ret != Gst.FlowReturn.OK:
            print(f"[WARNING] video push-buffer returned {video_ret}", file=sys.stderr)
            return False

        created_unix_ns = time.time_ns()
        payload = make_json_payload(self.counter, created_unix_ns, pts_ms)

        json_buf = Gst.Buffer.new_allocate(None, len(payload), None)
        json_buf.fill(0, payload)
        json_buf.pts = frame_pts
        json_buf.dts = frame_pts
        json_buf.duration = frame_duration

        json_ret = self.jsonsrc.emit("push-buffer", json_buf)
        if json_ret != Gst.FlowReturn.OK:
            print(f"[WARNING] JSON push-buffer returned {json_ret}", file=sys.stderr)
            return False

        if self.counter % max(1, self.fps) == 0:
            print(
                f"[INFO] pushed JSON #{self.counter}, "
                f"pts_ms={pts_ms:.2f}, "
                f"size={len(payload)} bytes"
            )

        self.counter += 1
        return True

    def run(self) -> None:
        self.loop = GLib.MainLoop()

        print(
            f"[INFO] Sending H.264 + JSON metadata in MPEG-TS to udp://{self.host}:{self.port}"
        )
        print(
            f"[INFO] Video={self.width}x{self.height}@{self.fps}, "
            f"bitrate={self.bitrate_kbps} kbps, JSON=per video frame"
        )

        ret = self.pipeline.set_state(Gst.State.PLAYING)
        print(f"[INFO] set_state PLAYING returned: {ret.value_nick}")

        GLib.timeout_add(100, self.start_frame_timer)

        def handle_signal(sig, frame):
            print("\n[INFO] Stopping sender...")
            self.stop()

        signal.signal(signal.SIGINT, handle_signal)
        signal.signal(signal.SIGTERM, handle_signal)

        try:
            self.loop.run()
        finally:
            self.pipeline.set_state(Gst.State.NULL)

    def stop(self) -> None:
        try:
            self.videosrc.emit("end-of-stream")
        except Exception:
            pass
        try:
            self.jsonsrc.emit("end-of-stream")
        except Exception:
            pass

        self.pipeline.set_state(Gst.State.NULL)

        if self.loop and self.loop.is_running():
            self.loop.quit()


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=5000)
    parser.add_argument("--width", type=int, default=1280)
    parser.add_argument("--height", type=int, default=720)
    parser.add_argument("--fps", type=int, default=30)
    parser.add_argument("--bitrate-kbps", type=int, default=2500)

    args = parser.parse_args()

    sender = H264JsonUdpSender(
        host=args.host,
        port=args.port,
        width=args.width,
        height=args.height,
        fps=args.fps,
        bitrate_kbps=args.bitrate_kbps,
    )

    sender.run()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Receiver

json_receiver.py listens for an MPEG-TS stream on UDP port 5000 by default. tsdemux creates dynamic pads for the incoming streams: the H.264 pad is linked to the video decode/display branch, while the first non-video pad is treated as the private JSON metadata stream and linked to an appsink.

For every metadata sample, the receiver decodes the bytes as UTF-8 JSON, prints the payload with timing information, compares metadata PTS with the latest decoded video PTS, and updates a textoverlay on the video output. Invalid JSON is logged as raw bytes so transport or payload issues are still visible.

flowchart LR
    udp["udpsrc<br/>port=5000<br/>video/mpegts packetsize=188"] --> demux["tsdemux<br/>latency=0"]

    demux -- "video/x-h264 dynamic pad" --> vqueue["queue<br/>leaky downstream"]
    vqueue --> parse["h264parse"]
    parse --> decoder["avdec_h264<br/>max-threads=1"]
    decoder --> convert["videoconvert"]
    convert --> probe["video PTS pad probe"]
    probe --> overlay["textoverlay<br/>RX JSON status"]
    overlay --> vsink["autovideosink<br/>or fakesink with --no-video<br/>sync=false"]

    demux -- "private metadata dynamic pad" --> mqueue["queue<br/>leaky downstream"]
    mqueue --> appsink["appsink meta_sink<br/>emit-signals=true<br/>drop=true"]
    appsink --> sample["on_metadata_sample()"]
    sample --> decode["decode UTF-8 JSON"]
    decode --> metrics["counter, wall latency,<br/>metadata/video PTS deltas"]
    metrics --> log["print payload and timing"]
    metrics --> overlay

Receiver code

#!/usr/bin/env python3
"""Receive H.264 video plus private JSON metadata from MPEG-TS over UDP."""

from __future__ import annotations

import argparse
import json
import signal
import sys
import time
from typing import Optional

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GLib", "2.0")
from gi.repository import GLib, Gst


Gst.init(None)


def get_element_or_raise(pipeline: Gst.Element, name: str) -> Gst.Element:
    elem = pipeline.get_by_name(name)
    if elem is None:
        raise RuntimeError(f"Could not find element named {name!r}")
    return elem


class TsVideoJsonReceiver:
    def __init__(self, port: int, show_video: bool = True) -> None:
        self.port = port
        self.show_video = show_video
        self.loop: Optional[GLib.MainLoop] = None
        self.json_count = 0
        self.video_linked = False
        self.metadata_linked = False
        self.latest_video_pts_ms: float | None = None

        video_sink = "autovideosink" if show_video else "fakesink"
        pipeline_desc = f"""
    udpsrc name=udpsrc port={port}
        caps="video/mpegts,systemstream=true,packetsize=188"
        ! tsdemux name=demux latency=0

    queue name=video_queue leaky=downstream max-size-buffers=3 max-size-time=0 max-size-bytes=0
        ! h264parse name=h264parse
        ! avdec_h264 name=decoder max-threads=1
        ! videoconvert name=videoconvert
        ! textoverlay name=receiver_overlay
                       text="RX JSON waiting"
                       valignment=bottom
                       halignment=right
                       font-desc="Sans, 20"
                       shaded-background=true
        ! {video_sink} name=video_sink sync=false

    queue name=meta_queue leaky=downstream max-size-buffers=100 max-size-time=0 max-size-bytes=0
        ! appsink name=meta_sink emit-signals=true sync=false drop=true max-buffers=100
"""

        self.pipeline = Gst.parse_launch(pipeline_desc)

        self.demux = get_element_or_raise(self.pipeline, "demux")
        self.video_queue = get_element_or_raise(self.pipeline, "video_queue")
        self.videoconvert = get_element_or_raise(self.pipeline, "videoconvert")
        self.receiver_overlay = get_element_or_raise(self.pipeline, "receiver_overlay")
        self.meta_queue = get_element_or_raise(self.pipeline, "meta_queue")
        self.meta_sink = get_element_or_raise(self.pipeline, "meta_sink")

        self.demux.connect("pad-added", self.on_demux_pad_added)
        self.meta_sink.connect("new-sample", self.on_metadata_sample)

        video_probe_pad = self.videoconvert.get_static_pad("src")
        if video_probe_pad is None:
            raise RuntimeError("Could not get videoconvert src pad")
        video_probe_pad.add_probe(Gst.PadProbeType.BUFFER, self.on_video_buffer)

        bus = self.pipeline.get_bus()
        bus.add_signal_watch()
        bus.connect("message", self.on_bus_message)

    def on_video_buffer(self, pad: Gst.Pad, info: Gst.PadProbeInfo) -> Gst.PadProbeReturn:
        buf = info.get_buffer()
        if buf is not None and buf.pts != Gst.CLOCK_TIME_NONE:
            self.latest_video_pts_ms = buf.pts / Gst.MSECOND
        return Gst.PadProbeReturn.OK

    def on_demux_pad_added(self, demux: Gst.Element, pad: Gst.Pad) -> None:
        caps = pad.get_current_caps()
        if caps is None:
            caps = pad.query_caps(None)

        caps_str = caps.to_string()
        pad_name = pad.get_name()

        print(f"[INFO] demux pad-added: name={pad_name}, caps={caps_str}")

        if caps_str.startswith("video/x-h264"):
            if self.video_linked:
                print("[INFO] Video already linked; ignoring extra video pad")
                return

            sink_pad = self.video_queue.get_static_pad("sink")
            result = pad.link(sink_pad)
            print(f"[INFO] video link result: {result.value_nick}")

            if result == Gst.PadLinkReturn.OK:
                self.video_linked = True
            return

        if not self.metadata_linked:
            sink_pad = self.meta_queue.get_static_pad("sink")
            result = pad.link(sink_pad)
            print(f"[INFO] metadata/JSON link result: {result.value_nick}")

            if result == Gst.PadLinkReturn.OK:
                self.metadata_linked = True
            return

        print(f"[INFO] Extra non-video pad ignored: name={pad_name}, caps={caps_str}")

    def on_metadata_sample(self, sink: Gst.Element) -> Gst.FlowReturn:
        sample = sink.emit("pull-sample")
        if sample is None:
            return Gst.FlowReturn.ERROR

        buf = sample.get_buffer()
        if buf is None:
            return Gst.FlowReturn.ERROR

        ok, map_info = buf.map(Gst.MapFlags.READ)
        if not ok:
            return Gst.FlowReturn.ERROR

        try:
            data = bytes(map_info.data)
        finally:
            buf.unmap(map_info)

        self.json_count += 1

        pts_ms = None
        if buf.pts != Gst.CLOCK_TIME_NONE:
            pts_ms = buf.pts / Gst.MSECOND

        duration_ms = None
        if buf.duration != Gst.CLOCK_TIME_NONE:
            duration_ms = buf.duration / Gst.MSECOND

        try:
            payload = json.loads(data.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            print(
                f"JSON #{self.json_count}: "
                f"invalid payload size={len(data)} bytes, "
                f"pts_ms={pts_ms}, "
                f"duration_ms={duration_ms}, "
                f"error={exc}, "
                f"hex={data.hex(' ')}",
                flush=True,
            )
            return Gst.FlowReturn.OK

        counter = payload.get("counter")
        created_unix_ns = payload.get("created_unix_ns")
        sender_pts_ms = payload.get("sender_pts_ms")

        receiver_time_ns = time.time_ns()
        wall_latency_ms = None
        if isinstance(created_unix_ns, int):
            wall_latency_ms = (receiver_time_ns - created_unix_ns) / 1_000_000.0

        metadata_pts_delta_ms = None
        if pts_ms is not None and isinstance(sender_pts_ms, (int, float)):
            metadata_pts_delta_ms = pts_ms - float(sender_pts_ms)

        video_meta_delta_ms = None
        if self.latest_video_pts_ms is not None and pts_ms is not None:
            video_meta_delta_ms = self.latest_video_pts_ms - pts_ms

        overlay_lines = [f"RX JSON #{counter}"]
        if wall_latency_ms is not None:
            overlay_lines.append(f"wall {wall_latency_ms:.2f} ms")
        if video_meta_delta_ms is not None:
            overlay_lines.append(f"video-meta {video_meta_delta_ms:.2f} ms")
        self.receiver_overlay.set_property("text", "\n".join(overlay_lines))

        print(
            f"JSON #{self.json_count}: "
            f"counter={counter}, "
            f"size={len(data)} bytes, "
            f"metadata_pts_ms={pts_ms}, "
            f"sender_pts_ms={sender_pts_ms}, "
            f"wall_latency_ms={wall_latency_ms}, "
            f"latest_video_pts_ms={self.latest_video_pts_ms}, "
            f"video_meta_delta_ms={video_meta_delta_ms}, "
            f"metadata_pts_delta_ms={metadata_pts_delta_ms}, "
            f"duration_ms={duration_ms}, "
            f"payload={payload}",
            flush=True,
        )

        return Gst.FlowReturn.OK

    def on_bus_message(self, bus: Gst.Bus, message: Gst.Message) -> None:
        msg_type = message.type

        if msg_type == Gst.MessageType.ERROR:
            err, debug = message.parse_error()
            print(f"[ERROR] {err}", file=sys.stderr)
            if debug:
                print(f"[DEBUG] {debug}", file=sys.stderr)
            self.stop()

        elif msg_type == Gst.MessageType.WARNING:
            err, debug = message.parse_warning()
            print(f"[WARNING] {err}", file=sys.stderr)
            if debug:
                print(f"[DEBUG] {debug}", file=sys.stderr)

        elif msg_type == Gst.MessageType.EOS:
            print("[INFO] EOS")
            self.stop()

    def run(self) -> None:
        self.loop = GLib.MainLoop()

        print(f"[INFO] Listening on UDP port {self.port}")

        ret = self.pipeline.set_state(Gst.State.PLAYING)
        print(f"[INFO] set_state PLAYING: {ret.value_nick}")

        def handle_signal(sig, frame):
            print("\n[INFO] stopping...")
            self.stop()

        signal.signal(signal.SIGINT, handle_signal)
        signal.signal(signal.SIGTERM, handle_signal)

        try:
            self.loop.run()
        finally:
            self.pipeline.set_state(Gst.State.NULL)

    def stop(self) -> None:
        self.pipeline.set_state(Gst.State.NULL)
        if self.loop and self.loop.is_running():
            self.loop.quit()


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=5000)
    parser.add_argument("--no-video", action="store_true")
    args = parser.parse_args()

    receiver = TsVideoJsonReceiver(
        port=args.port,
        show_video=not args.no_video,
    )
    receiver.run()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())