diff --git a/config.ini.example b/config.ini.example index 7de79a0..b678182 100644 --- a/config.ini.example +++ b/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] diff --git a/config.ini.minimal-example b/config.ini.minimal-example index b848230..d8d23a2 100644 --- a/config.ini.minimal-example +++ b/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 diff --git a/config.ini.quickstart b/config.ini.quickstart index eb29270..aab35f4 100644 --- a/config.ini.quickstart +++ b/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 '" diff --git a/docs/configuration.md b/docs/configuration.md index 6d69c62..57c31f5 100644 --- a/docs/configuration.md +++ b/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: ` 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)). diff --git a/modules/commands/path_command.py b/modules/commands/path_command.py index ff398e5..e76fb1e 100644 --- a/modules/commands/path_command.py +++ b/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]] diff --git a/modules/commands/test_command.py b/modules/commands/test_command.py index fe833a1..1dcab2c 100644 --- a/modules/commands/test_command.py +++ b/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}") diff --git a/modules/response_template.py b/modules/response_template.py new file mode 100644 index 0000000..531703d --- /dev/null +++ b/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) diff --git a/modules/utils.py b/modules/utils.py index 8f5c2bf..3973b70 100644 --- a/modules/utils.py +++ b/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. diff --git a/tests/unit/test_response_template.py b/tests/unit/test_response_template.py new file mode 100644 index 0000000..c2ae19c --- /dev/null +++ b/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"