public/agent_gsm
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

- attempt to make MOSes better

Browse Source
This commit is contained in:
Dmytro Bogovych
2023-09-17 19:37:15 +03:00
parent 4696b0e690
commit e512f7a643
5 changed files with 235 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/agent_gsm.py
View File

@@ -152,17 +152,20 @@ def run_analyze(file_test: str, file_reference: str, number: str) -> bool:
return False
try:
bounds_signal : SignalBoundaries = detect_degraded_signal(Path(file_test), Path(file_reference))
# bounds_signal.offset_start = 0
# bounds_signal.offset_finish = 0
print(f'Found signal bounds: {bounds_signal}')
bounds_signal : SignalBoundaries = None
if is_caller:
bounds_signal.offset_start = 10.0 # Skip ringtones
bounds_signal.offset_finish = 1.0 # Eat possible end tone
elif is_answerer:
bounds_signal.offset_start = 0.0
bounds_signal.offset_finish = 1.0 # Eat possible end tone
# PVQA report
pvqa_mos, pvqa_report, pvqa_rfactor = utils_sevana.find_pvqa_mos(file_test, bounds_signal.offset_start, bounds_signal.offset_finish)
utils.log(f'PVQA MOS: {pvqa_mos}, PVQA R-factor: {pvqa_rfactor}')
# AQuA report
bounds_reference : SignalBoundaries = detect_reference_signal(Path(file_reference))
bounds_reference : SignalBoundaries = SignalBoundaries()
bounds_reference.offset_start = 0
bounds_reference.offset_finish = 0
@@ -258,7 +261,7 @@ def make_call(target: str):
timelimit_seconds=ref_time_length,
target=target)
run_analyze(CONFIG.RecordFile, CONFIG.ReferenceAudio, target)
run_analyze(CONFIG.RecordFile, CONFIG.PreparedReferenceAudio, target)
except Exception as e:
utils.log_error(f'BT I/O failed finally. Error: {str(e)}')

23
src/utils_sevana.py
View File

@@ -12,8 +12,6 @@ import time
import urllib
from pathlib import Path
from colorama import Fore, Style
from utils_cache import InfoCache
PVQA_CMD = "{pvqa} --license {pvqa_lic} --config {pvqa_cfg} --mode analysis --channel 0 " \
"--report {output} --input {input} --cut-begin {cut_begin} --cut-end {cut_end}"
@@ -21,10 +19,8 @@ PVQA_CMD = "{pvqa} --license {pvqa_lic} --config {pvqa_cfg} --mode analysis --ch
PVQA_CMD_LIC_SERVER = "{pvqa} --license-server {pvqa_lic} --config {pvqa_cfg} --mode analysis --channel 0 " \
"--report {output} --input {input}"
AQUA_CMD = ("{aqua} {aqua_lic} -mode files -src file \"{reference}\" -tstf \"{input}\" -avlp off -smtnrm on "
"-decor off -mprio off -acr auto -npnt auto -voip on -enorm rms -g711 off "
"-spfrcor on -grad off -tmc on -hist-pitch on on -hist-levels on on on -miter 1 -specp 32 {spectrum} "
"-ratem %%m -fau {faults} -output json -trim r 15 -cut-tst {cut_begin} {cut_end} -cut-src {cut_begin_src} {cut_end_src}")
AQUA_CMD = ("{aqua} {aqua_lic} -mode files -src file \"{reference}\" -tstf \"{input}\" -config {aqua_config} "
"-specp 32 {spectrum} -fau {faults} -cut-tst {cut_begin} {cut_end} -cut-src {cut_begin_src} {cut_end_src}")
PVQA_PATH = ""
PVQA_LIC_PATH = "pvqa.lic"
@@ -32,6 +28,7 @@ PVQA_CFG_PATH = "pvqa.cfg"
AQUA_PATH = ""
AQUA_LIC_PATH = "aqua-wb.lic"
AQUA_CFG_PATH = "aqua.cfg"
SILER_PATH = ""
@@ -78,9 +75,11 @@ def load_config_and_licenses(server: str):
load_file(utils.join_host_and_path(server, '/deploy/pvqa.cfg'), PVQA_CFG_PATH)
load_file(utils.join_host_and_path(server, '/deploy/pvqa.lic'), PVQA_LIC_PATH)
load_file(utils.join_host_and_path(server, '/deploy/aqua-wb.lic'), AQUA_LIC_PATH)
load_file(utils.join_host_and_path(server, '/deploy/aqua.cfg'), AQUA_CFG_PATH)
except Exception as e:
utils.log_error(f'Failed to fetch new licenses and config. Skipping it.')
def find_binaries(bin_directory: Path, license_server: str = None) -> bool:
# Update path to pvqa/aqua-wb
global PVQA_CFG_PATH, PVQA_LIC_PATH, AQUA_LIC_PATH, PVQA_PATH, AQUA_PATH, PVQA_CMD, AQUA_CMD, SILER_PATH, SPEECH_DETECTOR_PATH
@@ -94,6 +93,7 @@ def find_binaries(bin_directory: Path, license_server: str = None) -> bool:
PVQA_LIC_PATH = bin_directory / PVQA_LIC_PATH
PVQA_CFG_PATH = bin_directory / PVQA_CFG_PATH
AQUA_PATH = bin_directory / platform_prefix / AQUA_PATH
AQUA_CFG_PATH = bin_directory / AQUA_CFG_PATH
AQUA_LIC_PATH = bin_directory / AQUA_LIC_PATH
SILER_PATH = bin_directory / platform_prefix / SILER_PATH
SPEECH_DETECTOR_PATH = bin_directory / platform_prefix / SPEECH_DETECTOR_PATH
@@ -109,9 +109,16 @@ def find_binaries(bin_directory: Path, license_server: str = None) -> bool:
PVQA_CFG_PATH = Path(utils.get_script_path()) / 'pvqa.cfg'
if not PVQA_CFG_PATH.exists():
utils.log_error(f'Failed to find pvqa config.')
utils.log_error(f'Failed to find PVQA config file.')
return False
if not AQUA_CFG_PATH.exists():
AQUA_CFG_PATH = Path(utils.get_script_path()) / 'aqua.cfg'
if not AQUA_CFG_PATH.exists():
utils.log_error(f'Failed to find AQuA config file.')
return False
if not AQUA_PATH.exists():
utils.log_error(f'Failed to find aqua-wb binary.')
return False
@@ -238,7 +245,7 @@ def find_aqua_mos(good_path, test_path, test_file_offset_begin: float = 0.0, tes
good_file_offset_begin: float = 0.0, good_file_offset_end: float = 0.0):
try:
out_data = ""
cmd = AQUA_CMD.format(aqua=AQUA_PATH, aqua_lic=AQUA_LIC_PATH,
cmd = AQUA_CMD.format(aqua=AQUA_PATH, aqua_lic=AQUA_LIC_PATH, aqua_config = AQUA_CFG_PATH,
reference=good_path, input=test_path, spectrum=AQUA_SPECTRUM,
faults=AQUA_FAULTS,
cut_begin=int(test_file_offset_begin * 1000), cut_end=int(test_file_offset_end * 1000),