Two visualizers (#18)

jani_visualizer plots the automata defined in a jani file.
trace_visualizer plots the content of a trace csv in a compact way.

---------

Signed-off-by: Christian Henkel <[email protected]>
Signed-off-by: Marco Lampacrescia <[email protected]>
Signed-off-by: Michaela Klauck <[email protected]>
Co-authored-by: Marco Lampacrescia <[email protected]>
Co-authored-by: Michaela Klauck <[email protected]>
Co-authored-by: Christian Henkel <[email protected]>

.github/workflows/test.yml

@@ -71,9 +71,17 @@ jobs:
             - name: Install packages
               run: | 
                 source colcon_ws/install/setup.bash  # TODO: remove after the release of bt_tools
-                pip install ${{ matrix.os == 'ubuntu-24.04' && '--break-system-packages' || '' }} jani_generator/.[dev]
                 pip install ${{ matrix.os == 'ubuntu-24.04' && '--break-system-packages' || '' }} as2fm_common/.[dev]
+                pip install ${{ matrix.os == 'ubuntu-24.04' && '--break-system-packages' || '' }} jani_generator/.[dev]
                 pip install ${{ matrix.os == 'ubuntu-24.04' && '--break-system-packages' || '' }} scxml_converter/.[dev]
+                pip install ${{ matrix.os == 'ubuntu-24.04' && '--break-system-packages' || '' }} visualizers/jani_visualizer/.[dev]
+                pip install ${{ matrix.os == 'ubuntu-24.04' && '--break-system-packages' || '' }} visualizers/trace_visualizer/.[dev]
+            # this solves
+            # E   ValueError: numpy.dtype size changed, may indicate binary incompatibility. Expected 96 from C header, got 88 from PyObject
+            - name: Downgrade numpy, networkx to match
+              run: | 
+                pip install numpy==1.26.4 networkx==2.8.8
+              if: ${{ matrix.os == 'ubuntu-22.04' }}
             # lint packages
             # TODO: add linting    
             # run the tests
@@ -82,4 +90,4 @@ jobs:
                 export PATH=$PATH:${{ steps.get_smc_storm.outputs.SMC_STORM_PATH }}
                 # source /opt/ros/${{ matrix.ros-distro }}/setup.bash
                 source colcon_ws/install/setup.bash  # TODO: remove after the release of bt_tools
-                pytest-3 -vs as2fm_common jani_generator scxml_converter
+                pytest-3 -vs as2fm_common jani_generator scxml_converter visualizers/jani_visualizer visualizers/trace_visualizer

.gitignore

@@ -1,6 +1,7 @@
 .vscode
 *.egg-info/
 */build/
+*/*/build/
 *.pyc
 
 # Sphinx

as2fm_common/pyproject.toml

@@ -1,5 +1,5 @@
 [build-system]
-requires = ["setuptools>=61.0.0", "wheel"]
+requires = ["setuptools>=61.0.0", "wheel", "pip>=24.2"]
 build-backend = "setuptools.build_meta"
 
 [project]

jani_generator/pyproject.toml

@@ -1,5 +1,5 @@
 [build-system]
-requires = ["setuptools>=61.0.0", "wheel"]
+requires = ["setuptools>=61.0.0", "wheel", "pip>=24.2"]
 build-backend = "setuptools.build_meta"
 
 [project]

scxml_converter/pyproject.toml

@@ -1,5 +1,5 @@
 [build-system]
-requires = ["setuptools>=61.0.0", "wheel"]
+requires = ["setuptools>=61.0.0", "wheel", "pip>=24.2"]
 build-backend = "setuptools.build_meta"
 
 [project]

visualizers/jani_visualizer/pyproject.toml

@@ -0,0 +1,34 @@
+[build-system]
+requires = ["setuptools>=61.0.0", "wheel", "pip>=24.2"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "jani_visualizer"
+version = "0.0.1"
+description = ""
+readme = "README.md"
+authors = [
+    {name = "Christian Henkel", email = "[email protected]"},
+    {name = "Marco Lampacrescia", email = "[email protected]"}
+]
+license = {file = "LICENSE"}
+classifiers = [
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.12",
+]
+keywords = []
+dependencies = [
+    "webcolors",
+    "plantuml",
+]
+
+requires-python = ">=3.7"
+
+[project.optional-dependencies]
+dev = ["pytest", "pytest-cov", "pycodestyle", "flake8", "mypy", "isort", "bumpver"]
+
+[project.scripts]
+jani_to_plantuml = "jani_visualizer.main:main_jani_to_plantuml"
+
+[isort]
+profile = "google"

visualizers/jani_visualizer/src/jani_visualizer/main.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+
+# Copyright (c) 2024 - for information on the respective copyright owner
+# see the NOTICE file
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+import os
+import json
+import plantuml
+
+from jani_visualizer.visualizer import PlantUMLAutomata
+
+def main_jani_to_plantuml():
+    parser = argparse.ArgumentParser(
+        description='Converts a `*.jani` file to a `*.plantuml` file.')
+    parser.add_argument('input_fname', type=str, help='The input jani file.')
+    parser.add_argument('output_plantuml_fname', type=str, help='The output plantuml file.')
+    parser.add_argument('output_svg_fname', type=str, help='The output svg file.')
+    parser.add_argument('--no-syncs', action='store_true', 
+                        help='Don\'t connects transitions that are synchronized.')
+    parser.add_argument('--no-assignments', action='store_true',
+                        help='Don\'t show assignments on the edges.')
+    parser.add_argument('--no-guard', action='store_true',
+                        help='Don\'t show guards on the edges.')
+    args = parser.parse_args()
+
+    assert os.path.isfile(args.input_fname), f"File {args.input_fname} must exist."
+    try:
+        with open(args.input_fname, 'r') as f:
+            jani_dict = json.load(f)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Error while reading the input file {args.input_fname}") from e
+
+    assert not os.path.isfile(args.output_plantuml_fname), \
+        f"File {args.output_plantuml_fname} must not exist."
+
+    assert not os.path.isfile(args.output_svg_fname), \
+        f"File {args.output_svg_fname} must not exist."
+
+    pua = PlantUMLAutomata(jani_dict)
+    puml_str = pua.to_plantuml(
+            with_assignments=not args.no_assignments,
+            with_guards=not args.no_guard,
+            with_syncs=not args.no_syncs
+        )
+    with open(args.output_plantuml_fname, 'w') as f:
+        f.write(puml_str)
+
+    plantuml.PlantUML('http://www.plantuml.com/plantuml/img/').processes_file(
+        args.output_plantuml_fname, outfile=args.output_svg_fname)
+    url = plantuml.PlantUML('http://www.plantuml.com/plantuml/img/').get_url(puml_str)
+    print(f"{url=}")

visualizers/jani_visualizer/src/jani_visualizer/visualizer.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+
+# Copyright (c) 2024 - for information on the respective copyright owner
+# see the NOTICE file
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pprint
+from typing import Union
+from colorsys import hsv_to_rgb
+from webcolors import rgb_to_hex
+
+
+def _compact_assignments(assignments: Union[dict, list, str, int]) -> str:
+    out: str = ""
+    if isinstance(assignments, dict):
+        if 'ref' in assignments:
+            assert 'value' in assignments, \
+                "The value must be present if ref is present."
+            out += f"{assignments['ref']}=({_compact_assignments(assignments['value'])})\n"
+        elif 'op' in assignments:
+            if 'left' in assignments and 'right' in assignments:
+                out += f"{_compact_assignments(assignments['left'])} {assignments['op']} {_compact_assignments(assignments['right'])}"
+            elif 'exp' in assignments:
+                out += f"{assignments['op']}({_compact_assignments(assignments['exp'])})"
+            else:
+                raise ValueError(f"Unknown assignment: {assignments}")
+        elif assignments.keys() == {'exp'}:
+            out += f"({_compact_assignments(assignments['exp'])})"
+        else:
+            raise ValueError(f"Unknown assignment: {assignments}")
+    elif isinstance(assignments, list):
+        for assignment in assignments:
+            out += _compact_assignments(assignment)
+    elif isinstance(assignments, str):
+        out += assignments
+    elif isinstance(assignments, int):
+        out += str(assignments)
+    else:
+        raise ValueError(f"Unknown assignment: {assignments}")
+    return out
+
+
+def _unique_name(automaton_name: str, location_name: str) -> str:
+    out = f"{automaton_name}_{location_name}"
+    for repl in ("-", ".", "/"):
+        out = out.replace(repl, "_")
+    return out
+
+
+class PlantUMLAutomata:
+    """This represents jani automata in plantuml format."""
+
+    def __init__(self, jani_dict: dict):
+        self.jani_dict = jani_dict
+        self.jani_automata = jani_dict['automata']
+        assert isinstance(self.jani_automata, list), \
+            "The automata must be a list."
+        assert len(self.jani_automata) >= 1, \
+            "At least one automaton must be present."
+
+    def _preprocess_syncs(self):
+        """Preprocess the synchronizations."""
+        assert 'system' in self.jani_dict, \
+            "The system must be present."
+        assert 'syncs' in self.jani_dict['system'], \
+            "The system must have syncs."
+        n_syncs = len(self.jani_dict["system"]["syncs"])
+        automata = [a['name'] for a in self.jani_automata]
+
+        # define colors for the syncs
+        colors = []
+        for i in range(n_syncs):
+            h = i / n_syncs
+            r, g, b = hsv_to_rgb(h, 1, 0.8)
+            color = rgb_to_hex((int(r * 255), int(g * 255), int(b * 255)))
+            colors.append(color)
+
+        # produce a dict with automaton, action -> color
+        colors_per_action = {}
+        for i, sync in enumerate(self.jani_dict["system"]["syncs"]):
+            synchronise = sync["synchronise"]
+            assert len(synchronise) == len(automata), \
+                "The synchronisation must have the same number of elements as the automata."
+            for action, automaton in zip(synchronise, automata):
+                if action == None:
+                    continue
+                if automaton not in colors_per_action:
+                    colors_per_action[automaton] = {}
+                colors_per_action[automaton][action] = colors[i]
+        return colors_per_action
+
+    def to_plantuml(self,
+                    with_assignments: bool = False,
+                    with_guards: bool = False,
+                    with_syncs: bool = False,
+                    ) -> str:
+        colors_per_action = self._preprocess_syncs()
+
+        puml: str = "@startuml\n"
+        puml += "scale 500 width\n"
+
+        for automaton in self.jani_automata:
+            # add a box for the automaton
+            automaton_name = automaton['name']
+            puml += f"package {automaton_name} {{\n"
+            for location in automaton['locations']:
+                loc_name = _unique_name(automaton_name, location['name'])
+                puml += f"    usecase \"{location['name']}\" as {loc_name}\n"
+            for edge in automaton['edges']:
+                source = _unique_name(automaton_name, edge['location'])
+                assert len(edge['destinations']) == 1, \
+                    "Only one destination is supported."
+                destination = edge['destinations'][0]
+                target = _unique_name(automaton_name, destination['location'])
+                edge_label = ""
+                color = "#000"  # black by default
+
+                # Assignments
+                if (
+                    with_assignments and
+                    'assignments' in destination and
+                    len(destination['assignments']) > 0
+                ):
+                    assignments_str = _compact_assignments(destination['assignments']).strip()
+                    edge_label += f"⏬{assignments_str}\n"
+
+                # Guards
+                if (
+                    with_guards and
+                    'guard' in edge
+                ):
+                    guard_str = _compact_assignments(edge['guard']).strip()
+                    edge_label += f"💂{guard_str}\n"
+
+                # Syncs
+                if (
+                    with_syncs and
+                    'action' in edge
+                ):
+                    action = edge['action']
+                    if automaton['name'] in colors_per_action and action in colors_per_action[automaton['name']]:
+                        color = colors_per_action[automaton['name']][action]
+                    edge_label += f"🔗{action}\n"
+
+                edge_label = '  \\n\\\n'.join(edge_label.split('\n'))
+                if len(edge_label.strip()) > 0:
+                    puml += f"    {source} -[{color}]-> {target} : {edge_label}\n"
+                else:
+                    puml += f"    {source} -[{color}]-> {target}\n"
+            puml += "}\n"
+
+        return puml