Skip to content

Instantly share code, notes, and snippets.

@SpirosArk
Forked from alumae/test_buffer.py
Last active March 24, 2021 07:44
Show Gist options
  • Save SpirosArk/5ec7bd79855e3efc758d0a7802ae459c to your computer and use it in GitHub Desktop.
Save SpirosArk/5ec7bd79855e3efc758d0a7802ae459c to your computer and use it in GitHub Desktop.
import gi
import sys
gi.require_version('Gst', '1.0')
from gi.repository import Gst
Gst.init(None)
appsrc = Gst.ElementFactory.make("appsrc", "appsrc")
filesink = Gst.ElementFactory.make("filesink", "filesink")
filesink.set_property("location", "test.dat")
pipeline = Gst.Pipeline()
pipeline.add(appsrc)
pipeline.add(filesink)
appsrc.link(filesink)
pipeline.set_state(Gst.State.PLAYING)
data = "1234" * 12
print ("Using data: %s" % data)
buf = Gst.Buffer.new_wrapped(data.encode())
appsrc.emit("push-buffer", buf)
pipeline.send_event(Gst.Event.new_eos())
result = open("test.dat").read()
print ("Result : %s" % result)
pipeline.set_state(Gst.State.NULL)
@SpirosArk
Copy link
Author

Some extra guidance on what i need for the program to do:
I need the buffer to push the data through the pipeline and write the data at filesink (where there is a file).
And then i expect from it to give me the same result as data = "1234" * 12

@SpirosArk
Copy link
Author

SpirosArk commented Mar 24, 2021

Update: Code is finally working. instead of new_allocate i used new_wrapped in order for the data to been wrapped immediately.

data.encode() was used because data is not in range of 0 to 255, so if you tried int(data) there were overflow issues.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment