diff --git a/README.md b/README.md index 1a25499..f3f1b76 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,30 @@ It also features some quality of life improvements for ROS outputs. ros2 action list

+- ### **```ros2 param list``` coloring** + Node headers are colored with the group badge; parameter names are rendered dim so the structure is easy to scan. `--param-type` type annotations are dimmed automatically. + +

+ros2 param list +

+ +- ### **```ros2 param describe``` coloring** + The group badge and parameter name are highlighted; field labels (`Type:`, `Description:`) are dimmed; section headers like `Constraints:` are bolded so the output structure is immediately readable. + +

+ros2 param describe +

+ +- ### **Parameter change alert** + When a node's parameter changes at runtime — via `ros2 param set` or any parameter service client — an inline notification appears in the launch terminal showing the node, parameter name, and old→new value. Two styles: compact inline and a full-width inverted block that's hard to miss in busy logs. + +

+param change alert inline +

+

+param change alert inverted +

+ - ### **Truly non-invasive** Shell-level pipe; you won't loose autocompletion or aliases for launch files diff --git a/dendROS/dendROS.sh b/dendROS/dendROS.sh index fbac6be..660d615 100644 --- a/dendROS/dendROS.sh +++ b/dendROS/dendROS.sh @@ -77,6 +77,12 @@ ros2() { elif [[ "$1" == "action" && "$2" == "list" ]]; then "$_ROS2_BIN" "$@" | python3 "${_DENDROS_DIR}/dendros_action_list.py" return ${PIPESTATUS[0]} + elif [[ "$1" == "param" && "$2" == "list" ]]; then + "$_ROS2_BIN" "$@" | python3 "${_DENDROS_DIR}/dendros_param_list.py" "${@:3}" + return ${PIPESTATUS[0]} + elif [[ "$1" == "param" && "$2" == "describe" ]]; then + "$_ROS2_BIN" "$@" | python3 "${_DENDROS_DIR}/dendros_param_describe.py" "${@:3}" + return ${PIPESTATUS[0]} else "$_ROS2_BIN" "$@" fi diff --git a/dendROS/dendROS_pipe.py b/dendROS/dendROS_pipe.py index 0fe84a1..9d125bb 100755 --- a/dendROS/dendROS_pipe.py +++ b/dendROS/dendROS_pipe.py @@ -34,6 +34,7 @@ from lib.global_config import load_global_config, get_node_colors_path import lib.crash_alert as ca import lib.traceback_color as tc +import lib.param_watcher as pw _DEBUG = os.environ.get('DENDROS_DEBUG', '') not in ('', '0') @@ -150,6 +151,12 @@ def main(): if config_path: _save_node_colors(color_map, tag_map, style_map) + param_alert = bool(global_cfg.get('param_change_alert', False)) + param_alert_scope = global_cfg.get('param_change_alert_scope', 'tracked') + param_alert_style = global_cfg.get('param_change_alert_style', 'inline') + if param_alert: + pw.setup(color_map, tag_map, param_alert_scope) + show_tag = defaults.get('show_group_tag', True) color_mode = defaults.get('color_mode', 'tag_only') tag_position = defaults.get('tag_position', 'after') @@ -364,6 +371,11 @@ def _iter_stdin(): _emit(_colorize(line)) + if param_alert: + for notif in pw.drain(color_map, tag_map, style_map, tag_style, show_tag, + param_alert_style): + _emit(notif) + if ca._crash_alert_enabled: if new_death: ca.print_alert_banner() diff --git a/dendROS/dendros_config.py b/dendROS/dendros_config.py index 3a8822e..798c0ee 100644 --- a/dendROS/dendros_config.py +++ b/dendROS/dendros_config.py @@ -43,10 +43,13 @@ ("init_color", "Init: color", "cycle", ["palette", "null"]), ("init_color_bold", "Init: bold colors", "cycle", [False, True]), ("init_label", "Init: auto label", "cycle", [False, True]), - ("crash_alert", "Crash alert", "cycle", [False, True]), - ("crash_alert_color", "Alert color", "cycle", ["node", "red"]), - ("crash_alert_interval", "Alert interval (s)", "text", None), - ("traceback_color", "Traceback color", "cycle", ["fancy", "red", "off"]), + ("crash_alert", "Crash alert", "cycle", [False, True]), + ("crash_alert_color", "Alert color", "cycle", ["node", "red"]), + ("crash_alert_interval", "Alert interval (s)", "text", None), + ("traceback_color", "Traceback color", "cycle", ["fancy", "red", "off"]), + ("param_change_alert", "Param change alert", "cycle", [False, True]), + ("param_change_alert_scope", "Param alert scope", "cycle", ["tracked", "all"]), + ("param_change_alert_style", "Param alert style", "cycle", ["inline", "inverted"]), ] _DESCS = { @@ -140,6 +143,18 @@ "fancy — bold red header/exception, dim red frame lines (default)", "red — entire traceback in bold red | off — no coloring (white)", ), + "param_change_alert": ( + "on — print an inline notification in the launch terminal whenever a parameter changes at runtime", + "off — parameter changes are silent; use ros2 param get to check values manually", + ), + "param_change_alert_scope": ( + "tracked — only notify for nodes that have a color group in a dendROS.yaml config", + "all — notify for every node on the ROS graph (includes nodes with no config entry)", + ), + "param_change_alert_style": ( + "inline — compact single line: [dendROS] param [TAG] /node param_name → value", + "inverted — reverse-video block with node color as background; harder to miss", + ), } _VAL_LABEL = {True: "on", False: "off", None: "null"} diff --git a/dendROS/dendros_param_describe.py b/dendROS/dendros_param_describe.py new file mode 100644 index 0000000..5ecc4bf --- /dev/null +++ b/dendROS/dendros_param_describe.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Colorize and format ros2 param describe output using dendROS node colors.""" + +import os +import re +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +try: + import yaml +except ImportError: + yaml = None + +from lib.config_loader import load_config, merge_color_maps, resolve_node, resolve_node_style +from lib.colors import RESET, _resolve_color +from lib.global_config import load_global_config, get_node_colors_path + +_PARAM_NAME_RE = re.compile(r'^(Parameter name):\s+(.*)') + + +def _load_shared_colors(): + if yaml is None: + return {}, {}, {} + path = get_node_colors_path() + if not os.path.isfile(path): + return {}, {}, {} + try: + with open(path) as f: + data = yaml.safe_load(f) or {} + return ( + data.get('color_map', {}), + data.get('tag_map', {}), + data.get('style_map', {}), + ) + except Exception: + return {}, {}, {} + + +def _scan_configs(): + if yaml is None: + return [] + seen, paths = set(), [] + for prefix in os.environ.get('AMENT_PREFIX_PATH', '').split(':'): + share = os.path.join(prefix, 'share') if prefix else '' + if not share or not os.path.isdir(share): + continue + try: + for pkg in sorted(os.listdir(share)): + candidate = os.path.join(share, pkg, 'config', 'dendROS.yaml') + if os.path.isfile(candidate) and candidate not in seen: + seen.add(candidate) + paths.append(candidate) + except OSError: + pass + return paths + + +def _badge(label, ansi_code, style): + if style == 'inverted': + return f'\033[{ansi_code};7m[{label}]{RESET}' + return f'\033[{ansi_code}m[{label}]{RESET}' + + +def _parse_node_arg(argv): + """Return the first positional argument that looks like a node path (starts with '/').""" + for arg in argv: + if arg.startswith('/'): + return arg + return None + + +def _format_key_value(indent, key, value): + """Dim the key label; bold the section header when there is no value.""" + if value: + return f'{indent}\033[2m{key}:\033[0m {value}' + return f'{indent}\033[1m{key}:\033[0m' + + +def _colorize_line(raw, ansi_code, label, node_style, + unmatched_ansi, unmatched_tag, tag_style, + show_tag, dim_unmatched): + """Return a formatted version of one output line.""" + if not raw.strip(): + return raw + + # ── Parameter name block header ────────────────────────────────────────── + m = _PARAM_NAME_RE.match(raw) + if m: + param_name = m.group(2).strip() + dim_label = f'\033[2mParameter name:\033[0m' + + if ansi_code: + colored = f'\033[{ansi_code};1m{param_name}\033[0m' + if show_tag and label: + return f'{_badge(label, ansi_code, node_style)} {dim_label} {colored}' + return f'{dim_label} {colored}' + + if unmatched_ansi: + colored = f'\033[{unmatched_ansi};1m{param_name}\033[0m' + if show_tag and unmatched_tag: + return f'{_badge(unmatched_tag, unmatched_ansi, tag_style)} {dim_label} {colored}' + return f'{dim_label} {colored}' + + if dim_unmatched: + return f'{dim_label} \033[2m{param_name}\033[0m' + + return f'{dim_label} {param_name}' + + # ── Key: value lines (all indented fields) ──────────────────────────────── + stripped = raw.lstrip() + indent = raw[:len(raw) - len(stripped)] + + if ':' in stripped: + key, _, rest = stripped.partition(':') + value = rest.lstrip() + return _format_key_value(indent, key, value) + + return raw + + +def main(): + cfg = load_global_config() + show_tag = cfg['show_tag_cli'] + tag_style = cfg['tag_style'] + unmatched_clr = cfg['unmatched_color'] + unmatched_tag = cfg['unmatched_tag'] + dim_unmatched = cfg['dim_unmatched'] + unmatched_ansi = _resolve_color(unmatched_clr) if unmatched_clr else None + + color_map, tag_map, style_map = _load_shared_colors() + + if not color_map: + config_paths = _scan_configs() + tuples = [] + for path in config_paths: + try: + c, t, m, s, k, _ = load_config(path) + tuples.append((c, t, m, s, k)) + except Exception: + pass + if tuples: + c0, t0, m0, s0, k0 = tuples[0] + color_map, tag_map, _, style_map, _ = merge_color_maps( + c0, t0, m0, s0, k0, tuples[1:] + ) + + node_arg = _parse_node_arg(sys.argv[1:]) + if node_arg: + ansi_code, label = resolve_node(node_arg, color_map, tag_map) + node_style = (resolve_node_style(node_arg, style_map) or tag_style) if ansi_code else tag_style + else: + ansi_code, label, node_style = None, None, tag_style + + for line in sys.stdin: + raw = line.rstrip('\n') + out = _colorize_line( + raw, ansi_code, label, node_style, + unmatched_ansi, unmatched_tag, tag_style, + show_tag, dim_unmatched, + ) + sys.stdout.write(out + '\n') + sys.stdout.flush() + + +if __name__ == '__main__': + main() diff --git a/dendROS/dendros_param_list.py b/dendROS/dendros_param_list.py new file mode 100644 index 0000000..cd5c4d7 --- /dev/null +++ b/dendROS/dendros_param_list.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +"""Colorize ros2 param list output using dendROS node colors.""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +try: + import yaml +except ImportError: + yaml = None + +from lib.config_loader import load_config, merge_color_maps, resolve_node, resolve_node_style +from lib.colors import RESET, _resolve_color +from lib.global_config import load_global_config, get_node_colors_path + + +def _load_shared_colors(): + if yaml is None: + return {}, {}, {} + path = get_node_colors_path() + if not os.path.isfile(path): + return {}, {}, {} + try: + with open(path) as f: + data = yaml.safe_load(f) or {} + return ( + data.get('color_map', {}), + data.get('tag_map', {}), + data.get('style_map', {}), + ) + except Exception: + return {}, {}, {} + + +def _scan_configs(): + if yaml is None: + return [] + seen, paths = set(), [] + for prefix in os.environ.get('AMENT_PREFIX_PATH', '').split(':'): + share = os.path.join(prefix, 'share') if prefix else '' + if not share or not os.path.isdir(share): + continue + try: + for pkg in sorted(os.listdir(share)): + candidate = os.path.join(share, pkg, 'config', 'dendROS.yaml') + if os.path.isfile(candidate) and candidate not in seen: + seen.add(candidate) + paths.append(candidate) + except OSError: + pass + return paths + + +def _badge(label, ansi_code, style): + if style == 'inverted': + return f'\033[{ansi_code};7m[{label}]{RESET}' + return f'\033[{ansi_code}m[{label}]{RESET}' + + +def _parse_node_arg(argv): + """Return the first positional argument that looks like a node path (starts with '/').""" + for arg in argv: + if arg.startswith('/'): + return arg + return None + + +def _split_param_type(param): + """Split 'name (type)' → (name, type_str) or (name, None).""" + if param.endswith(')') and ' (' in param: + name, rest = param.rsplit(' (', 1) + return name, rest[:-1] + return param, None + + +def main(): + cfg = load_global_config() + show_tag = cfg['show_tag_cli'] + tag_style = cfg['tag_style'] + unmatched_clr = cfg['unmatched_color'] + unmatched_tag = cfg['unmatched_tag'] + dim_unmatched = cfg['dim_unmatched'] + unmatched_ansi = _resolve_color(unmatched_clr) if unmatched_clr else None + + color_map, tag_map, style_map = _load_shared_colors() + + if not color_map: + config_paths = _scan_configs() + tuples = [] + for path in config_paths: + try: + c, t, m, s, k, _ = load_config(path) + tuples.append((c, t, m, s, k)) + except Exception: + pass + if tuples: + c0, t0, m0, s0, k0 = tuples[0] + color_map, tag_map, _, style_map, _ = merge_color_maps( + c0, t0, m0, s0, k0, tuples[1:] + ) + + current_ansi = None # color code of the most recent node header + current_style = None # tag_style of the most recent node header + + # Pre-seed color from the node argument so bare output (no header line) is colored. + # When a node header IS in the output the header parser will simply overwrite this. + node_arg = _parse_node_arg(sys.argv[1:]) + if node_arg: + _ansi, _label = resolve_node(node_arg, color_map, tag_map) + if _ansi: + current_ansi = _ansi + current_style = resolve_node_style(node_arg, style_map) or tag_style + elif unmatched_ansi: + current_ansi = unmatched_ansi + current_style = tag_style + + for line in sys.stdin: + raw = line.rstrip('\n') + stripped = raw.lstrip() + + if not raw: + sys.stdout.write('\n') + continue + + indent = raw[: len(raw) - len(stripped)] + + # Node header: '/node_name:' — no leading whitespace, ends with ':' + if not indent and stripped.endswith(':'): + node_name = stripped[:-1] # drop the trailing ':' + ansi_code, label = resolve_node(node_name, color_map, tag_map) + + if ansi_code: + current_ansi = ansi_code + node_style = resolve_node_style(node_name, style_map) or tag_style + current_style = node_style + colored_name = f'\033[{ansi_code}m{node_name}:{RESET}' + if show_tag and label: + badge = _badge(label, ansi_code, node_style) + out = f'{badge} {colored_name}' + else: + out = colored_name + elif unmatched_ansi: + current_ansi = unmatched_ansi + current_style = tag_style + colored_name = f'\033[{unmatched_ansi}m{node_name}:{RESET}' + if show_tag and unmatched_tag: + badge = _badge(unmatched_tag, unmatched_ansi, tag_style) + out = f'{badge} {colored_name}' + else: + out = colored_name + elif dim_unmatched: + current_ansi = None + current_style = None + out = f'\033[2m{node_name}:{RESET}' + else: + current_ansi = None + current_style = None + out = raw + + sys.stdout.write(out + '\n') + sys.stdout.flush() + continue + + # Param line: indented under a node header + param_name, type_str = _split_param_type(stripped) + type_part = f' (\033[2m{type_str}{RESET})' if type_str else '' + + if current_ansi: + out = f'{indent}\033[{current_ansi}m\033[2m{param_name}{RESET}{type_part}' + elif dim_unmatched: + out = f'{indent}\033[2m{param_name}{RESET}{type_part}' + else: + out = f'{indent}{param_name}{type_part}' + + sys.stdout.write(out + '\n') + sys.stdout.flush() + + +if __name__ == '__main__': + main() diff --git a/dendROS/lib/global_config.py b/dendROS/lib/global_config.py index c12f432..3000924 100644 --- a/dendROS/lib/global_config.py +++ b/dendROS/lib/global_config.py @@ -27,12 +27,15 @@ "init_color": "palette", "init_color_bold": False, "init_label": False, - "crash_alert": True, - "crash_alert_color": "node", - "crash_alert_interval": 30, - "traceback_color": "fancy", - "tag_style": "normal", - "show_default_services": True, + "crash_alert": True, + "crash_alert_color": "node", + "crash_alert_interval": 30, + "traceback_color": "fancy", + "tag_style": "normal", + "show_default_services": True, + "param_change_alert": True, + "param_change_alert_scope": "tracked", + "param_change_alert_style": "inline", } @@ -72,4 +75,4 @@ def save_global_config(cfg): path = get_global_config_path() os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as f: - yaml.dump({k: cfg[k] for k in DEFAULTS}, f, default_flow_style=False) + yaml.dump({k: cfg.get(k, DEFAULTS[k]) for k in DEFAULTS}, f, default_flow_style=False) diff --git a/dendROS/lib/param_watcher.py b/dendROS/lib/param_watcher.py new file mode 100644 index 0000000..27fb528 --- /dev/null +++ b/dendROS/lib/param_watcher.py @@ -0,0 +1,274 @@ +"""Background watcher for /parameter_events — queues param change notifications. + +Architecture +------------ +A single daemon thread runs `ros2 topic echo /parameter_events` as a subprocess +and parses its YAML output (message chunks separated by `---` lines). Each +chunk that represents a genuine runtime change (changed_parameters non-empty, +not a CLI daemon node) is pushed to `_queue`. + +The main pipe thread calls `drain()` between stdin lines to pop all pending +events and get back formatted notification strings. No locking is needed +because only the daemon thread writes to _queue and only the main thread reads +from it (queue.Queue is already thread-safe). +""" + +import atexit +import os +import queue +import shutil +import subprocess +import threading + +try: + import yaml +except ImportError: + yaml = None + +_RESET = '\033[0m' +# White-bg + black-text strip used in the inverted alert style +_WB = '\033[107;30m' + +# [dendROS] header matches the logo title: "[dend" in logo-blue, "ROS]" in logo-orange +_DENDROS = '\033[38;2;0;75;107;1m[dend\033[38;2;224;127;0;1mROS]\033[0m' + +# Inverted-style header: logo colors as BACKGROUNDS, black text (hollow/cutout letters). +# \033[30m sets black fg once; subsequent bg changes leave fg unchanged. +_DENDROS_INV = ( + '\033[48;2;0;75;107;1m\033[30m[dend' # logo-blue bg + bold + black text + '\033[48;2;224;127;0;1mROS]' # logo-orange bg + bold (fg stays black) + '\033[0m\033[107;30m' # reset → white bg + black text for rest +) + + +def _fg_to_bg(code: str) -> str: + """Convert a foreground ANSI SGR code string to its background equivalent. + + Standard 30-37 → 40-47, bright 90-97 → 100-107, 24-bit 38;2;R;G;B → 48;2;R;G;B. + Bold and other modifiers are dropped so the caller can add its own text attributes. + """ + parts = code.split(';') + out = [] + i = 0 + while i < len(parts): + p = parts[i] + n = int(p) if p.isdigit() else -1 + if 30 <= n <= 37: + out.append(str(n + 10)); i += 1 + elif 90 <= n <= 97: + out.append(str(n + 10)); i += 1 + elif n == 38 and i + 1 < len(parts): + if parts[i + 1] == '2' and i + 4 < len(parts): + out += ['48', '2'] + parts[i + 2:i + 5]; i += 5 + elif parts[i + 1] == '5' and i + 2 < len(parts): + out += ['48', '5', parts[i + 2]]; i += 3 + else: + i += 1 + else: + i += 1 # drop bold and other non-color modifiers + return ';'.join(out) if out else '0' + +_queue = queue.Queue() +_proc: 'subprocess.Popen | None' = None +_param_cache: dict = {} # (node, param_name) → last-seen value string + +# ParameterType enum → value field name in the YAML message +_TYPE_FIELDS = { + 1: 'bool_value', + 2: 'integer_value', + 3: 'double_value', + 4: 'string_value', + 5: 'byte_array_value', + 6: 'bool_array_value', + 7: 'integer_array_value', + 8: 'double_array_value', + 9: 'string_array_value', +} +_MAX_VALUE_LEN = 60 + + +def _find_ros2(): + b = shutil.which('ros2') + if b: + return b + distro = os.environ.get('ROS_DISTRO', '') + if distro: + c = f'/opt/ros/{distro}/bin/ros2' + if os.path.isfile(c): + return c + return None + + +def _extract_value(v): + """Return a human-readable string for a ParameterValue dict.""" + if not isinstance(v, dict): + return str(v) if v is not None else '' + field = _TYPE_FIELDS.get(v.get('type', 0)) + if not field: + return '' + val = v.get(field, '') + if isinstance(val, bool): + return 'true' if val else 'false' + if isinstance(val, list): + s = str(val) + return (s[:_MAX_VALUE_LEN] + '…') if len(s) > _MAX_VALUE_LEN else s + return str(val) + + +def _process_chunk(lines, color_map, tag_map, scope, resolve_node_fn): + """Parse one YAML message chunk; enqueue any changed-parameter events.""" + try: + msg = yaml.safe_load('\n'.join(lines)) + except Exception: + return + if not isinstance(msg, dict): + return + + node = msg.get('node', '') + # /_ros2cli_XXXXX nodes are transient CLI tool nodes — always skip + if not node or node.startswith('/_ros2cli_'): + return + + changed = msg.get('changed_parameters') or [] + if not changed: + return + + if scope == 'tracked': + code, _ = resolve_node_fn(node, color_map, tag_map) + if not code: + return + + for param in changed: + if not isinstance(param, dict): + continue + name = param.get('name', '') + if not name: + continue + new_val = _extract_value(param.get('value', {})) + cache_key = (node, name) + old_val = _param_cache.get(cache_key) # None on first ever change + _param_cache[cache_key] = new_val + _queue.put((node, name, old_val, new_val)) + + +def _watch(color_map, tag_map, scope): + """Background thread: stream /parameter_events and fill the queue.""" + global _proc + ros2 = _find_ros2() + if not ros2: + return + + try: + _proc = subprocess.Popen( + [ros2, 'topic', 'echo', '/parameter_events'], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + ) + except Exception: + return + + from lib.config_loader import resolve_node # deferred — avoids import cycles + + chunk: list[str] = [] + try: + for raw in _proc.stdout: + line = raw.rstrip('\n') + if line == '---': + if chunk: + _process_chunk(chunk, color_map, tag_map, scope, resolve_node) + chunk = [] + else: + chunk.append(line) + except Exception: + pass + + +@atexit.register +def _cleanup(): + if _proc is not None and _proc.poll() is None: + try: + _proc.terminate() + except Exception: + pass + + +def setup(color_map, tag_map, scope: str): + """Start the background watcher. No-op when yaml or ros2 binary is unavailable.""" + if yaml is None or not _find_ros2(): + return + t = threading.Thread(target=_watch, args=(color_map, tag_map, scope), daemon=True) + t.start() + + +def _fmt_inline(node, name, old_val, value, code, label, node_style, show_tag): + """Rainbow header + plain 'param' + node color identity + bold param data.""" + old_part = f'{old_val} → ' if old_val is not None else '? → ' + if code: + node_str = f'\033[{code}m{node}{_RESET}' + if show_tag and label: + badge = (f'\033[{code};7m[{label}]{_RESET}' + if node_style == 'inverted' + else f'\033[{code}m[{label}]{_RESET}') + node_str = f'{badge} {node_str}' + else: + node_str = node # untracked in "all" mode + + return ( + f'{_DENDROS} param' + f' {node_str} \033[1m{name}\033[0m: {old_part}{value}\n' + ) + + +def _fmt_inverted(node, name, old_val, value, code, label, node_style, show_tag): + """[dendROS] header + continuous white-bg strip from 'param' to EOL. + + Layout: + [blue/orange][dend|ROS][reset][WB] param [node-bg;black][TAG] /node[WB] [bold]name[/bold]: old→new [K][reset] + + The white-bg (_WB) starts immediately after [dendROS] and is never interrupted + by a bare reset — the node-identity island switches to its explicit bg and then + returns to _WB, so every space in the line (including between sections) is white. + \033[K fills the remainder of the console line with the white bg. + """ + old_part = f'{old_val} → ' if old_val is not None else '? → ' + + if code: + node_bg = _fg_to_bg(code) + if show_tag and label: + node_section = f'\033[{node_bg};30m[{label}] {node}{_WB}' + else: + node_section = f'\033[{node_bg};30m{node}{_WB}' + else: + node_section = node # untracked — plain black text on the white bg + + return ( + f'{_DENDROS_INV} param {node_section}' + f' \033[1m{name}\033[22m: {old_part}{value} \033[K{_RESET}\n' + ) + + +def drain(color_map, tag_map, style_map, tag_style: str, show_tag: bool, + alert_style: str = 'inline') -> list[str]: + """Pop all pending change events; return formatted notification lines (with \\n). + + Must be called from the main thread only. + """ + from lib.config_loader import resolve_node, resolve_node_style + + lines: list[str] = [] + while True: + try: + node, name, old_val, value = _queue.get_nowait() + except queue.Empty: + break + + code, label = resolve_node(node, color_map, tag_map) + node_style = (resolve_node_style(node, style_map) or tag_style) if code else tag_style + + if alert_style == 'inverted': + lines.append(_fmt_inverted(node, name, old_val, value, code, label, node_style, show_tag)) + else: + lines.append(_fmt_inline(node, name, old_val, value, code, label, node_style, show_tag)) + + return lines diff --git a/docs/assets/images/screenshots/param_alert_inline.png b/docs/assets/images/screenshots/param_alert_inline.png new file mode 100644 index 0000000..ee49d54 Binary files /dev/null and b/docs/assets/images/screenshots/param_alert_inline.png differ diff --git a/docs/assets/images/screenshots/param_alert_inverted.png b/docs/assets/images/screenshots/param_alert_inverted.png new file mode 100644 index 0000000..7c49bcc Binary files /dev/null and b/docs/assets/images/screenshots/param_alert_inverted.png differ diff --git a/docs/assets/images/screenshots/param_describe.png b/docs/assets/images/screenshots/param_describe.png new file mode 100644 index 0000000..3cb91a9 Binary files /dev/null and b/docs/assets/images/screenshots/param_describe.png differ diff --git a/docs/assets/images/screenshots/param_list.png b/docs/assets/images/screenshots/param_list.png new file mode 100644 index 0000000..6fc5989 Binary files /dev/null and b/docs/assets/images/screenshots/param_list.png differ diff --git a/docs/global-config.md b/docs/global-config.md index 5ec72ec..2047b89 100644 --- a/docs/global-config.md +++ b/docs/global-config.md @@ -93,6 +93,14 @@ Settings are written to `~/.config/dendROS/defaults.yaml` and apply across all p |---|---|---| | **Traceback color** | `fancy` / `red` / `off` | `fancy` = bold red header + dim red frames; `red` = all bold red; `off` = passthrough. See [Traceback Highlighting](traceback-highlighting.md). | +### Parameter change alert + +| Setting | Values | Description | +|---|---|---| +| **Param change alert** | `on` / `off` | Print an inline notification whenever a node's parameter changes at runtime. See [Parameter Change Alert](param-change-alert.md). | +| **Param alert scope** | `tracked` / `all` | `tracked` = only nodes with a config group; `all` = entire ROS graph. | +| **Param alert style** | `inline` / `inverted` | `inline` = compact colored line; `inverted` = full white-background strip, harder to miss in busy logs. | + ### Init defaults | Setting | Values | Description | @@ -121,10 +129,13 @@ show_default_services: true debug: false config_merge: true colorize_launch_msgs: true -crash_alert: false +crash_alert: true crash_alert_color: node crash_alert_interval: 30 traceback_color: fancy +param_change_alert: true +param_change_alert_scope: tracked +param_change_alert_style: inline init_modify_build: true init_on_existing: abort init_color: palette diff --git a/docs/index.md b/docs/index.md index 0c08308..a4f2ce4 100644 --- a/docs/index.md +++ b/docs/index.md @@ -87,6 +87,21 @@ DendROS shadows the `ros2` command with a shell function. When you run `ros2 lau

Action list output colored by owning node group, with the same badge and style options available in all other CLI commands.

+
📝
+ros2 param list colors +

Node headers colored with group badges; parameter names dimmed. Type annotations from --param-type rendered dim automatically.

+
+
+
🔬
+ros2 param describe colors +

Badge and bold colored param name at the top; field labels dimmed; section headers like Constraints: bolded for instant readability.

+
+
+
🔔
+Parameter change alert +

Inline notification whenever a node's parameter changes at runtime — shows node, param name, and old→new value directly in the launch log.

+
+
🚫
Truly non-invasive

No launch file changes. Exit codes preserved. DENDROS_DISABLE=1 bypasses everything instantly.

diff --git a/docs/param-change-alert.md b/docs/param-change-alert.md new file mode 100644 index 0000000..8227b0d --- /dev/null +++ b/docs/param-change-alert.md @@ -0,0 +1,93 @@ +# Parameter Change Alert + +When a node's parameter is changed at runtime via `ros2 param set` or any ROS 2 parameter service client, DendROS prints an inline notification in the launch terminal — so you always know when and what changed without leaving the log view. + +Enabled by default. Disable with: + +```bash +dendros config # → "Param change alert" → off +``` + +or in `~/.config/dendROS/defaults.yaml`: + +```yaml +param_change_alert: false +``` + +--- + +## What it looks like + +### Inline + +
+
+
+
+
+
+
+
ros2 launch my_pkg bringup.launch.py
+
+
+

+Inline parameter change alert +

+
+
+ +### Inverted + +
+
+
+
+
+
+
+
ros2 launch my_pkg bringup.launch.py
+
+
+

+Inverted parameter change alert +

+
+
+ +--- + +## Scope + +| `param_change_alert_scope` | Behavior | +|---|---| +| `tracked` (default) | Only nodes that appear in a `dendROS.yaml` group generate notifications | +| `all` | Every node on the ROS graph generates notifications, including unmatched nodes | + +--- + +## Alert styles + +| `param_change_alert_style` | Appearance | +|---|---| +| `inline` (default) | Compact single line: `[dendROS]` tag + colored node identity + bold param name | +| `inverted` | Full white-background strip from `[dendROS]` to end of line; node identity shown as a colored background island; harder to miss in busy logs | + +--- + +## Configuring via `dendros config` + +| TUI field | Values | Description | +|---|---|---| +| **Param change alert** | `on` / `off` | Enable or disable the feature. | +| **Param alert scope** | `tracked` / `all` | Which nodes trigger notifications. | +| **Param alert style** | `inline` / `inverted` | Visual style of the notification line. | + +--- + +## Notes + +- Notifications are delayed until the next log line from the launch process arrives (drain is called after each line). If the process is silent, notifications queue up and appear on the next output. +- Transient CLI daemon nodes (`/_ros2cli_*`) are always filtered out regardless of scope. +- Startup parameter declarations (`new_parameters`) are ignored — only `changed_parameters` events generate alerts. +- The background thread terminates automatically when the launch process exits. +- Covers: `ros2 param set`, programmatic `node->set_parameters()`, and any parameter service client that goes through the ROS 2 parameter service. diff --git a/docs/param-describe.md b/docs/param-describe.md new file mode 100644 index 0000000..26bd981 --- /dev/null +++ b/docs/param-describe.md @@ -0,0 +1,37 @@ +# ros2 param describe Colorization + +When you run `ros2 param describe /node param_name`, DendROS colorizes the output: the group badge and param name are highlighted, labels are dimmed so they recede, and section headers like `Constraints:` are rendered bold so the structure is immediately readable. + +--- + +## What it looks like + +
+
+
+
+
+
+
+
ros2 param describe /battery_monitor start_type_description_service
+
+
+

+ros2 param describe colorized output +

+
+
+ +--- + + +## Badge and style options + +| Setting | Effect on param describe | +|---|---| +| `show_tag_cli: true` | Badge shown to the left of the `Parameter name:` line | +| `tag_style: inverted` | Badge rendered with colored background | +| `unmatched_color` | Unmatched node uses the fallback color for the param name | +| `dim_unmatched` | Unmatched param name dimmed (only when `unmatched_color: null`) | + +--- diff --git a/docs/param-list.md b/docs/param-list.md new file mode 100644 index 0000000..73746c9 --- /dev/null +++ b/docs/param-list.md @@ -0,0 +1,40 @@ +# ros2 param list Colorization + +When you run `ros2 param list`, DendROS automatically colorizes node headers with your group colors and badges, and dims the parameter names so the structure is easier to scan at a glance. The `--param-type` flag is fully supported. + +--- + +## What it looks like + +
+
+
+
+
+
+
+
ros2 param list --param-type
+
+
+

+ros2 param list colorized output +

+
+
+ +--- + + +## Badge and style options + +| Setting | Effect on param list | +|---|---| +| `show_tag_cli: true` | Badge shown to the left of the node header | +| `tag_style: inverted` | Badge rendered with colored background | +| Per-group `show_tag: false` | Badge suppressed for that group only | +| `unmatched_color` | Unmatched node headers shown in the fallback color | +| `unmatched_tag` | Badge shown next to unmatched headers (requires `unmatched_color`) | +| `dim_unmatched` | Unmatched headers dimmed (only when `unmatched_color: null`) | + +--- + diff --git a/docs/reference.md b/docs/reference.md index ef25e2c..6a073b2 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -76,6 +76,9 @@ Stored in `~/.config/dendROS/defaults.yaml`, managed via `dendros config`: | `crash_alert_color` | `node` | `node` = use group color; `red` = always bold red. | | `crash_alert_interval` | `30` | Seconds between periodic banner reprints. `0` = only on new crashes. | | `traceback_color` | `fancy` | `fancy` = bold red header + dim red frames; `red` = all bold red; `off` = passthrough. | +| `param_change_alert` | `true` | Print inline notification on runtime parameter changes. | +| `param_change_alert_scope` | `tracked` | `tracked` = only config-listed nodes; `all` = entire graph. | +| `param_change_alert_style` | `inline` | `inline` = compact line; `inverted` = white-background strip. | | `init_modify_build` | `true` | `dendros init`: patch build files. | | `init_on_existing` | `abort` | `dendros init`: `abort`, `merge`, or `overwrite`. | | `init_color` | `palette` | `dendros init`: `palette` or `null`. | @@ -96,6 +99,8 @@ The `ros2()` shell wrapper intercepts specific subcommands; everything else call | `ros2 node info …` | Output piped through `dendros_node_info.py` — node name, sections, and entries colorized by group. See [ros2 node info](node-info.md). | | `ros2 service list` | Output piped through `dendros_service_list.py` — services colored by owning node; default system services dimmed. See [ros2 service list](service-list.md). | | `ros2 action list` | Output piped through `dendros_action_list.py` — actions colored by owning node. See [ros2 action list](action-list.md). | +| `ros2 param list` | Output piped through `dendros_param_list.py` — node headers colored, param lines dimmed. See [ros2 param list](param-list.md). | +| `ros2 param describe` | Output piped through `dendros_param_describe.py` — badge + colored param name, dim labels, bold section headers. See [ros2 param describe](param-describe.md). | | Everything else | Passed directly to the real `ros2` binary, untouched. | --- diff --git a/install.sh b/install.sh index d1c22cc..eeac30b 100755 --- a/install.sh +++ b/install.sh @@ -52,14 +52,19 @@ $SUDO cp "${SCRIPT_DIR}/dendROS/dendros_node_info.py" "$INSTALL_DIR/" $SUDO cp "${SCRIPT_DIR}/dendROS/dendros_node_list.py" "$INSTALL_DIR/" $SUDO cp "${SCRIPT_DIR}/dendROS/dendros_service_list.py" "$INSTALL_DIR/" $SUDO cp "${SCRIPT_DIR}/dendROS/dendros_action_list.py" "$INSTALL_DIR/" +$SUDO cp "${SCRIPT_DIR}/dendROS/dendros_param_list.py" "$INSTALL_DIR/" +$SUDO cp "${SCRIPT_DIR}/dendROS/dendros_param_describe.py" "$INSTALL_DIR/" $SUDO cp "${SCRIPT_DIR}/dendROS/dendROS.sh" "$INSTALL_DIR/" $SUDO cp -r "${SCRIPT_DIR}/dendROS/lib" "$INSTALL_DIR/" $SUDO chmod +x "$INSTALL_DIR/dendROS_pipe.py" $SUDO chmod +x "$INSTALL_DIR/dendros_config.py" $SUDO chmod +x "$INSTALL_DIR/dendros_init.py" $SUDO chmod +x "$INSTALL_DIR/dendros_node_list.py" +$SUDO chmod +x "$INSTALL_DIR/dendros_node_info.py" $SUDO chmod +x "$INSTALL_DIR/dendros_service_list.py" $SUDO chmod +x "$INSTALL_DIR/dendros_action_list.py" +$SUDO chmod +x "$INSTALL_DIR/dendros_param_list.py" +$SUDO chmod +x "$INSTALL_DIR/dendros_param_describe.py" $SUDO chmod 644 "$INSTALL_DIR/dendROS.sh" echo -e "${GREEN}[DendROS] Files installed to ${INSTALL_DIR}${RESET}" diff --git a/mkdocs.yml b/mkdocs.yml index d7bad7e..2874ae2 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -75,10 +75,13 @@ nav: - Features: - Crash Alert: crash-alert.md - Traceback Highlighting: traceback-highlighting.md + - Parameter Change Alert: param-change-alert.md - ros2 node list: node-list.md - ros2 node info: node-info.md - ros2 service list: service-list.md - ros2 action list: action-list.md + - ros2 param list: param-list.md + - ros2 param describe: param-describe.md - Tools: - dendros init: dendros-init.md - dendros config: global-config.md diff --git a/test/unit/conftest.py b/test/unit/conftest.py index 15ac633..c84fc03 100644 --- a/test/unit/conftest.py +++ b/test/unit/conftest.py @@ -23,11 +23,13 @@ # ── ANSI helpers ───────────────────────────────────────────────────────────── ANSI_RE = re.compile(r'\033\[([0-9;]*)m') +# Broader regex that strips any CSI sequence (e.g. \033[K for erase-to-EOL) +_FULL_ANSI_RE = re.compile(r'\033\[[^a-zA-Z]*[a-zA-Z]') def strip_ansi(s: str) -> str: """Remove all ANSI escape sequences from s.""" - return ANSI_RE.sub('', s) + return _FULL_ANSI_RE.sub('', s) def ansi_codes(s: str) -> list: diff --git a/test/unit/test_global_config.py b/test/unit/test_global_config.py index c63df2f..e4e4f70 100644 --- a/test/unit/test_global_config.py +++ b/test/unit/test_global_config.py @@ -90,6 +90,9 @@ def test_loads_existing_file(self, tmp_config): "traceback_color": "red", "tag_style": "inverted", "show_default_services": False, + "param_change_alert": True, + "param_change_alert_scope": "all", + "param_change_alert_style": "inverted", } with open(tmp_config, "w") as f: yaml.dump(data, f) @@ -172,6 +175,9 @@ def test_roundtrip_custom_values(self, tmp_config): "traceback_color": "off", "tag_style": "inverted", "show_default_services": False, + "param_change_alert": True, + "param_change_alert_scope": "all", + "param_change_alert_style": "inverted", } save_global_config(custom) result = load_global_config() diff --git a/test/unit/test_param_describe.py b/test/unit/test_param_describe.py new file mode 100644 index 0000000..c4656f7 --- /dev/null +++ b/test/unit/test_param_describe.py @@ -0,0 +1,273 @@ +"""Tests for dendros_param_describe.py colorization and formatting.""" + +import os +import sys +import subprocess + +import pytest +import yaml + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) +PARAM_DESCRIBE_PATH = os.path.join(REPO_ROOT, 'dendROS', 'dendros_param_describe.py') + +from conftest import assert_segment_colored, assert_segment_uncolored, colored_segments, strip_ansi + +# ── Fixtures ────────────────────────────────────────────────────────────────── + +_SIMPLE = [ + 'Parameter name: use_sim_time', + ' Type: boolean', + ' Constraints:', +] + +_WITH_DESC = [ + 'Parameter name: my_param', + ' Type: string', + ' Description: Controls the output rate.', + ' Constraints:', + ' Read only: true', +] + +_MULTI = [ + 'Parameter name: use_sim_time', + ' Type: boolean', + ' Constraints:', + 'Parameter name: publish_rate', + ' Type: double', + ' Description: Rate in Hz.', + ' Constraints:', + ' Min value: 0.0', + ' Max value: 100.0', + ' Step: 0.0', +] + + +# ── Helper ──────────────────────────────────────────────────────────────────── + +def run_describe(tmp_prefix, lines, global_cfg=None, node_colors=None, + argv_extra=None, timeout=10): + """Run dendros_param_describe.py with text lines as stdin.""" + env = os.environ.copy() + env['AMENT_PREFIX_PATH'] = tmp_prefix + env.pop('ROS_DISTRO', None) + env['HOME'] = tmp_prefix + + cfg_dir = os.path.join(tmp_prefix, '.config', 'dendROS') + os.makedirs(cfg_dir, exist_ok=True) + + if global_cfg: + with open(os.path.join(cfg_dir, 'defaults.yaml'), 'w') as f: + yaml.dump(global_cfg, f) + if node_colors: + with open(os.path.join(cfg_dir, 'node_colors.yaml'), 'w') as f: + yaml.dump(node_colors, f) + + stdin = '\n'.join(lines) + '\n' + cmd = [sys.executable, PARAM_DESCRIBE_PATH] + (argv_extra or []) + result = subprocess.run( + cmd, input=stdin.encode(), capture_output=True, env=env, timeout=timeout, + ) + return result.stdout.decode(), result.stderr.decode(), result.returncode + + +# ── Parameter name header ───────────────────────────────────────────────────── + +class TestParamDescribeHeader: + """'Parameter name: X' line: dim label, bold+node-colored param name.""" + + def test_param_name_label_dimmed(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + assert '\033[2mParameter name:\033[0m' in stdout + + def test_param_name_value_bold_node_color(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + assert '\033[32;1muse_sim_time\033[0m' in stdout + + def test_param_name_unmatched_passthrough(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/unknown']) + # label still dimmed, name not colored + assert '\033[2mParameter name:\033[0m' in stdout + assert '\033[32;1m' not in stdout + + def test_param_name_no_node_arg(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc) + # no color on param name but label is still formatted + assert '\033[2mParameter name:\033[0m' in stdout + + def test_namespaced_node_resolved_by_basename(self, tmp_path): + nc = {'color_map': {'talker': '33'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/my_ns/talker']) + assert '\033[33;1muse_sim_time\033[0m' in stdout + + def test_wildcard_pattern(self, tmp_path): + nc = {'color_map': {'nav2_*': '35'}, 'tag_map': {'nav2_*': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/nav2_planner']) + assert '\033[35;1muse_sim_time\033[0m' in stdout + + def test_multi_param_each_header_colored(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _MULTI, node_colors=nc, + argv_extra=['/talker']) + headers = [l for l in stdout.splitlines() if 'Parameter name:' in strip_ansi(l)] + assert len(headers) == 2 + for h in headers: + assert '\033[32;1m' in h + + +# ── Field formatting ────────────────────────────────────────────────────────── + +class TestParamDescribeFields: + """Indented field lines: key dimmed, value plain; section headers bold.""" + + def test_type_key_dimmed(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + assert '\033[2mType:\033[0m' in stdout + + def test_type_value_plain(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + type_line = [l for l in stdout.splitlines() if 'Type:' in strip_ansi(l)][0] + # 'boolean' should appear outside any color code + segs = colored_segments(type_line) + assert any('boolean' in t and c is None for t, c in segs) + + def test_description_key_dimmed(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _WITH_DESC, node_colors=nc, + argv_extra=['/talker']) + assert '\033[2mDescription:\033[0m' in stdout + + def test_constraints_header_bold(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + assert '\033[1mConstraints:\033[0m' in stdout + + def test_read_only_key_dimmed(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _WITH_DESC, node_colors=nc, + argv_extra=['/talker']) + assert '\033[2mRead only:\033[0m' in stdout + + def test_range_constraints_formatted(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _MULTI, node_colors=nc, + argv_extra=['/talker']) + assert '\033[2mMin value:\033[0m' in stdout + assert '\033[2mMax value:\033[0m' in stdout + assert '\033[2mStep:\033[0m' in stdout + + def test_formatting_applied_without_node_color(self, tmp_path): + """Field formatting applies even when the node has no config color.""" + stdout, _, _ = run_describe(str(tmp_path), _WITH_DESC) + assert '\033[2mType:\033[0m' in stdout + assert '\033[1mConstraints:\033[0m' in stdout + assert '\033[2mDescription:\033[0m' in stdout + + def test_indent_preserved(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _WITH_DESC, node_colors=nc, + argv_extra=['/talker']) + lines = stdout.splitlines() + type_line = [strip_ansi(l) for l in lines if 'Type:' in strip_ansi(l)][0] + constraint_line = [strip_ansi(l) for l in lines if 'Read only:' in strip_ansi(l)][0] + assert type_line.startswith(' ') + assert constraint_line.startswith(' ') + + def test_empty_lines_preserved(self, tmp_path): + lines = _SIMPLE + ['', 'Parameter name: other_param', ' Type: integer', ' Constraints:'] + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), lines, node_colors=nc, + argv_extra=['/talker']) + assert '' in stdout.splitlines() + + +# ── Tag badge ───────────────────────────────────────────────────────────────── + +class TestParamDescribeTag: + + def test_tag_shown_before_param_name(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + header = [l for l in stdout.splitlines() if 'Parameter name:' in strip_ansi(l)][0] + assert '[TLK]' in header + assert header.index('[TLK]') < header.index('Parameter name:') + + def test_tag_hidden_when_show_tag_cli_false(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker'], + global_cfg={'show_tag_cli': False}) + assert '[TLK]' not in stdout + + def test_inverted_tag_style(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, + 'style_map': {'talker': 'inverted'}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/talker']) + assert '\033[32;7m[TLK]' in stdout + + def test_tag_not_shown_on_field_lines(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _WITH_DESC, node_colors=nc, + argv_extra=['/talker']) + for line in stdout.splitlines(): + if 'Type:' in strip_ansi(line) or 'Description:' in strip_ansi(line): + assert '[TLK]' not in line + + def test_unmatched_tag(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/unknown'], + global_cfg={'unmatched_color': 'white', 'unmatched_tag': '?'}) + assert '[?]' in stdout + + +# ── Unmatched / dim_unmatched ───────────────────────────────────────────────── + +class TestParamDescribeUnmatched: + + def test_unmatched_color_applied_to_param_name(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/unknown'], + global_cfg={'unmatched_color': 'cyan'}) + assert '\033[' in stdout + # at least the param name should be colored + header = [l for l in stdout.splitlines() if 'Parameter name:' in strip_ansi(l)][0] + assert '\033[' in header + + def test_dim_unmatched_param_name(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, node_colors=nc, + argv_extra=['/unknown'], + global_cfg={'dim_unmatched': True}) + assert '\033[2muse_sim_time\033[0m' in stdout + + +# ── AMENT_PREFIX_PATH fallback ──────────────────────────────────────────────── + +class TestParamDescribeFallback: + + def test_fallback_scan_colors_param_name(self, tmp_path): + cfg_dir = tmp_path / 'share' / 'my_pkg' / 'config' + cfg_dir.mkdir(parents=True) + (cfg_dir / 'dendROS.yaml').write_text(yaml.dump({ + 'groups': {'loc': {'color': 'bold blue', 'label': 'LOC', + 'nodes': ['talker']}} + })) + stdout, _, _ = run_describe(str(tmp_path), _SIMPLE, argv_extra=['/talker']) + assert '\033[' in stdout diff --git a/test/unit/test_param_list.py b/test/unit/test_param_list.py new file mode 100644 index 0000000..c207505 --- /dev/null +++ b/test/unit/test_param_list.py @@ -0,0 +1,334 @@ +"""Tests for dendros_param_list.py colorization.""" + +import os +import sys +import subprocess + +import pytest +import yaml + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) +PARAM_LIST_PATH = os.path.join(REPO_ROOT, 'dendROS', 'dendros_param_list.py') + +from conftest import assert_segment_colored, assert_segment_uncolored, colored_segments, strip_ansi + + +# ── Helper ──────────────────────────────────────────────────────────────────── + +def run_param_list(tmp_prefix, lines, global_cfg=None, node_colors=None, + argv_extra=None, timeout=10): + """Run dendros_param_list.py with text lines as stdin; return (stdout, stderr, rc). + + argv_extra: list of extra args appended to the command (mirrors ${@:3} from the shell). + """ + env = os.environ.copy() + env['AMENT_PREFIX_PATH'] = tmp_prefix + env.pop('ROS_DISTRO', None) + env['HOME'] = tmp_prefix + + cfg_dir = os.path.join(tmp_prefix, '.config', 'dendROS') + os.makedirs(cfg_dir, exist_ok=True) + + if global_cfg: + with open(os.path.join(cfg_dir, 'defaults.yaml'), 'w') as f: + yaml.dump(global_cfg, f) + if node_colors: + with open(os.path.join(cfg_dir, 'node_colors.yaml'), 'w') as f: + yaml.dump(node_colors, f) + + stdin = '\n'.join(lines) + '\n' + cmd = [sys.executable, PARAM_LIST_PATH] + (argv_extra or []) + result = subprocess.run( + cmd, + input=stdin.encode(), + capture_output=True, + env=env, + timeout=timeout, + ) + return result.stdout.decode(), result.stderr.decode(), result.returncode + + +# ── Node header colorization ────────────────────────────────────────────────── + +class TestParamNodeHeader: + """Node header lines (/node_name:) are colored with the group's color.""" + + def test_matched_node_header_colored(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:'], node_colors=nc) + assert '\033[32m/talker:\033[0m' in stdout + + def test_unmatched_node_header_passthrough(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/unknown_node:'], node_colors=nc) + assert_segment_uncolored(stdout, '/unknown_node:') + + def test_multiple_node_headers(self, tmp_path): + nc = {'color_map': {'talker': '32', 'listener': '34'}, + 'tag_map': {'talker': '', 'listener': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/talker:', ' p1', '/listener:', ' p2'], node_colors=nc) + assert '\033[32m/talker:\033[0m' in stdout + assert '\033[34m/listener:\033[0m' in stdout + + def test_namespaced_node_header(self, tmp_path): + nc = {'color_map': {'talker': '33'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/my_ns/talker:'], node_colors=nc) + assert '\033[33m/my_ns/talker:\033[0m' in stdout + + def test_wildcard_pattern(self, tmp_path): + nc = {'color_map': {'nav2_*': '35'}, 'tag_map': {'nav2_*': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/nav2_planner:'], node_colors=nc) + assert '\033[35m/nav2_planner:\033[0m' in stdout + + def test_no_config_passthrough(self, tmp_path): + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:']) + assert '\033[' not in stdout + assert '/talker:' in stdout + + def test_empty_lines_preserved(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/talker:', '', ' p1'], node_colors=nc) + assert '' in stdout.splitlines() + + +# ── Param line colorization (dimmed) ───────────────────────────────────────── + +class TestParamLines: + """Param lines (indented) use the current node's color, dimmed.""" + + def test_param_colored_dim_with_node_color(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:', ' use_sim_time'], + node_colors=nc) + assert '\033[32m\033[2muse_sim_time\033[0m' in stdout + + def test_param_indent_preserved(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:', ' use_sim_time'], + node_colors=nc) + line = [l for l in stdout.splitlines() if 'use_sim_time' in l][0] + plain = strip_ansi(line) + assert plain.startswith(' ') + + def test_param_under_unmatched_node_passthrough(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/unknown:', ' some_param'], node_colors=nc) + assert_segment_uncolored(stdout, 'some_param') + + def test_params_inherit_node_color_change(self, tmp_path): + nc = {'color_map': {'talker': '32', 'listener': '34'}, + 'tag_map': {'talker': '', 'listener': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), + ['/talker:', ' p1', '/listener:', ' p2'], + node_colors=nc) + assert '\033[32m\033[2mp1\033[0m' in stdout + assert '\033[34m\033[2mp2\033[0m' in stdout + + def test_multiple_params_same_node(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/talker:', ' param_a', ' param_b'], node_colors=nc) + assert '\033[32m\033[2mparam_a\033[0m' in stdout + assert '\033[32m\033[2mparam_b\033[0m' in stdout + + def test_param_no_config_passthrough(self, tmp_path): + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:', ' use_sim_time']) + assert '\033[' not in stdout + + +# ── --param-type flag: type annotation dimmed ───────────────────────────────── + +class TestParamListTypeFlag: + """--param-type appends ' (type)'; type content must be rendered dim.""" + + def test_type_dimmed_for_matched_param(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/talker:', ' use_sim_time (bool)'], node_colors=nc) + assert '(\033[2mbool\033[0m)' in stdout + + def test_type_dimmed_for_unmatched_param(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/unknown:', ' foo (integer)'], node_colors=nc) + assert '(\033[2minteger\033[0m)' in stdout + + def test_param_name_and_type_both_present(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/talker:', ' my_param (string)'], node_colors=nc) + assert '\033[32m\033[2mmy_param\033[0m' in stdout + assert '(\033[2mstring\033[0m)' in stdout + + def test_no_type_no_parens(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:', ' my_param'], + node_colors=nc) + assert '(' not in strip_ansi(stdout) + + +# ── Tag badge ───────────────────────────────────────────────────────────────── + +class TestParamListTag: + """Tag badge appears left of the node header when show_tag_cli is true.""" + + def test_tag_shown_before_node_header(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:'], node_colors=nc) + assert '[TLK]' in stdout + line = [l for l in stdout.splitlines() if '/talker:' in strip_ansi(l)][0] + assert line.index('[TLK]') < line.index('/talker:') + + def test_tag_hidden_when_show_tag_cli_false(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:'], + node_colors=nc, + global_cfg={'show_tag_cli': False}) + assert '[TLK]' not in stdout + + def test_inverted_tag_style(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, + 'style_map': {'talker': 'inverted'}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:'], node_colors=nc) + assert '\033[32;7m[TLK]' in stdout + + def test_unmatched_tag(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/unknown:'], node_colors=nc, + global_cfg={'unmatched_color': 'white', 'unmatched_tag': '?'}) + assert '[?]' in stdout + + def test_tag_not_shown_on_param_lines(self, tmp_path): + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': 'TLK'}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:', ' p1'], node_colors=nc) + param_line = [l for l in stdout.splitlines() if 'p1' in strip_ansi(l)][0] + assert '[TLK]' not in param_line + + +# ── Unmatched / dim_unmatched ───────────────────────────────────────────────── + +class TestParamListUnmatched: + + def test_unmatched_color_applied_to_header(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/unknown:'], node_colors=nc, + global_cfg={'unmatched_color': 'cyan'}) + assert '\033[' in stdout + + def test_dim_unmatched_header(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/unknown:'], node_colors=nc, + global_cfg={'dim_unmatched': True}) + assert '\033[2m/unknown:\033[0m' in stdout + + def test_dim_unmatched_param(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list( + str(tmp_path), ['/unknown:', ' my_param'], node_colors=nc, + global_cfg={'dim_unmatched': True}) + assert '\033[2mmy_param\033[0m' in stdout + + def test_passthrough_when_no_config(self, tmp_path): + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['/unknown:', ' p1'], node_colors=nc) + assert_segment_uncolored(stdout, '/unknown:') + assert_segment_uncolored(stdout, 'p1') + + +# ── AMENT_PREFIX_PATH fallback ──────────────────────────────────────────────── + +class TestParamListFallback: + + def test_fallback_scan_colors_header(self, tmp_path): + cfg_dir = tmp_path / 'share' / 'my_pkg' / 'config' + cfg_dir.mkdir(parents=True) + (cfg_dir / 'dendROS.yaml').write_text(yaml.dump({ + 'groups': {'loc': {'color': 'bold blue', 'label': 'LOC', + 'nodes': ['talker']}} + })) + stdout, _, _ = run_param_list(str(tmp_path), ['/talker:']) + assert '\033[' in stdout + + +# ── ros2 param list /node_name — bare output format ────────────────────────── + +class TestParamListNodeArg: + """When a node path is passed as argv, bare param lines (no header, no indent) + are colored using that node's color, pre-seeded before reading stdin.""" + + def test_bare_params_colored_via_node_arg(self, tmp_path): + """Bare param names (no header, no indentation) colored when node arg given.""" + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time', 'publish_rate'], + node_colors=nc, argv_extra=['/talker']) + assert '\033[32m\033[2muse_sim_time\033[0m' in stdout + assert '\033[32m\033[2mpublish_rate\033[0m' in stdout + + def test_bare_params_with_type_flag(self, tmp_path): + """--param-type type dimming works for bare output.""" + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time (bool)'], + node_colors=nc, argv_extra=['/talker']) + assert '\033[32m\033[2muse_sim_time\033[0m' in stdout + assert '(\033[2mbool\033[0m)' in stdout + + def test_node_arg_with_namespace(self, tmp_path): + """Namespaced node path resolved by basename matching.""" + nc = {'color_map': {'talker': '33'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time'], + node_colors=nc, argv_extra=['/my_ns/talker']) + assert '\033[33m\033[2muse_sim_time\033[0m' in stdout + + def test_node_arg_unmatched_passthrough(self, tmp_path): + """Unknown node arg leaves bare params uncolored.""" + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time'], + node_colors=nc, argv_extra=['/unknown_node']) + assert_segment_uncolored(stdout, 'use_sim_time') + + def test_node_arg_unmatched_dim(self, tmp_path): + """dim_unmatched applies to bare params from an unmatched node arg.""" + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time'], + node_colors=nc, argv_extra=['/unknown_node'], + global_cfg={'dim_unmatched': True}) + assert '\033[2muse_sim_time\033[0m' in stdout + + def test_node_arg_unmatched_color(self, tmp_path): + """unmatched_color pre-seeds bare params from an unmatched node arg.""" + nc = {'color_map': {}, 'tag_map': {}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time'], + node_colors=nc, argv_extra=['/unknown_node'], + global_cfg={'unmatched_color': 'cyan'}) + assert '\033[' in stdout + + def test_header_in_output_overrides_node_arg(self, tmp_path): + """When output includes a node header, header color wins (even if arg differs).""" + nc = {'color_map': {'talker': '32', 'listener': '34'}, + 'tag_map': {'talker': '', 'listener': ''}, 'style_map': {}} + # arg says /talker but output has /listener: header — listener color should apply + stdout, _, _ = run_param_list(str(tmp_path), ['/listener:', ' p1'], + node_colors=nc, argv_extra=['/talker']) + assert '\033[34m\033[2mp1\033[0m' in stdout + + def test_flag_args_ignored_for_node_detection(self, tmp_path): + """Flags like --param-type passed in argv_extra do not confuse node detection.""" + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time (bool)'], + node_colors=nc, + argv_extra=['--param-type', '/talker']) + assert '\033[32m\033[2muse_sim_time\033[0m' in stdout + + def test_no_node_arg_bare_params_passthrough(self, tmp_path): + """Without a node arg, bare param names are passed through without color.""" + nc = {'color_map': {'talker': '32'}, 'tag_map': {'talker': ''}, 'style_map': {}} + stdout, _, _ = run_param_list(str(tmp_path), ['use_sim_time'], node_colors=nc) + assert_segment_uncolored(stdout, 'use_sim_time') diff --git a/test/unit/test_param_watcher.py b/test/unit/test_param_watcher.py new file mode 100644 index 0000000..26e6086 --- /dev/null +++ b/test/unit/test_param_watcher.py @@ -0,0 +1,484 @@ +"""Tests for lib/param_watcher.py — param change notification subsystem.""" + +import os +import queue +import sys + +import pytest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) +sys.path.insert(0, os.path.join(REPO_ROOT, 'dendROS')) + +import lib.param_watcher as pw +from conftest import strip_ansi + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +def _resolve_node_mock(node, color_map, tag_map): + """Minimal stand-in for lib.config_loader.resolve_node (exact basename match).""" + basename = node.rsplit('/', 1)[-1] + code = color_map.get(node) or color_map.get(basename) + label = tag_map.get(node) or tag_map.get(basename) if code else None + return code, label + + +def _make_param_event(node, changed=None, new=None, deleted=None): + """Build a list of YAML lines representing one ParameterEvent message.""" + lines = [f'node: {node}', 'stamp:', ' sec: 0', ' nanosec: 0'] + for section, params in [('new_parameters', new), ('changed_parameters', changed), + ('deleted_parameters', deleted)]: + if params: + lines.append(f'{section}:') + for name, type_id, val_field, val in params: + lines += [ + f'- name: {name}', + f' value:', + f' type: {type_id}', + f' {val_field}: {val}', + ] + else: + lines.append(f'{section}: []') + return lines + + +def _drain_all(color_map, tag_map, style_map=None, tag_style='normal', show_tag=True, + alert_style='inline'): + """Drain the queue using the real drain() but with a mock resolve_node.""" + return pw.drain(color_map, tag_map, style_map or {}, tag_style, show_tag, alert_style) + + +def _flush_queue(): + while True: + try: + pw._queue.get_nowait() + except queue.Empty: + break + + +# ── _extract_value ──────────────────────────────────────────────────────────── + +class TestExtractValue: + + def test_bool_true(self): + assert pw._extract_value({'type': 1, 'bool_value': True}) == 'true' + + def test_bool_false(self): + assert pw._extract_value({'type': 1, 'bool_value': False}) == 'false' + + def test_integer(self): + assert pw._extract_value({'type': 2, 'integer_value': 42}) == '42' + + def test_double(self): + assert pw._extract_value({'type': 3, 'double_value': 3.14}) == '3.14' + + def test_string(self): + assert pw._extract_value({'type': 4, 'string_value': 'hello'}) == 'hello' + + def test_bool_array(self): + result = pw._extract_value({'type': 6, 'bool_array_value': [True, False]}) + assert 'True' in result or 'False' in result + + def test_long_array_truncated(self): + big = list(range(100)) + result = pw._extract_value({'type': 7, 'integer_array_value': big}) + assert '…' in result + assert len(result) <= pw._MAX_VALUE_LEN + 2 + + def test_unknown_type_returns_empty(self): + assert pw._extract_value({'type': 99}) == '' + + def test_non_dict_input(self): + result = pw._extract_value('bare_string') + assert result == 'bare_string' + + +# ── _process_chunk ──────────────────────────────────────────────────────────── + +class TestProcessChunk: + + def setup_method(self): + _flush_queue() + pw._param_cache.clear() + + def _process(self, lines, color_map=None, tag_map=None, scope='tracked'): + cm = color_map or {} + tm = tag_map or {} + pw._process_chunk(lines, cm, tm, scope, _resolve_node_mock) + + def test_changed_param_enqueued(self): + lines = _make_param_event( + '/talker', + changed=[('use_sim_time', 1, 'bool_value', 'true')], + ) + color_map = {'talker': '32'} + self._process(lines, color_map) + node, name, old_val, value = pw._queue.get_nowait() + assert node == '/talker' + assert name == 'use_sim_time' + assert value == 'true' + assert old_val is None # first-ever change for this param + + def test_new_params_not_enqueued(self): + """new_parameters (startup declarations) must be silently ignored.""" + lines = _make_param_event( + '/talker', + new=[('use_sim_time', 1, 'bool_value', 'false')], + ) + self._process(lines, {'talker': '32'}) + assert pw._queue.empty() + + def test_deleted_params_not_enqueued(self): + lines = _make_param_event( + '/talker', + deleted=[('old_param', 4, 'string_value', 'gone')], + ) + self._process(lines, {'talker': '32'}) + assert pw._queue.empty() + + def test_cli_daemon_node_filtered(self): + lines = _make_param_event( + '/_ros2cli_12345', + changed=[('use_sim_time', 1, 'bool_value', 'false')], + ) + self._process(lines, {}) + assert pw._queue.empty() + + def test_untracked_node_filtered_in_tracked_scope(self): + lines = _make_param_event( + '/unknown_node', + changed=[('my_param', 2, 'integer_value', '5')], + ) + self._process(lines, {'talker': '32'}, scope='tracked') + assert pw._queue.empty() + + def test_untracked_node_passes_in_all_scope(self): + lines = _make_param_event( + '/unknown_node', + changed=[('my_param', 2, 'integer_value', '5')], + ) + self._process(lines, {'talker': '32'}, scope='all') + assert not pw._queue.empty() + node, name, old_val, value = pw._queue.get_nowait() + assert node == '/unknown_node' + assert name == 'my_param' + + def test_multiple_changed_params_all_enqueued(self): + lines = _make_param_event( + '/talker', + changed=[ + ('param_a', 1, 'bool_value', 'true'), + ('param_b', 2, 'integer_value', '7'), + ], + ) + self._process(lines, {'talker': '32'}) + assert pw._queue.qsize() == 2 + + def test_invalid_yaml_silently_ignored(self): + self._process(['not: valid: yaml: :::'], {'talker': '32'}) + # should not raise; queue stays empty + assert pw._queue.empty() + + def test_empty_node_field_filtered(self): + lines = ['node: ', 'changed_parameters:', '- name: p', ' value:', ' type: 1', + ' bool_value: true', 'new_parameters: []', 'deleted_parameters: []'] + self._process(lines, {}, scope='all') + assert pw._queue.empty() + + def test_namespaced_node_matched_by_basename(self): + lines = _make_param_event( + '/my_ns/talker', + changed=[('rate', 3, 'double_value', '10.0')], + ) + self._process(lines, {'talker': '32'}, scope='tracked') + assert not pw._queue.empty() + + def test_wildcard_pattern_matched(self): + """Wildcard patterns in color_map are resolved by _resolve_node_mock (exact only) + — this test verifies that a direct match reaches the queue.""" + lines = _make_param_event( + '/nav2_planner', + changed=[('max_vel', 3, 'double_value', '1.5')], + ) + self._process(lines, {'nav2_planner': '35'}, scope='tracked') + assert not pw._queue.empty() + + +# ── drain() formatting ──────────────────────────────────────────────────────── + +class TestDrain: + + def setup_method(self): + _flush_queue() + pw._param_cache.clear() + + def _push(self, node, name, value, old_val=None): + pw._queue.put((node, name, old_val, value)) + + def test_drain_empty_queue_returns_empty_list(self): + result = _drain_all({}, {}) + assert result == [] + + def test_notification_contains_node_name(self): + self._push('/talker', 'use_sim_time', 'true') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert len(lines) == 1 + assert '/talker' in strip_ansi(lines[0]) + + def test_notification_contains_param_name(self): + self._push('/talker', 'use_sim_time', 'true') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert 'use_sim_time' in strip_ansi(lines[0]) + + def test_notification_contains_value(self): + self._push('/talker', 'use_sim_time', 'true') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert 'true' in strip_ansi(lines[0]) + + def test_notification_contains_dendros_header(self): + self._push('/talker', 'rate', '10.0') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert '[dendROS]' in strip_ansi(lines[0]) + + def test_dendros_header_logo_colors(self): + """[dendROS] must split on the logo palette: [dend in logo-blue, ROS] in logo-orange.""" + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert '[dendROS]' in strip_ansi(lines[0]) + # Logo-blue (0,75,107) for '[dend' + assert '\033[38;2;0;75;107;1m[dend' in lines[0] + # Logo-orange (224,127,0) for 'ROS]' + assert '\033[38;2;224;127;0;1mROS]' in lines[0] + + def test_node_colored(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert '\033[32m/talker\033[0m' in lines[0] + + def test_tag_shown_when_label_present(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': 'TLK'}) + assert '[TLK]' in lines[0] + + def test_tag_hidden_when_show_tag_false(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': 'TLK'}, show_tag=False) + assert '[TLK]' not in lines[0] + + def test_inverted_tag_style(self): + self._push('/talker', 'p', 'v') + lines = _drain_all( + {'talker': '32'}, {'talker': 'TLK'}, + style_map={'talker': 'inverted'}, tag_style='inverted', + ) + assert '\033[32;7m[TLK]\033[0m' in lines[0] + + def test_param_name_bold(self): + self._push('/talker', 'my_param', 'val') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert '\033[1mmy_param\033[0m' in lines[0] + + def test_param_keyword_plain_white(self): + """'param' keyword must appear without any ANSI color code (plain/white).""" + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert 'param' in strip_ansi(lines[0]) + assert '\033[35mparam\033[0m' not in lines[0] + assert '\033[2mparam\033[0m' not in lines[0] + + def test_untracked_node_no_color(self): + """In 'all' mode, untracked nodes appear without ANSI color wrapping the node name.""" + self._push('/unknown_node', 'p', 'v') + lines = _drain_all({}, {}) + assert '/unknown_node' in lines[0] + import re + assert not re.search(r'\033\[[0-9;]+m/unknown_node\033\[0m', lines[0]) + + def test_multiple_events_all_drained(self): + for i in range(5): + self._push('/talker', f'param_{i}', str(i)) + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert len(lines) == 5 + + def test_line_ends_with_newline(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert lines[0].endswith('\n') + + def test_arrow_separator_present(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert '→' in lines[0] + + def test_unknown_old_value_shown_as_question_mark(self): + """First-ever change (old_val=None) should show '? →' in output.""" + self._push('/talker', 'p', 'new', old_val=None) + lines = _drain_all({'talker': '32'}, {'talker': ''}) + assert '? →' in strip_ansi(lines[0]) + + def test_known_old_value_shown(self): + self._push('/talker', 'p', 'new', old_val='old') + lines = _drain_all({'talker': '32'}, {'talker': ''}) + plain = strip_ansi(lines[0]) + assert 'old → new' in plain + + def test_process_chunk_caches_and_reports_old_value(self): + """Second change via _process_chunk should carry the first change's value as old.""" + import queue as _queue_mod + pw._param_cache.clear() + _flush_queue() + + def _resolve(node, cm, tm): + return cm.get(node.rsplit('/', 1)[-1]), None + + lines1 = _make_param_event('/talker', changed=[('rate', 3, 'double_value', '5.0')]) + pw._process_chunk(lines1, {'talker': '32'}, {}, 'all', _resolve) + _, _, old1, val1 = pw._queue.get_nowait() + assert old1 is None + assert val1 == '5.0' + + lines2 = _make_param_event('/talker', changed=[('rate', 3, 'double_value', '10.0')]) + pw._process_chunk(lines2, {'talker': '32'}, {}, 'all', _resolve) + _, _, old2, val2 = pw._queue.get_nowait() + assert old2 == '5.0' + assert val2 == '10.0' + + +# ── Inverted alert style ────────────────────────────────────────────────────── + +class TestInvertedAlertStyle: + + def setup_method(self): + _flush_queue() + pw._param_cache.clear() + + def _push(self, node, name, value, old_val=None): + pw._queue.put((node, name, old_val, value)) + + def test_inverted_sections_use_correct_backgrounds(self): + """Node identity uses explicit node-color bg; everything else on white bg.""" + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + # Node identity: explicit green bg + black text (_fg_to_bg('32') == '42') + assert '\033[42;30m' in lines[0] + # White bg present (for param section and separating spaces) + assert '\033[107;30m' in lines[0] + # No reverse-video for node identity + assert '\033[32;7m' not in lines[0] + + def test_inverted_contains_node_name(self): + self._push('/talker', 'use_sim_time', 'true') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '/talker' in strip_ansi(lines[0]) + + def test_inverted_contains_param_name(self): + self._push('/talker', 'use_sim_time', 'true') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert 'use_sim_time' in strip_ansi(lines[0]) + + def test_inverted_contains_value(self): + self._push('/talker', 'use_sim_time', 'true') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert 'true' in strip_ansi(lines[0]) + + def test_inverted_tag_and_node_use_explicit_node_bg(self): + """Tag and node name together in one colored-bg island (_fg_to_bg of node code).""" + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': 'TLK'}, alert_style='inverted') + assert '[TLK]' in strip_ansi(lines[0]) + # _fg_to_bg('32') == '42' → green bg + black text; returns to white bg (_WB) after + assert '\033[42;30m[TLK] /talker\033[107;30m' in lines[0] + + def test_inverted_tag_hidden_when_show_tag_false(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': 'TLK'}, + show_tag=False, alert_style='inverted') + assert '[TLK]' not in strip_ansi(lines[0]) + + def test_inverted_no_param_bold_marker(self): + """Inverted block wraps everything; no standalone bold param marker needed.""" + self._push('/talker', 'my_param', 'val') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + # Content must still contain the param name + assert 'my_param' in strip_ansi(lines[0]) + + def test_inverted_untracked_uses_explicit_white_bg(self): + """Untracked nodes (all mode) get the white bg strip; no reverse-video or color bg.""" + self._push('/unknown', 'p', 'v') + lines = _drain_all({}, {}, alert_style='inverted') + assert '\033[107;30m' in lines[0] # white bg present + assert '\033[7m' not in lines[0] # no reverse-video at all + # No node-color bg escape (no fg→bg conversion applied for untracked) + import re + assert not re.search(r'\033\[4[0-9];30m', lines[0]) + + def test_inverted_block_reset_at_end(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '\033[0m' in lines[0] + + def test_inverted_extends_to_eol_with_erase_sequence(self): + """\\033[K (erase to EOL) must be present to fill background to console edge.""" + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '\033[K' in lines[0] + + def test_inverted_dendros_header_present(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '[dendROS]' in strip_ansi(lines[0]) + + def test_inverted_dendros_header_logo_colors(self): + """[dend on logo-blue bg, ROS] on logo-orange bg; black text (hollow/cutout).""" + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '\033[48;2;0;75;107;1m' in lines[0] # logo-blue background + assert '\033[48;2;224;127;0;1m' in lines[0] # logo-orange background + # Black text applied before [dend (hollow cutout look) + assert '\033[48;2;0;75;107;1m\033[30m[dend' in lines[0] + + def test_inverted_arrow_present(self): + self._push('/talker', 'p', 'v') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '→' in strip_ansi(lines[0]) + + def test_inverted_same_content_as_inline(self): + """Both styles must expose the same semantic content in plain text.""" + self._push('/talker', 'use_sim_time', 'true', old_val='false') + inline_lines = _drain_all({'talker': '32'}, {'talker': 'CTR'}, alert_style='inline') + self._push('/talker', 'use_sim_time', 'true', old_val='false') + inv_lines = _drain_all({'talker': '32'}, {'talker': 'CTR'}, alert_style='inverted') + assert strip_ansi(inline_lines[0]).strip() == strip_ansi(inv_lines[0]).strip() + + def test_inverted_old_value_unknown_shows_question_mark(self): + self._push('/talker', 'p', 'new', old_val=None) + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert '? →' in strip_ansi(lines[0]) + + def test_inverted_known_old_value_shown(self): + self._push('/talker', 'p', 'new', old_val='old') + lines = _drain_all({'talker': '32'}, {'talker': ''}, alert_style='inverted') + assert 'old → new' in strip_ansi(lines[0]) + + +# ── Global config defaults ──────────────────────────────────────────────────── + +class TestGlobalConfigDefaults: + + def test_param_change_alert_default_true(self): + from lib.global_config import DEFAULTS + assert DEFAULTS['param_change_alert'] is True + + def test_param_change_alert_scope_default_tracked(self): + from lib.global_config import DEFAULTS + assert DEFAULTS['param_change_alert_scope'] == 'tracked' + + def test_param_change_alert_style_default_inline(self): + from lib.global_config import DEFAULTS + assert DEFAULTS['param_change_alert_style'] == 'inline' + + def test_all_keys_present_in_defaults(self): + from lib.global_config import DEFAULTS + assert 'param_change_alert' in DEFAULTS + assert 'param_change_alert_scope' in DEFAULTS + assert 'param_change_alert_style' in DEFAULTS