Initial setup of fprime-gds to work with OpenMCT

src/fprime_gds/executables/cli.py

@@ -782,3 +782,41 @@ def get_arguments(self) -> Dict[Tuple[str, ...], Dict[str, Any]]:
 
     def handle_arguments(self, args, **kwargs):
         return args
+
+class OpenMCTTelemetryPollerParser(ParserBase):
+    """ Parser for OpenMCT Server arguments  """
+
+    def get_arguments(self) -> Dict[Tuple[str, ...], Dict[str, Any]]:
+        """ Return arguments on whether to use OpenMCT, post to a specific URI, and publish telemetry at a specific frequency"""
+        return {
+            ("--openmct",): {
+                "dest": "openmct",
+                "action": 'store_true',
+                "help": "Determines whethers OpenMCT will be used for Telemetry Visualization",
+                "required": False
+            },
+            ("--openmct-uri",): {
+                "dest": "openmct_uri",
+                "type": str,
+                "default": "http://127.0.0.1:4052/fprime_telem",
+                "help": "URI of the OpenMCT Server. The URI at which the F-Prime telemetry will be broadcasted.",
+                "required": False
+            },
+            ("--openmct-telem-rate",): {
+                "dest": "openmct_telem_rate",
+                "type": float,
+                "default": 1,
+                "help": "Rate(in Hz) at which we want to poll the F-Prime Telemetry Pipeline for new telemetry",
+                "required": False
+            },
+            ("--openmct-dir",): {
+                "dest": "openmct_dir",
+                "type": str, 
+                "default": "",
+                "help": "Directory where the OpenMCT Installation is located",
+                "required": False
+            }
+        }
+
+    def handle_arguments(self, args, **kwargs):
+        return args

src/fprime_gds/executables/run_deployment.py

@@ -13,9 +13,22 @@
     GdsParser,
     ParserBase,
     StandardPipelineParser,
+    OpenMCTTelemetryPollerParser
 )
 from fprime_gds.executables.utils import AppWrapperException, run_wrapped_application
 
+# Try to Import FPrime-OpenMCT Python Packages
+try: 
+    import fprime_openmct
+except ImportError:
+    fprime_openmct = None
+
+if fprime_openmct is not None: 
+    from fprime_openmct.config_server import ServerConfig
+    from fprime_openmct.fprime_to_openmct import TopologyAppDictionaryJSONifier
+
+
+
 BASE_MODULE_ARGUMENTS = [sys.executable, "-u", "-m"]
 
 
@@ -27,7 +40,7 @@ def parse_args():
     :return: parsed argument namespace
     """
     # Get custom handlers for all executables we are running
-    arg_handlers = [StandardPipelineParser, GdsParser, BinaryDeployment, CommParser]
+    arg_handlers = [StandardPipelineParser, GdsParser, BinaryDeployment, CommParser, OpenMCTTelemetryPollerParser]
     # Parse the arguments, and refine through all handlers
     args, parser = ParserBase.parse_args(arg_handlers, "Run F prime deployment and GDS")
     return args
@@ -148,6 +161,32 @@ def launch_comm(parsed_args):
     app_cmd = BASE_MODULE_ARGUMENTS + ["fprime_gds.executables.comm"] + arguments
     return launch_process(app_cmd, name=f'comm[{parsed_args.adapter}] Application', launch_time=1)
 
+def launch_openmct(parsed_args):
+    """ Launch OpenMCT Node Server 
+
+    Args:
+        parsed_args: parsed argument namespace
+    Return:
+        launched process
+    """
+    openmct_server = ServerConfig()
+
+    app_cmd = openmct_server.launch_openmct_server()
+
+    return app_cmd 
+
+def poll_telem(parsed_args):
+    """ Poll and Send Telemetry to OpenMCT 
+
+    Args:
+        parsed_args: parsed argument namespace
+    Return:
+        launched process
+    """
+
+    app_cmd = BASE_MODULE_ARGUMENTS + ["fprime_openmct.fprime_telem_poller"] + StandardPipelineParser().reproduce_cli_args(parsed_args) + OpenMCTTelemetryPollerParser().reproduce_cli_args(parsed_args)
+
+    return launch_process(app_cmd, name='OpenMCT Poller', launch_time=1) 
 
 def main():
     """
@@ -174,6 +213,23 @@ def main():
     # Launch the desired GUI package
     launchers.append(launch_html)
 
+    # Check if OpenMCT is set to be used. If true, generate OpenMCT States and Launch the OpenMCT Server
+    if parsed_args.openmct:
+
+        if fprime_openmct is None:
+            raise ImportError('FPrime-OpenMCT Bridge not installed. Please install fprime_openmct with pip')
+
+        # Try Generating the OpenMCT JSON and Initial States
+        top_dict = TopologyAppDictionaryJSONifier(str(parsed_args.dictionary))
+        top_dict.writeOpenMCTJSON('FPrimeDeploymentTopologyAppDictionary')
+        top_dict.writeInitialStatesJSON('initial_states')
+
+        # Try Launching OpenMCT Server
+        launchers.append(launch_openmct)
+
+        #Try polling for telemetry
+        launchers.append(poll_telem)
+
     # Launch launchers and wait for the last app to finish
     try:
         procs = [launcher(parsed_args) for launcher in launchers]