experiment with transporting uncompressed a/v together in a matroska stream

8 files changed, 80 insertions, 272 deletions

diff --git a/voctocore/lib/audio/mix.py b/voctocore/lib/audiomix.py
index bd803b2..bd803b2 100644
--- a/voctocore/lib/audio/mix.py
+++ b/voctocore/lib/audiomix.py
diff --git a/voctocore/lib/audio/rawoutput.py b/voctocore/lib/avrawoutput.py
index e268827..70243bd 100644
--- a/voctocore/lib/audio/rawoutput.py
+++ b/voctocore/lib/avrawoutput.py
@@ -4,8 +4,8 @@ from gi.repository import GObject, Gst
 
 from lib.config import Config
 
-class AudioRawOutput(object):
-	log = logging.getLogger('AudioRawOutput')
+class AVRawOutput(object):
+	log = logging.getLogger('AVRawOutput')
 
 	name = None
 	port = None
@@ -16,28 +16,44 @@ class AudioRawOutput(object):
 
 	currentConnections = []
 
-	def __init__(self, channel, port, caps):
-		self.log = logging.getLogger('AudioRawOutput['+channel+']')
+	def __init__(self, channel, port):
+		self.log = logging.getLogger('AVRawOutput['+channel+']')
 
 		self.channel = channel
 		self.port = port
-		self.caps = caps
 
 		pipeline = """
-			interaudiosrc channel={channel} !
-			{caps} !
-			gdppay !
-			multifdsink resend-streamheader=false name=fd
+			interaudiosrc channel=audio_{channel} !
+			{acaps} !
+			queue !
+			mux.
+
+			intervideosrc channel=video_{channel} !
+			{vcaps} !
+			textoverlay halignment=left valignment=top ypad=75 text=AVRawOutput !
+			timeoverlay halignment=left valignment=top ypad=75 xpad=400 !
+			queue !
+			mux.
+
+			matroskamux
+				name=mux
+				streamable=true
+				writing-app=Voctomix-AVRawOutput !
+
+			multifdsink
+				sync-method=next-keyframe
+				name=fd
 		""".format(
 			channel=self.channel,
-			caps=self.caps
+			acaps=Config.get('mix', 'audiocaps'),
+			vcaps=Config.get('mix', 'videocaps')
 		)
-		self.log.debug('Launching Pipeline:\n%s', pipeline)
+		self.log.debug('Launching Output-Pipeline:\n%s', pipeline)
 		self.receiverPipeline = Gst.parse_launch(pipeline)
 		self.receiverPipeline.bus.add_signal_watch()
 		self.receiverPipeline.set_state(Gst.State.PLAYING)
 
-		self.log.debug('Binding to Mirror-Socket on [::]:%u', port)
+		self.log.debug('Binding to Output-Socket on [::]:%u', port)
 		self.boundSocket = socket.socket(socket.AF_INET6)
 		self.boundSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 		self.boundSocket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, False)
@@ -59,6 +75,7 @@ class AudioRawOutput(object):
 
 		return True
 
+	# FIXME handle disconnects
 	def disconnect(self, receiverPipeline, currentConnection):
 		self.currentConnections.remove(currentConnection)
 		self.log.info('Disconnected Receiver, now %u Receiver connected', len(self.currentConnections))
diff --git a/voctocore/lib/audio/src.py b/voctocore/lib/avsource.py
index 03c7bcc..ff36470 100644
--- a/voctocore/lib/audio/src.py
+++ b/voctocore/lib/avsource.py
@@ -4,8 +4,8 @@ from gi.repository import GObject, Gst
 
 from lib.config import Config
 
-class AudioSrc(object):
-	log = logging.getLogger('AudioSrc')
+class AVSource(object):
+	log = logging.getLogger('AVSource')
 
 	name = None
 	port = None
@@ -16,12 +16,11 @@ class AudioSrc(object):
 	boundSocket = None
 	currentConnection = None
 
-	def __init__(self, name, port, caps):
-		self.log = logging.getLogger('AudioSrc['+name+']')
+	def __init__(self, name, port):
+		self.log = logging.getLogger('AVSource['+name+']')
 
 		self.name = name
 		self.port = port
-		self.caps = caps
 
 		self.log.debug('Binding to Source-Socket on [::]:%u', port)
 		self.boundSocket = socket.socket(socket.AF_INET6)
@@ -43,22 +42,35 @@ class AudioSrc(object):
 
 		pipeline = """
 			fdsrc fd={fd} !
-			gdpdepay !
-			{caps} !
+			matroskademux name=demux
+
+			demux. ! 
+			{acaps} !
+			queue !
+			tee name=atee
+
+			atee. ! queue ! interaudiosink channel=audio_{name}_mixer
+			atee. ! queue ! interaudiosink channel=audio_{name}_mirror
+
+			demux. ! 
+			{vcaps} !
+			textoverlay halignment=left valignment=top ypad=25 text=AVSource !
+			timeoverlay halignment=left valignment=top ypad=25 xpad=400 !
 			queue !
-			tee name=tee
+			tee name=vtee
 
-			tee. ! queue ! interaudiosink channel=audio_{name}_mixer
-			tee. ! queue ! interaudiosink channel=audio_{name}_mirror
+			vtee. ! queue ! intervideosink channel=video_{name}_mixer
+			vtee. ! queue ! intervideosink channel=video_{name}_mirror
 		""".format(
 			fd=conn.fileno(),
 			name=self.name,
-			caps=self.caps
+			acaps=Config.get('mix', 'audiocaps'),
+			vcaps=Config.get('mix', 'videocaps')
 		)
-		self.log.debug('Launching Source-Receiver-Pipeline:\n%s', pipeline)
+		self.log.debug('Launching Source-Pipeline:\n%s', pipeline)
 		self.receiverPipeline = Gst.parse_launch(pipeline)
 
-		self.log.debug('Binding End-of-Stream-Signal on Source-Receiver-Pipeline')
+		self.log.debug('Binding End-of-Stream-Signal on Source-Pipeline')
 		self.receiverPipeline.bus.add_signal_watch()
 		self.receiverPipeline.bus.connect("message::eos", self.on_eos)
 		self.receiverPipeline.bus.connect("message::error", self.on_error)
@@ -69,12 +81,12 @@ class AudioSrc(object):
 		return True
 
 	def on_eos(self, bus, message):
-		self.log.info('Received End-of-Stream-Signal on Source-Receiver-Pipeline')
+		self.log.info('Received End-of-Stream-Signal on Source-Pipeline')
 		if self.currentConnection is not None:
 			self.disconnect()
 
 	def on_error(self, bus, message):
-		self.log.info('Received Error-Signal on Source-Receiver-Pipeline')
+		self.log.info('Received Error-Signal on Source-Pipeline')
 		(code, debug) = message.parse_error()
 		self.log.debug('Error-Details: #%u: %s', code, debug)
 
diff --git a/voctocore/lib/commands.py b/voctocore/lib/commands.py
index 9b35bf2..9673e57 100644
--- a/voctocore/lib/commands.py
+++ b/voctocore/lib/commands.py
@@ -2,7 +2,7 @@
 import logging
 
 from lib.config import Config
-from lib.video.mix import CompositeModes
+from lib.videomix import CompositeModes
 
 class ControlServerCommands():
 	log = logging.getLogger('ControlServerCommands')
@@ -12,26 +12,26 @@ class ControlServerCommands():
 
 	def __init__(self, pipeline):
 		self.pipeline = pipeline
-		self.vnames = Config.getlist('sources', 'video')
+		self.sources = Config.getlist('mix', 'sources')
 
-	def decodeVideoSrcName(self, src_name_or_id):
+	def decodeSourceName(self, src_name_or_id):
 		if isinstance(src_name_or_id, str):
 			try:
-				return self.vnames.index(src_name_or_id)
+				return self.sources.index(src_name_or_id)
 			except Exception as e:
-				raise IndexError("video-source %s unknown" % src_name_or_id)
+				raise IndexError("source %s unknown" % src_name_or_id)
 
-		if src_name_or_id < 0 or src_name_or_id >= len(self.vnames):
-			raise IndexError("video-source %s unknown" % src_name_or_id)
+		if src_name_or_id < 0 or src_name_or_id >= len(self.sources):
+			raise IndexError("source %s unknown" % src_name_or_id)
 
 
 	def set_video_a(self, src_name_or_id):
-		src_id = self.decodeVideoSrcName(src_name_or_id)
+		src_id = self.decodeSourceName(src_name_or_id)
 		self.pipeline.vmixer.setVideoA(src_id)
 		return True
 
 	def set_video_b(self, src_name_or_id):
-		src_id = self.decodeVideoSrcName(src_name_or_id)
+		src_id = self.decodeSourceName(src_name_or_id)
 		self.pipeline.vmixer.setVideoB(src_id)
 		return True
 
diff --git a/voctocore/lib/pipeline.py b/voctocore/lib/pipeline.py
index 4e4a714..03fdd3d 100644
--- a/voctocore/lib/pipeline.py
+++ b/voctocore/lib/pipeline.py
@@ -4,84 +4,35 @@ from gi.repository import Gst
 
 # import library components
 from lib.config import Config
-from lib.video.src import VideoSrc
-from lib.video.rawoutput import VideoRawOutput
-from lib.video.mix import VideoMix
-
-from lib.audio.src import AudioSrc
-from lib.audio.rawoutput import AudioRawOutput
-from lib.audio.mix import AudioMix
+from lib.avsource import AVSource
+from lib.avrawoutput import AVRawOutput
 
 class Pipeline(object):
 	"""mixing, streaming and encoding pipeline constuction and control"""
 	log = logging.getLogger('Pipeline')
 
-	vsources = []
-	vmirrors = []
-	vpreviews = []
-	vmixer = None
-	vmixerout = None
-
-	asources = []
-	amirrors = []
-	apreviews = []
-	amixer = None
-	amixerout = None
+	sources = []
+	mirrors = []
 
 	def __init__(self):
-		self.log.debug('creating Video-Pipeline')
-		self.initVideo()
-
-		self.log.debug('creating Audio-Pipeline')
-		self.initAudio()
-
-	def initVideo(self):
-		caps = Config.get('mix', 'videocaps')
-		self.log.info('Video-Caps configured to: %s', caps)
+		self.log.info('Video-Caps configured to: %s', Config.get('mix', 'videocaps'))
+		self.log.info('Audio-Caps configured to: %s', Config.get('mix', 'audiocaps'))
 
-		names = Config.getlist('sources', 'video')
+		names = Config.getlist('mix', 'sources')
 		if len(names) < 1:
-			raise RuntimeException("At least one Video-Source must be configured!")
+			raise RuntimeException("At least one AVSource must be configured!")
 
+		self.log.info('Creating %u Creating AVSources: %s', len(names), names)
 		for idx, name in enumerate(names):
 			port = 10000 + idx
-			self.log.info('Creating Video-Source %s at tcp-port %u', name, port)
+			self.log.info('Creating AVSource %s at tcp-port %u', name, port)
 
-			source = VideoSrc(name, port, caps)
-			self.vsources.append(source)
+			source = AVSource(name, port)
+			self.sources.append(source)
 
 
 			port = 13000 + idx
-			self.log.info('Creating Mirror-Output for Video-Source %s at tcp-port %u', name, port)
-
-			mirror = VideoRawOutput('video_%s_mirror' % name, port, caps)
-			self.vmirrors.append(mirror)
-
-		self.log.debug('Creating Video-Mixer')
-		self.vmixer = VideoMix()
-
-		port = 11000
-		self.log.debug('Creating Video-Mixer-Output at tcp-port %u', port)
-		self.vmixerout = VideoRawOutput('video_mix', port, caps)
-
-	def initAudio(self):
-		caps = Config.get('mix', 'audiocaps')
-		self.log.info('Audio-Caps configured to: %s', caps)
-
-		names = Config.getlist('sources', 'audio')
-		if len(names) < 1:
-			raise RuntimeException("At least one Audio-Source must be configured!")
-
-		for idx, name in enumerate(names):
-			port = 20000 + idx
-			self.log.info('Creating Audio-Source %s at tcp-port %u', name, port)
-
-			source = AudioSrc(name, port, caps)
-			self.asources.append(source)
-
-
-			port = 23000 + idx
-			self.log.info('Creating Mirror-Output for Audio-Source %s at tcp-port %u', name, port)
+			self.log.info('Creating Mirror-Output for AVSource %s at tcp-port %u', name, port)
 
-			mirror = AudioRawOutput('audio_%s_mirror' % name, port, caps)
-			self.amirrors.append(mirror)
+			mirror = AVRawOutput('%s_mirror' % name, port)
+			self.mirrors.append(mirror)
diff --git a/voctocore/lib/video/rawoutput.py b/voctocore/lib/video/rawoutput.py
deleted file mode 100644
index bb9efc4..0000000
--- a/voctocore/lib/video/rawoutput.py
+++ /dev/null
@@ -1,84 +0,0 @@
-#!/usr/bin/python3
-import logging, socket
-from gi.repository import GObject, Gst
-
-from lib.config import Config
-
-class VideoRawOutput(object):
-	log = logging.getLogger('VideoRawOutput')
-
-	name = None
-	port = None
-	caps = None
-
-	boundSocket = None
-
-	receiverPipelines = []
-	currentConnections = []
-
-	def __init__(self, channel, port, caps):
-		self.log = logging.getLogger('VideoRawOutput['+channel+']')
-
-		self.channel = channel
-		self.port = port
-		self.caps = caps
-
-		self.log.debug('Binding to Mirror-Socket on [::]:%u', port)
-		self.boundSocket = socket.socket(socket.AF_INET6)
-		self.boundSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-		self.boundSocket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, False)
-		self.boundSocket.bind(('::', port))
-		self.boundSocket.listen(1)
-
-		self.log.debug('Setting GObject io-watch on Socket')
-		GObject.io_add_watch(self.boundSocket, GObject.IO_IN, self.on_connect)
-
-	def on_connect(self, sock, *args):
-		conn, addr = sock.accept()
-		self.log.info("Incomming Connection from %s", addr)
-
-		pipeline = """
-			intervideosrc channel={channel} !
-			{caps} !
-			textoverlay text={channel} halignment=left valignment=top ypad=225 !
-			gdppay !
-			fdsink fd={fd}
-		""".format(
-			fd=conn.fileno(),
-			channel=self.channel,
-			caps=self.caps
-		)
-		self.log.debug('Launching Pipeline:\n%s', pipeline)
-		receiverPipeline = Gst.parse_launch(pipeline)
-
-		def on_eos(bus, message):
-			self.log.info('Received End-of-Stream-Signal on Source-Receiver-Pipeline')
-			self.disconnect(receiverPipeline, conn)
-
-		def on_error(bus, message):
-			self.log.info('Received Error-Signal on Source-Receiver-Pipeline')
-
-			(error, debug) = message.parse_error()
-			self.log.debug('Error-Message %s\n%s', error.message, debug)
-
-			self.disconnect(receiverPipeline, conn)
-
-		self.log.debug('Binding End-of-Stream-Signal on Pipeline')
-		receiverPipeline.bus.add_signal_watch()
-		receiverPipeline.bus.connect("message::eos", on_eos)
-		receiverPipeline.bus.connect("message::error", on_error)
-
-		receiverPipeline.set_state(Gst.State.PLAYING)
-
-		self.receiverPipelines.append(receiverPipeline)
-		self.currentConnections.append(conn)
-
-		self.log.info('Now %u Receiver connected', len(self.currentConnections))
-
-		return True
-
-	def disconnect(self, receiverPipeline, currentConnection):
-		receiverPipeline.set_state(Gst.State.NULL)
-		self.receiverPipelines.remove(receiverPipeline)
-		self.currentConnections.remove(currentConnection)
-		self.log.info('Disconnected Receiver, now %u Receiver connected', len(self.currentConnections))
diff --git a/voctocore/lib/video/src.py b/voctocore/lib/video/src.py
deleted file mode 100644
index cd36d8c..0000000
--- a/voctocore/lib/video/src.py
+++ /dev/null
@@ -1,88 +0,0 @@
-#!/usr/bin/python3
-import logging, socket
-from gi.repository import GObject, Gst
-
-from lib.config import Config
-
-class VideoSrc(object):
-	log = logging.getLogger('VideoSrc')
-
-	name = None
-	port = None
-	caps = None
-
-	receiverPipeline = None
-
-	boundSocket = None
-	currentConnection = None
-
-	def __init__(self, name, port, caps):
-		self.log = logging.getLogger('VideoSrc['+name+']')
-
-		self.name = name
-		self.port = port
-		self.caps = caps
-
-		self.log.debug('Binding to Source-Socket on [::]:%u', port)
-		self.boundSocket = socket.socket(socket.AF_INET6)
-		self.boundSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-		self.boundSocket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, False)
-		self.boundSocket.bind(('::', port))
-		self.boundSocket.listen(1)
-
-		self.log.debug('Setting GObject io-watch on Socket')
-		GObject.io_add_watch(self.boundSocket, GObject.IO_IN, self.on_connect)
-
-	def on_connect(self, sock, *args):
-		conn, addr = sock.accept()
-		self.log.info("Incomming Connection from %s", addr)
-
-		if self.currentConnection is not None:
-			self.log.warn("Another Source is already connected")
-			return True
-
-		pipeline = """
-			fdsrc fd={fd} !
-			gdpdepay !
-			{caps} !
-			textoverlay text=video_{name}_fd halignment=left valignment=top ypad=125 !
-			queue !
-			tee name=tee
-
-			tee. ! queue ! intervideosink channel=video_{name}_mixer
-			tee. ! queue ! intervideosink channel=video_{name}_mirror
-		""".format(
-			fd=conn.fileno(),
-			name=self.name,
-			caps=self.caps
-		)
-		self.log.debug('Launching Source-Receiver-Pipeline:\n%s', pipeline)
-		self.receiverPipeline = Gst.parse_launch(pipeline)
-
-		self.log.debug('Binding End-of-Stream-Signal on Source-Receiver-Pipeline')
-		self.receiverPipeline.bus.add_signal_watch()
-		self.receiverPipeline.bus.connect("message::eos", self.on_eos)
-		self.receiverPipeline.bus.connect("message::error", self.on_error)
-
-		self.receiverPipeline.set_state(Gst.State.PLAYING)
-
-		self.currentConnection = conn
-		return True
-
-	def on_eos(self, bus, message):
-		self.log.info('Received End-of-Stream-Signal on Source-Receiver-Pipeline')
-		if self.currentConnection is not None:
-			self.disconnect()
-
-	def on_error(self, bus, message):
-		self.log.info('Received Error-Signal on Source-Receiver-Pipeline')
-		(code, debug) = message.parse_error()
-		self.log.debug('Error-Details: #%u: %s', code, debug)
-
-		if self.currentConnection is not None:
-			self.disconnect()
-
-	def disconnect(self):
-		self.receiverPipeline.set_state(Gst.State.NULL)
-		self.receiverPipeline = None
-		self.currentConnection = None
diff --git a/voctocore/lib/video/mix.py b/voctocore/lib/videomix.py
index bcac7d2..bcac7d2 100644
--- a/voctocore/lib/video/mix.py
+++ b/voctocore/lib/videomix.py