自定义Python GStreamer插件注册失败:无效ELF头错误求助
问题:Python GStreamer插件注册时出现"invalid ELF header"错误
运行以下命令时触发错误:
GST_PLUGIN_PATH=$GST_PLUGIN_PATH:$PWD gst-inspect-1.0 audiotestsrc_py.py
错误输出:
(gst-inspect-1.0:82): GStreamer-WARNING **: 18:57:46.322: Failed to load plugin 'audiotestsrc_py.py': audiotestsrc_py.py: invalid ELF header Could not load plugin file: Opening module failed: audiotestsrc_py.py: invalid ELF header
插件代码
import gi gi.require_version('Gst', '1.0') gi.require_version('GstBase', '1.0') gi.require_version('GstAudio', '1.0') from gi.repository import Gst, GLib, GObject, GstBase, GstAudio import numpy as np OCAPS = Gst.Caps.from_string ( 'audio/x-raw, format=F32LE, layout=interleaved, rate=44100, channels=2') SAMPLESPERBUFFER = 1024 DEFAULT_FREQ = 440 DEFAULT_VOLUME = 0.8 DEFAULT_MUTE = False DEFAULT_IS_LIVE = False class AudioTestSrc(GstBase.BaseSrc): __gstmetadata__ = ('CustomSrc','Src', \ 'Custom test src element', 'Mathieu Duponchelle') __gproperties__ = { "freq": (int, "Frequency", "Frequency of test signal", 1, GLib.MAXINT, DEFAULT_FREQ, GObject.ParamFlags.READWRITE ), "volume": (float, "Volume", "Volume of test signal", 0.0, 1.0, DEFAULT_VOLUME, GObject.ParamFlags.READWRITE ), "mute": (bool, "Mute", "Mute the test signal", DEFAULT_MUTE, GObject.ParamFlags.READWRITE ), "is-live": (bool, "Is live", "Whether to act as a live source", DEFAULT_IS_LIVE, GObject.ParamFlags.READWRITE ), } __gsttemplates__ = Gst.PadTemplate.new("src", Gst.PadDirection.SRC, Gst.PadPresence.ALWAYS, OCAPS) def __init__(self): GstBase.BaseSrc.__init__(self) self.info = GstAudio.AudioInfo() self.freq = DEFAULT_FREQ self.volume = DEFAULT_VOLUME self.mute = DEFAULT_MUTE self.set_live(DEFAULT_IS_LIVE) self.set_format(Gst.Format.TIME) def do_set_caps(self, caps): self.info.from_caps(caps) self.set_blocksize(self.info.bpf * SAMPLESPERBUFFER) return True def do_get_property(self, prop): if prop.name == 'freq': return self.freq elif prop.name == 'volume': return self.volume elif prop.name == 'mute': return self.mute elif prop.name == 'is-live': return self.is_live else: raise AttributeError('unknown property %s' % prop.name) def do_set_property(self, prop, value): if prop.name == 'freq': self.freq = value elif prop.name == 'volume': self.volume = value elif prop.name == 'mute': self.mute = value elif prop.name == 'is-live': self.set_live(value) else: raise AttributeError('unknown property %s' % prop.name) def do_start (self): self.next_sample = 0 self.next_byte = 0 self.next_time = 0 self.accumulator = 0 self.generate_samples_per_buffer = SAMPLESPERBUFFER return True def do_gst_base_src_query(self, query): if query.type == Gst.QueryType.LATENCY: latency = Gst.util_uint64_scale_int(self.generate_samples_per_buffer, Gst.SECOND, self.info.rate) is_live = self.is_live query.set_latency(is_live, latency, Gst.CLOCK_TIME_NONE) res = True else: res = GstBase.BaseSrc.do_query(self, query) return res def do_get_times(self, buf): end = 0 start = 0 if self.is_live: ts = buf.pts if ts != Gst.CLOCK_TIME_NONE: duration = buf.duration if duration != Gst.CLOCK_TIME_NONE: end = ts + duration start = ts else: start = Gst.CLOCK_TIME_NONE end = Gst.CLOCK_TIME_NONE return start, end def do_create(self, offset, length): if length == -1: samples = SAMPLESPERBUFFER else: samples = int(length / self.info.bpf) self.generate_samples_per_buffer = samples bytes_ = samples * self.info.bpf next_sample = self.next_sample + samples next_byte = self.next_byte + bytes_ next_time = Gst.util_uint64_scale_int(next_sample, Gst.SECOND, self.info.rate) if not self.mute: r = np.repeat( np.arange(self.accumulator, self.accumulator + samples), self.info.channels) data = ((np.sin(2 * np.pi * r * self.freq / self.info.rate) * self.volume) .astype(np.float32)) else: data = [0] * bytes_ buf = Gst.Buffer.new_wrapped(bytes(data)) buf.offset = self.next_sample buf.offset_end = next_sample buf.pts = self.next_time buf.duration = next_time - self.next_time self.next_time = next_time self.next_sample = next_sample self.next_byte = next_byte self.accumulator += samples self.accumulator %= self.info.rate / self.freq return (Gst.FlowReturn.OK, buf) __gstelementfactory__ = ("audiotestsrc_py", Gst.Rank.NONE, AudioTestSrc)
用户已查阅相关技术资料,但仍无法解决该问题。
解决方案
错误原因
GStreamer默认会将GST_PLUGIN_PATH下的文件当作二进制ELF插件加载,直接传入.py脚本时,它会尝试将文本脚本解析为二进制文件,从而抛出"invalid ELF header"错误。Python GStreamer插件需要通过GStreamer的Python插件加载器识别,而非直接作为插件文件传入。
具体解决步骤
安装Python插件加载器
在Debian/Ubuntu系统中执行以下命令安装:sudo apt install gstreamer1.0-python3-plugin-loader其他系统可通过对应包管理工具安装同类组件(如Fedora为
gstreamer1-python)。正确加载插件的命令
无需指定.py文件,设置环境变量让GStreamer找到Python插件,然后查询代码中定义的工厂名(audiotestsrc_py):GST_PLUGIN_PATH="$PWD" GST_PYTHON_PATH="$PWD" gst-inspect-1.0 audiotestsrc_pyGST_PLUGIN_PATH:指定GStreamer扫描插件的目录GST_PYTHON_PATH:让Python加载器定位你的脚本文件- 最后参数为代码中
__gstelementfactory__定义的工厂名,而非脚本文件名
备选目录规范方式
在当前目录创建python子目录,将脚本放入其中,直接运行:GST_PLUGIN_PATH="$PWD" gst-inspect-1.0 audiotestsrc_pyGStreamer的Python加载器会自动扫描
GST_PLUGIN_PATH下的python子目录,识别其中的Python插件。
内容的提问来源于stack exchange,提问作者Ali Soliman
相关产品推荐
相关产品推荐

