From 031632245e147e4532b252f91e31631570aa0fd3 Mon Sep 17 00:00:00 2001 From: MaZderMind Date: Mon, 11 May 2015 00:32:50 +0200 Subject: a rudimentary audio pipeline --- voctocore/lib/audio/mix.py | 19 ++++++++ voctocore/lib/audio/rawoutput.py | 83 +++++++++++++++++++++++++++++++ voctocore/lib/audio/src.py | 102 +++++++++++++++++++++++++++++++++++++++ voctocore/lib/commands.py | 6 ++- voctocore/lib/controlserver.py | 30 +++++++----- voctocore/lib/pipeline.py | 35 +++++++++++++- 6 files changed, 262 insertions(+), 13 deletions(-) create mode 100644 voctocore/lib/audio/mix.py create mode 100644 voctocore/lib/audio/rawoutput.py create mode 100644 voctocore/lib/audio/src.py (limited to 'voctocore/lib') 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) -- cgit v1.2.3