a rudimentary audio pipeline

12 files changed, 285 insertions, 15 deletions

diff --git a/voctocore/README.md b/voctocore/README.md
index 7fa069a..14bd785 100644
--- a/voctocore/README.md
+++ b/voctocore/README.md
@@ -33,4 +33,9 @@ TCP-Port 9999
 
 < set_video_a blafoo
 > error "blafoo" is no known src
+
+…
+
+> signal video_a cam1
+
 ````
diff --git a/voctocore/lib/audio/mix.py b/voctocore/lib/audio/mix.py
new file mode 100644
index 0000000..bd803b2
--- /dev/null
+++ b/voctocore/lib/audio/mix.py
@@ -0,0 +1,19 @@
+#!/usr/bin/python3
+import logging
+from gi.repository import Gst
+from enum import Enum
+
+from lib.config import Config
+
+class AudioMix(object):
+	log = logging.getLogger('VideoMix')
+
+	mixingPipeline = None
+
+	caps = None
+	names = []
+
+	selectedSource = 0
+
+	def __init__(self):
+		pass
diff --git a/voctocore/lib/audio/rawoutput.py b/voctocore/lib/audio/rawoutput.py
new file mode 100644
index 0000000..1ae22e3
--- /dev/null
+++ b/voctocore/lib/audio/rawoutput.py
@@ -0,0 +1,83 @@
+#!/usr/bin/python3
+import logging, socket
+from gi.repository import GObject, Gst
+
+from lib.config import Config
+
+class AudioRawOutput(object):
+	log = logging.getLogger('AudioRawOutput')
+
+	name = None
+	port = None
+	caps = None
+
+	boundSocket = None
+
+	receiverPipelines = []
+	currentConnections = []
+
+	def __init__(self, channel, port, caps):
+		self.log = logging.getLogger('AudioRawOutput['+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 = """
+			interaudiosrc channel={channel} !
+			{caps} !
+			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/audio/src.py b/voctocore/lib/audio/src.py
new file mode 100644
index 0000000..83805b9
--- /dev/null
+++ b/voctocore/lib/audio/src.py
@@ -0,0 +1,102 @@
+#!/usr/bin/python3
+import logging, socket
+from gi.repository import GObject, Gst
+
+from lib.config import Config
+
+class AudioSrc(object):
+	log = logging.getLogger('AudioSrc')
+
+	name = None
+	port = None
+	caps = None
+
+	distributionPipeline = None
+	receiverPipeline = None
+
+	boundSocket = None
+	currentConnection = None
+
+	def __init__(self, name, port, caps):
+		self.log = logging.getLogger('AudioSrc['+name+']')
+
+		self.name = name
+		self.port = port
+		self.caps = caps
+
+		pipeline = """
+			interaudiosrc channel=audio_{name}_in !
+			{caps} !
+			queue !
+			tee name=tee
+
+			tee. ! queue ! interaudiosink channel=audio_{name}_mirror
+			tee. ! queue ! interaudiosink channel=audio_{name}_preview
+			tee. ! queue ! interaudiosink channel=audio_{name}_mixer
+		""".format(
+			name=self.name,
+			caps=self.caps
+		)
+
+		self.log.debug('Launching Source-Distribution-Pipeline:\n%s', pipeline)
+		self.distributionPipeline = Gst.parse_launch(pipeline)
+		self.distributionPipeline.set_state(Gst.State.PLAYING)
+
+		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} !
+			interaudiosink channel=audio_{name}_in
+		""".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/commands.py b/voctocore/lib/commands.py
index 05db54c..9b35bf2 100644
--- a/voctocore/lib/commands.py
+++ b/voctocore/lib/commands.py
@@ -36,6 +36,10 @@ class ControlServerCommands():
 		return True
 
 	def set_composite_mode(self, composite_mode):
-		mode = CompositeModes[composite_mode]
+		try:
+			mode = CompositeModes[composite_mode]
+		except KeyError as e:
+			raise KeyError("composite-mode %s unknown" % composite_mode)
+
 		self.pipeline.vmixer.setCompositeMode(mode)
 		return True
diff --git a/voctocore/lib/controlserver.py b/voctocore/lib/controlserver.py
index 2b96152..79aa617 100644
--- a/voctocore/lib/controlserver.py
+++ b/voctocore/lib/controlserver.py
@@ -1,5 +1,5 @@
 #!/usr/bin/python3
-import socket, logging
+import socket, logging, traceback
 from gi.repository import GObject
 
 from lib.commands import ControlServerCommands
@@ -90,15 +90,23 @@ class ControlServer():
 			return False, 'unknown command %s' % command
 
 
-		# fetch the function-pointer
-		f = getattr(self.commands, command)
+		try:
+			# fetch the function-pointer
+			f = getattr(self.commands, command)
 
-		# call the function
-		ret = f(*args)
+			# call the function
+			ret = f(*args)
 
-		# if it returned an iterable, probably (Success, Message), pass that on
-		if hasattr(ret, '__iter__'):
-			return ret
-		else:
-			# otherwise construct a tuple
-			return (ret, None)
+			# if it returned an iterable, probably (Success, Message), pass that on
+			if hasattr(ret, '__iter__'):
+				return ret
+			else:
+				# otherwise construct a tuple
+				return (ret, None)
+
+		except Exception as e:
+			self.log.error("Trapped Exception in Remote-Communication: %s", e)
+			traceback.print_exc()
+
+			# In case of an Exception, return that
+			return False, str(e)
diff --git a/voctocore/lib/pipeline.py b/voctocore/lib/pipeline.py
index a4ada04..4e4a714 100644
--- a/voctocore/lib/pipeline.py
+++ b/voctocore/lib/pipeline.py
@@ -8,6 +8,10 @@ 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
+
 class Pipeline(object):
 	"""mixing, streaming and encoding pipeline constuction and control"""
 	log = logging.getLogger('Pipeline')
@@ -18,11 +22,18 @@ class Pipeline(object):
 	vmixer = None
 	vmixerout = None
 
+	asources = []
+	amirrors = []
+	apreviews = []
+	amixer = None
+	amixerout = None
+
 	def __init__(self):
 		self.log.debug('creating Video-Pipeline')
 		self.initVideo()
 
-		self.log.debug('creating Control-Server')
+		self.log.debug('creating Audio-Pipeline')
+		self.initAudio()
 
 	def initVideo(self):
 		caps = Config.get('mix', 'videocaps')
@@ -52,3 +63,25 @@ class Pipeline(object):
 		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)
+
+			mirror = AudioRawOutput('audio_%s_mirror' % name, port, caps)
+			self.amirrors.append(mirror)
diff --git a/voctocore/scripts/audio-play-cam1-mirror.sh b/voctocore/scripts/audio-play-cam1-mirror.sh
new file mode 100755
index 0000000..e2d5905
--- /dev/null
+++ b/voctocore/scripts/audio-play-cam1-mirror.sh
@@ -0,0 +1,5 @@
+#!/bin/sh
+gst-launch-1.0 \
+	tcpclientsrc host=localhost port=23000 !\
+	gdpdepay !\
+	alsasink sync=false
diff --git a/voctocore/scripts/audio-play-cam2-mirror.sh b/voctocore/scripts/audio-play-cam2-mirror.sh
new file mode 100755
index 0000000..6881e43
--- /dev/null
+++ b/voctocore/scripts/audio-play-cam2-mirror.sh
@@ -0,0 +1,5 @@
+#!/bin/sh
+gst-launch-1.0 \
+	tcpclientsrc host=localhost port=23001 !\
+	gdpdepay !\
+	alsasink sync=false
diff --git a/voctocore/scripts/audio-source-cam1.sh b/voctocore/scripts/audio-source-cam1.sh
index 0ca10d8..f42bc3f 100755
--- a/voctocore/scripts/audio-source-cam1.sh
+++ b/voctocore/scripts/audio-source-cam1.sh
@@ -1,6 +1,6 @@
 #!/bin/sh
 gst-launch-1.0 \
-	audiotestsrc !\
+	audiotestsrc freq=440 !\
 	audio/x-raw,format=S16LE,channels=2,layout=interleaved,rate=48000 !\
 	gdppay !\
 	tcpclientsink host=localhost port=20000
diff --git a/voctocore/scripts/audio-source-cam2.sh b/voctocore/scripts/audio-source-cam2.sh
new file mode 100755
index 0000000..84de3cb
--- /dev/null
+++ b/voctocore/scripts/audio-source-cam2.sh
@@ -0,0 +1,6 @@
+#!/bin/sh
+gst-launch-1.0 \
+	audiotestsrc freq=330 !\
+	audio/x-raw,format=S16LE,channels=2,layout=interleaved,rate=48000 !\
+	gdppay !\
+	tcpclientsink host=localhost port=20001
diff --git a/voctocore/scripts/video-play-cam2-mirror.sh b/voctocore/scripts/video-play-cam2-mirror.sh
index 75e65c7..b4ff9e5 100755
--- a/voctocore/scripts/video-play-cam2-mirror.sh
+++ b/voctocore/scripts/video-play-cam2-mirror.sh
@@ -1,5 +1,5 @@
 #!/bin/sh
 gst-launch-1.0 \
-	tcpclientsrc host=localhost port=13000 !\
+	tcpclientsrc host=localhost port=13001 !\
 	gdpdepay !\
 	xvimagesink