dump pcm info

 * prototype for TplgFormatter
 * process commandline arguement
 * dump pcm info

Signed-off-by: Yongan Lu <[email protected]>

tools/tplgtool-ng/tplgtool.py

@@ -2,7 +2,9 @@
 
 import argparse
 import enum
-from construct import this, ListContainer, Struct, Enum, FlagsEnum, Const, Switch, Array, Bytes, GreedyRange, FocusedSeq, Padded, Padding, PaddedString, Flag, Byte, Int16ul, Int32ul, Int64ul, Terminated
+import os
+import typing
+from construct import this, Container, ListContainer, Struct, Enum, FlagsEnum, Const, Switch, Array, Bytes, GreedyRange, FocusedSeq, Padded, Padding, PaddedString, Flag, Byte, Int16ul, Int32ul, Int64ul, Terminated
 
 class TplgType(enum.IntEnum):
     "File and Block header data types. `SND_SOC_TPLG_TYPE_`"
@@ -369,8 +371,73 @@ def parse_file(self, filename: str) -> ListContainer:
         "Parse the TPLG binary file."
         return self.sections.parse_file(filename)
 
+class TplgFormatter:
+    "the TplgFormater class can organize the topology data and format output."
+
+    def __init__(self, parsed_tplg: ListContainer):
+        self._tplg = self._group_block(parsed_tplg)
+
+    @staticmethod
+    def _group_block(parsed_tplg: ListContainer) -> "dict[str, list[list[Container]]]":
+        tplg = {
+            "pcm_list": [],
+            "widget_list": [],
+            "graph_list": [],
+            "link_list": []
+        }
+        # the last item is tplg name, ignore it
+        for item in parsed_tplg:
+            tplgtype = int(item["header"]["type"])
+            blocks: ListContainer = item["blocks"]
+            if tplgtype == TplgType.PCM:
+                tplg["pcm_list"].append(sorted(blocks, key=lambda pcm: pcm["pcm_id"]))
+            elif tplgtype == TplgType.DAPM_WIDGET:
+                tplg["widget_list"].append(blocks)
+            elif tplgtype == TplgType.DAPM_GRAPH:
+                tplg["graph_list"].append(blocks)
+            elif tplgtype in [TplgType.DAI_LINK, TplgType.BACKEND_LINK]:
+                tplg["link_list"].append(sorted(blocks, key=lambda link: link["id"]))
+        return tplg
+
+    @staticmethod
+    def get_pcm_fmt(pcm: Container) -> "list[list[str]]":
+        r"""
+        Return a list of formats lists:
+            [0]: playback stream formats
+            [1]: capture stream formats
+        """
+        return [[fmt for (fmt, flag) in cap["formats"].items() if flag and not fmt.startswith("_")] for cap in pcm["caps"]]
+
+    @staticmethod
+    def get_pcm_type(pcm: Container):
+        if pcm["playback"] == 1 and pcm["capture"] == 1:
+            return "both"
+        if pcm["playback"] == 1:
+            return "playback"
+        if pcm["capture"] == 1:
+            return "capture"
+        return "None"
+
+    def format_pcm(self):
+        for pcms in self._tplg["pcm_list"]:
+            for pcm in pcms:
+                name = pcm["pcm_name"]
+                id = pcm["pcm_id"]
+                pcm_type = TplgFormatter.get_pcm_type(pcm)
+                fmt_list = TplgFormatter.get_pcm_fmt(pcm)
+                cap_index = int(pcm_type == "capture") # 1 for capture, 0 for playback
+                cap = pcm["caps"][cap_index]
+                fmt = fmt_list[cap_index][0]
+                rate_min = cap["rate_min"]
+                rate_max = cap["rate_max"]
+                ch_min = cap["channels_min"]
+                ch_max = cap["channels_max"]
+                yield f"pcm={name};id={id};type={pcm_type};fmt={fmt};rate_min={rate_min};rate_max={rate_max};ch_min={ch_min};ch_max={ch_max};"
+
 if __name__ == "__main__":
 
+    supported_dump = ['pcm', 'graph']
+
     def parse_cmdline():
         parser = argparse.ArgumentParser(add_help=True, formatter_class=argparse.RawTextHelpFormatter,
             description='A Topology Reader totally Written in Python.')
@@ -387,9 +454,32 @@ def parse_cmdline():
             "https://graphviz.gitlab.io/_pages/doc/info/output.html for all supported formats")
         parser.add_argument('-V', '--live_view', action="store_true", help="generate and view topology graph")
 
-        return vars(parser.parse_args())
+        return parser.parse_args()
+
+    def arg_split(value: str) -> "list[str]":
+        "Split comma-delimited arguements value."
+        return [v.strip() for v in value.split(',')]
+
+    def get_tplg_paths(tplgroot: typing.Union[str, None], filenames: str) -> "list[str]":
+        from glob import iglob
+
+        tplgroot = '.' if tplgroot is None else tplgroot.strip()
+        if not os.path.isdir(tplgroot):
+            raise NotADirectoryError(tplgroot)
+        return [fullname for filename in arg_split(filenames) for fullname in iglob(os.path.join(tplgroot, filename))]
+
+    def dump_pcm_info(parsed_tplgs: "list[ListContainer]"):
+        for tplg in parsed_tplgs:
+            formatter = TplgFormatter(tplg)
+            for info in formatter.format_pcm():
+                print(info)
 
     cmd_args = parse_cmdline()
+    tplg_paths = get_tplg_paths(cmd_args.tplgroot, cmd_args.filename)
+
     tplgFormat = TplgBinaryFormat()
-    tplgStruct = tplgFormat.parse_file(cmd_args['filename'])
-    print(tplgStruct)
+    parsed_tplg_list = map(tplgFormat.parse_file, tplg_paths)
+
+    dump_types = supported_dump if cmd_args.dump is None else arg_split(cmd_args.dump)
+    if 'pcm' in dump_types:
+        dump_pcm_info(parsed_tplg_list)