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 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 204 insertions(+) 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/audio') 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 -- cgit v1.2.3