agent_gsm/src/utils_dtmf.py

110 lines
2.7 KiB
Python

#!/usr/bin/python3
import argparse
import os
import sys
import typing
import time
import subprocess
import select
import multiprocessing
import signal
import utils
import uiautomator2 as u2
import utils_mcon
# Exit codes
EXIT_SUCCESS = 0
EXIT_ERROR = 1
AUTOMATOR = None
def gsm_attach_automator():
# Run stock dialer as way to preload automator stack
utils.log("Connecting to device...")
d = u2.connect()
# Preload GSM helper app
utils.log("Preloading GSM helper app")
d.app_start("biz.sevana.qualtestgsm")
# Wait timeout for UI element is 60.0s
d.implicitly_wait(60.0)
# Preload stock dialer
# utils.log("Preloading stock dialer")
# d.app_start("com.skype.raider")
return d
def gsm_switch_to_dtmf_panel(d):
# As stub for now - use Skype Contact click
# d(resourceId="com.skype.raider:id/vm_name", text=contact_name).click()
return None
# Send DTMF string
def send_dtmf(dtmf: str):
global AUTOMATOR
gsm_switch_to_dtmf_panel(AUTOMATOR)
for c in dtmf:
utils_mcon.gsm_send_digit(c)
# Number of finished calls
CALL_COUNTER = multiprocessing.Value('i', 0)
def on_call_finished(file):
# Increase finished calls counter
CALL_COUNTER.value = CALL_COUNTER.value + 1
def make_call(target: str, dtmf: str):
global CALL_COUNTER
# Start subprocess to monitor events from Qualtest GSM
start_handler = lambda file_record, file_play, number: send_dtmf(dtmf)
finish_handler = lambda file_record, file_play, number: on_call_finished()
PROCESS_MONITOR = multiprocessing.Process(target=utils_mcon.gsm_monitor,
args=(None, None, start_handler, finish_handler, None))
PROCESS_MONITOR.start()
# Initiate GSM phone call via adb
utils_mcon.gsm_make_call(target)
# Wait for call finish with some timeout. Kill monitoring process on finish.
while CALL_COUNTER.value == 0:
time.sleep(1)
PROCESS_MONITOR.terminate()
if __name__ == '__main__':
# Default exit code
retcode = EXIT_SUCCESS
# Command line parameters
parser = argparse.ArgumentParser()
parser.add_argument("--target", help="target number")
parser.add_argument("--dtmf", help="DTMF string to send after call established")
args = parser.parse_args()
# Check if we have to make a call
try:
if args.target:
# Preload automator framework
AUTOMATOR = gsm_attach_automator()
# Start call
make_call(args.target, args.dtmf)
except Exception as e:
utils.log_error(e)
retcode = EXIT_ERROR
# Exit code 0 (success)
sys.exit(retcode)