RTP-PTP-Testquelle.py

#!/usr/bin/env python3
import logging
import sys

import gi

gi.require_version('Gst', '1.0')
gi.require_version('GstNet', '1.0')
from gi.repository import Gst, GstNet, GObject

logging.basicConfig(level=logging.DEBUG)

Gst.init([])
GObject.threads_init()

logging.debug('creating GObject-MainLoop')
mainloop = GObject.MainLoop()

# audiotestsrc ! audio/x-raw,channels=2,rate=48000,format=S16LE ! …
pipeline = """
    filesrc location=foo.wav ! wavparse ! audioconvert ! audio/x-raw,channels=2,rate=48000,format=S16LE !
        capssetter caps=audio/x-raw,format=S16BE !
        rtpL16pay timestamp-offset=0    !
        udpsink host=10.0.0.20 port=5004
"""
# 10.0.0.20

logging.debug('lounching pipeline \n' + pipeline)
pipeline_el = Gst.parse_launch(pipeline)

ptp_domain = 0
ptp_interfaces = ['enp0s25']

logging.info('initializing PTP-Subsystem for Network-Interface(s) %s', ptp_interfaces)
success = GstNet.ptp_init(GstNet.PTP_CLOCK_ID_NONE, ptp_interfaces)
if success:
    logging.debug('successfully initializing PTP-Subsystem')
else:
    logging.error('failed to initializing PTP-Subsystem')
    sys.exit(42)

logging.debug('obtaining PTP-Clock for domain %u', ptp_domain)
ptp_clock = GstNet.PtpClock.new('PTP-Master', ptp_domain)
if ptp_clock is not None:
    logging.debug('obtained PTP-Clock for domain %u', ptp_domain)
else:
    logging.error('failed to obtain PTP-Clock')
    sys.exit(42)

logging.debug('waiting for PTP-Clock to sync')
ptp_clock.wait_for_sync(Gst.CLOCK_TIME_NONE)
logging.info('successfully synced PTP-Clock')

pipeline_el.use_clock(ptp_clock)

logging.debug('Launching Mixing-Pipeline')
pipeline_el.set_state(Gst.State.PLAYING)

logging.info('running GObject-MainLoop')
mainloop.run()

Test-Alsa-Senke.sh

gst-launch-1.0 udpsrc address=127.0.0.1 port=5004 !\
  application/x-rtp,clock-rate=48000,channels=2,encoding-name=L16 !\
  rtpL16depay !\
  capssetter caps=audio/x-raw,format=S16LE !\
  alsasink