dump topology graph

Signed-off-by: Yongan Lu <yongan.lu@intel.com>

tools/tplgtool-ng/tplgtool.py

@@ -3,6 +3,7 @@
 import argparse
 import enum
 import os
+import sys
 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
 
@@ -431,7 +432,40 @@ def format_pcm(self):
             ch_max = cap["channels_max"]
             yield f"pcm={name};id={pcm_id};type={pcm_type};fmt={fmt};rate_min={rate_min};rate_max={rate_max};ch_min={ch_min};ch_max={ch_max};"
 
+    def draw_graph(self, outfile: str, outdir: str = '.', file_format: str = "png", live_view: bool = False):
+        from graphviz import Digraph
+        import tempfile
+
+        nodes = self._nodes_for_graph()
+        graph = Digraph("Topology Graph", format=file_format)
+        for nodename, _ in nodes:
+            graph.node(nodename)
+        for edge in self._graph_list:
+            graph.edge(edge["source"], edge["sink"])
+        if live_view:
+            # if run the tool over ssh, live view feature will be disabled
+            if 'DISPLAY' not in os.environ.keys():
+                print("No available GUI over ssh, unable to view the graph", file=sys.stderr)
+            else:
+                graph.view(filename=outfile, directory=tempfile.gettempdir(), cleanup=True)
+        else:
+            graph.render(filename=outfile, directory=outdir, cleanup=True)
+
+    def _nodes_for_graph(self) -> typing.Iterable[typing.Tuple[str, Container]]:
+        for widget in self._widget_list:
+            name = self._node_name_in_graph(widget)
+            if name != None:
+                yield (name, widget)
+
+    def _node_name_in_graph(self, widget: Container) -> typing.Union[str, None]:
+        for edge in self._graph_list:
+            for ename in [edge["source"], edge["sink"]]:
+                if ename in [widget["widget"]["name"], widget["widget"]["sname"]]:
+                    return ename
+        return None
+
 if __name__ == "__main__":
+    from pathlib import Path
 
     supported_dump = ['pcm', 'graph']
 
@@ -465,18 +499,18 @@ def get_tplg_paths(tplgroot: typing.Union[str, None], filenames: str) -> "list[s
             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 = TplgAdaptor(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()
-    parsed_tplg_list = list(map(tplgFormat.parse_file, tplg_paths))
+    parsed_tplgs = list(zip(tplg_paths, 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)
+    for file, tplg in parsed_tplgs:
+        adaptor = TplgAdaptor(tplg)
+        if 'pcm' in dump_types:
+            for info in adaptor.format_pcm():
+                print(info)
+        if 'graph' in dump_types:
+            basename = Path(file).stem
+            adaptor.draw_graph(basename, outdir=cmd_args.directory, file_format=cmd_args.format, live_view=cmd_args.live_view)