292 lines
12 KiB
Python
292 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# FrugalVox: experimental, straightforward, no-nonsense IVR framework on top of pyVoIP (patched) and TTS engines
|
|
# Created by Luxferre in 2023, released into public domain
|
|
# Deps: PyYAML, NumPy, espeak-ng/flite/libttspico, SoX, patched pyVoIP (see https://github.com/tayler6000/pyVoIP/issues/107#issuecomment-1440231926)
|
|
# All configuration is in config.yaml
|
|
|
|
import sys
|
|
import os
|
|
import signal
|
|
import tempfile
|
|
import yaml
|
|
import wave
|
|
import time
|
|
from datetime import datetime # for logging
|
|
import traceback # for logging
|
|
import socket # for local IP detection
|
|
import numpy as np # for in-band DTMF detection
|
|
import importlib.util # for action modules import
|
|
from pyVoIP.VoIP import VoIPPhone, InvalidStateError, CallState
|
|
|
|
# global parameters
|
|
progname = 'FrugalVox v0.0.1'
|
|
config = {} # placeholder for config object
|
|
configfile = './config.yaml' # default config yaml path (relative to the workdir)
|
|
if len(sys.argv) > 1:
|
|
configfile = sys.argv[1]
|
|
configfile = os.path.realpath(configfile)
|
|
kernelroot = os.path.realpath(os.path.dirname(__file__)) # absolute path to the kernel directory
|
|
configroot = os.path.dirname(configfile)
|
|
sys.path.append(kernelroot) # make the kernel module findable
|
|
if configroot != kernelroot:
|
|
sys.path.append(configroot) # make the modules in configuration directory findable
|
|
audio_buf_len = 160 # analyze this amount of raw audio data bytes
|
|
emptybuf = b'\x80' * audio_buf_len
|
|
DTMF_TABLE = {
|
|
'1': [1209, 697],
|
|
'2': [1336, 697],
|
|
'3': [1477, 697],
|
|
'A': [1633, 697],
|
|
'4': [1209, 770],
|
|
'5': [1336, 770],
|
|
'6': [1477, 770],
|
|
'B': [1633, 770],
|
|
'7': [1209, 852],
|
|
'8': [1336, 852],
|
|
'9': [1477, 852],
|
|
'C': [1633, 852],
|
|
'*': [1209, 941],
|
|
'0': [1336, 941],
|
|
'#': [1477, 941],
|
|
'D': [1633, 941]
|
|
}
|
|
DTMF_GEN_CMD = 'sox -n -D -b 8 -r 8000 %s synth 0.2 sin %s sin %s remix - gain -0.1' # command template to generate DTMF tone clips (order: file, f1, f2)
|
|
ivrconfig = None # placeholder for IVR auth config
|
|
calls = {} # placeholder for all realtime call instances
|
|
|
|
# helper methods
|
|
|
|
def logevent(msg):
|
|
dts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
print('[%s] %s' % (dts, msg))
|
|
|
|
def load_audio(fname): # load audio data from a WAV PCM file
|
|
f = wave.open(fname, 'rb')
|
|
frames = f.getnframes()
|
|
data = f.readframes(frames)
|
|
f.close()
|
|
return data
|
|
|
|
def load_yaml(fname): # load an object from a YAML file
|
|
yf = open(fname, 'r')
|
|
yc = yf.read()
|
|
yf.close()
|
|
return yaml.safe_load(yc)
|
|
|
|
def tts_to_file(text, fname, conf): # render the text to a file
|
|
fh, tname = tempfile.mkstemp('.wav', 'fvx-')
|
|
os.close(fh)
|
|
rate = int(conf['rate'])
|
|
volume = int(conf['volume'])
|
|
pitch = int(conf['pitch'])
|
|
ecmd = conf['cmd']['synth'] % (conf['voice'], volume, pitch, rate, tname, text)
|
|
os.system(ecmd) # render to the temporary file
|
|
# now, resample the synthesized file to Unsigned 8-bit 8Khz mono PCM
|
|
smpcmd = conf['cmd']['transcode'] % (tname, fname)
|
|
os.system(smpcmd)
|
|
os.remove(tname)
|
|
|
|
def tts_to_buf(text, conf): # render the text directly to a buffer
|
|
fh, fname = tempfile.mkstemp('.wav', 'fvx-')
|
|
os.close(fh)
|
|
tts_to_file(text, fname, conf)
|
|
buf = load_audio(fname)
|
|
os.remove(fname)
|
|
return buf
|
|
|
|
def gen_dtmf(f1, f2): # render two sine frequencies to a file
|
|
fh, tname = tempfile.mkstemp('.wav', 'fvx-')
|
|
os.close(fh)
|
|
os.system(DTMF_GEN_CMD % (tname, f1, f2))
|
|
buf = load_audio(tname)
|
|
os.remove(tname)
|
|
return buf
|
|
|
|
def get_caller_addr(call): # extract caller's SIP address from the call request headers
|
|
return call.request.headers['From']['address']
|
|
|
|
def get_callee_addr(call): # extract destination SIP address from the call request headers
|
|
return call.request.headers['To']['address']
|
|
|
|
def flush_input_audio(call): # clear the call's RTP input buffer
|
|
abuf = None
|
|
for i in range(625): # because 625 * 160 = 100000 (pyVoIP's internal buffer size)
|
|
abuf = call.read_audio(audio_buf_len, False)
|
|
|
|
def playbuf(buf, call): # properly play audio buffer on the call
|
|
blen = len(buf) / 8000
|
|
call.write_audio(buf)
|
|
time.sleep(blen)
|
|
|
|
def playclips(clipset, call): # properly play clips on the call
|
|
for clipname in clipset:
|
|
playbuf(clips[clipname], call)
|
|
|
|
def hangup(call): # call hangup wrapper
|
|
global calls
|
|
if call.call_id in calls:
|
|
del calls[call.call_id]
|
|
try:
|
|
call.hangup()
|
|
except InvalidStateError:
|
|
pass
|
|
logevent('Call with %s terminated' % get_caller_addr(call))
|
|
|
|
# in-band DTMF detector
|
|
|
|
def isNumberInArray(array, number):
|
|
offset = 5
|
|
for i in range(number - offset, number + offset):
|
|
if i in array:
|
|
return True
|
|
return False
|
|
|
|
def detect_dtmf(buf): # Detect a DTMF digit in the audio buffer using FFT
|
|
data = np.frombuffer(buf, dtype=np.uint8)
|
|
ftdata = np.fft.fft(data)
|
|
ftlen = len(ftdata)
|
|
for i in range(ftlen):
|
|
ftdata[i] = int(np.absolute(ftdata[i]))
|
|
lb = 20 * np.average(ftdata) # lower bound for filtering
|
|
freqs = []
|
|
for i in range(ftlen):
|
|
if ftdata[i] > lb:
|
|
freqs.append(i)
|
|
for d, fpair in DTMF_TABLE.items(): # Detect and return the digit
|
|
if isNumberInArray(freqs, fpair[0]) and isNumberInArray(freqs, fpair[1]):
|
|
return d
|
|
|
|
# IVR command handler (for authenticated and authorized action runs)
|
|
|
|
def command_handler(act, modulefile, call, userid):
|
|
global clips
|
|
global calls
|
|
global config
|
|
actid = act[0]
|
|
params = act[1:]
|
|
logevent('Running action %s from the module %s with params (%s)' % (actid, modulefile, ', '.join(params)))
|
|
(modname, ext) = os.path.splitext(os.path.basename(modulefile))
|
|
spec = importlib.util.spec_from_file_location(modname, modulefile)
|
|
actionmodule = importlib.util.module_from_spec(spec)
|
|
sys.modules[modname] = actionmodule
|
|
spec.loader.exec_module(actionmodule)
|
|
actionmodule.run_action(actid, params, call, userid, config, clips, calls)
|
|
|
|
# main call handler
|
|
|
|
def main_call_handler(call): # call object as the argument
|
|
global clips
|
|
global ivrconfig
|
|
global calls
|
|
calls[call.call_id] = call # register the call in the list
|
|
logevent('New incoming call from %s' % get_caller_addr(call))
|
|
try:
|
|
call.answer()
|
|
authdone = True
|
|
userid = '0000' # default for the unauthorized
|
|
actionsallowed = '*'
|
|
if ivrconfig['auth'] == True: # drop all permissions and prompt for the PIN
|
|
authdone = False
|
|
actionsallowed = {}
|
|
playclips(ivrconfig['authpromptclips'], call)
|
|
else: # prompt for the first command
|
|
playclips(ivrconfig['cmdpromptclips'], call)
|
|
cmdbuf = '' # command buffer
|
|
cache_digit = None # in-band digit cache
|
|
while call.state == CallState.ANSWERED: # main event loop
|
|
audiobuf = call.read_audio(audio_buf_len, False) # nonblocking audio buffer read
|
|
digit = call.get_dtmf() # get a single out-of-band DTMF digit
|
|
if digit == '' and audiobuf != emptybuf: # no out-of-band digit, try in-band detection
|
|
ib_digit = detect_dtmf(audiobuf)
|
|
if ib_digit != cache_digit:
|
|
if ib_digit == None: # digit transmission ended
|
|
digit = cache_digit # save the digit
|
|
cache_digit = None # reset the cache
|
|
else: # digit transmission started
|
|
cache_digit = ib_digit
|
|
if digit == '#': # end of the command
|
|
if authdone: # we're authenticated, let's authorize the action
|
|
actionparts = cmdbuf.split('*')
|
|
actionroot = actionparts[0]
|
|
letthrough = False
|
|
if actionsallowed == '*' or (actionroot in actionsallowed):
|
|
letthrough = True
|
|
if letthrough: # authorized
|
|
if actionroot in ivrconfig['actions']: # command exists
|
|
actionmod = os.path.realpath(os.path.join(configroot, ivrconfig['actions'][actionroot])) # resolve the action module file
|
|
command_handler(actionparts, actionmod, call, userid) # pass control to the command handler along with the call instance
|
|
else: # command doesn't exist, notify the caller
|
|
playclips(ivrconfig['cmdfailclips'], call)
|
|
logevent('Attempt to execute a non-existing action %s with the user ID %s' % (cmdbuf, userid))
|
|
else: # notify the caller that the command doesn't exist and log the event
|
|
playclips(ivrconfig['cmdfailclips'], call)
|
|
logevent('Attempt to execute an unauthorized action %s with the user ID %s' % (cmdbuf, userid))
|
|
playclips(ivrconfig['cmdpromptclips'], call) # prompt for the next command
|
|
flush_input_audio(call)
|
|
else: # we expect the first command to be our user PIN
|
|
if cmdbuf in ivrconfig['users']: # PIN found, confirm auth and prompt for the command
|
|
authdone = True
|
|
userid = cmdbuf
|
|
actionsallowed = ivrconfig['users'][userid]
|
|
playclips(ivrconfig['cmdpromptclips'], call) # prompt for the next command
|
|
else: # PIN not found, alert the caller, log the failed entry and hang up
|
|
playclips(ivrconfig['authfailclips'], call)
|
|
logevent('Attempt to enter with invalid PIN %s' % cmdbuf)
|
|
hangup(call)
|
|
cmdbuf = '' # clear command buffer
|
|
elif digit != '': # append the digit to the command buffer
|
|
cmdbuf += digit
|
|
hangup(call)
|
|
except InvalidStateError: # usually this means the call was hung up mid-action
|
|
hangup(call)
|
|
except SystemExit: # in case the service has been stopped or restarted
|
|
hangup(call)
|
|
except Exception as e:
|
|
print('Unknown error: ', sys.exc_info())
|
|
traceback.print_exc()
|
|
hangup(call)
|
|
|
|
# signal handler for graceful process termination
|
|
|
|
def sighandler(signum, frame):
|
|
global phone
|
|
logevent('Stopping the SIP client...')
|
|
phone.stop()
|
|
logevent('SIP client stopped, bye!')
|
|
|
|
# entry point
|
|
|
|
if __name__ == '__main__':
|
|
logevent('Starting %s' % progname)
|
|
config = load_yaml(configfile)
|
|
ivrconfig = config['ivr']
|
|
logevent('Configuration loaded from %s' % configfile)
|
|
clipDir = os.path.realpath(os.path.join(configroot, config['clips']['dir']))
|
|
logevent('Loading static audio clips')
|
|
clips = config['clips']['files']
|
|
for k, fname in clips.items():
|
|
clips[k] = load_audio(os.path.join(clipDir, fname))
|
|
logevent('Rendering TTS phrases')
|
|
for pname, phrase in config['tts']['phrases'].items():
|
|
clips[pname] = tts_to_buf(phrase, config['tts'])
|
|
logevent('Rendering DTMF clips')
|
|
clips['dtmf'] = {}
|
|
for digit, fpair in DTMF_TABLE.items():
|
|
clips['dtmf'][digit] = gen_dtmf(fpair[0], fpair[1])
|
|
logevent('All clips loaded to memory buffers from %s' % clipDir)
|
|
logevent('Initializing SIP phone part')
|
|
sip = config['sip']
|
|
sipport = int(sip['port'])
|
|
localname = socket.gethostname()
|
|
localip = (([ip for ip in socket.gethostbyname_ex(localname)[2] if not ip.startswith('127.')] or [[(s.connect((sip['host'], sipport)), s.getsockname()[0], s.close()) for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]]) + [None])[0]
|
|
if localip == None:
|
|
localip = socket.gethostbyname(localname)
|
|
logevent('Local IP detected: %s' % localip)
|
|
phone = VoIPPhone(sip['host'], sipport, sip['username'], sip['password'], myIP=localip, rtpPortLow=int(sip['rtpPortLow']), rtpPortHigh=int(sip['rtpPortHigh']), callCallback=main_call_handler)
|
|
# register the SIGINT and SIGTERM handlers to gracefully stop the phone instance
|
|
signal.signal(signal.SIGINT, sighandler)
|
|
signal.signal(signal.SIGTERM, sighandler)
|
|
phone.start()
|
|
logevent('SIP client started')
|