diff --git a/tools/sof-tplgreader.py b/tools/sof-tplgreader.py index dbcee19b..7239821e 100755 --- a/tools/sof-tplgreader.py +++ b/tools/sof-tplgreader.py @@ -19,6 +19,13 @@ def __comp_pipeline(self, pipeline): def _key2str(self, cap, key): return cap["%s_min" % (key)], cap["%s_max" % (key)] + @staticmethod + def attach_comp_to_pipeline(comps, comp_index, comp_name, pipeline_dict): + comp = comps[comp_index] + if comp != []: + comp_names = [i['name'] for i in comp] + pipeline_dict[comp_name.lower()] = " ".join(comp_names) + # fork & split from TplgFormatter def loadFile(self, filename, sofcard=0): tplg_parser = TplgParser() @@ -37,19 +44,18 @@ def loadFile(self, filename, sofcard=0): exit(1) pgas = formatter.find_comp_for_pcm(pcm, 'PGA') eqs = formatter.find_comp_for_pcm(pcm, 'EQ') + kwds = formatter.find_comp_for_pcm(pcm, 'KPBM') + asrcs = formatter.find_comp_for_pcm(pcm, 'ASRC') pipeline_dict = {} pipeline_dict['pcm'] = pcm["pcm_name"] pipeline_dict['id'] = str(pcm["pcm_id"]) pipeline_dict['type'] = TplgFormatter.get_pcm_type(pcm) cap = pcm["caps"][pcm['capture']] - pga = pgas[pcm['capture']] - if pga != []: - pga_names = [i['name'] for i in pga] - pipeline_dict['pga'] = " ".join(pga_names) - eq = eqs[pcm['capture']] - if eq != []: - eq_names = [i['name'] for i in eq] - pipeline_dict['eq'] = " ".join(eq_names) + # acquire component from pipeline graph, and add to pipeline dict + clsTPLGReader.attach_comp_to_pipeline(pgas, pcm['capture'], "PGA", pipeline_dict) + clsTPLGReader.attach_comp_to_pipeline(eqs, pcm['capture'], "EQ", pipeline_dict) + clsTPLGReader.attach_comp_to_pipeline(kwds, pcm['capture'], "KPBM", pipeline_dict) + clsTPLGReader.attach_comp_to_pipeline(asrcs, pcm['capture'], "ASRC", pipeline_dict) # supported formats of playback pipeline in formats[0] # supported formats of capture pipeline in formats[1] formats = TplgFormatter.get_pcm_fmt(pcm) @@ -64,14 +70,9 @@ def loadFile(self, filename, sofcard=0): pb_pipeline_dict = pipeline_dict.copy() pb_pipeline_dict["type"] = "playback" cap = pcm["caps"][pcm['playback']] - pga = pgas[0] # idx 0 is playback PGA - if pga != []: - pga_names = [i['name'] for i in pga] - pb_pipeline_dict['pga'] = " ".join(pga_names) - eq = eqs[0] - if eq != []: - eq_names = [i['name'] for i in eq] - pb_pipeline_dict['eq'] = " ".join(eq_names) + clsTPLGReader.attach_comp_to_pipeline(pgas, 0, "PGA", pb_pipeline_dict) + clsTPLGReader.attach_comp_to_pipeline(eqs, 0, "EQ", pb_pipeline_dict) + clsTPLGReader.attach_comp_to_pipeline(asrcs, 0, "ASRC", pb_pipeline_dict) pb_pipeline_dict["fmts"] = " ".join(formats[pcm['playback']]) pb_pipeline_dict['fmt'] = pb_pipeline_dict['fmts'].split(' ')[0] pb_pipeline_dict['rate_min'], pb_pipeline_dict['rate_max'] = self._key2str(cap, 'rate') @@ -292,20 +293,27 @@ def func_getPipeline(tplgObj, tplgName, sdcard_id, sort): pipeline_lst = [] tplg_root = "" + # default, we want every pipeline + filter_str = "type:any" + # As KWD pipeline is so special, and our current test case cannot deal with KWD pipeline, + # Let's hide the KWD pipeline from other pipelines, except user really want KWD pipeline + # with '-f kpbm" if ret_args['filter'] is not None and len(ret_args['filter']) > 0: - # parse filter string into two structures, one is dict list for each filter item, - # and another is logic operation list. - filter_lst = [] - op_lst = [] - for filter_elem in ret_args['filter'].split('|'): - for item in filter_elem.split('&'): - key, _, value = item.partition(':') - filter_lst.append({key.strip():value.strip().split(',')}) - for char in ret_args['filter']: - if char == '|' or char == "&": - op_lst.append(char) - filter_dict['filter'] = filter_lst - filter_dict['op'] = op_lst + if "kpbm" in ret_args["filter"]: filter_str = ret_args['filter'] + else: filter_str = ret_args['filter'] + "& ~kpbm" + # parse filter string into two structures, one is dict list for each filter item, + # and another is logic operation list. + filter_lst = [] + op_lst = [] + for filter_elem in filter_str.split('|'): + for item in filter_elem.split('&'): + key, _, value = item.partition(':') + filter_lst.append({key.strip():value.strip().split(',')}) + for char in filter_str: + if char == '|' or char == "&": + op_lst.append(char) + filter_dict['filter'] = filter_lst + filter_dict['op'] = op_lst if ret_args['ignore'] is not None and len(ret_args['ignore']) > 0: for emStr in ret_args['ignore']: diff --git a/tools/tplgtool.py b/tools/tplgtool.py index 087342a8..92dcb358 100755 --- a/tools/tplgtool.py +++ b/tools/tplgtool.py @@ -685,6 +685,31 @@ def find_node_by_name(name, node_list): sys.exit(1) return None + @staticmethod + def recursive_search_comp(node, comp_type, comp_list, direction): + def comp_in_list(comp, comp_list): + for elem in comp_list: + if elem == comp["widget"]: + return True + return False + + if node is None: + return + + if type(node) == list: + for elem in node: + TplgFormatter.recursive_search_comp(elem, comp_type, comp_list, direction) + + if type(node) != list and direction == "forward": + if node["widget"]["name"].startswith(comp_type): + if not comp_in_list(node, comp_list): comp_list.append(node["widget"]) + TplgFormatter.recursive_search_comp(node["sink"], comp_type, comp_list, direction) + + if type(node) != list and direction == "backward": + if node["widget"]["name"].startswith(comp_type): + if not comp_in_list(node, comp_list): comp_list.append(node["widget"]) + TplgFormatter.recursive_search_comp(node["source"], comp_type, comp_list, direction) + # find specified type of components connected to ref_node @staticmethod def find_connected_comp(ref_node, comp_type): @@ -693,20 +718,8 @@ def find_connected_comp(ref_node, comp_type): comp_type = comp_type.upper() # to upper case comp_list = [] node = ref_node - # Currently, multiplex node is not supported, once met - # multiplex node, end while loop - while type(node) != list and node["source"] != None: - if node["widget"]["name"].startswith(comp_type) or \ - node["widget"]["sname"].startswith(comp_type): - comp_list.append(node["widget"]) - node = node["source"] - - node = ref_node - while type(node) != list and node["sink"] != None: - if node["widget"]["name"].startswith(comp_type) or \ - node["widget"]["sname"].startswith(comp_type): - comp_list.append(node["widget"]) - node = node["sink"] + TplgFormatter.recursive_search_comp(node, comp_type, comp_list, "forward") + TplgFormatter.recursive_search_comp(node, comp_type, comp_list, "backward") return comp_list # find specified components for PCM @@ -801,11 +814,11 @@ def init_node(graph, head): # this function is used to deal with multiple connections. def inner_init(graph, head): if head["sink"] != None and type(head) != list: - graph.node(name=head["name"],) + graph.node(name=head["name"]) init_node(graph, head["sink"]) elif head["sink"] != None: for node in head["sink"]: - init_node(graph, node,) + init_node(graph, node) else : graph.node(name=head["name"])