init
This commit is contained in:
Luxferre
@@ -0,0 +1,277 @@
|
||||
#!/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]
|
||||
}
|
||||
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 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 clips and compiling TTS phrases')
|
||||
clips = config['clips']['files']
|
||||
for k, fname in clips.items():
|
||||
clips[k] = load_audio(os.path.join(clipDir, fname))
|
||||
for pname, phrase in config['tts']['phrases'].items():
|
||||
clips[pname] = tts_to_buf(phrase, config['tts'])
|
||||
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')
|
||||
Reference in New Issue
Block a user