| ''' | |
| Element that transforms audio samples to video frames representing | |
| the waveform. | |
| Requires matplotlib, numpy and numpy_ringbuffer | |
| Example pipeline: | |
| gst-launch-1.0 audiotestsrc ! audioplot window-duration=0.01 ! videoconvert ! autovideosink | |
| ''' | |
| import gi | |
| gi.require_version('Gst', '1.0') | |
| gi.require_version('GstBase', '1.0') | |
| gi.require_version('GstAudio', '1.0') | |
| gi.require_version('GstVideo', '1.0') | |
| from gi.repository import Gst, GLib, GObject, GstBase, GstAudio, GstVideo | |
| try: | |
| import numpy as np | |
| import matplotlib.patheffects as pe | |
| from numpy_ringbuffer import RingBuffer | |
| from matplotlib import pyplot as plt | |
| from matplotlib.backends.backend_agg import FigureCanvasAgg | |
| except ImportError: | |
| Gst.error('audioplot requires numpy, numpy_ringbuffer and matplotlib') | |
| raise | |
| Gst.init_python() | |
| AUDIO_FORMATS = [f.strip() for f in | |
| GstAudio.AUDIO_FORMATS_ALL.strip('{ }').split(',')] | |
| ICAPS = Gst.Caps(Gst.Structure('audio/x-raw', | |
| format=Gst.ValueList(AUDIO_FORMATS), | |
| layout='interleaved', | |
| rate=Gst.IntRange(range(1, GLib.MAXINT)), | |
| channels=Gst.IntRange(range(1, GLib.MAXINT)))) | |
| OCAPS = Gst.Caps(Gst.Structure('video/x-raw', | |
| format='ARGB', | |
| width=Gst.IntRange(range(1, GLib.MAXINT)), | |
| height=Gst.IntRange(range(1, GLib.MAXINT)), | |
| framerate=Gst.FractionRange(Gst.Fraction(1, 1), | |
| Gst.Fraction(GLib.MAXINT, 1)))) | |
| DEFAULT_WINDOW_DURATION = 1.0 | |
| DEFAULT_WIDTH = 640 | |
| DEFAULT_HEIGHT = 480 | |
| DEFAULT_FRAMERATE_NUM = 25 | |
| DEFAULT_FRAMERATE_DENOM = 1 | |
| class AudioPlotFilter(GstBase.BaseTransform): | |
| __gstmetadata__ = ('AudioPlotFilter', 'Filter', | |
| 'Plot audio waveforms', 'Mathieu Duponchelle') | |
| __gsttemplates__ = (Gst.PadTemplate.new("src", | |
| Gst.PadDirection.SRC, | |
| Gst.PadPresence.ALWAYS, | |
| OCAPS), | |
| Gst.PadTemplate.new("sink", | |
| Gst.PadDirection.SINK, | |
| Gst.PadPresence.ALWAYS, | |
| ICAPS)) | |
| __gproperties__ = { | |
| "window-duration": (float, | |
| "Window Duration", | |
| "Duration of the sliding window, in seconds", | |
| 0.01, | |
| 100.0, | |
| DEFAULT_WINDOW_DURATION, | |
| GObject.ParamFlags.READWRITE | |
| ) | |
| } | |
| def __init__(self): | |
| GstBase.BaseTransform.__init__(self) | |
| self.window_duration = DEFAULT_WINDOW_DURATION | |
| def do_get_property(self, prop): | |
| if prop.name == 'window-duration': | |
| return self.window_duration | |
| else: | |
| raise AttributeError('unknown property %s' % prop.name) | |
| def do_set_property(self, prop, value): | |
| if prop.name == 'window-duration': | |
| self.window_duration = value | |
| else: | |
| raise AttributeError('unknown property %s' % prop.name) | |
| def do_transform(self, inbuf, outbuf): | |
| if not self.h: | |
| self.h, = self.ax.plot(np.array(self.ringbuffer), | |
| lw=0.5, | |
| color='k', | |
| path_effects=[pe.Stroke(linewidth=1.0, | |
| foreground='g'), | |
| pe.Normal()]) | |
| else: | |
| self.h.set_ydata(np.array(self.ringbuffer)) | |
| self.fig.canvas.restore_region(self.background) | |
| self.ax.draw_artist(self.h) | |
| self.fig.canvas.blit(self.ax.bbox) | |
| s = self.agg.tostring_argb() | |
| outbuf.fill(0, s) | |
| outbuf.pts = self.next_time | |
| outbuf.duration = self.frame_duration | |
| self.next_time += self.frame_duration | |
| return Gst.FlowReturn.OK | |
| def __append(self, data): | |
| arr = np.array(data) | |
| end = self.thinning_factor * int(len(arr) / self.thinning_factor) | |
| arr = np.mean(arr[:end].reshape(-1, self.thinning_factor), 1) | |
| self.ringbuffer.extend(arr) | |
| def do_generate_output(self): | |
| inbuf = self.queued_buf | |
| _, info = inbuf.map(Gst.MapFlags.READ) | |
| res, data = self.converter.convert(GstAudio.AudioConverterFlags.NONE, | |
| info.data) | |
| data = memoryview(data).cast('i') | |
| nsamples = len(data) - self.buf_offset | |
| if nsamples == 0: | |
| self.buf_offset = 0 | |
| inbuf.unmap(info) | |
| return Gst.FlowReturn.OK, None | |
| if self.cur_offset + nsamples < self.next_offset: | |
| self.__append(data[self.buf_offset:]) | |
| self.buf_offset = 0 | |
| self.cur_offset += nsamples | |
| inbuf.unmap(info) | |
| return Gst.FlowReturn.OK, None | |
| consumed = self.next_offset - self.cur_offset | |
| self.__append(data[self.buf_offset:self.buf_offset + consumed]) | |
| inbuf.unmap(info) | |
| _, outbuf = GstBase.BaseTransform.do_prepare_output_buffer(self, inbuf) | |
| ret = self.do_transform(inbuf, outbuf) | |
| self.next_offset += self.samplesperbuffer | |
| self.cur_offset += consumed | |
| self.buf_offset += consumed | |
| return ret, outbuf | |
| def do_transform_caps(self, direction, caps, filter_): | |
| if direction == Gst.PadDirection.SRC: | |
| res = ICAPS | |
| else: | |
| res = OCAPS | |
| if filter_: | |
| res = res.intersect(filter_) | |
| return res | |
| def do_fixate_caps(self, direction, caps, othercaps): | |
| if direction == Gst.PadDirection.SRC: | |
| return othercaps.fixate() | |
| else: | |
| so = othercaps.get_structure(0).copy() | |
| so.fixate_field_nearest_fraction("framerate", | |
| DEFAULT_FRAMERATE_NUM, | |
| DEFAULT_FRAMERATE_DENOM) | |
| so.fixate_field_nearest_int("width", DEFAULT_WIDTH) | |
| so.fixate_field_nearest_int("height", DEFAULT_HEIGHT) | |
| ret = Gst.Caps.new_empty() | |
| ret.append_structure(so) | |
| return ret.fixate() | |
| def do_set_caps(self, icaps, ocaps): | |
| in_info = GstAudio.AudioInfo.new_from_caps(icaps) | |
| out_info = GstVideo.VideoInfo().new_from_caps(ocaps) | |
| self.convert_info = GstAudio.AudioInfo() | |
| self.convert_info.set_format(GstAudio.AudioFormat.S32, | |
| in_info.rate, | |
| in_info.channels, | |
| in_info.position) | |
| self.converter = GstAudio.AudioConverter.new(GstAudio.AudioConverterFlags.NONE, | |
| in_info, | |
| self.convert_info, | |
| None) | |
| self.fig = plt.figure() | |
| dpi = self.fig.get_dpi() | |
| self.fig.patch.set_alpha(0.3) | |
| self.fig.set_size_inches(out_info.width / float(dpi), | |
| out_info.height / float(dpi)) | |
| self.ax = plt.Axes(self.fig, [0., 0., 1., 1.]) | |
| self.fig.add_axes(self.ax) | |
| self.ax.set_axis_off() | |
| self.ax.set_ylim((GLib.MININT, GLib.MAXINT)) | |
| self.agg = self.fig.canvas.switch_backends(FigureCanvasAgg) | |
| self.h = None | |
| samplesperwindow = int(in_info.rate * in_info.channels * self.window_duration) | |
| self.thinning_factor = max(int(samplesperwindow / out_info.width - 1), 1) | |
| cap = int(samplesperwindow / self.thinning_factor) | |
| self.ax.set_xlim([0, cap]) | |
| self.ringbuffer = RingBuffer(capacity=cap) | |
| self.ringbuffer.extend([0.0] * cap) | |
| self.frame_duration = Gst.util_uint64_scale_int(Gst.SECOND, | |
| out_info.fps_d, | |
| out_info.fps_n) | |
| self.next_time = self.frame_duration | |
| self.agg.draw() | |
| self.background = self.fig.canvas.copy_from_bbox(self.ax.bbox) | |
| self.samplesperbuffer = Gst.util_uint64_scale_int(in_info.rate * in_info.channels, | |
| out_info.fps_d, | |
| out_info.fps_n) | |
| self.next_offset = self.samplesperbuffer | |
| self.cur_offset = 0 | |
| self.buf_offset = 0 | |
| return True | |
| GObject.type_register(AudioPlotFilter) | |
| __gstelementfactory__ = ("audioplot", Gst.Rank.NONE, AudioPlotFilter) |