feat(test): Test_Command response_format with piped path filters

Add format_piped_template (pathbytes_min, prefix_if_nonempty), utils
bytes-per-hop helpers, and [Test_Command] format priority over Keywords.

config.ini.example

@@ -1372,6 +1372,13 @@ enabled = true
 require_path_bytes_greater_or_equal_to = 0
 # Optional response when rejected by path-byte requirement; leave empty for silent reject
 require_path_bytes_failure_response =
+
+# Optional: full ack template for test/t. When set, overrides [Keywords] test for this command only.
+# Same placeholders as Keywords (sender, phrase, phrase_part, connection_info, path, hops, hops_label,
+# timestamp, elapsed, snr, path_distance, firstlast_distance).
+# Pipe filters (feed-style), e.g. only show cumulative path distance when hops are multibyte (2+ bytes per hop):
+# response_format = ack @[{sender}]{phrase_part} | {connection_info}{path_distance|pathbytes_min:2|prefix_if_nonempty: | Path Dist: }| F/L Dist: {firstlast_distance} | {elapsed} | Rec: {timestamp}
+# Filters: pathbytes_min:N (alias pathbytes:N) clears the field unless bytes_per_hop >= N; prefix_if_nonempty:L prepends literal L when value non-empty after prior filters. If L contains |, put prefix_if_nonempty last in that placeholder (it consumes the rest of the chain as its literal).
 # channels =
 
 [Trace_Command]

config.ini.minimal-example

@@ -308,6 +308,8 @@ enabled = true
 # true: Test command is available (responds to 'test' or 't')
 # false: Test command is disabled
 enabled = true
+# Optional: overrides [Keywords] test; supports {field|pathbytes_min:2|prefix_if_nonempty: | Path Dist: } style filters (see config.ini.example [Test_Command])
+# response_format =
 
 [Path_Command]
 # Enable or disable the path command

config.ini.quickstart

@@ -46,6 +46,7 @@ admin_commands = repeater,webviewer,reload,channelpause
 
 [Keywords]
 test = "ack @[{sender}]{phrase_part} | {connection_info} | Received at: {timestamp}"
+# Optional: set [Test_Command] response_format to override this for test/t (see config.ini.example).
 ping = "Pong!"
 pong = "Ping!"
 help = "Bot Help: test (or t), ping, version, help, hello, cmd, advert, wx, aqi, sun, moon, solar, hfcond, satpass, prefix, path, sports, dice, roll, stats | More: 'help <command>'"

docs/configuration.md

@@ -100,6 +100,7 @@ Many commands and features have their own section. Options there control whether
 Examples of sections that configure specific commands or features:
 
 - **`[Path_Command]`** – Path decoding and repeater selection. See [Path Command](path-command-config.md) for all options.
+- **`[Test_Command]`** – `test` / `t` behavior. Optional **`response_format`** overrides the legacy **`[Keywords] test`** string. Templates support the same placeholders as Keywords, plus **feed-style pipe filters** on placeholders (e.g. `{path_distance|pathbytes_min:2}`) implemented in `modules/response_template.py`—see comments under `[Test_Command]` in `config.ini.example`.
 - **`[Prefix_Command]`** – Prefix lookup, prefix best, range limits.
 - **`[Cmd_Command]`** – `cmd` behavior. Set `cmd_reference_url` to return `Full command reference: <url>` instead of the generated compact command list.
 - **`[Weather]`** – Used by the `wx` / `gwx` commands and the Weather Service plugin (see [Weather Service](weather-service.md)).

modules/commands/path_command.py

@@ -12,7 +12,7 @@ from typing import Any, Callable, Optional
 
 from ..models import MeshMessage
 from ..security_utils import sanitize_name
-from ..utils import calculate_distance, parse_path_string
+from ..utils import bytes_per_hop_from_routing_and_nodes, calculate_distance, parse_path_string
 from .base_command import BaseCommand
 
 
@@ -182,14 +182,7 @@ class PathCommand(BaseCommand):
         self, node_ids: list[str], routing_info: Optional[dict[str, Any]]
     ) -> int:
         """Bytes per hop from packet metadata or inferred from hex node width."""
-        if routing_info:
-            bph = routing_info.get('bytes_per_hop')
-            if isinstance(bph, int) and 1 <= bph <= 3:
-                return bph
-        if not node_ids:
-            return 1
-        widths = [len(n) // 2 for n in node_ids]
-        return min(widths)
+        return bytes_per_hop_from_routing_and_nodes(routing_info, node_ids)
 
     def _should_resolve_repeater_names(
         self, node_ids: list[str], routing_info: Optional[dict[str, Any]]

modules/commands/test_command.py

@@ -10,6 +10,7 @@ from datetime import datetime
 from typing import Any, Optional
 
 from ..models import MeshMessage
+from ..response_template import format_piped_template
 from ..utils import calculate_distance, extract_path_node_ids_from_message
 from .base_command import BaseCommand
 
@@ -141,6 +142,12 @@ class TestCommand(BaseCommand):
         Returns:
             Optional[str]: The configured or default response format string.
         """
+        if self.bot.config.has_section('Test_Command'):
+            raw = self.bot.config.get('Test_Command', 'response_format', fallback='')
+            if raw:
+                cleaned = self._strip_quotes_from_config(raw).strip()
+                if cleaned:
+                    return cleaned
         if self.bot.config.has_section('Keywords'):
             format_str = self.bot.config.get('Keywords', 'test', fallback=None)
             if format_str:
@@ -683,19 +690,26 @@ class TestCommand(BaseCommand):
             path_distance = self._calculate_path_distance(message)
             firstlast_distance = self._calculate_firstlast_distance(message)
             phrase_part = f": {phrase}" if phrase else ""
-            return response_format.format(
-                sender=message.sender_id or self.translate('common.unknown_sender'),
-                phrase=phrase,
-                phrase_part=phrase_part,
-                connection_info=connection_info,
-                path=path_display,
-                hops=hops_str,
-                hops_label=hops_label,
-                timestamp=timestamp,
-                elapsed=elapsed,
-                snr=message.snr or self.translate('common.unknown'),
-                path_distance=path_distance or "",
-                firstlast_distance=firstlast_distance or ""
+            fields = {
+                'sender': message.sender_id or self.translate('common.unknown_sender'),
+                'phrase': phrase,
+                'phrase_part': phrase_part,
+                'connection_info': connection_info,
+                'path': path_display,
+                'hops': hops_str,
+                'hops_label': hops_label,
+                'timestamp': timestamp,
+                'elapsed': elapsed,
+                'snr': str(message.snr) if message.snr is not None else self.translate('common.unknown'),
+                'path_distance': path_distance or '',
+                'firstlast_distance': firstlast_distance or '',
+            }
+            return format_piped_template(
+                response_format,
+                fields,
+                message=message,
+                logger=self.logger,
+                prefix_hex_chars=getattr(self.bot, 'prefix_hex_chars', 2),
             )
         except (KeyError, ValueError) as e:
             self.logger.warning(f"Error formatting test response: {e}")

modules/response_template.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Piped placeholders for command response templates (feed-style ``{field|filter:args}``).
+
+Used by :class:`~modules.commands.test_command.TestCommand` and extensible for other
+commands. Same brace limitation as feed formatting: no nested ``{}`` inside a placeholder.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Callable
+
+from .utils import message_path_bytes_per_hop
+
+FilterFn = Callable[[str, dict[str, Any], str], str]
+
+
+def _filter_pathbytes_min(value: str, ctx: dict[str, Any], args: str) -> str:
+    """Clear *value* unless message path uses at least *N* bytes per hop (N in 1..3)."""
+    message = ctx.get('message')
+    if message is None:
+        return ''
+    try:
+        n = int(args.strip())
+    except ValueError:
+        return value
+    if n < 1 or n > 3:
+        return value
+    prefix_hex = int(ctx.get('prefix_hex_chars') or 2)
+    bph = message_path_bytes_per_hop(message, prefix_hex_chars=prefix_hex)
+    if bph < n:
+        return ''
+    return value
+
+
+def _filter_prefix_if_nonempty(value: str, ctx: dict[str, Any], args: str) -> str:
+    """Prepend *args* literal to *value* only when *value* is non-empty after prior filters."""
+    if not value:
+        return ''
+    return args + value
+
+
+RESPONSE_TEMPLATE_FILTERS: dict[str, FilterFn] = {
+    'pathbytes_min': _filter_pathbytes_min,
+    'pathbytes': _filter_pathbytes_min,
+    'prefix_if_nonempty': _filter_prefix_if_nonempty,
+}
+
+
+def _field_and_filter_specs(inner: str) -> tuple[str, list[tuple[str, str]]]:
+    """Split ``inner`` into field name and ``(filter_name, args)`` pairs.
+
+    Pipe ``|`` separates filters. ``prefix_if_nonempty`` is special: its argument may
+    contain ``|`` (e.g. `` | Path Dist: ``), so once that filter is reached we merge
+    all remaining segments and treat the rest as its args. ``prefix_if_nonempty`` must
+    be last in the chain if the literal includes a pipe.
+    """
+    raw_parts = inner.split('|')
+    field_name = raw_parts[0].strip()
+    if len(raw_parts) < 2:
+        return field_name, []
+    specs: list[tuple[str, str]] = []
+    i = 1
+    while i < len(raw_parts):
+        if raw_parts[i].lstrip().startswith('prefix_if_nonempty'):
+            merged = '|'.join(raw_parts[i:])
+            if re.match(r'^\s*prefix_if_nonempty\s*$', merged):
+                specs.append(('prefix_if_nonempty', ''))
+                break
+            m = re.match(r'^\s*prefix_if_nonempty\s*:(.*)$', merged, flags=re.DOTALL)
+            if m:
+                specs.append(('prefix_if_nonempty', m.group(1)))
+            else:
+                specs.append(('prefix_if_nonempty', ''))
+            break
+        segment = raw_parts[i].strip()
+        name, sep, arg = segment.partition(':')
+        name = name.strip()
+        arg = arg if sep else ''
+        specs.append((name, arg))
+        i += 1
+    return field_name, specs
+
+
+def format_piped_template(
+    template: str,
+    fields: dict[str, str],
+    *,
+    message: Any = None,
+    logger: Any = None,
+    prefix_hex_chars: int = 2,
+) -> str:
+    """Replace ``{field}`` and ``{field|filter:arg|...}`` using *fields* and optional *message*.
+
+    Args:
+        template: Raw template string from config.
+        fields: Mapping of placeholder names to string values (e.g. ``sender``, ``path_distance``).
+        message: Triggering mesh message; required for ``pathbytes`` / ``pathbytes_min`` filters.
+        logger: Optional logger for unknown filter warnings.
+        prefix_hex_chars: Bot prefix width for inferring bytes per hop from legacy path text.
+
+    Returns:
+        Fully expanded string.
+    """
+    ctx: dict[str, Any] = {
+        'message': message,
+        'logger': logger,
+        'prefix_hex_chars': prefix_hex_chars,
+    }
+
+    def replace_placeholder(match: re.Match[str]) -> str:
+        inner_raw = match.group(1)
+        if '|' not in inner_raw:
+            return str(fields.get(inner_raw.strip(), ''))
+        field_name, filter_specs = _field_and_filter_specs(inner_raw)
+        value = str(fields.get(field_name, ''))
+        for name, arg in filter_specs:
+            fn = RESPONSE_TEMPLATE_FILTERS.get(name)
+            if fn is None:
+                if logger is not None:
+                    logger.warning(f"Unknown response template filter {name!r} in {{{inner_raw}}}")
+                continue
+            value = fn(value, ctx, arg)
+        return value
+
+    return re.sub(r"\{([^}]+)\}", replace_placeholder, template)

modules/utils.py

@@ -1818,6 +1818,55 @@ def extract_path_node_ids_from_message(message: Any) -> list[str]:
     return []
 
 
+def _normalized_message_path_string(message: Any) -> str:
+    """Strip route suffix and hop-count suffix from message.path for continuous-hex parsing."""
+    path_string = (getattr(message, 'path', None) or '').strip()
+    if not path_string or 'Direct' in path_string or '0 hops' in path_string:
+        return ''
+    if ' via ROUTE_TYPE_' in path_string:
+        path_string = path_string.split(' via ROUTE_TYPE_')[0]
+    path_string = re.sub(r'\s*\([^)]*hops?[^)]*\)', '', path_string, flags=re.IGNORECASE).strip()
+    return path_string
+
+
+def bytes_per_hop_from_routing_and_nodes(
+    routing_info: Optional[dict[str, Any]],
+    node_ids: list[str],
+) -> int:
+    """Bytes per hop from packet routing metadata, else inferred from hex node width.
+
+    When ``routing_info`` includes ``bytes_per_hop`` in 1..3, that value wins.
+    Otherwise uses minimum half-byte width among ``node_ids`` (comma or path_nodes).
+    Returns ``1`` when no nodes (direct / unknown).
+    """
+    if routing_info:
+        bph = routing_info.get('bytes_per_hop')
+        if isinstance(bph, int) and 1 <= bph <= 3:
+            return bph
+    if node_ids:
+        return min(len(n) // 2 for n in node_ids)
+    return 1
+
+
+def message_path_bytes_per_hop(message: Any, *, prefix_hex_chars: int = 2) -> int:
+    """Best-effort bytes per hop for the message path (RF metadata or inferred from path text).
+
+    Uses ``routing_info.bytes_per_hop`` when present (1..3). Otherwise prefers
+    :func:`extract_path_node_ids_from_message`, then comma/continuous hex via
+    :func:`node_ids_from_path_string` using ``prefix_hex_chars`` for legacy paths.
+
+    Returns ``1`` when no usable path (direct / unparseable) so conservative gates
+    (e.g. ``pathbytes_min:2``) do not treat unknown as multibyte.
+    """
+    routing_info = getattr(message, 'routing_info', None)
+    node_ids = extract_path_node_ids_from_message(message)
+    if not node_ids:
+        ps = _normalized_message_path_string(message)
+        if ps:
+            node_ids = node_ids_from_path_string(ps, prefix_hex_chars)
+    return bytes_per_hop_from_routing_and_nodes(routing_info, node_ids)
+
+
 def node_ids_from_path_string(path_str: str, prefix_hex_chars: int = 2) -> list[str]:
     """Parse path display string into node IDs: multi-byte comma tokens, else fixed-width scan.
 

tests/unit/test_response_template.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Unit tests for piped response templates and message_path_bytes_per_hop."""
+
+import configparser
+from unittest.mock import MagicMock, Mock
+
+import pytest
+
+from modules.commands.test_command import TestCommand as MeshTestCommand
+from modules.models import MeshMessage
+from modules.response_template import format_piped_template
+from modules.utils import message_path_bytes_per_hop
+
+
+@pytest.mark.unit
+def test_message_path_bytes_per_hop_from_routing():
+    msg = MeshMessage(
+        content="test",
+        channel="c",
+        routing_info={"bytes_per_hop": 2, "path_length": 1, "path_nodes": ["0102"]},
+    )
+    assert message_path_bytes_per_hop(msg) == 2
+
+
+@pytest.mark.unit
+def test_message_path_bytes_per_hop_infers_from_nodes():
+    msg = MeshMessage(
+        content="test",
+        channel="c",
+        path="01,02,03 (3 hops)",
+        routing_info=None,
+    )
+    assert message_path_bytes_per_hop(msg) == 1
+
+
+@pytest.mark.unit
+def test_format_piped_template_plain_field():
+    out = format_piped_template("a={x}|end", {"x": "hi"}, message=None)
+    assert out == "a=hi|end"
+
+
+@pytest.mark.unit
+def test_pathbytes_min_clears_when_below_threshold():
+    msg = MeshMessage(
+        content="test",
+        channel="c",
+        routing_info={"bytes_per_hop": 1, "path_length": 2, "path_nodes": ["01", "02"]},
+    )
+    out = format_piped_template(
+        "d={path_distance|pathbytes_min:2}",
+        {"path_distance": "10.0km (1 segs)"},
+        message=msg,
+    )
+    assert out == "d="
+
+
+@pytest.mark.unit
+def test_pathbytes_min_keeps_multibyte():
+    msg = MeshMessage(
+        content="test",
+        channel="c",
+        routing_info={"bytes_per_hop": 2, "path_length": 1, "path_nodes": ["0102"]},
+    )
+    out = format_piped_template(
+        "d={path_distance|pathbytes:2}",
+        {"path_distance": "5.0km (1 segs)"},
+        message=msg,
+    )
+    assert out == "d=5.0km (1 segs)"
+
+
+@pytest.mark.unit
+def test_prefix_if_nonempty_literal_may_contain_pipe():
+    """Regression: args like ' | Path Dist: ' must not split into a fake 'Path Dist' filter."""
+    msg = MeshMessage(
+        content="test",
+        channel="c",
+        routing_info={"bytes_per_hop": 2, "path_length": 1, "path_nodes": ["0102"]},
+    )
+    out = format_piped_template(
+        "x={path_distance|pathbytes_min:2|prefix_if_nonempty: | Path Dist: }",
+        {"path_distance": "1km"},
+        message=msg,
+        logger=None,
+    )
+    assert out == "x= | Path Dist: 1km"
+
+
+@pytest.mark.unit
+def test_get_response_format_test_command_over_keywords():
+    bot = MagicMock()
+    bot.logger = Mock()
+    bot.config = configparser.ConfigParser()
+    bot.config.add_section("Bot")
+    bot.config.set("Bot", "bot_name", "TestBot")
+    bot.config.add_section("Channels")
+    bot.config.set("Channels", "monitor_channels", "general")
+    bot.config.set("Channels", "respond_to_dms", "true")
+    bot.config.add_section("Keywords")
+    bot.config.set("Keywords", "test", "from-keywords")
+    bot.config.add_section("Test_Command")
+    bot.config.set("Test_Command", "enabled", "true")
+    bot.config.set("Test_Command", "response_format", "from-test-cmd")
+    bot.config.add_section("Path_Command")
+    bot.config.set("Path_Command", "recency_weight", "0.2")
+    bot.translator = MagicMock()
+    bot.translator.translate = Mock(side_effect=lambda key, **kwargs: key)
+    bot.prefix_hex_chars = 2
+
+    cmd = MeshTestCommand(bot)
+    assert cmd.get_response_format() == "from-test-cmd"