A demo of how to use a GstAppSrc GStreamer element in python code to push raw audio data into a pipeline from application code.

python_gstappsrc.py

#!/usr/bin/env python3
"""
A demo of how to use a GstAppSrc GStreamer element in python code to push
raw audio data into a pipeline from application code.
"""

import sys
import math
from itertools import islice
import gi

gi.require_version('Gst', '1.0')
gi.require_version('GstApp', '1.0')
from gi.repository import Gst
from gi.repository import GstApp
from gi.repository import GLib

Gst.init_check(sys.argv)

class MyAppSrc(GstApp.AppSrc):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.sample_rate = 8000
        self.wave = self.sinwave(440)
        self.rawcaps = Gst.Caps.from_string(
            'audio/x-raw,format=U8,rate={},channels=1,layout=interleaved'.format(self.sample_rate)
        )

    def sinwave(self, freq):
        """ Generator function, yields samples of a sin wave of the specificed frequency, assuming
        the sample rate in self.sample_rate """
        samples_per_cycle = self.sample_rate/freq
        n = 0
        while True:
            n += 1
            yield int(
                127 * (math.sin(
                    math.pi*2*((n%samples_per_cycle)/samples_per_cycle)
                ) + 1.0)
            )

    def do_enough_data(self):
        print("Got enough samples")

    def do_need_data(self, need):
        print("Pushing {} more samples".format(need))
        buf = Gst.Buffer.new_wrapped(bytes(islice(self.wave, 4096)))
        self.push_buffer(buf)

    def link_with_raw_caps(self, dest):
        self.link_filtered(dest, self.rawcaps)

pipeline = Gst.ElementFactory.make("pipeline", None)

# Simple pipline, just our custom AppSrc connected dirrectly to an autoaudiosink
source = MyAppSrc()
sink = Gst.ElementFactory.make("autoaudiosink")

pipeline.add(source)
pipeline.add(sink)

# for this custom AppSrc element, we use our custom link_with_raw_caps instead of any of the
# normal link functions, so that we can be sure the output caps match our audio stream
source.link_with_raw_caps(sink)

pipeline.set_state(Gst.State.PLAYING)

GLib.MainLoop().run()