diff --git a/ros2cli/package.xml b/ros2cli/package.xml index 19a2a10ce..529d641b2 100644 --- a/ros2cli/package.xml +++ b/ros2cli/package.xml @@ -18,6 +18,7 @@ Michael Jeronimo python3-argcomplete + python3-colorama python3-packaging python3-psutil rclpy diff --git a/ros2cli/ros2cli/cli.py b/ros2cli/ros2cli/cli.py index 6926b5392..5d9074537 100644 --- a/ros2cli/ros2cli/cli.py +++ b/ros2cli/ros2cli/cli.py @@ -21,6 +21,7 @@ from rclpy.executors import ExternalShutdownException +from ros2cli.color import ColorState from ros2cli.command import add_subparsers_on_demand @@ -42,6 +43,14 @@ def main(*, script_name='ros2', argv=None, description=None, extension=None): 'Do not force line buffering in stdout and instead use the python default buffering, ' 'which might be affected by PYTHONUNBUFFERED/-u and depends on whatever stdout is ' 'interactive or not')) + parser.add_argument( + '--color', + action='store_true', + default=False, + help=( + 'Enable color output. ' + 'Color can also be enabled persistently by setting the ' + 'ROS2CLI_COLOR_OUTPUT=1 environment variable.')) # add arguments for command extension(s) if extension: @@ -66,6 +75,9 @@ def main(*, script_name='ros2', argv=None, description=None, extension=None): # parse the command line arguments args = parser.parse_args(args=argv) + # apply color setting: --color flag takes precedence over ROS2CLI_COLOR_OUTPUT. + ColorState.set_from_args(args.color) + if not args.use_python_default_buffering: # Make the output always line buffered. # TextIoWrapper has a reconfigure() method, call that if available. diff --git a/ros2cli/ros2cli/color.py b/ros2cli/ros2cli/color.py new file mode 100644 index 000000000..bcebdfe59 --- /dev/null +++ b/ros2cli/ros2cli/color.py @@ -0,0 +1,160 @@ +# Copyright 2026 Peng Wang +# +# 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. + +""" +Color output support for ros2cli. + +Built on `colorama `_ for cross-platform +ANSI support. Color output is **disabled by default**. + +Two ways to enable color: + +* ``--color`` command-line flag — enable color for a single invocation. +* ``ROS2CLI_COLOR_OUTPUT`` environment variable set to a non-empty, non-``0`` + value (e.g. ``ROS2CLI_COLOR_OUTPUT=1``) — enable color for all ros2cli + commands in the current shell session. + +In both cases color is only active when stdout is an interactive terminal +(TTY). Output piped to another command or redirected to a file will never +contain ANSI escape codes. + +The ``--color`` flag takes precedence over ``ROS2CLI_COLOR_OUTPUT``. + +Usage:: + + from ros2cli.color import bold, cyan, yellow, green, red + + print(cyan('/my_node')) + print(green('/chatter') + ': ' + yellow('std_msgs/msg/String')) + print(bold(cyan(args.node_name))) + +All functions are transparent when color is disabled — they return the +input string unchanged, so call-sites never need to branch on color state. +""" + +import os +import sys + +import colorama + +# Enable ANSI escape processing on Windows (no-op on Linux/macOS). +colorama.just_fix_windows_console() + +# colorama does not expose an underline constant; isolate the raw escape here. +_ANSI_UNDERLINE = '\x1b[4m' + + +class ColorState: + """ + Manage the global color-enabled flag for ros2cli. + + Color is **disabled by default**. It can be enabled via the ``--color`` + command-line flag or the ``ROS2CLI_COLOR_OUTPUT`` environment variable. + """ + + _enabled: 'bool | None' = None + + @classmethod + def is_enabled(cls) -> bool: + """Return whether color output is currently enabled.""" + if cls._enabled is None: + cls._enabled = cls._resolve() + return cls._enabled + + @classmethod + def _resolve(cls) -> bool: + """ + Compute the effective color state from the environment. + + Color is enabled when ``ROS2CLI_COLOR_OUTPUT`` is set to a non-empty, + non-``"0"`` value **and** stdout is a TTY. The TTY check prevents + ANSI escape codes from leaking into pipes or redirected files. + """ + val = os.environ.get('ROS2CLI_COLOR_OUTPUT') + if val in (None, '', '0'): + return False + return cls._is_tty() + + @classmethod + def _is_tty(cls) -> bool: + """Return True when stdout is an interactive terminal.""" + try: + return sys.stdout.isatty() + except AttributeError: + return False + + @classmethod + def set_from_args(cls, color: bool) -> None: + """Apply the value of the ``--color`` command-line flag.""" + if color: + cls._enabled = cls._is_tty() + + @classmethod + def reset(cls) -> None: + """Reset to undetermined state. For use in tests only.""" + cls._enabled = None + + +def _c(text: str, *codes: str) -> str: + """Apply colorama *codes* to *text*, or return *text* unchanged.""" + if not ColorState.is_enabled(): + return text + return ''.join(codes) + text + colorama.Style.RESET_ALL + + +def bold(text: str) -> str: + """Apply bold/bright style.""" + return _c(text, colorama.Style.BRIGHT) + + +def dim(text: str) -> str: + """Apply dim/faint style.""" + return _c(text, colorama.Style.DIM) + + +def underline(text: str) -> str: + """Apply underline style.""" + return _c(text, _ANSI_UNDERLINE) + + +def red(text: str) -> str: + return _c(text, colorama.Fore.RED) + + +def green(text: str) -> str: + return _c(text, colorama.Fore.GREEN) + + +def yellow(text: str) -> str: + return _c(text, colorama.Fore.YELLOW) + + +def blue(text: str) -> str: + return _c(text, colorama.Fore.BLUE) + + +def magenta(text: str) -> str: + return _c(text, colorama.Fore.MAGENTA) + + +def cyan(text: str) -> str: + return _c(text, colorama.Fore.CYAN) + + +def white(text: str) -> str: + return _c(text, colorama.Fore.WHITE) + + +def black(text: str) -> str: + return _c(text, colorama.Fore.BLACK) diff --git a/ros2cli/test/test_color.py b/ros2cli/test/test_color.py new file mode 100644 index 000000000..57c7103e9 --- /dev/null +++ b/ros2cli/test/test_color.py @@ -0,0 +1,235 @@ +# Copyright 2026 Peng Wang +# +# 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 os +from unittest.mock import patch + +import colorama +import pytest + +from ros2cli.color import black +from ros2cli.color import blue +from ros2cli.color import bold +from ros2cli.color import ColorState +from ros2cli.color import cyan +from ros2cli.color import dim +from ros2cli.color import green +from ros2cli.color import magenta +from ros2cli.color import red +from ros2cli.color import underline +from ros2cli.color import white +from ros2cli.color import yellow + + +@pytest.fixture(autouse=True) +def reset_color_state(): + """Reset ColorState after every test.""" + yield + ColorState.reset() + + +class TestColorStateDefault: + """Color is disabled by default (no env var, no --color flag).""" + + def test_disabled_by_default(self): + with patch.dict(os.environ, {}, clear=True): + ColorState.reset() + assert ColorState.is_enabled() is False + + def test_disabled_when_env_var_absent(self): + env = {k: v for k, v in os.environ.items() if k != 'ROS2CLI_COLOR_OUTPUT'} + with patch.dict(os.environ, env, clear=True): + ColorState.reset() + assert ColorState.is_enabled() is False + + +class TestRosColorOutput: + """Test ROS2CLI_COLOR_OUTPUT environment variable (requires TTY).""" + + def test_enabled_when_set_to_1_with_tty(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': '1'}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + assert ColorState.is_enabled() is True + + def test_disabled_when_set_to_1_without_tty(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': '1'}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': False}): + ColorState.reset() + assert ColorState.is_enabled() is False + + def test_enabled_for_truthy_values_with_tty(self): + for val in ('1', '2', 'true', 'yes'): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': val}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + assert ColorState.is_enabled() is True, \ + f'ROS2CLI_COLOR_OUTPUT={val!r} with TTY should enable color' # noqa: E501 + + def test_disabled_for_zero(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': '0'}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + assert ColorState.is_enabled() is False + + def test_disabled_for_empty_string(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': ''}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + assert ColorState.is_enabled() is False + + +class TestSetFromArgs: + """Test ColorState.set_from_args() — the --color flag.""" + + def test_flag_true_enables_color_with_tty(self): + with patch.dict(os.environ, {}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + ColorState.set_from_args(True) + assert ColorState.is_enabled() is True + + def test_flag_true_disabled_without_tty(self): + with patch.dict(os.environ, {}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': False}): + ColorState.reset() + ColorState.set_from_args(True) + assert ColorState.is_enabled() is False + + def test_flag_false_falls_back_to_env_enabled_with_tty(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': '1'}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + ColorState.set_from_args(False) + assert ColorState.is_enabled() is True + + def test_flag_false_falls_back_to_env_disabled(self): + with patch.dict(os.environ, {}, clear=True): + ColorState.reset() + ColorState.set_from_args(False) + assert ColorState.is_enabled() is False + + +class TestAtomicEnabled: + """Test each atomic color/style function when color is enabled.""" + + def setup_method(self): + ColorState._enabled = True + + def _check(self, func, *expected_codes, text='hello'): + result = func(text) + for code in expected_codes: + assert code in result + assert text in result + assert result.endswith(colorama.Style.RESET_ALL) + + def test_bold(self): self._check(bold, colorama.Style.BRIGHT) + def test_dim(self): self._check(dim, colorama.Style.DIM) + def test_black(self): self._check(black, colorama.Fore.BLACK) + def test_red(self): self._check(red, colorama.Fore.RED) + def test_green(self): self._check(green, colorama.Fore.GREEN) + def test_yellow(self): self._check(yellow, colorama.Fore.YELLOW) + def test_blue(self): self._check(blue, colorama.Fore.BLUE) + def test_magenta(self): self._check(magenta, colorama.Fore.MAGENTA) + def test_cyan(self): self._check(cyan, colorama.Fore.CYAN) + def test_white(self): self._check(white, colorama.Fore.WHITE) + + def test_underline(self): + result = underline('x') + assert 'x' in result and result.endswith(colorama.Style.RESET_ALL) + + def test_reset_appended_once(self): + result = bold('x') + assert result.endswith(colorama.Style.RESET_ALL) + assert result.count(colorama.Style.RESET_ALL) == 1 + + +class TestAtomicDisabled: + """Test each atomic color/style function is transparent when color is disabled.""" + + def setup_method(self): + ColorState._enabled = False + + def test_all_transparent(self): + text = 'test' + for func in (bold, dim, underline, black, red, green, yellow, blue, + magenta, cyan, white): + result = func(text) + assert result == text, f'{func.__name__}() should be transparent when disabled' + assert '\x1b' not in result + + +class TestNesting: + """Test composition of color functions (nesting).""" + + def setup_method(self): + ColorState._enabled = True + + def test_bold_cyan(self): + result = bold(cyan('/my_node')) + assert colorama.Style.BRIGHT in result + assert colorama.Fore.CYAN in result + assert '/my_node' in result + + def test_nesting_disabled_is_transparent(self): + ColorState._enabled = False + assert bold(cyan('x')) == 'x' + + def test_bold_yellow(self): + result = bold(yellow('WARNING')) + assert colorama.Style.BRIGHT in result + assert colorama.Fore.YELLOW in result + + +class TestEnvVarIntegration: + """Integration tests for env var and --color flag end-to-end.""" + + def test_default_no_color(self): + with patch.dict(os.environ, {}, clear=True): + ColorState.reset() + result = bold(cyan('/my_node')) + assert result == '/my_node' + + def test_ros_color_output_colorizes_with_tty(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': '1'}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + result = bold(cyan('/my_node')) + assert '\x1b' in result + assert '/my_node' in result + + def test_ros_color_output_no_color_without_tty(self): + with patch.dict(os.environ, {'ROS2CLI_COLOR_OUTPUT': '1'}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': False}): + ColorState.reset() + result = bold(cyan('/my_node')) + assert result == '/my_node' + + def test_color_flag_colorizes_with_tty(self): + with patch.dict(os.environ, {}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': True}): + ColorState.reset() + ColorState.set_from_args(True) + result = bold(cyan('/my_node')) + assert '\x1b' in result + assert '/my_node' in result + + def test_color_flag_no_color_without_tty(self): + with patch.dict(os.environ, {}, clear=True): + with patch('sys.stdout', **{'isatty.return_value': False}): + ColorState.reset() + ColorState.set_from_args(True) + result = bold(cyan('/my_node')) + assert result == '/my_node' diff --git a/ros2node/ros2node/verb/info.py b/ros2node/ros2node/verb/info.py index e06250887..8ff4e3b92 100644 --- a/ros2node/ros2node/verb/info.py +++ b/ros2node/ros2node/verb/info.py @@ -14,6 +14,10 @@ import sys +from ros2cli.color import blue +from ros2cli.color import bold +from ros2cli.color import cyan +from ros2cli.color import green from ros2cli.helpers import interactive_select from ros2cli.node.strategy import add_arguments from ros2cli.node.strategy import NodeStrategy @@ -30,7 +34,11 @@ def print_names_and_types(names_and_types): - print(*[2 * ' ' + s.name + ': ' + ', '.join(s.types) for s in names_and_types], sep='\n') + lines = [ + 2 * ' ' + s.name + ': ' + bold(blue(', '.join(s.types))) + for s in names_and_types + ] + print(*lines, sep='\n') class InfoVerb(VerbExtension): @@ -75,30 +83,30 @@ def main(self, *, args): num_nodes=count, node_name=args.node_name), file=sys.stderr) if count > 0: - print(args.node_name) + print(bold(cyan(args.node_name))) subscribers = get_subscriber_info( node=node, remote_node_name=args.node_name, include_hidden=args.include_hidden) - print(' Subscribers:') + print(green(' Subscribers:')) print_names_and_types(subscribers) publishers = get_publisher_info( node=node, remote_node_name=args.node_name, include_hidden=args.include_hidden) - print(' Publishers:') + print(green(' Publishers:')) print_names_and_types(publishers) service_servers = get_service_server_info( node=node, remote_node_name=args.node_name, include_hidden=args.include_hidden) - print(' Service Servers:') + print(green(' Service Servers:')) print_names_and_types(service_servers) service_clients = get_service_client_info( node=node, remote_node_name=args.node_name, include_hidden=args.include_hidden) - print(' Service Clients:') + print(green(' Service Clients:')) print_names_and_types(service_clients) actions_servers = get_action_server_info( node=node, remote_node_name=args.node_name, include_hidden=args.include_hidden) - print(' Action Servers:') + print(green(' Action Servers:')) print_names_and_types(actions_servers) actions_clients = get_action_client_info( node=node, remote_node_name=args.node_name, include_hidden=args.include_hidden) - print(' Action Clients:') + print(green(' Action Clients:')) print_names_and_types(actions_clients) else: return "Unable to find node '" + args.node_name + "'"