sof-test: read /proc instead of parsing TPLG file for legacy HDA platforms

case-lib/lib.sh

@@ -264,3 +264,8 @@ sudo sync -f || true
 if [ ! "$DMESG_LOG_START_LINE" ]; then
     DMESG_LOG_START_LINE=$(wc -l /var/log/kern.log|awk '{print $1;}')
 fi
+
+is_sof_used()
+{
+    grep -q "sof" /proc/asound/cards;
+}

case-lib/pipeline.sh

@@ -1,11 +1,33 @@
 #!/bin/bash
 
+# This function will evaluate pipeline parameter related shell code generated by
+# - sof-tplgreader.py (for SOF, pipeline parameters dumped from topology)
+# - sof-dump-status.py (for legacy HDA, pipeline paramters dumped from proc)
+# Args: $1: SOF topology path
+#       $2: Pipeline filter in string form
+# Note: for legacy HDA, topology is not present, $1 will be empty.
 func_pipeline_export()
 {
-    # no parameter input the function
-    if [ $# -lt 1 ]; then
-        die "Topology file name is not specified, unable to run command: $SCRIPT_NAME"
+
+    # function parameter check
+    if [ $# -ne 2 ]; then
+        die "Not enough parameters, expect two parameters: topology path and pipeline filter"
     fi
+
+    # For legacy HDA platform, there is no topology, we have to export pipeline
+    # parameters from proc file system.
+    is_sof_used || {
+        filter_str="$2"
+        dlogi "No SOF sound card found, exporting pipeline parameters from proc file system"
+        tmp_pipeline_params=$(mktemp /tmp/pipeline-params.XXXXXXXX)
+        sof-dump-status.py -e "$filter_str" > "$tmp_pipeline_params" ||
+            die "Failed to export pipeline parameters from proc file system"
+        # shellcheck disable=SC1090
+        source "$tmp_pipeline_params" || return 1
+        rm "$tmp_pipeline_params"
+        return 0
+    }
+
     # got tplg_file, verify file exist
     tplg_path=$(func_lib_get_tplg_path "$1") || {
         die "Topology $1 not found, check the TPLG environment variable or specify topology path with -t"

tools/common.py

@@ -0,0 +1,30 @@
+def format_pipeline(pipeline, noKey=False):
+    output = ""
+    for key, value in pipeline.items():
+        if noKey is True:
+            output += str(value) + " "
+        else:
+            output += key + "=" + str(value) + ";"
+    return output.strip()
+
+# This function will generate shell code according to pipeline parameters,
+# then pipeline parameters can be accessed from test case by sourcing or
+# executing the generated code.
+def export_pipeline(pipeline_lst):
+    length = len(pipeline_lst)
+    keyword = 'PIPELINE'
+    # clear up the older define
+    print('unset %s_COUNT' % (keyword))
+    print('unset %s_LST' % (keyword))
+    print('declare -g %s_COUNT' % (keyword))
+    print('declare -ag %s_LST' % (keyword))
+    print('%s_COUNT=%d' % (keyword, length))
+    for idx in range(0, length):
+        # store pipeline
+        print('%s_LST[%d]="%s"' % (keyword, idx, format_pipeline(pipeline_lst[idx])))
+        # store pipeline to each list
+        print('unset %s_%d' % (keyword, idx))
+        print('declare -Ag %s_%d' % (keyword, idx))
+        for key, value in pipeline_lst[idx].items():
+            print('%s_%d["%s"]="%s"' % (keyword, idx, key, value))
+    return 0

tools/sof-dump-status.py

@@ -2,6 +2,7 @@
 
 import subprocess
 import os
+from common import format_pipeline, export_pipeline
 
 class clsSYSCardInfo():
     def __init__(self):
@@ -364,12 +365,8 @@ def export_proc_sound(proc_card):
             return
 
     def dump_cardinfo_pcm(card_info):
-        pcm_lst = card_info.get('pcm')
-        def _getStr(tmp_dict, key):
-            return '%s=%s'%(key, tmp_dict.get(key))
-        for pcm in pcm_lst:
-            print('%s;%s;%s;' % (_getStr(pcm, 'id'), _getStr(pcm, 'pcm'), _getStr(pcm, 'type')))
-        return 0
+        for pipeline in card_info.get('pcm'):
+            print(format_pipeline(pipeline))
 
     def dump_dapm(dapm, filter = "all"):
         if filter == "all" and len(dapm['dapm_lst']) == 0:
@@ -413,6 +410,11 @@ def dump_dapm(dapm, filter = "all"):
     parser.add_argument('-P', '--fwpath', action='store_true', help='get firmware path according to DMI info')
     parser.add_argument('-S', '--dsp_status', type=int, help='get current dsp power status, should specify sof card number')
     parser.add_argument('-d', '--dapm', choices=['all', 'on', 'off', 'standby'], help='get current dapm status, this option need root permission to access debugfs')
+    # The filter string here is compatible with the filter string for pipeline from topology,
+    # and takes the form of 'type:playback & pga:any & ~asrc:any | id:5'.
+    parser.add_argument('-e', '--export', type=str, help='export pipeline parameters of specified type from proc file system,\n'
+    'to specify pipeline type, use "-e type:playback", complex string like\n'
+    '"type:playback & pga:any" can be used, but only "type" is processed')
     parser.add_argument('--version', action='version', version='%(prog)s 1.0')
 
     ret_args = vars(parser.parse_args())
@@ -437,6 +439,41 @@ def dump_dapm(dapm, filter = "all"):
             print(run_status['status'])
         exit(0)
 
+    if ret_args['export'] is not None:
+        sysinfo.loadProcSound()
+        pipeline_lst = []
+        for (card_id, card_info) in sysinfo.proc_card.items():
+            for pcm in card_info['pcm']:
+                # There are limited pipeline parameters in the proc file system,
+                # add some default parameters to make use of sof-test for legacy HDA test
+                pcm['fmt'] = 'S16_LE'
+                pcm['fmts'] = 'S16_LE S24_LE S32_LE'
+                pcm['rate'] = '48000'
+                pcm['channel'] = '2'
+                pcm['dev'] = 'hw:{},{}'.format(card_id, pcm['id'])
+            pipeline_lst.extend(card_info['pcm'])
+        # The filter string may be very complex due to the compatibility with topology pipeline
+        # filter string, but we only implement a simple type filter here.
+        try:
+            # ret_args['export'] contains filter string, and it may looks like 'type:capture & pga:any | id:3'
+            # extract filter elements, and drop logic operator
+            filter_elements = [elem for elem in ret_args['export'].split(' ') if ':' in elem]
+            # extract requested pipeline type from filter elements, and ignore others.
+            requested_type = [f.split(':')[1] for f in filter_elements if f.strip().startswith('type')][0]
+        except:
+            print('Invalid filter string')
+            exit(1)
+        # requested_type may take one of the values: playback, capture, any
+        if requested_type in ['playback', 'capture']:
+            export_pipeline([p for p in pipeline_lst if p['type'] == requested_type])
+        elif requested_type == 'any':
+            export_pipeline(pipeline_lst)
+        else:
+            print('Unknown requested pipeline type: %s' % requested_type)
+            print('Available requested pipeline types are: playback, capture, any')
+            exit(1)
+        exit(0)
+
     if ret_args.get('id') is not None:
         sysinfo.loadProcSound()
         card_info = sysinfo.proc_card.get(str(ret_args['id']))

tools/sof-tplgreader.py

@@ -4,6 +4,7 @@
 import os
 import re
 from tplgtool import TplgParser, TplgFormatter
+from common import format_pipeline, export_pipeline
 
 class clsTPLGReader:
     def __init__(self):
@@ -243,33 +244,6 @@ def getPipeline(self, sort=False):
         return self._output_lst
 
 if __name__ == "__main__":
-    def func_dump_pipeline(pipeline, noKey=False):
-        output = ""
-        for key, value in pipeline.items():
-            if noKey is True:
-                output += str(value) + " "
-            else:
-                output += key + "=" + str(value) + ";"
-        return output.strip()
-
-    def func_export_pipeline(pipeline_lst):
-        length = len(pipeline_lst)
-        keyword = 'PIPELINE'
-        # clear up the older define
-        print('unset %s_COUNT' % (keyword))
-        print('unset %s_LST' % (keyword))
-        print('declare -g %s_COUNT' % (keyword))
-        print('declare -ag %s_LST' % (keyword))
-        print('%s_COUNT=%d' % (keyword, length))
-        for idx in range(0, length):
-            # store pipeline
-            print('%s_LST[%d]="%s"' % (keyword, idx, func_dump_pipeline(pipeline_lst[idx])))
-            # store pipeline to each list
-            print('unset %s_%d' % (keyword, idx))
-            print('declare -Ag %s_%d' % (keyword, idx))
-            for key, value in pipeline_lst[idx].items():
-                print('%s_%d["%s"]="%s"' % (keyword, idx, key, value))
-        return 0
 
     def func_getPipeline(tplgObj, tplgName, sdcard_id, sort):
         if tplgObj.loadFile(tplgName, sdcard_id) != 0:
@@ -376,15 +350,15 @@ def parse_and_set_block_keyword(block_str, tplgreader):
         pipeline_lst += func_getPipeline(tplgreader, f, ret_args['sofcard'], ret_args['sort'])[:]
 
     if ret_args['export'] is True:
-        exit(func_export_pipeline(pipeline_lst))
+        exit(export_pipeline(pipeline_lst))
 
     if ret_args['count'] is True:
         if ret_args['value'] is True:
             print(len(pipeline_lst))
         else:
             print("Pipeline Count: %d" % (len(pipeline_lst)))
     elif ret_args['index'] is not None and ret_args['index'] < len(pipeline_lst):
-        print(func_dump_pipeline(pipeline_lst[ret_args['index']], ret_args['value']))
+        print(format_pipeline(pipeline_lst[ret_args['index']], ret_args['value']))
     else:
         for pipeline in pipeline_lst:
-            print(func_dump_pipeline(pipeline, ret_args['value']))
+            print(format_pipeline(pipeline, ret_args['value']))