voice-gateway-service/pjproject/tests/pjsua/mod_media_playrec.py

# PLAYFILE -> RECFILE:
# Input file is played and is recorded to output, then compare them.
# Useful to tes clock rates compatibility and resample quality
#    null-audio
#    port 1: wav file input xxxxxx.clock_rate.wav, e.g: test1.8.wav
#    port 2: wav file ouput xxxxxx.clock_rate.wav, e.g: res1.8.wav
#    wav input must be more than 3 seconds long

import time
import sys
import re
import subprocess
import inc_const as const
import inc_util as util
from inc_cfg import *

# Load configuration
cfg_file = util.load_module_from_file("cfg_file", ARGS[1])

# WAV similarity calculator
COMPARE_WAV_EXE = ""
if sys.platform.find("win32")!=-1:
    COMPARE_WAV_EXE = "tools/cmp_wav.exe"
    G_INUNIX = False
else:
    COMPARE_WAV_EXE = "tools/cmp_wav"
    G_INUNIX = True


# Threshold to declare degradation is too high when result is lower than this value
COMPARE_THRESHOLD = 2

# COMPARE params
input_filename  = ""	# Input filename
output_filename = ""	# Output filename

# Test body function
def test_func(t):
    global input_filename
    global output_filename

    endpt = t.process[0]
    
    # Get input file name
    input_filename = re.compile(const.MEDIA_PLAY_FILE).search(endpt.inst_param.arg).group(1)
    endpt.trace("Input file = " + input_filename)

    # Get output file name
    output_filename = re.compile(const.MEDIA_REC_FILE).search(endpt.inst_param.arg).group(1)
    endpt.trace("Output file = " + output_filename)

    # Find appropriate clock rate for the input file
    clock_rate = re.compile(".+(\.\d+\.wav)$").match(output_filename).group(1)
    if (clock_rate==None):
        endpt.trace("Cannot compare input & output, incorrect output filename format")
        return
    input_filename = re.sub("\.\d+\.wav$", clock_rate, input_filename)
    endpt.trace("WAV file to be compared with output = " + input_filename)

    # Connect input-output file
    endpt.sync_stdout()

    endpt.send("cc 1 2")
    endpt.expect(const.MEDIA_CONN_PORT_SUCCESS)

    # Wait
    time.sleep(3)

    endpt.sync_stdout()

    # Disconnect input-output file
    endpt.send("cd 1 2")
    endpt.expect(const.MEDIA_DISCONN_PORT_SUCCESS)


# Post body function
def post_func(t):
    global input_filename
    global output_filename

    endpt = t.process[0]

    # Check WAV similarity
    fullcmd = COMPARE_WAV_EXE + " " + input_filename + " " + output_filename + " " + "3000"
    endpt.trace("Popen " + fullcmd)
    cmp_proc = subprocess.Popen(fullcmd, shell=G_INUNIX, stdout=subprocess.PIPE, universal_newlines=True)

    # Parse similarity ouput
    line = cmp_proc.stdout.readline()
    mo_sim_val = re.match(".+=\s+(\d+)", line)
    if (mo_sim_val == None):
        raise TestError("Error comparing WAV files")
        return

    # Evaluate the similarity value
    sim_val = mo_sim_val.group(1)
    if (int(sim_val) >= COMPARE_THRESHOLD):
        endpt.trace("WAV similarity = " + sim_val)
    else:
        raise TestError("WAV degraded heavily, similarity = " + sim_val)


# Here where it all comes together
test = cfg_file.test_param
test.test_func = test_func
test.post_func = post_func