mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-12-29 11:40:38 +00:00
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
|
||
|
|
'''
|
||
|
|
Simple example to demonstrate using Gst.DeviceMonitor and `transcodebin` to
|
||
|
|
record audio from a microphone into an .oga file
|
||
|
|
'''
|
||
|
|
import gi
|
||
|
|
import sys
|
||
|
|
gi.require_version('Gst', '1.0')
|
||
|
|
gi.require_version('GstPbutils', '1.0')
|
||
|
|
gi.require_version('GLib', '2.0')
|
||
|
|
gi.require_version('GObject', '2.0')
|
||
|
|
from gi.repository import GLib, GObject, Gst, GstPbutils
|
||
|
|
|
||
|
|
def bus_call(bus, message, loop):
|
||
|
|
t = message.type
|
||
|
|
Gst.debug_bin_to_dot_file_with_ts(pipeline, Gst.DebugGraphDetails.ALL, "test")
|
||
|
|
if t == Gst.MessageType.EOS:
|
||
|
|
sys.stdout.write("End-of-stream\n")
|
||
|
|
loop.quit()
|
||
|
|
elif t == Gst.MessageType.ERROR:
|
||
|
|
err, debug = message.parse_error()
|
||
|
|
sys.stderr.write("Error: %s: %s\n" % (err, debug))
|
||
|
|
loop.quit()
|
||
|
|
return True
|
||
|
|
|
||
|
|
def stop(loop, pipeline):
|
||
|
|
_, position = pipeline.query_position(Gst.Format.TIME)
|
||
|
|
print("Position: %s\r" % Gst.TIME_ARGS(position))
|
||
|
|
|
||
|
|
if position > 10 * Gst.SECOND:
|
||
|
|
loop.quit()
|
||
|
|
print("Stopping after 10 seconds")
|
||
|
|
return False
|
||
|
|
|
||
|
|
return True
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
Gst.init(sys.argv)
|
||
|
|
|
||
|
|
if len(sys.argv) != 2:
|
||
|
|
print("Missing <output-location> parametter")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
monitor = Gst.DeviceMonitor.new()
|
||
|
|
monitor.add_filter("Audio/Source", None)
|
||
|
|
monitor.start()
|
||
|
|
|
||
|
|
# This is happening synchonously, use the GstBus based API and
|
||
|
|
# monitor.start() to avoid blocking the main thread.
|
||
|
|
devices = monitor.get_devices()
|
||
|
|
|
||
|
|
if not devices:
|
||
|
|
print("No microphone found...")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
default = [d for d in devices if d.get_properties().get_value("is-default") is True]
|
||
|
|
if len(default) == 1:
|
||
|
|
device = default[0]
|
||
|
|
else:
|
||
|
|
print("Avalaible microphones:")
|
||
|
|
for i, d in enumerate(devices):
|
||
|
|
print("%d - %s" % (i, d.get_display_name()))
|
||
|
|
res = int(input("Select device: "))
|
||
|
|
device = devices[res]
|
||
|
|
|
||
|
|
pipeline = Gst.ElementFactory.make("pipeline", None)
|
||
|
|
source = device.create_element()
|
||
|
|
transcodebin = Gst.ElementFactory.make("transcodebin", None)
|
||
|
|
Gst.util_set_object_arg(transcodebin, "profile", "video/ogg:audio/x-opus")
|
||
|
|
filesink = Gst.ElementFactory.make("filesink", None)
|
||
|
|
filesink.props.location = sys.argv[1]
|
||
|
|
|
||
|
|
pipeline.add(source, transcodebin, filesink)
|
||
|
|
source.link(transcodebin)
|
||
|
|
transcodebin.link(filesink)
|
||
|
|
|
||
|
|
pipeline.set_state(Gst.State.PLAYING)
|
||
|
|
|
||
|
|
bus = pipeline.get_bus()
|
||
|
|
bus.add_signal_watch()
|
||
|
|
|
||
|
|
loop = GLib.MainLoop()
|
||
|
|
GLib.timeout_add_seconds(1, stop, loop, pipeline)
|
||
|
|
bus.connect ("message", bus_call, loop)
|
||
|
|
loop.run()
|
||
|
|
|
||
|
|
pipeline.set_state(Gst.State.NULL)
|
||
|
|
pipeline.get_state(Gst.CLOCK_TIME_NONE)
|