Feature: FPP Visualize (#147)

* Visualize with hardcoded dependencies

* move visualize under fpp/

* Use fprime_visual Flask app

* assume layout tools in PATH

* spelling

* formatting

* Soft dependency on fprime_visual

* Add --cache-dir option

* formatting

* Review changes

* Verbose permission error

* Use tempfile if working-dir not specified

* spelling

* Fix is_supported parameters

* Formatting

.github/actions/spelling/expect.txt

@@ -75,6 +75,7 @@ filepath
 firest
 floordiv
 FPGA
+fpl
 fpp
 fprime
 fromkeys
@@ -230,6 +231,7 @@ stylesheet
 subdir
 subparser
 Subproc
+subtopology
 sys
 tcanham
 Tcp

src/fprime/fpp/cli.py

@@ -35,6 +35,29 @@ def run_fpp_check(
     )
 
 
+def run_fpp_to_xml(
+    build: "Build",
+    parsed: argparse.Namespace,
+    _: Dict[str, str],
+    __: Dict[str, str],
+    ___: List[str],
+):
+    """Run the fpp-to-xml utility
+
+    Args:
+        build: build directory output
+        parsed: parsed input arguments
+        _: unused cmake_args
+        __: unused make_args
+        ___: unused pass-through arguments
+    """
+    FppUtility("fpp-to-xml").execute(
+        build,
+        parsed.path,
+        args=({}, ["--directory", parsed.directory] if parsed.directory else []),
+    )
+
+
 def add_fpp_parsers(
     subparsers, common: argparse.ArgumentParser
 ) -> Tuple[Dict[str, Callable], Dict[str, argparse.ArgumentParser]]:
@@ -56,4 +79,11 @@ def add_fpp_parsers(
     check_parser.add_argument(
         "-u", "--unconnected", default=None, help="write unconnected ports to file"
     )
-    return {"fpp-check": run_fpp_check}, {"fpp-check": check_parser}
+    fpp_to_xml_parser = subparsers.add_parser(
+        "fpp-to-xml", help="Runs fpp-to-xml utility", parents=[common], add_help=False
+    )
+    fpp_to_xml_parser.add_argument("--directory", default=None, help="Output directory")
+    return {"fpp-check": run_fpp_check, "fpp-to-xml": run_fpp_to_xml}, {
+        "fpp-check": check_parser,
+        "fpp-to-xml": fpp_to_xml_parser,
+    }

src/fprime/fpp/common.py

@@ -7,6 +7,8 @@
 import itertools
 import subprocess
 from pathlib import Path
+import shutil
+import sys
 from typing import Dict, List, Tuple
 
 from fprime.common.error import FprimeException
@@ -43,6 +45,10 @@ def __init__(self, name):
         super().__init__(TargetScope.LOCAL)
         self.utility = name
 
+    def is_supported(self, _=None, __=None):
+        """Returns whether this utility is supported"""
+        return bool(shutil.which(self.utility))
+
     @staticmethod
     def get_locations_file(builder: Build) -> Path:
         """Returns the location of the FPP locations file
@@ -98,6 +104,14 @@ def execute(
             args: extra arguments to supply to the utility
         """
         # First refresh the cache but only if it detects it needs too
+
+        if not self.is_supported():
+            print(
+                f"[ERROR] Cannot find executable: {self.utility}.",
+                file=sys.stderr,
+            )
+            return 1
+
         builder.cmake.cmake_refresh_cache(builder.build_dir, False)
 
         # Read files and arguments

src/fprime/fpp/visualize.py

@@ -0,0 +1,167 @@
+""" fprime.fpp.visualize: Command line targets for fprime-util visualize
+
+@author thomas-bc
+"""
+import argparse
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Callable, Dict, List, Tuple
+
+from fprime.fpp.common import FppUtility
+
+try:
+    from fprime_visual.flask.app import construct_app
+except ImportError:
+    construct_app = None
+
+
+def run_fprime_visualize(
+    build: "Build",
+    parsed: argparse.Namespace,
+    _: Dict[str, str],
+    __: Dict[str, str],
+    ___: List[str],
+):
+    """Run pipeline of utilities to generate visualization. This includes:
+    - fpp-to-xml
+    - fpl-convert-xml
+    - fpl-layout
+    - start fprime-visual Flask app to serve visualization
+
+    Args:
+        build: build directory output
+        parsed: parsed input arguments
+        _: unused cmake_args
+        __: unused make_args
+        ___: unused pass-through arguments
+    """
+    if construct_app is None:
+        raise ModuleNotFoundError(
+            "fprime-visual is not installed. Please install with `pip install fprime-visual`"
+        )
+
+    if not (shutil.which("fpl-convert-xml") and shutil.which("fpl-layout")):
+        raise FileNotFoundError(
+            "fpl-layout is not installed. Please install with `pip install fprime-fpp>1.2.0`"
+        )
+
+    # Set up working directory using specified directory, or create a temporary one
+    if parsed.working_dir:
+        viz_cache = Path(parsed.working_dir).resolve()
+    else:
+        viz_cache = Path(
+            tempfile.TemporaryDirectory(prefix="fprime-visual-").name
+        ).resolve()
+
+    # Set sub-paths for different types of generated files
+    xml_cache = (viz_cache / "xml").resolve()
+    extract_cache = (viz_cache / "extracted").resolve()
+    try:
+        xml_cache.mkdir(parents=True, exist_ok=True)
+        extract_cache.mkdir(parents=True, exist_ok=True)
+    except PermissionError:
+        raise PermissionError(
+            f"Unable to write to {viz_cache.resolve()}. Use --working-dir to set a different location."
+        )
+
+    # Run fpp-to-xml
+    FppUtility("fpp-to-xml").execute(
+        build,
+        parsed.path,
+        args=(
+            {},
+            ["--directory", str(xml_cache)],
+        ),
+    )
+    topology_match = list(xml_cache.glob("*TopologyAppAi.xml"))
+    if len(topology_match) == 1:
+        topology_xml = topology_match[0]
+    else:
+        raise Exception(
+            f"Found {len(topology_match)} '*TopologyAppAi.xml' topology files - expected 1"
+        )
+
+    print(f"Generated topology XML file: {topology_xml.resolve()}")
+
+    topology_txt = viz_cache / "Topology.txt"
+    topology_json = viz_cache / "Topology.json"
+
+    # Execute: fpl-convert-xml Topology.xml > Topology.txt
+    with open(topology_txt.resolve(), "w") as txt_file:
+        subprocess.run(
+            ["fpl-convert-xml", topology_xml.resolve()], stdout=txt_file, check=True
+        )
+
+    # Execute: fpl-layout < Topology.txt > Topology.json
+    with open(topology_json.resolve(), "w") as json_file:
+        with open(topology_txt.resolve(), "r") as txt_file:
+            subprocess.run(["fpl-layout"], stdin=txt_file, stdout=json_file, check=True)
+
+    print("Extracting subtopologies...")
+    # Execute: fpl-extract-xml -d extracted/ Topology.xml
+    subprocess.run(
+        ["fpl-extract-xml", "-d", extract_cache.resolve(), topology_xml.resolve()],
+        check=True,
+    )
+    subtopologies = list(extract_cache.glob("*.xml"))
+    for subtopology in subtopologies:
+        # Execute: fpl-convert-xml subtopology.xml > subtopology.txt
+        subtopology_txt = extract_cache / f"{subtopology.stem}.txt"
+        with open(subtopology_txt.resolve(), "w") as txt_file:
+            subprocess.run(
+                ["fpl-convert-xml", subtopology.resolve()], stdout=txt_file, check=True
+            )
+        # Execute: fpl-layout < subtopology.txt > subtopology.json
+        subtopology_json = viz_cache / f"{subtopology.stem}.json"
+        with open(subtopology_json.resolve(), "w") as json_file:
+            with open(subtopology_txt.resolve(), "r") as txt_file:
+                subprocess.run(
+                    ["fpl-layout"], stdin=txt_file, stdout=json_file, check=True
+                )
+
+    print("[INFO] Starting fprime-visual server...")
+    print(f"[INFO] Serving files in {str(viz_cache.resolve())}")
+    config = {"SOURCE_DIRS": [str(viz_cache.resolve())]}
+    app = construct_app(config)
+    try:
+        app.run(port=parsed.gui_port)
+    except KeyboardInterrupt:
+        print("[INFO] CTRL-C received. Exiting.")
+    return 0
+
+
+def add_fpp_viz_parsers(
+    subparsers, common: argparse.ArgumentParser
+) -> Tuple[Dict[str, Callable], Dict[str, argparse.ArgumentParser]]:
+    """Sets up the fprime-viz command line parsers
+
+    Creates command line parsers for fprime-viz commands and associates these commands to processing functions for those fpp
+    commands.
+
+    Args:
+        subparsers: subparsers to add to
+        common: common parser for all fprime-util commands
+
+    Returns:
+        Tuple of dictionary mapping command name to processor, and command to parser
+    """
+    viz_parser = subparsers.add_parser(
+        "visualize",
+        help="Runs visualization pipeline",
+        parents=[common],
+        add_help=False,
+    )
+    viz_parser.add_argument(
+        "--gui-port",
+        help="Set the GUI port for fprime-visual [default: %(default)s]",
+        required=False,
+        default=7000,
+    )
+    viz_parser.add_argument(
+        "--working-dir",
+        help="Set the directory to store layout files in (default to ephemeral location)",
+        required=False,
+    )
+    return {"visualize": run_fprime_visualize}, {"visualize": viz_parser}

src/fprime/util/cli.py

@@ -18,6 +18,7 @@
 from fprime.util.build_helper import load_build
 from fprime.util.commands import run_code_format, run_hash_to_file, run_info, run_new
 from fprime.util.help_text import HelpText
+from fprime.fpp.visualize import add_fpp_viz_parsers
 
 
 def utility_entry(args):
@@ -301,10 +302,13 @@ def parse_args(args):
         subparsers, common_parser, HelpText
     )
     fpp_runners, fpp_parsers = add_fpp_parsers(subparsers, common_parser)
+    viz_runners, viz_parsers = add_fpp_viz_parsers(subparsers, common_parser)
     parsers.update(fbuild_parsers)
     parsers.update(fpp_parsers)
+    parsers.update(viz_parsers)
     runners.update(fbuild_runners)
     runners.update(fpp_runners)
+    runners.update(viz_runners)
     runners.update(add_special_parsers(subparsers, common_parser, HelpText))
 
     # Parse and prepare to run

src/fprime/util/code_formatter.py

@@ -55,7 +55,7 @@ def __init__(self, executable: str, style_file: "Path", options: Dict):
         self.allowed_extensions = ALLOWED_EXTENSIONS.copy()
         self._files_to_format: List[Path] = []
 
-    def is_supported(self) -> bool:
+    def is_supported(self, _=None, __=None) -> bool:
         return bool(shutil.which(self.executable))
 
     def allow_extension(self, file_ext: str) -> None: