Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
#!/usr/bin/env python
# This is the loquacious daemon. It is designed to record audio clips from JACK upon noticing the microphones are on via Redis (simply use scripts to set the redis key 'mics' to 'on' or 'off').
# Uses JACK Timemachine in console mode in the background to handle the actual work of recording audio.
# Once it's done grabbing audio it pushes the filename (full path) to a Redis list. You can then write code to deal with new entries to the list and retrieve them; for instance, in our system we log each new recording as an Aircheck, and after each show, generate a podcast from the recorded data.
# Note you will need to cull old files on your own - the script does not deal with this. Something like:
# 0 * * * * find /home/insanity/loquaciousd/ -mtime +1 -delete
# in your crontab will do the job (removing all airchecks older than a day).
# The record path is specified in the spawnProcess call arguments.
import redis
import sys
import os
from twisted.internet import protocol, reactor, task
import re
class TimemachineProcessProtocol(protocol.ProcessProtocol):
def __init__(self):
self.recording = False
def connectionMade(self):
print "connectionMade!"
def outReceived(self, data):
print "outReceived! with %d bytes!" % len(data)
print data
fn = re.match('opened \'(.+)\'', data)
if fn and
print "Got filename "
# Cool - we're done recording and we have our filename. Let's roll with that.
r = redis.Redis(host='some-redis-host', port=6379, db=0)
def errReceived(self, data):
print "errReceived! with %d bytes!" % len(data)
print data
def inConnectionLost(self):
print "inConnectionLost! stdin is closed! (we probably did it)"
def processExited(self, reason):
print "processExited, status %d" % (reason.value.exitCode,)
def processEnded(self, reason):
print "processEnded, status %d" % (reason.value.exitCode,)
print "quitting"
def start(self):
print "start recording"
self.recording = True
def stop(self):
print "stop recording"
self.recording = False
proc = TimemachineProcessProtocol()
reactor.spawnProcess(proc, '/usr/bin/timemachine', ['-i', '-n', 'Loquaciousd', '-t', '4', '-p', '/home/insanity/loquaciousd/audio-', '-f', 'wav', 'system:capture_1', 'system:capture_2'], env=os.environ, uid=0)
def changeStates():
print "Checking redis."
r = redis.Redis(host='some-redis-host', port=6379, db=0)
mic_state = r.get("mics")
if mic_state == "on" and proc.recording == False:
elif mic_state == "off" and proc.recording == True:
print "Doing nothing."
print "Whoops, failed miserably"
l = task.LoopingCall(changeStates)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.