Add --hide-unmonitored flag to ncurses frontend (#20)

Add a --hide-unmonitored flag to the ncurses frontend. This makes it possible for the user to start greenwave monitor with only the topics that are being monitored visible.

Tests were also added to verify the argument parsing is functional with the frontend.

Signed-off-by: Blake McHale <bmchale@nvidia.com>

Author: bmchalenv

greenwave_monitor/CMakeLists.txt

@@ -110,6 +110,12 @@ if(BUILD_TESTING)
     WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
   )
 
+  # Add ncurses frontend tests
+  ament_add_pytest_test(test_ncurses_frontend_argparse test/test_ncurses_frontend_argparse.py
+    TIMEOUT 120
+    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
+  )
+
   # Add gtests
   ament_add_gtest(test_message_diagnostics test/test_message_diagnostics.cpp
     TIMEOUT 60

greenwave_monitor/greenwave_monitor/ncurses_frontend.py

@@ -24,6 +24,7 @@
 with real-time diagnostics including publication rates, latency, and status.
 """
 
+import argparse
 import curses
 import signal
 import threading
@@ -40,7 +41,7 @@
 class GreenwaveNcursesFrontend(Node):
     """Ncurses frontend for Greenwave Monitor."""
 
-    def __init__(self):
+    def __init__(self, hide_unmonitored: bool = False):
         """Initialize the ncurses frontend node."""
         super().__init__('greenwave_ncurses_frontend')
 
@@ -59,7 +60,7 @@ def __init__(self):
         self.input_buffer = ''
         self.status_message = ''
         self.status_timeout = 0
-        self.show_only_monitored = False
+        self.hide_unmonitored = hide_unmonitored
 
         # Initialize UI adaptor
         self.ui_adaptor = GreenwaveUiAdaptor(self)
@@ -98,7 +99,7 @@ def update_visible_topics(self):
         """Update the visible topics list based on current filters."""
         all_topic_names = list(self.all_topics)
 
-        if self.show_only_monitored and self.ui_adaptor:
+        if self.hide_unmonitored and self.ui_adaptor:
             # Filter to only show topics that have diagnostic data (are being monitored)
             filtered_topics = []
             for topic_name in all_topic_names:
@@ -289,10 +290,10 @@ def curses_main(stdscr, node):
                         status_message = f'Error: {msg}'
                     status_timeout = current_time + 3.0
             elif key == ord('h') or key == ord('H'):
-                node.show_only_monitored = not node.show_only_monitored
+                node.hide_unmonitored = not node.hide_unmonitored
                 with node.topics_lock:
                     node.update_visible_topics()
-                mode_text = 'monitored only' if node.show_only_monitored else 'all topics'
+                mode_text = 'monitored only' if node.hide_unmonitored else 'all topics'
                 status_message = f'Showing {mode_text}'
                 status_timeout = current_time + 3.0
 
@@ -434,11 +435,16 @@ def curses_main(stdscr, node):
             status_line = ("Format: Hz [tolerance%] - Examples: '30' (30Hz±5% default) "
                            "or '30 10' (30Hz±10%) - ESC=cancel, Enter=confirm")
         else:
-            mode_text = 'monitored only' if node.show_only_monitored else 'all topics'
+            if node.hide_unmonitored:
+                mode_text = 'monitored only'
+                mode_help_text = 'show unmonitored'
+            else:
+                mode_text = 'all topics'
+                mode_help_text = 'hide unmonitored'
             status_line = (
                 f'Showing {start_idx + 1} - {num_shown} of {len(visible_topics)} '
                 f'topics ({mode_text}). Enter=toggle, f=set freq, c=clear freq, '
-                f'h=hide unmonitored, q=quit')
+                f'h={mode_help_text}, q=quit')
 
         try:
             stdscr.addstr(height - 2, 0, status_line[:width - 1])
@@ -448,10 +454,24 @@ def curses_main(stdscr, node):
         stdscr.refresh()
 
 
+def parse_args(args=None):
+    """Parse command-line arguments."""
+    parser = argparse.ArgumentParser(
+        description='Ncurses-based frontend for Greenwave Monitor'
+    )
+    parser.add_argument(
+        '--hide-unmonitored',
+        action='store_true',
+        help='Hide unmonitored topics on initialization'
+    )
+    return parser.parse_known_args(args)
+
+
 def main(args=None):
     """Entry point for the ncurses frontend application."""
-    rclpy.init(args=args)
-    node = GreenwaveNcursesFrontend()
+    parsed_args, ros_args = parse_args(args)
+    rclpy.init(args=ros_args)
+    node = GreenwaveNcursesFrontend(hide_unmonitored=parsed_args.hide_unmonitored)
     thread = None
 
     def signal_handler(signum, frame):

greenwave_monitor/scripts/ncurses_dashboard

@@ -23,6 +23,7 @@
 # Parse command line arguments
 DEMO_MODE=false
 LOG_DIR=""
+HIDE_UNMONITORED=false
 MONITOR_ARGS=()
 
 show_help() {
@@ -33,15 +34,16 @@ show_help() {
     echo "OPTIONS:"
     echo "  --demo, --test             Launch demo publisher nodes for testing"
     echo "  --log-dir DIR              Enable logging to specified directory"
+    echo "  --hide-unmonitored         Hide unmonitored topics on initialization"
     echo "  --help, -h                 Show this help message"
     echo ""
     echo "MONITOR_ARGS are passed directly to the greenwave_monitor node"
     echo ""
     echo "Controls in ncurses interface:"
-    echo "  a = Add topic to monitoring"
-    echo "  r = Remove selected topic"
-    echo "  f = Set expected frequency (format: topic_name hz tolerance)"
+    echo "  enter/space = toggle topic monitoring"
+    echo "  f = Set expected frequency for selected topic (format: hz tolerance%)"
     echo "  c = Clear frequency settings for selected topic"
+    echo "  h = Toggle hiding unmonitored topics"
     echo "  ↑/↓ = Navigate topics"
     echo "  q = Quit"
 }
@@ -56,6 +58,10 @@ while [[ $# -gt 0 ]]; do
             LOG_DIR="$2"
             shift 2
             ;;
+        --hide-unmonitored)
+            HIDE_UNMONITORED=true
+            shift
+            ;;
         --help|-h)
             show_help
             exit 0
@@ -115,7 +121,12 @@ echo "Monitor process started with PID: $MONITOR_PID"
 # Launch ncurses frontend in the foreground
 echo "Starting ncurses TUI..."
 echo "Controls: a=Add Topic, r=Remove, f=Set Frequency, c=Clear Freq, q=Quit"
-python3 -m greenwave_monitor.ncurses_frontend
+# NOTE: add proper argument parsing to the ncurses frontend if more than one argument is added here
+FRONTEND_ARGS=()
+if [ "$HIDE_UNMONITORED" = "true" ]; then
+    FRONTEND_ARGS+=("--hide-unmonitored")
+fi
+python3 -m greenwave_monitor.ncurses_frontend "${FRONTEND_ARGS[@]}"
 
 # Note: We don't need to explicitly exit here because the trap will handle cleanup
-# when the ncurses frontend exits
+# when the ncurses frontend exits

greenwave_monitor/test/test_ncurses_frontend_argparse.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# 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.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for ncurses frontend argument parsing."""
+
+from greenwave_monitor.ncurses_frontend import parse_args
+
+
+class TestParseArgs:
+    """Test argument parsing for ncurses frontend."""
+
+    def test_default_hide_unmonitored_false(self):
+        """Test that hide_unmonitored defaults to False."""
+        parsed_args, _ = parse_args([])
+        assert parsed_args.hide_unmonitored is False
+
+    def test_hide_unmonitored_long_flag(self):
+        """Test --hide-unmonitored flag enables hide_unmonitored."""
+        parsed_args, _ = parse_args(['--hide-unmonitored'])
+        assert parsed_args.hide_unmonitored is True
+
+    def test_ros_args_passthrough(self):
+        """Test that ROS arguments are passed through."""
+        parsed_args, ros_args = parse_args(
+            ['--hide-unmonitored', '--ros-args', '-r', '__node:=my_node']
+        )
+        assert parsed_args.hide_unmonitored is True
+        assert '--ros-args' in ros_args
+        assert '-r' in ros_args
+        assert '__node:=my_node' in ros_args