Skip to content

Commit

Permalink
add stream compression options: lzo and lz4
Browse files Browse the repository at this point in the history 
git-svn-id: https://xpra.org/svn/Xpra/trunk@12704 3bb7dfac-3a0b-4e04-842a-767bc560f471
  • Loading branch information
@totaam
totaam committed May 27, 2016
1 parent 4f52b38 commit 8d17597
Show file tree
Hide file tree
Showing 4 changed files with 92 additions and 33 deletions.
85 changes: 57 additions & 28 deletions src/xpra/sound/gstreamer_util.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,10 @@ def force_enabled(codec_name):
#RTP = "rtp"
RAW = "raw"

#stream compression
LZ4 = "lz4"
LZO = "lzo"

FLAC_GDP = FLAC+"+"+GDP
OPUS_GDP = OPUS+"+"+GDP
SPEEX_GDP = SPEEX+"+"+GDP
Expand All @@ -105,41 +109,49 @@ def force_enabled(codec_name):
VORBIS_MKA = VORBIS+"+"+MKA
AAC_GDP = AAC+"+"+GDP
AAC_MPEG4 = AAC+"+"+MPEG4
RAW_GDP = RAW+"+"+GDP
RAW_GDP = RAW+"+"+GDP
WAV_LZ4 = WAV+"+"+LZ4
WAV_LZO = WAV+"+"+LZO
RAW_GDP_LZ4 = RAW+"+"+GDP+"+"+LZ4
RAW_GDP_LZO = RAW+"+"+GDP+"+"+LZO


#format: encoder, container-formatter, decoder, container-parser
#we keep multiple options here for the same encoding
#and will populate the ones that are actually available into the "CODECS" dict
CODEC_OPTIONS = [
(VORBIS , "vorbisenc", "gdppay", "vorbisdec", "gdpdepay"),
(VORBIS_MKA , "vorbisenc", "webmmux", "vorbisdec", "matroskademux"),
(VORBIS , "vorbisenc", "gdppay", "vorbisdec", "gdpdepay", None),
(VORBIS_MKA , "vorbisenc", "webmmux", "vorbisdec", "matroskademux", None),
#fails silently - no idea why:
#(VORBIS_OGG , "vorbisenc", "oggmux", "vorbisparse ! vorbisdec", "oggdemux"),
#does not work - no idea why:
#(FLAC , "flacenc", "oggmux", "flacparse ! flacdec", "oggdemux"),
#this only works in gstreamer 0.10 and is filtered out during initialization:
(FLAC , "flacenc", "oggmux", "flacdec", "oggdemux"),
(FLAC_GDP , "flacenc", "gdppay", "flacparse ! flacdec", "gdpdepay"),
(MP3 , "lamemp3enc", None, "mp3parse ! mad", None),
(MP3 , "lamemp3enc", None, "mpegaudioparse ! mad", None),
(WAV , "wavenc", None, "wavparse", None),
(OPUS , "opusenc", "oggmux", "opusdec", "oggdemux"),
(OPUS_GDP , "opusenc", "gdppay", "opusdec", "gdpdepay"),
(FLAC , "flacenc", "oggmux", "flacdec", "oggdemux", None),
(FLAC_GDP , "flacenc", "gdppay", "flacparse ! flacdec", "gdpdepay", None),
(MP3 , "lamemp3enc", None, "mp3parse ! mad", None, None),
(MP3 , "lamemp3enc", None, "mpegaudioparse ! mad", None, None),
(WAV , "wavenc", None, "wavparse", None, None),
(WAV_LZ4 , "wavenc", None, "wavparse", None, "lz4"),
(WAV_LZO , "wavenc", None, "wavparse", None, "lzo"),
(OPUS , "opusenc", "oggmux", "opusdec", "oggdemux", None),
(OPUS_GDP , "opusenc", "gdppay", "opusdec", "gdpdepay", None),
#for rtp, we would need to send the caps:
#(OPUS_RTP , "opusenc", "rtpopuspay", "opusdec", "rtpopusdepay"),
#(OPUS_RTP , "opusenc", "rtpopuspay", "opusparse ! opusdec", "rtpopusdepay"),
#this causes "could not link opusenc0 to webmmux0"
#(OPUS_WEBM , "opusenc", "webmmux", "opusdec", "matroskademux"),
#(OPUS_WEBM , "opusenc", "webmmux", "opusparse ! opusdec", "matroskademux"),
(SPEEX , "speexenc", "oggmux", "speexdec", "oggdemux"),
(SPEEX_GDP , "speexenc", "gdppay", "speexdec", "gdpdepay"),
(WAVPACK , "wavpackenc", None, "wavpackparse ! wavpackdec", None),
(AAC_GDP , "faac", "gdppay", "faad", "gdpdepay"),
(AAC_GDP , "avenc_aac", "gdppay", "avdec_aac", "gdpdepay"),
(AAC_MPEG4 , "faac", "mp4mux", "faad", "qtdemux"),
(AAC_MPEG4 , "avenc_aac", "mp4mux", "avdec_aac", "qtdemux"),
(RAW_GDP , None, "gdppay", None, "gdpdepay"),
(SPEEX , "speexenc", "oggmux", "speexdec", "oggdemux", None),
(SPEEX_GDP , "speexenc", "gdppay", "speexdec", "gdpdepay", None),
(WAVPACK , "wavpackenc", None, "wavpackparse ! wavpackdec", None, None),
(AAC_GDP , "faac", "gdppay", "faad", "gdpdepay", None),
(AAC_GDP , "avenc_aac", "gdppay", "avdec_aac", "gdpdepay", None),
(AAC_MPEG4 , "faac", "mp4mux", "faad", "qtdemux", None),
(AAC_MPEG4 , "avenc_aac", "mp4mux", "avdec_aac", "qtdemux", None),
(RAW_GDP , None, "gdppay", None, "gdpdepay", None),
(RAW_GDP_LZ4, None, "gdppay", None, "gdpdepay", "lz4"),
(RAW_GDP_LZO, None, "gdppay", None, "gdpdepay", "lzo"),
]

MUX_OPTIONS = [
Expand Down Expand Up @@ -225,7 +237,7 @@ def force_enabled(codec_name):
SPEEX_GDP : 0,
}

CODEC_ORDER = [OPUS_GDP, OPUS, VORBIS, VORBIS_MKA, FLAC_GDP, FLAC, MP3, AAC_GDP, AAC_MPEG4, RAW_GDP, WAV, WAVPACK, SPEEX_GDP, SPEEX]
CODEC_ORDER = [OPUS_GDP, OPUS, VORBIS, VORBIS_MKA, FLAC_GDP, FLAC, MP3, AAC_GDP, AAC_MPEG4, RAW_GDP_LZ4, RAW_GDP_LZO, RAW_GDP, WAV_LZ4, WAV_LZO, WAV, WAVPACK, SPEEX_GDP, SPEEX]


gst = None
Expand Down Expand Up @@ -484,15 +496,26 @@ def get_codecs():
log("skipping opus with GStreamer 0.10")
continue
#verify we have all the elements needed:
if has_plugins(*elements[1:]):
#ie: FLAC, "flacenc", "oggmux", "flacdec", "oggdemux" = elements
encoding, encoder, muxer, decoder, demuxer = elements
CODECS[encoding] = (encoder, muxer, decoder, demuxer)
#ie: FLAC, "flacenc", "oggmux", "flacdec", "oggdemux", None = elements
try:
encoder, muxer, decoder, demuxer, stream_comp = elements[1:]
except ValueError as e:
log.error("Error: invalid codec entry: %s", e)
log.error(" %s", elements)
continue
if stream_comp:
assert stream_comp in ("lz4", "lzo")
from xpra.net.compression import use_lz4
if not use_lz4:
log("skipping %s: missing lz4", encoding)
continue
if has_plugins(encoder, muxer, decoder, demuxer):
CODECS[encoding] = (encoder, muxer, decoder, demuxer, stream_comp)
log("initialized sound codecs:")
for k in [x for x in CODEC_ORDER if x in CODECS]:
def ci(v):
return "%-22s" % (v or "")
log("* %-10s : %s", k, csv([ci(v) for v in CODECS[k]]))
log("* %-12s : %s", k, csv([ci(v) for v in CODECS[k]]))
return CODECS

def get_muxers():
Expand All @@ -510,18 +533,24 @@ def get_demuxers():
return demuxers


def get_stream_compressor(name):
codecs = get_codecs()
assert name in codecs, "invalid codec: %s (should be one of: %s)" % (name, codecs.keys())
_, _, _, _, stream_comp = codecs.get(name)
return stream_comp

def get_encoder_formatter(name):
codecs = get_codecs()
assert name in codecs, "invalid codec: %s (should be one of: %s)" % (name, codecs.keys())
encoder, formatter, _, _ = codecs.get(name)
encoder, formatter, _, _, _ = codecs.get(name)
assert encoder is None or has_plugins(encoder), "encoder %s not found" % encoder
assert formatter is None or has_plugins(formatter), "formatter %s not found" % formatter
return encoder, formatter

def get_decoder_parser(name):
codecs = get_codecs()
assert name in codecs, "invalid codec: %s (should be one of: %s)" % (name, codecs.keys())
_, _, decoder, parser = codecs.get(name)
_, _, decoder, parser, _ = codecs.get(name)
assert decoder is None or has_plugins(decoder), "decoder %s not found" % decoder
assert parser is None or has_plugins(parser), "parser %s not found" % parser
return decoder, parser
Expand All @@ -530,14 +559,14 @@ def has_encoder(name):
codecs = get_codecs()
if name not in codecs:
return False
encoder, fmt, _, _ = codecs.get(name)
encoder, fmt, _, _, _ = codecs.get(name)
return has_plugins(encoder, fmt)

def has_decoder(name):
codecs = get_codecs()
if name not in codecs:
return False
_, _, decoder, parser = codecs.get(name)
_, _, decoder, parser, _ = codecs.get(name)
return has_plugins(decoder, parser)

def has_codec(name):
Expand Down
19 changes: 18 additions & 1 deletion src/xpra/sound/sink.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,10 @@

from xpra.sound.sound_pipeline import SoundPipeline
from xpra.gtk_common.gobject_util import one_arg_signal, gobject
from xpra.sound.gstreamer_util import plugin_str, get_decoder_parser, get_queue_time, normv, get_codecs, get_default_sink, get_sink_plugins, \
from xpra.sound.gstreamer_util import plugin_str, get_decoder_parser, get_stream_compressor, get_queue_time, normv, get_codecs, get_default_sink, get_sink_plugins, \
MP3, CODEC_ORDER, gst, QUEUE_LEAK, GST_QUEUE_NO_LEAK, MS_TO_NS, DEFAULT_SINK_PLUGIN_OPTIONS
from xpra.gtk_common.gobject_compat import import_glib
from xpra.net.compression import decompress_by_name

from xpra.scripts.config import InitExit
from xpra.util import csv
Expand Down Expand Up @@ -80,6 +81,7 @@ def __init__(self, sink_type=None, sink_options={}, codecs=get_codecs(), codec_o
decoder, parser = get_decoder_parser(codec)
SoundPipeline.__init__(self, codec)
self.container_format = (parser or "").replace("demux", "").replace("depay", "")
self.stream_compressor = get_stream_compressor(codec)
self.sink_type = sink_type
self.levels = deque(maxlen=100)
self.volume = None
Expand Down Expand Up @@ -411,10 +413,24 @@ def can_push_buffer(self):
return False
return True


def uncompress_data(self, data, metadata):
if not data or not metadata:
return data
compress = metadata.get("compress")
if not compress:
return data
assert compress in ("lz4", "lzo")
v = decompress_by_name(data, compress)
#log("decompressed %s data: %i bytes into %i bytes", compress, len(data), len(v))
return v


def add_data0(self, data, metadata=None, packet_metadata=()):
if not self.can_push_buffer():
return
self.last_data = data
data = self.uncompress_data(data, metadata)
now = time.time()
clt = self.queue.get_property("current-level-time")//MS_TO_NS
delta = QUEUE_TIME//MS_TO_NS-clt
Expand Down Expand Up @@ -447,6 +463,7 @@ def fadein():
def add_data1(self, data, metadata=None, packet_metadata=()):
if not self.can_push_buffer():
return
data = self.uncompress_data(data, metadata)
for x in packet_metadata:
self.do_add_data(x)
if self.do_add_data(data, metadata):
Expand Down
9 changes: 8 additions & 1 deletion src/xpra/sound/sound_pipeline.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ class SoundPipeline(gobject.GObject):

def __init__(self, codec):
gobject.GObject.__init__(self)
self.stream_compressor = None
self.codec = codec
self.codec_description = ""
self.codec_mode = ""
Expand Down Expand Up @@ -169,14 +170,20 @@ def start(self):
self.idle_emit("new-stream", self.codec)
self.update_state("active")
self.pipeline.set_state(gst.STATE_PLAYING)
if self.stream_compressor:
self.info["stream-compressor"] = self.stream_compressor
self.emit_info()
#we may never get the stream start, synthesize codec event so we get logging:
parts = self.codec.split("+")
self.timeout_add(1000, self.new_codec_description, parts[0])
if len(parts)>1:
if len(parts)>1 and parts[1]!=self.stream_compressor:
self.timeout_add(1000, self.new_container_description, parts[1])
elif self.container_format:
self.timeout_add(1000, self.new_container_description, self.container_format)
if self.stream_compressor:
def logsc():
self.gstloginfo("using stream compression %s", self.stream_compressor)
self.timeout_add(1000, logsc)
log("SoundPipeline.start() done")

def stop(self):
Expand Down
12 changes: 9 additions & 3 deletions src/xpra/sound/src.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,9 @@
from xpra.util import csv, AtomicInteger
from xpra.sound.sound_pipeline import SoundPipeline
from xpra.gtk_common.gobject_util import n_arg_signal, gobject
from xpra.sound.gstreamer_util import get_source_plugins, plugin_str, get_encoder_formatter, get_encoder_default_options, normv, get_codecs, get_gst_version, get_queue_time, \
from xpra.sound.gstreamer_util import get_source_plugins, plugin_str, get_encoder_formatter, get_stream_compressor, get_encoder_default_options, normv, get_codecs, get_gst_version, get_queue_time, \
MP3, CODEC_ORDER, MUXER_DEFAULT_OPTIONS, ENCODER_NEEDS_AUDIOCONVERT, SOURCE_NEEDS_AUDIOCONVERT, MS_TO_NS, GST_QUEUE_LEAK_DOWNSTREAM
from xpra.net.compression import compressed_wrapper
from xpra.scripts.config import InitExit
from xpra.log import Logger
log = Logger("sound")
Expand Down Expand Up @@ -70,7 +71,7 @@ def __init__(self, src_type=None, src_options={}, codecs=get_codecs(), codec_opt
raise InitExit(1, "no matching codecs between arguments '%s' and supported list '%s'" % (csv(codecs), csv(get_codecs().keys())))
codec = matching[0]
encoder, fmt = get_encoder_formatter(codec)
self.container_format = (fmt or "").replace("mux", "").replace("pay", "")
SoundPipeline.__init__(self, codec)
self.queue = None
self.caps = None
self.volume = None
Expand All @@ -81,7 +82,8 @@ def __init__(self, src_type=None, src_options={}, codecs=get_codecs(), codec_opt
self.buffer_latency = False
self.jitter_queue = None
self.file = None
SoundPipeline.__init__(self, codec)
self.container_format = (fmt or "").replace("mux", "").replace("pay", "")
self.stream_compressor = get_stream_compressor(codec)
src_options["name"] = "src"
source_str = plugin_str(src_type, src_options)
#FIXME: this is ugly and relies on the fact that we don't pass any codec options to work!
Expand Down Expand Up @@ -284,6 +286,10 @@ def emit_buffer0(self, buf):
return self.emit_buffer(buf.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))
metadata["compress"] = self.stream_compressor
f = self.file
if f:
for x in self.pending_metadata:
Expand Down

0 comments on commit 8d17597

Please sign in to comment.