From 55f1909bce3d9a6efb18fc8ac74864534807bb7b Mon Sep 17 00:00:00 2001 From: Antoine Martin Date: Fri, 9 Dec 2016 16:07:39 +0000 Subject: [PATCH] 2.0: drop gstreamer 0.10 support git-svn-id: https://xpra.org/svn/Xpra/trunk@14504 3bb7dfac-3a0b-4e04-842a-767bc560f471 --- src/xpra/platform/win32/paths.py | 13 +-- src/xpra/sound/gstreamer_util.py | 192 +++++++++---------------------- src/xpra/sound/sink.py | 141 +++-------------------- src/xpra/sound/sound_pipeline.py | 20 +--- src/xpra/sound/src.py | 157 +++++++++---------------- 5 files changed, 136 insertions(+), 387 deletions(-) diff --git a/src/xpra/platform/win32/paths.py b/src/xpra/platform/win32/paths.py index b50fdee68e..b167d8032e 100644 --- a/src/xpra/platform/win32/paths.py +++ b/src/xpra/platform/win32/paths.py @@ -129,11 +129,8 @@ def do_get_app_dir(): return default_get_app_dir() def do_get_sound_command(): - from xpra.sound.gstreamer_util import GSTREAMER1 - if GSTREAMER1: - from xpra.platform.paths import get_app_dir - app_dir = get_app_dir() - #is there a python3 bundled sound subdirectory - sound_exe = os.path.join(app_dir, "Sound", "xpra_cmd.exe") - return [sound_exe] - return ["xpra_cmd.exe"] + from xpra.platform.paths import get_app_dir + app_dir = get_app_dir() + #is there a python3 bundled sound subdirectory + sound_exe = os.path.join(app_dir, "Sound", "xpra_cmd.exe") + return [sound_exe] diff --git a/src/xpra/sound/gstreamer_util.py b/src/xpra/sound/gstreamer_util.py index 18e4ce592b..85347a712b 100755 --- a/src/xpra/sound/gstreamer_util.py +++ b/src/xpra/sound/gstreamer_util.py @@ -45,7 +45,6 @@ def get_queue_time(default_value=450, prefix=""): OSX = sys.platform.startswith("darwin") ALLOW_SOUND_LOOP = envbool("XPRA_ALLOW_SOUND_LOOP", False) -GSTREAMER1 = envbool("XPRA_GSTREAMER1", True) PULSEAUDIO_DEVICE_NAME = os.environ.get("XPRA_PULSEAUDIO_DEVICE_NAME", "") USE_DEFAULT_DEVICE = envbool("XPRA_USE_DEFAULT_DEVICE", True) def force_enabled(codec_name): @@ -96,7 +95,6 @@ def force_enabled(codec_name): (FLAC , "flacenc", None, "flacparse ! flacdec", None), #this only works in gstreamer 0.10 and is filtered out during initialization: (FLAC_OGG , "flacenc", "oggmux", "flacparse ! flacdec", "oggdemux"), - (MP3 , "lamemp3enc", None, "mp3parse ! mad", None), (MP3 , "lamemp3enc", None, "mpegaudioparse ! mad", None), (MP3_MPEG4 , "lamemp3enc", "mp4mux", "mp3parse ! mad", "qtdemux"), (MP3_MPEG4 , "lamemp3enc", "mp4mux", "mpegaudioparse ! mad", "qtdemux"), @@ -157,20 +155,12 @@ def force_enabled(codec_name): #"vorbisenc" : {"perfect-timestamp" : 1}, } ENCODER_DEFAULT_OPTIONS = { - 0 : { - "opusenc" : { - "cbr" : 0, - "complexity" : 0 - }, - }, - 1 : { - #FIXME: figure out when it is safe to apply the "bitrate-type" setting: - "opusenc" : { - #only available with 1.6 onwards? - #"bitrate-type" : 2, #constrained vbr - "complexity" : 0 - }, - }, + #FIXME: figure out when it is safe to apply the "bitrate-type" setting: + "opusenc" : { + #only available with 1.6 onwards? + #"bitrate-type" : 2, #constrained vbr + "complexity" : 0 + }, } #we may want to review this if/when we implement UDP transport: MUXER_DEFAULT_OPTIONS = { @@ -209,7 +199,6 @@ def force_enabled(codec_name): gst = None has_gst = None -gst_major_version = None gst_vinfo = None pygst_version = "" @@ -255,120 +244,54 @@ def new_buffer(data): pygst_version = Gst.get_pygst_version() return Gst -def import_gst0_10(): - log("import_gst0_10()") - global gst_version, pygst_version - import pygst - log("import_gst0_10() pygst=%s", pygst) - pygst.require("0.10") - #initializing gstreamer parses sys.argv - #which interferes with our own command line arguments - #so we temporarily hide it, - #also import with stderr redirection in place - #to avoid gobject warnings: - from xpra.os_util import HideStdErr, HideSysArgv - with HideStdErr(): - with HideSysArgv(): - import gst - gst_version = gst.gst_version - pygst_version = gst.pygst_version - gst.new_buffer = gst.Buffer - if not hasattr(gst, 'MESSAGE_STREAM_START'): - #a None value is better than nothing: - #(our code can assume it exists - just never matches) - gst.MESSAGE_STREAM_START = None - return gst - - -def get_version_str(version): - if version==1: - return "1.0" - else: - return "0.10" - def import_gst(): - global gst, has_gst, gst_vinfo, gst_major_version + global gst, has_gst, gst_vinfo if has_gst is not None: return gst - PYTHON3 = sys.version_info[0]>=3 - if PYTHON3: - imports = [ (import_gst1, 1) ] - elif GSTREAMER1: - imports = [ - (import_gst1, 1), - (import_gst0_10, 0), - ] - else: - imports = [ - (import_gst0_10, 0), - (import_gst1, 1), - ] - errs = {} - saved_sys_path = sys.path[:] - saved_os_environ = os.environ.copy() - for import_function, MV in imports: - #restore os.environ and sys.path - sys.path = saved_sys_path[:] - os.environ.clear() - os.environ.update(saved_os_environ) - vstr = get_version_str(MV) - #hacks to locate gstreamer plugins on win32 and osx: - if WIN32: - frozen = hasattr(sys, "frozen") and sys.frozen in ("windows_exe", "console_exe", True) - log("gstreamer_util: frozen=%s", frozen) - if frozen: - #on win32, we keep separate trees - #because GStreamer 0.10 and 1.x were built using different and / or incompatible version of the same libraries: - from xpra.platform.paths import get_app_dir - gst_dir = os.path.join(get_app_dir(), "gstreamer-%s" % vstr) #ie: C:\Program Files\Xpra\gstreamer-0.10 - os.environ["GST_PLUGIN_PATH"] = gst_dir - if MV==1: - gst_bin_dir = os.path.join(gst_dir, "bin") #ie: C:\Program Files\Xpra\gstreamer-0.10\bin - os.environ["PATH"] = os.pathsep.join(x for x in (gst_bin_dir, os.environ.get("PATH", "")) if x) - sys.path.insert(0, gst_bin_dir) - scanner = os.path.join(gst_bin_dir, "gst-plugin-scanner.exe") - if os.path.exists(scanner): - os.environ["GST_PLUGIN_SCANNER"] = scanner - gi_dir = os.path.join(get_app_dir(), "girepository-%s" % vstr) - os.environ["GI_TYPELIB_PATH"] = gi_dir - elif OSX: - bundle_contents = os.environ.get("GST_BUNDLE_CONTENTS") - log("OSX: GST_BUNDLE_CONTENTS=%s", bundle_contents) - if bundle_contents: - os.environ["GST_PLUGIN_PATH"] = os.path.join(bundle_contents, "Resources", "lib", "gstreamer-%s" % vstr) - os.environ["GST_PLUGIN_SCANNER"] = os.path.join(bundle_contents, "Resources", "bin", "gst-plugin-scanner-%s" % vstr) - if MV==1: - gi_dir = os.path.join(bundle_contents, "Resources", "lib", "girepository-%s" % vstr) - os.environ["GI_TYPELIB_PATH"] = gi_dir - if MV<1: - #we should not be loading the gi bindings - try: - del os.environ["GI_TYPELIB_PATH"] - except: - pass - log("GStreamer %s environment: %s", vstr, dict((k,v) for k,v in os.environ.items() if (k.startswith("GST") or k.startswith("GI") or k=="PATH"))) - log("GStreamer %s sys.path=%s", vstr, csv(sys.path)) + #hacks to locate gstreamer plugins on win32 and osx: + if WIN32: + frozen = hasattr(sys, "frozen") and sys.frozen in ("windows_exe", "console_exe", True) + log("gstreamer_util: frozen=%s", frozen) + if frozen: + #on win32, we keep separate trees + #because GStreamer 0.10 and 1.x were built using different and / or incompatible version of the same libraries: + from xpra.platform.paths import get_app_dir + gst_dir = os.path.join(get_app_dir(), "gstreamer-1.0") #ie: C:\Program Files\Xpra\gstreamer-0.10 + os.environ["GST_PLUGIN_PATH"] = gst_dir + if MV==1: + gst_bin_dir = os.path.join(gst_dir, "bin") #ie: C:\Program Files\Xpra\gstreamer-0.10\bin + os.environ["PATH"] = os.pathsep.join(x for x in (gst_bin_dir, os.environ.get("PATH", "")) if x) + sys.path.insert(0, gst_bin_dir) + scanner = os.path.join(gst_bin_dir, "gst-plugin-scanner.exe") + if os.path.exists(scanner): + os.environ["GST_PLUGIN_SCANNER"] = scanner + gi_dir = os.path.join(get_app_dir(), "girepository-1.0") + os.environ["GI_TYPELIB_PATH"] = gi_dir + elif OSX: + bundle_contents = os.environ.get("GST_BUNDLE_CONTENTS") + log("OSX: GST_BUNDLE_CONTENTS=%s", bundle_contents) + if bundle_contents: + os.environ["GST_PLUGIN_PATH"] = os.path.join(bundle_contents, "Resources", "lib", "gstreamer-1.0") + os.environ["GST_PLUGIN_SCANNER"] = os.path.join(bundle_contents, "Resources", "bin", "gst-plugin-scanner-1.0") + if MV==1: + gi_dir = os.path.join(bundle_contents, "Resources", "lib", "girepository-1.0") + os.environ["GI_TYPELIB_PATH"] = gi_dir + log("GStreamer 1.x environment: %s", dict((k,v) for k,v in os.environ.items() if (k.startswith("GST") or k.startswith("GI") or k=="PATH"))) + log("GStreamer 1.x sys.path=%s", csv(sys.path)) - try: - log("trying to import GStreamer %s using %s", get_version_str(MV), import_function) - _gst = import_function() - v = _gst.version() - if v[-1]==0: - v = v[:-1] - gst_vinfo = ".".join((str(x) for x in v)) - gst_major_version = MV - gst = _gst - break - except Exception as e: - log("Warning failed to import GStreamer %s", vstr, exc_info=True) - errs[vstr] = e - if gst: - log("Python GStreamer version %s for Python %s.%s", gst_vinfo, sys.version_info[0], sys.version_info[1]) - else: - log.warn("Warning: failed to import GStreamer:") - for vstr,e in errs.items(): - log.warn(" GStreamer %s: %s", vstr, e) + try: + _gst = import_gst1() + v = _gst.version() + if v[-1]==0: + v = v[:-1] + gst_vinfo = ".".join((str(x) for x in v)) + gst = _gst + except Exception as e: + log("Warning failed to import GStreamer 1.x", exc_info=True) + log.warn("Warning: failed to import GStreamer 1.x:") + log.warn(" %s", e) + return None has_gst = gst is not None return gst @@ -413,11 +336,11 @@ def has_plugins(*names): return len(missing)==0 def get_encoder_default_options(encoder): - global gst_major_version, ENCODER_DEFAULT_OPTIONS_COMMON, ENCODER_DEFAULT_OPTIONS + global ENCODER_DEFAULT_OPTIONS_COMMON, ENCODER_DEFAULT_OPTIONS #strip the muxer: enc = encoder.split("+")[0] options = ENCODER_DEFAULT_OPTIONS_COMMON.get(enc, {}).copy() - options.update(ENCODER_DEFAULT_OPTIONS.get(gst_major_version).get(enc, {})) + options.update(ENCODER_DEFAULT_OPTIONS.get(enc, {})) return options @@ -436,7 +359,7 @@ def get_decoders(): return DECODERS def init_codecs(): - global CODECS, ENCODERS, DECODERS, gst_major_version + global CODECS, ENCODERS, DECODERS if CODECS is not None or not has_gst: return CODECS or {} #populate CODECS: @@ -496,21 +419,10 @@ def validate_encoding(elements): #and we have no way of knowing what version they have at this point, so just disable those: log("avoiding %s with gdp muxer - gstreamer version %s is too old", encoding, get_gst_version()) return False - elif encoding.startswith(FLAC): - #flac problems: - if WIN32 and gst_major_version==0: - #the gstreamer 0.10 builds on win32 use the outdated oss build, - #which includes outdated flac libraries with known CVEs, - #so avoid using those: - log("avoiding outdated flac module (likely buggy on win32 with gstreamer 0.10)") - return False elif WIN32 and encoding in (SPEEX_OGG, ): log("skipping %s on win32", encoding) return False elif encoding.startswith(OPUS): - if gst_major_version<1: - log("skipping %s with GStreamer 0.10", encoding) - return False if encoding==OPUS_MKA and get_gst_version()<(1, 8): #this causes "could not link opusenc0 to webmmux0" #(not sure which versions are affected, but 1.8.x is not) diff --git a/src/xpra/sound/sink.py b/src/xpra/sound/sink.py index 9e777e20c4..648f07d8a0 100755 --- a/src/xpra/sound/sink.py +++ b/src/xpra/sound/sink.py @@ -30,12 +30,8 @@ "qos" : True } -SINK_DEFAULT_ATTRIBUTES = {0 : { - "pulsesink" : {"client" : "Xpra"} - }, - 1 : { - "pulsesink" : {"client-name" : "Xpra"} - }, +SINK_DEFAULT_ATTRIBUTES = { + "pulsesink" : {"client-name" : "Xpra"}, } QUEUE_SILENT = envbool("XPRA_QUEUE_SILENT", False) @@ -123,11 +119,10 @@ def __init__(self, sink_type=None, sink_options={}, codecs=get_decoders(), codec "max-size-time=%s" % QUEUE_TIME, "leaky=%s" % QUEUE_LEAK])) sink_attributes = SINK_SHARED_DEFAULT_ATTRIBUTES.copy() - from xpra.sound.gstreamer_util import gst_major_version, get_gst_version + from xpra.sound.gstreamer_util import get_gst_version #anything older than this may cause problems (ie: centos 6.x) #because the attributes may not exist - if get_gst_version()>=(0, 10, 36): - sink_attributes.update(SINK_DEFAULT_ATTRIBUTES.get(gst_major_version, {}).get(sink_type, {})) + sink_attributes.update(SINK_DEFAULT_ATTRIBUTES.get(sink_type, {})) get_options_cb = DEFAULT_SINK_PLUGIN_OPTIONS.get(sink_type.replace("sink", "")) if get_options_cb: v = get_options_cb() @@ -141,29 +136,14 @@ def __init__(self, sink_type=None, sink_options={}, codecs=get_decoders(), codec self.volume = self.pipeline.get_by_name("volume") self.src = self.pipeline.get_by_name("src") self.queue = self.pipeline.get_by_name("queue") - if get_gst_version()<(1, ): - self.add_data = self.add_data0 - else: - self.add_data = self.add_data1 if self.queue: - if not QUEUE_SILENT: - if get_gst_version()<(1, ): - self.queue.connect("overrun", self.queue_overrun0) - self.queue.connect("underrun", self.queue_underrun0) - self.queue.connect("running", self.queue_running0) - self.queue.connect("pushing", self.queue_pushing0) - else: - self.queue.connect("overrun", self.queue_overrun1) - self.queue.connect("underrun", self.queue_underrun1) - self.queue.connect("running", self.queue_running1) - self.queue.connect("pushing", self.queue_pushing1) + if QUEUE_SILENT: + self.queue.set_property("silent", False) else: - #older versions may not have the "silent" attribute, - #in which case we will emit the signals for nothing - try: - self.queue.set_property("silent", False) - except Exception as e: - log("cannot silence the queue %s: %s", self.queue, e) + self.queue.connect("overrun", self.queue_overrun) + self.queue.connect("underrun", self.queue_underrun) + self.queue.connect("running", self.queue_running) + self.queue.connect("pushing", self.queue_pushing) def __repr__(self): return "SoundSink('%s' - %s)" % (self.pipeline_str, self.state) @@ -203,27 +183,12 @@ def adjust_volume(self): return True - def _queue_pushing(self, *args): - self.queue_state = "pushing" - self.emit_info() - return True - - def queue_pushing0(self, *args): - gstlog("queue_pushing0") - return self._queue_pushing() - - def queue_pushing1(self, *args): - gstlog("queue_pushing1") - return self._queue_pushing() - - - def queue_running0(self, *args): - gstlog("queue_running") - self.queue_state = "running" + def queue_pushing(self, *args): + gstlog("queue_pushing") self.emit_info() return True - def queue_running1(self, *args): + def queue_running(self, *args): gstlog("queue_running") self.queue_state = "running" self.set_min_level() @@ -231,39 +196,12 @@ def queue_running1(self, *args): self.emit_info() return True - def queue_underrun0(self, *args): - now = time.time() - gstlog("queue_underrun0") - self.queue_state = "underrun" - self.last_underrun = now - clt = self.queue.get_property("current-level-time")//MS_TO_NS - mintt = self.queue.get_property("min-threshold-time")//MS_TO_NS - gstlog("underrun: clt=%s mintt=%s state=%s", clt, mintt, self.state) - if clt==0 and mintt==0 and self.state in ("running", "active"): - if self.last_data: - self.add_data(self.last_data) - #this is going to cause scratchy sound, - #temporarily lower the volume: - def fadeout(): - gstlog("fadeout") - self.target_volume = 0.0 - self.start_adjust_volume(1) - def fadein(): - gstlog("fadein") - self.target_volume = self.normal_volume - self.start_adjust_volume(10) - fadeout() - glib.timeout_add(300, fadein) - return 1 - self.emit_info() - return 1 - - def queue_underrun1(self, *args): + def queue_underrun(self, *args): now = time.time() if self.queue_state=="starting" or 1000*(now-self.start_time)2: self.last_underrun = now @@ -281,16 +219,7 @@ def get_level_range(self, mintime=2, maxtime=10): return maxl-minl return 0 - def queue_overrun0(self, *args): - clt = self.queue.get_property("current-level-time")//MS_TO_NS - log("queue_overrun0 level=%ims", clt) - now = time.time() - self.last_overrun = now - self.overrun_events.append(now) - self.overruns += 1 - return 1 - - def queue_overrun1(self, *args): + def queue_overrun(self, *args): now = time.time() if self.queue_state=="starting" or 1000*(now-self.start_time)0: - #with gstreamer 1.x, we don't always get the "audio-codec" message.. - #so print the codec from here instead (and assume gstreamer is using what we told it to) - #after a delay, just in case we do get the real "audio-codec" message! - self.timeout_add(500, self.new_codec_description, self.codec.split("+")[0]) + #with gstreamer 1.x, we don't always get the "audio-codec" message.. + #so print the codec from here instead (and assume gstreamer is using what we told it to) + #after a delay, just in case we do get the real "audio-codec" message! + self.timeout_add(500, self.new_codec_description, self.codec.split("+")[0]) elif t in (gst.MESSAGE_ASYNC_DONE, gst.MESSAGE_NEW_CLOCK): gstlog("%s", message) elif t == gst.MESSAGE_STATE_CHANGED: @@ -324,14 +323,7 @@ def on_message(self, bus, message): self.update_state(state) self.idle_emit("state-changed", state) elif t == gst.MESSAGE_DURATION: - if gst_major_version==0: - d = message.parse_duration() - try: - v = d[1] - if v>0: - gstlog("duration changed: %s", v) - except: - gstlog("duration changed: %s", d) + gstlog("duration changed: %s", message) elif t == gst.MESSAGE_LATENCY: gstlog("latency message from %s: %s", message.src, message) elif t == gst.MESSAGE_INFO: diff --git a/src/xpra/sound/src.py b/src/xpra/sound/src.py index 3aae974e80..9fa42333c3 100755 --- a/src/xpra/sound/src.py +++ b/src/xpra/sound/src.py @@ -115,52 +115,36 @@ def __init__(self, src_type=None, src_options={}, codecs=get_encoders(), codec_o self.queue.set_property("silent", True) except Exception as e: log("cannot make queue silent: %s", e) - try: - if get_gst_version()<(1,0): - self.sink.set_property("enable-last-buffer", False) - else: - self.sink.set_property("enable-last-sample", False) - except Exception as e: - log("failed to disable last buffer: %s", e) + self.sink.set_property("enable-last-sample", False) self.skipped_caps = set() if JITTER>0: self.jitter_queue = Queue() - try: - #Gst 1.0: - self.sink.connect("new-sample", self.on_new_sample) - self.sink.connect("new-preroll", self.on_new_preroll1) - except: - #Gst 0.10: - self.sink.connect("new-buffer", self.on_new_buffer) - self.sink.connect("new-preroll", self.on_new_preroll0) + #Gst 1.0: + self.sink.connect("new-sample", self.on_new_sample) + self.sink.connect("new-preroll", self.on_new_preroll) self.src = self.pipeline.get_by_name("src") - try: - for x in ("actual-buffer-time", "actual-latency-time"): - #don't comment this out, it is used to verify the attributes are present: - gstlog("initial %s: %s", x, self.src.get_property(x)) - self.buffer_latency = True - except Exception as e: - log.info("source %s does not support 'buffer-time' or 'latency-time':", self.src_type) - log.info(" %s", e) - else: - #if the env vars have been set, try to honour the settings: - global BUFFER_TIME, LATENCY_TIME - if BUFFER_TIME>0: - if BUFFER_TIME=0: - self.src.set_property(attr, v*1000) - gstlog("overriding with: %s=%i", attr, v) - except Exception as e: - log.warn("source %s does not support '%s': %s", self.src_type, attr, e) - settime("buffer-time", BUFFER_TIME) - settime("latency-time", LATENCY_TIME) + for x in ("actual-buffer-time", "actual-latency-time"): + #don't comment this out, it is used to verify the attributes are present: + gstlog("initial %s: %s", x, self.src.get_property(x)) + self.buffer_latency = True + #if the env vars have been set, try to honour the settings: + global BUFFER_TIME, LATENCY_TIME + if BUFFER_TIME>0: + if BUFFER_TIME=0: + self.src.set_property(attr, v*1000) + gstlog("overriding with: %s=%i", attr, v) + except Exception as e: + log.warn("source %s does not support '%s': %s", self.src_type, attr, e) + settime("buffer-time", BUFFER_TIME) + settime("latency-time", LATENCY_TIME) gen = generation.increase() if SAVE_TO_FILE is not None: parts = codec.split("+") @@ -197,17 +181,17 @@ def get_info(self): return info - def on_new_preroll1(self, appsink): + def on_new_preroll(self, appsink): sample = appsink.emit('pull-preroll') - gstlog('new preroll1: %s', sample) - return self.emit_buffer1(sample) + gstlog('new preroll: %s', sample) + return self.emit_buffer(sample) def on_new_sample(self, bus): #Gst 1.0 sample = self.sink.emit("pull-sample") - return self.emit_buffer1(sample) + return self.emit_buffer(sample) - def emit_buffer1(self, sample): + def emit_buffer(self, sample): buf = sample.get_buffer() #info = sample.get_info() size = buf.get_size() @@ -224,65 +208,12 @@ def emit_buffer1(self, sample): if pts==-1 and duration==-1 and BUNDLE_METADATA and len(self.pending_metadata)<10: self.pending_metadata.append(data) return 0 - return self.emit_buffer(data, { + return self._emit_buffer(data, { "timestamp" : pts, "duration" : duration, }) - - def on_new_preroll0(self, appsink): - buf = appsink.emit('pull-preroll') - gstlog('new preroll0: %s bytes', len(buf)) - return self.emit_buffer0(buf) - - def on_new_buffer(self, bus): - #pygst 0.10 - buf = self.sink.emit("pull-buffer") - return self.emit_buffer0(buf) - - - def caps_to_dict(self, caps): - if not caps: - return {} - d = {} - try: - for cap in caps: - name = cap.get_name() - capd = {} - for k in cap.keys(): - v = cap[k] - if type(v) in (str, int): - capd[k] = cap[k] - elif k not in self.skipped_caps: - log("skipping %s cap key %s=%s of type %s", name, k, v, type(v)) - d[name] = capd - except Exception as e: - log.error("Error parsing '%s':", caps) - log.error(" %s", e) - return d - - def emit_buffer0(self, buf): - """ convert pygst structure into something more generic for the wire """ - #none of the metadata is really needed at present, but it may be in the future: - #metadata = {"caps" : buf.get_caps().to_string(), - # "size" : buf.size, - # "timestamp" : buf.timestamp, - # "duration" : buf.duration, - # "offset" : buf.offset, - # "offset_end": buf.offset_end} - log("emit buffer: %s bytes, timestamp=%s", len(buf.data), buf.timestamp//MS_TO_NS) - metadata = { - "timestamp" : normv(buf.timestamp), - "duration" : normv(buf.duration) - } - d = self.caps_to_dict(buf.get_caps()) - if not self.caps or self.caps!=d: - self.caps = d - self.info["caps"] = self.caps - metadata["caps"] = self.caps - return self.emit_buffer(buf.data, metadata) - - def emit_buffer(self, data, metadata={}): + def _emit_buffer(self, data, metadata={}): if self.stream_compressor and data: data = compressed_wrapper("sound", data, level=9, zlib=False, lz4=(self.stream_compressor=="lz4"), lzo=(self.stream_compressor=="lzo"), can_inline=True) #log("compressed using %s from %i bytes down to %i bytes", self.stream_compressor, len(odata), len(data)) @@ -313,6 +244,28 @@ def emit_buffer(self, data, metadata={}): log("emit_buffer data=%s, len=%i, metadata=%s", type(data), len(data), metadata) return self.do_emit_buffer(data, metadata) + + def caps_to_dict(self, caps): + if not caps: + return {} + d = {} + try: + for cap in caps: + name = cap.get_name() + capd = {} + for k in cap.keys(): + v = cap[k] + if type(v) in (str, int): + capd[k] = cap[k] + elif k not in self.skipped_caps: + log("skipping %s cap key %s=%s of type %s", name, k, v, type(v)) + d[name] = capd + except Exception as e: + log.error("Error parsing '%s':", caps) + log.error(" %s", e) + return d + + def flush_jitter_queue(self): while not self.jitter_queue.empty(): d,m = self.jitter_queue.get(False)