#!/usr/bin/env python # # Copyright 2016 The Chromium OS Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Script functions as a web app and wrapper for the cras_router program.""" import re import subprocess import logging import cherrypy # Node Format: [Stable_Id, ID, Vol, Plugged, L/R_swapped, Time, Type, Name] ID_INDEX = 1 PLUGGED_INDEX = 3 TYPE_INDEX = 6 NAME_INDEX = 7 def get_plugged_nodes(plugged_nodes, lines, is_input): start_str = 'Input Nodes:' if is_input else 'Output Nodes:' end_str = 'Attached clients:' if is_input else 'Input Devices:' for i in range(lines.index(start_str) + 2, lines.index(end_str)): node = filter(None, re.split(r'\s+|\*+', lines[i])) # check for nodes that are plugged nodes and loopback if node[PLUGGED_INDEX] == 'yes' and node[TYPE_INDEX][:4] != 'POST': key = node[TYPE_INDEX] + ' ' + node[NAME_INDEX] plugged_nodes[key] = node[ID_INDEX] class CrasRouterTest(object): """Cherrypy class that builds and runs the HTML for audio testing tool.""" @cherrypy.expose def index(self): """Builds up and displays the html for the audio testing tool. Returns: html that was built up based on plugged audio devices. """ # Stop program if currently being run. if 'process' in cherrypy.session: print 'Existing process' # If return code is None process is still running if cherrypy.session['process'].poll() is None: print 'Killing existing process' cherrypy.session['process'].kill() else: print 'Process already finished' html = """ Audio Test

Audio Closed Loop Test

Input Type

""" dump = subprocess.check_output(['cras_test_client', '--dump_s']) if not dump: return 'Could not connect to server' dump_lines = dump.split('\n') input_plugged_nodes = {} get_plugged_nodes(input_plugged_nodes, dump_lines, True) for name, node_id in input_plugged_nodes.items(): line = '' +name + '
\n' html += line html += """File

Output Type

""" output_plugged_nodes = {} get_plugged_nodes(output_plugged_nodes, dump_lines, False) for name, node_id in output_plugged_nodes.items(): line = '' +name + '
\n' html += line html += """File

Sample Rate

48,000 Hz
44,100 Hz

""" javascript = """ """ html += javascript return html @cherrypy.expose def start_test(self, input_type, output_type, input_file='', output_file='', rate=48000): """Capture audio from the input_type and plays it back to the output_type. Args: input_type: Node id for the selected input or 'file' for files output_type: Node id for the selected output or 'file' for files input_file: Path of the input if 'file' is input type output_file: Path of the output if 'file' is output type rate: Sample rate for the test. Returns: html for the tesing in progress page. """ print 'Beginning test' if input_type == 'file' or output_type == 'file': command = ['cras_test_client'] else: command = ['cras_router'] if input_type == 'file': command.append('--playback_file') command.append(str(input_file)) else: set_input = ['cras_test_client', '--select_input', str(input_type)] if subprocess.check_call(set_input): print 'Error setting input' if output_type == 'file': command.append('--capture_file') command.append(str(output_file)) else: set_output = ['cras_test_client', '--select_output', str(output_type)] if subprocess.check_call(set_output): print 'Error setting output' command.append('--rate') command.append(str(rate)) print 'Running commmand: ' + str(command) p = subprocess.Popen(command) cherrypy.session['process'] = p return """ Running test

Test in progress

Time Elapsed

""" if __name__ == '__main__': conf = { '/': { 'tools.sessions.on': True } } cherrypy.quickstart(CrasRouterTest(), '/', conf)