Skip to content

Fix errors overlaying on topics in ncurses frontend #35

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sgillen merged 4 commits into from
Feb 27, 2026
Merged

Fix errors overlaying on topics in ncurses frontend #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eaffcb0
Fix ncurses frontend
bmchalenv Feb 27, 2026
4e4e507
Silence stdout for ROS logs in ncurses
bmchalenv Feb 27, 2026
3fa661f
Fix display after setting expected freq
bmchalenv Feb 27, 2026
fe896e8
retrigger checks
bmchalenv Feb 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 18 additions & 9 deletions greenwave_monitor/greenwave_monitor/ncurses_frontend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,13 +113,11 @@ def update_visible_topics(self):

self.visible_topics.sort()

def toggle_topic_monitoring(self, topic_name: str):
def toggle_topic_monitoring(self, topic_name: str) -> tuple[bool, str]:
"""Toggle monitoring for a topic."""
if self.ui_adaptor:
self.ui_adaptor.toggle_topic_monitoring(topic_name)
self.show_status(f'Toggling monitoring for {topic_name}')
else:
self.show_status('UI adaptor not available')
return self.ui_adaptor.toggle_topic_monitoring(topic_name)
return False, 'UI adaptor not available'

def show_status(self, message: str):
"""Show a status message for 3 seconds."""
Expand All @@ -141,6 +139,7 @@ def curses_main(stdscr, node):
last_redraw = 0
status_message = ''
status_timeout = 0
status_is_error = False
input_mode = None
input_buffer = ''

Expand Down Expand Up @@ -231,12 +230,15 @@ def curses_main(stdscr, node):
success, msg = node.ui_adaptor.set_expected_frequency(
topic_name, hz, tolerance)
status_message = f'Set frequency for {topic_name}: {hz}Hz'
status_is_error = not success
if not success:
status_message = f'Error: {msg}'
else:
status_message = 'Invalid input format'
status_is_error = True
except ValueError:
status_message = 'Invalid frequency values'
status_is_error = True
status_timeout = current_time + 3.0
input_mode = None
input_buffer = ''
Expand Down Expand Up @@ -273,19 +275,23 @@ def curses_main(stdscr, node):
elif key == ord('\n') or key == ord(' '):
if 0 <= selected_row < len(node.visible_topics):
topic_name = node.visible_topics[selected_row]
node.toggle_topic_monitoring(topic_name)
status_message = f'Toggled monitoring for {topic_name}'
success, msg = node.toggle_topic_monitoring(topic_name)
status_message = msg if not success else f'Toggled monitoring for {topic_name}'
status_is_error = not success
status_timeout = current_time + 3.0
elif key == ord('f') or key == ord('F'):
if 0 <= selected_row < len(node.visible_topics):
input_mode = 'frequency'
input_buffer = ''
status_timeout = 0
status_is_error = False
elif key == ord('c') or key == ord('C'):
if 0 <= selected_row < len(node.visible_topics):
topic_name = node.visible_topics[selected_row]
success, msg = node.ui_adaptor.set_expected_frequency(
topic_name, clear=True)
status_message = f'Cleared frequency for {topic_name}'
status_is_error = not success
if not success:
status_message = f'Error: {msg}'
status_timeout = current_time + 3.0
Expand All @@ -295,6 +301,7 @@ def curses_main(stdscr, node):
node.update_visible_topics()
mode_text = 'monitored only' if node.hide_unmonitored else 'all topics'
status_message = f'Showing {mode_text}'
status_is_error = False
status_timeout = current_time + 3.0

# Get data safely
Expand Down Expand Up @@ -400,8 +407,9 @@ def curses_main(stdscr, node):
# Status message
if current_time < status_timeout:
try:
stdscr.addstr(height - 3, 0, status_message[:width-1],
curses.color_pair(COLOR_STATUS_MSG))
color = curses.color_pair(COLOR_ERROR) if status_is_error \
else curses.color_pair(COLOR_STATUS_MSG)
stdscr.addstr(height - 3, 0, status_message[:width-1], color)
except curses.error:
pass

Expand Down Expand Up @@ -470,6 +478,7 @@ def parse_args(args=None):
def main(args=None):
"""Entry point for the ncurses frontend application."""
parsed_args, ros_args = parse_args(args)
ros_args.extend(['--ros-args', '--disable-stdout-logs'])
rclpy.init(args=ros_args)
node = GreenwaveNcursesFrontend(hide_unmonitored=parsed_args.hide_unmonitored)
thread = None
Expand Down
32 changes: 17 additions & 15 deletions greenwave_monitor/greenwave_monitor/ui_adaptor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,45 +184,47 @@ def _on_diagnostics(self, msg: DiagnosticArray):
# Skip updating expected_frequencies if values aren't numeric
self.expected_frequencies.pop(topic_name, None)

def toggle_topic_monitoring(self, topic_name: str):
def toggle_topic_monitoring(self, topic_name: str) -> tuple[bool, str]:
"""Toggle monitoring for a topic."""
if not self.manage_topic_client.wait_for_service(timeout_sec=1.0):
return
return False, 'Could not connect to manage_topic service.'

request = ManageTopic.Request()
request.topic_name = topic_name

with self.data_lock:
request.add_topic = topic_name not in self.ui_diagnostics

action = 'start' if request.add_topic else 'stop'

try:
# Use asynchronous service call to prevent deadlock
future = self.manage_topic_client.call_async(request)
rclpy.spin_until_future_complete(self.node, future, timeout_sec=3.0)

if future.result() is None:
action = 'start' if request.add_topic else 'stop'
self.node.get_logger().error(
f'Failed to {action} monitoring: Service call timed out')
return
error_msg = f'Failed to {action} monitoring: Service call timed out'
self.node.get_logger().error(error_msg)
return False, error_msg

response = future.result()

with self.data_lock:
if not response.success:
action = 'start' if request.add_topic else 'stop'
self.node.get_logger().error(
f'Failed to {action} monitoring: {response.message}')
return
if not response.success:
error_msg = f'Failed to {action} monitoring: {response.message}'
self.node.get_logger().error(error_msg)
return False, error_msg

with self.data_lock:
if not request.add_topic and topic_name in self.ui_diagnostics:
del self.ui_diagnostics[topic_name]
if topic_name in self.expected_frequencies:
del self.expected_frequencies[topic_name]

return True, f'Successfully {"started" if request.add_topic else "stopped"} monitoring'

except Exception as e:
action = 'start' if request.add_topic else 'stop'
self.node.get_logger().error(f'Failed to {action} monitoring: {e}')
error_msg = f'Failed to {action} monitoring: {e}'
self.node.get_logger().error(error_msg)
return False, error_msg

def set_expected_frequency(self,
topic_name: str,
Expand Down