feat(template): add a hops_min filter for gating on route length

{firstlast_distance|prefix_if_nonempty: | F/L Dist: } prints "| F/L Dist: N/A"
on a direct message: the distance helpers return the literal "N/A" when there
is no path to measure, prefix_if_nonempty only asks whether the value is
non-empty, and "N/A" is. Suppressing that needed a gate, and pathbytes_min was
the only one available—so it was being used as a stand-in for "did this
message take any hops", which is not what it asks. It asks how the path is
encoded, so pathbytes_min:2 also discards a one-byte multi-hop path whose
distance is real and measurable.

hops_min:N asks about the route itself. hops_min:1 drops a clause on a direct
message and nothing else:

    hops_min:1   pathbytes_min:2
    direct         cleared      cleared
    1-byte 2 hops  kept         cleared   <- the difference
    2-byte 2 hops  kept         kept
    unknown        cleared      cleared

An unknown hop count clears the value: a gate that cannot confirm the route
should suppress rather than guess, matching pathbytes_min.

The hop-count logic moves to utils.message_hop_count so the filter and
BaseCommand.get_hops_display_values share one implementation instead of two.
It returns None for unknown rather than 0, since a gate must not read
"cannot tell" as "direct".
This commit is contained in:
agessaman
2026-08-25 10:03:30 -07:00
parent f9f65037a8
commit 7debb92e9e
9 changed files with 184 additions and 22 deletions
+10
View File
@@ -104,6 +104,16 @@ semantic versioning.
### Added
- `hops_min:N` response-template filter, alongside `pathbytes_min:N`. It clears a
field unless the message actually travelled at least N hops, so
`{firstlast_distance|hops_min:1|prefix_if_nonempty: | F/L Dist: }` drops the whole
clause on a direct message. The distance placeholders render `N/A` when there is no
path, and `prefix_if_nonempty` treats that as a value and prints its label, so a
gate was needed; `pathbytes_min` was the only one available and it asks how the path
is *encoded*, which meant throwing away a measurable one-byte multi-hop distance to
suppress the direct case. `hops_min` asks about the route instead. An unknown hop
count clears the field rather than guessing.
- `{packet_hash}` placeholder for `[Keywords]` responses, the test command's
`response_format` and the path command's `reply_prefix`: the 16-char MeshCore
packet identity hash (uppercase hex) of the packet that carried the request, so a
+8 -2
View File
@@ -1038,7 +1038,8 @@ enable_p_shortcut = true
# It renders empty unless every node in the chain has known coordinates and the bot has
# bot_latitude/bot_longitude set, so a partial sum is never mistaken for the full distance.
# Feed-style pipe filters are supported (same engine as [Test_Command] response_format),
# so a label can disappear along with an empty value.
# so a label can disappear along with an empty value. hops_min:1 drops a clause on a
# direct message, where {path_distance} would otherwise render "N/A".
# Only the first RF chunk includes the prefix when the reply is split.
# Example: @[{sender}]
# Example: {path_distance|prefix_if_nonempty:\U0001F4CF }
@@ -1556,7 +1557,12 @@ require_path_bytes_failure_response =
# {firstlast_distance} requires at least two path nodes with stored repeater/roomserver coordinates.
# 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).
# Filters: pathbytes_min:N (alias pathbytes:N) clears the field unless bytes_per_hop >= N; hops_min:N clears it unless the message travelled at least N hops; 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).
# hops_min asks about the route, pathbytes_min about how it is encoded. Use hops_min:1 to drop a
# distance clause on a direct message without also dropping a measurable one-byte multi-hop path.
# The distance placeholders render "N/A" on a direct message, which is non-empty, so
# prefix_if_nonempty alone will still print its label there -- gate with hops_min:1 first:
# response_format = ack @[{sender}]{phrase_part} | {path}{firstlast_distance|hops_min:1|prefix_if_nonempty: | F/L Dist: }{packet_hash|prefix_if_nonempty: | Hash: }
# channels =
[Trace_Command]
# Enable or disable the trace/tracer commands (link diagnostics)
+1 -1
View File
@@ -142,7 +142,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`.
- **`[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}`, `{firstlast_distance|hops_min:1}`) 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)).
+1
View File
@@ -26,6 +26,7 @@ These options only affect the **path** commands reply text and whether repeat
```ini
reply_prefix = "{path_distance|prefix_if_nonempty:📏 }\n"
```
- `hops_min:N` clears a field unless the message actually travelled at least N hops. `{path_distance}` renders `N/A` on a direct message, which `prefix_if_nonempty` treats as a value, so gate it first: `{path_distance|hops_min:1|prefix_if_nonempty:📏 }`. Unlike `pathbytes_min:N`, which asks how the path is *encoded*, this keeps a measurable one-byte multi-hop path.
**`minimum_path_bytes`** (integer `0``3`, default `0`)
+8 -18
View File
@@ -21,7 +21,12 @@ from ..command_prefix import (
from ..config_schema import LEGACY_ENABLED_ALIASES
from ..models import CHANNEL_REGIONAL_FLOOD_SCOPE_BODY_OVERHEAD, MeshMessage
from ..security_utils import validate_pubkey_format
from ..utils import format_elapsed_display, get_config_timezone, get_packet_hash_placeholder
from ..utils import (
format_elapsed_display,
get_config_timezone,
get_packet_hash_placeholder,
message_hop_count,
)
# Task-local override for the active translator. When set (via
# ``BaseCommand.respond_in_sender_language``), ``translate`` / ``translate_get_value``
@@ -1222,23 +1227,8 @@ class BaseCommand(ABC):
def get_hops_display_values(self, message: MeshMessage) -> tuple[str, str]:
"""Return hop count placeholders as numeric and pluralized strings."""
hops_val = getattr(message, 'hops', None)
routing_info = getattr(message, 'routing_info', None)
if not isinstance(hops_val, int) and routing_info is not None:
hops_val = routing_info.get('path_length')
if hops_val is None and routing_info.get('path_nodes'):
hops_val = len(routing_info['path_nodes'])
if not isinstance(hops_val, int):
path_str = message.path or ""
hop_match = re.search(r'\((\d+)\s*hops?', path_str, re.IGNORECASE)
if hop_match:
hops_val = int(hop_match.group(1))
elif re.search(r'\bdirect\b|\b0\s*hops?\b', path_str, re.IGNORECASE):
hops_val = 0
if not isinstance(hops_val, int):
hops_val = message_hop_count(message)
if hops_val is None:
return "?", "?"
hops_str = str(hops_val)
+29 -1
View File
@@ -10,7 +10,7 @@ from __future__ import annotations
import re
from typing import Any, Callable
from .utils import message_path_bytes_per_hop
from .utils import message_hop_count, message_path_bytes_per_hop
FilterFn = Callable[[str, dict[str, Any], str], str]
@@ -33,6 +33,33 @@ def _filter_pathbytes_min(value: str, ctx: dict[str, Any], args: str) -> str:
return value
def _filter_hops_min(value: str, ctx: dict[str, Any], args: str) -> str:
"""Clear *value* unless the message travelled at least *N* hops.
Asks about the route rather than how it is encoded, which is what separates
this from ``pathbytes_min``: a one-byte multi-hop path has a real, measurable
distance, and ``pathbytes_min:2`` would throw it away along with the direct
messages it was aimed at. ``hops_min:1`` is the way to drop a clause on a
direct message and nothing else.
An unknown hop count clears the value: a gate that cannot confirm the route
should suppress rather than guess, matching ``pathbytes_min``.
"""
message = ctx.get('message')
if message is None:
return ''
try:
n = int(args.strip())
except ValueError:
return value
if n < 0:
return value
hops = message_hop_count(message)
if hops is None or hops < 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:
@@ -43,6 +70,7 @@ def _filter_prefix_if_nonempty(value: str, ctx: dict[str, Any], args: str) -> st
RESPONSE_TEMPLATE_FILTERS: dict[str, FilterFn] = {
'pathbytes_min': _filter_pathbytes_min,
'pathbytes': _filter_pathbytes_min,
'hops_min': _filter_hops_min,
'prefix_if_nonempty': _filter_prefix_if_nonempty,
}
+29
View File
@@ -1915,6 +1915,35 @@ def bytes_per_hop_from_routing_and_nodes(
return 1
def message_hop_count(message: Any) -> Optional[int]:
"""Hop count for the message, or ``None`` when it cannot be determined.
Prefers ``message.hops``, then ``routing_info`` (``path_length``, else the
number of ``path_nodes``), then a count parsed from the path display string
(``"01,5f (2 hops)"``; ``"Direct"`` or ``"0 hops"`` mean zero).
``None`` means unknown, which is not the same as zero: callers that gate on
hop count should treat it as "cannot confirm" rather than "direct".
"""
hops_val = getattr(message, 'hops', None)
routing_info = getattr(message, 'routing_info', None)
if not isinstance(hops_val, int) and isinstance(routing_info, dict):
hops_val = routing_info.get('path_length')
if hops_val is None and routing_info.get('path_nodes'):
hops_val = len(routing_info['path_nodes'])
if not isinstance(hops_val, int):
path_str = getattr(message, 'path', None) or ""
hop_match = re.search(r'\((\d+)\s*hops?', path_str, re.IGNORECASE)
if hop_match:
hops_val = int(hop_match.group(1))
elif re.search(r'\bdirect\b|\b0\s*hops?\b', path_str, re.IGNORECASE):
hops_val = 0
return hops_val if isinstance(hops_val, int) else None
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).
+40
View File
@@ -23,6 +23,7 @@ from modules.utils import (
get_major_city_queries,
get_packet_hash_placeholder,
is_valid_timezone,
message_hop_count,
node_ids_from_path_string,
parse_location_string,
parse_path_string,
@@ -937,3 +938,42 @@ class TestCalculatePacketHashEdgeCases:
h = calculate_packet_hash("0800000000" + "00" + "ff")
assert h != "0000000000000000"
class TestMessageHopCount:
"""Tests for message_hop_count()."""
@staticmethod
def _msg(**kw):
m = Mock()
m.hops = kw.get("hops")
m.path = kw.get("path")
m.routing_info = kw.get("routing_info")
return m
def test_prefers_message_hops(self):
assert message_hop_count(self._msg(hops=3, path="01,02 (2 hops)")) == 3
def test_falls_back_to_routing_path_length(self):
assert message_hop_count(self._msg(routing_info={"path_length": 2})) == 2
def test_falls_back_to_counting_path_nodes(self):
msg = self._msg(routing_info={"path_nodes": ["01", "02", "03"]})
assert message_hop_count(msg) == 3
def test_falls_back_to_parsing_the_path_string(self):
assert message_hop_count(self._msg(path="01,5f (2 hops)")) == 2
assert message_hop_count(self._msg(path="0a (1 hop)")) == 1
def test_direct_path_string_is_zero_hops(self):
assert message_hop_count(self._msg(path="Direct")) == 0
assert message_hop_count(self._msg(path="Direct via ROUTE_TYPE_FLOOD")) == 0
assert message_hop_count(self._msg(path="0 hops")) == 0
def test_unknown_is_none_not_zero(self):
"""None means "cannot confirm", which callers must not read as direct."""
assert message_hop_count(self._msg()) is None
assert message_hop_count(self._msg(path="Unknown routing")) is None
def test_non_dict_routing_info_is_ignored(self):
assert message_hop_count(self._msg(routing_info="01,5f", path="Direct")) == 0
+58
View File
@@ -266,3 +266,61 @@ def test_test_command_response_omits_missing_packet_hash():
out = cmd.format_response(msg, "hash={packet_hash|prefix_if_nonempty:id:}.")
assert out == "hash=."
def _msg(**kw):
base = dict(content="test", channel="c")
base.update(kw)
return MeshMessage(**base)
@pytest.mark.unit
def test_hops_min_clears_on_a_direct_message():
out = format_piped_template(
"ack{d|hops_min:1|prefix_if_nonempty: | Dist: }",
{"d": "N/A"},
message=_msg(path="Direct", hops=0, routing_info={"path_length": 0, "bytes_per_hop": 2}),
)
assert out == "ack"
@pytest.mark.unit
def test_hops_min_keeps_a_single_byte_multihop_path():
"""The point of hops_min over pathbytes_min: a one-byte path still travelled,
so its distance is real and must not be discarded with the direct messages."""
msg = _msg(path="01,02 (2 hops)", hops=2,
routing_info={"path_length": 2, "path_nodes": ["01", "02"], "bytes_per_hop": 1})
assert format_piped_template("{d|hops_min:1}", {"d": "12.4km"}, message=msg) == "12.4km"
assert format_piped_template("{d|pathbytes_min:2}", {"d": "12.4km"}, message=msg) == ""
@pytest.mark.unit
def test_hops_min_threshold_is_inclusive():
msg = _msg(path="01,02 (2 hops)", hops=2)
assert format_piped_template("{d|hops_min:2}", {"d": "x"}, message=msg) == "x"
assert format_piped_template("{d|hops_min:3}", {"d": "x"}, message=msg) == ""
@pytest.mark.unit
def test_hops_min_zero_admits_a_direct_message():
msg = _msg(path="Direct", hops=0)
assert format_piped_template("{d|hops_min:0}", {"d": "x"}, message=msg) == "x"
@pytest.mark.unit
def test_hops_min_clears_when_the_hop_count_is_unknown():
"""A gate that cannot confirm the route suppresses rather than guesses."""
msg = _msg(path=None, hops=None, routing_info=None)
assert format_piped_template("{d|hops_min:1}", {"d": "x"}, message=msg) == ""
@pytest.mark.unit
def test_hops_min_without_a_message_clears():
assert format_piped_template("{d|hops_min:1}", {"d": "x"}, message=None) == ""
@pytest.mark.unit
def test_hops_min_with_an_unusable_argument_passes_the_value_through():
msg = _msg(path="Direct", hops=0)
assert format_piped_template("{d|hops_min:abc}", {"d": "x"}, message=msg) == "x"
assert format_piped_template("{d|hops_min:-1}", {"d": "x"}, message=msg) == "x"