feat(path_command): add reply prefix and minimum path bytes configuration

- Introduced `reply_prefix` to prepend a customizable string to path command replies, supporting placeholders for dynamic content.
- Added `minimum_path_bytes` setting to control the resolution of repeater names based on the number of bytes per hop, enhancing path command behavior.
- Updated relevant documentation and translations to reflect these new configuration options.
- Implemented unit tests to ensure correct functionality of the new features.
This commit is contained in:
agessaman
2026-05-12 17:23:22 -07:00
parent 5879838d8b
commit 13f22e4d7b
18 changed files with 280 additions and 62 deletions
+12
View File
@@ -876,6 +876,18 @@ require_path_bytes_failure_response =
# false: Only respond to "path", "decode", or "route" keywords
enable_p_shortcut = true
# Optional first line prepended to path command RF replies (Python str.format on the triggering message).
# Placeholders match Multitest/Keywords-style responses: {sender}, {connection_info}, {path},
# {timestamp}, {snr}, {rssi}. Only the first RF chunk includes the prefix when the reply is split.
# Example: @[{sender}]
reply_prefix =
# Bytes per hop required before resolving repeater names from the database.
# 0 or 1: always resolve names (default). 2: require 2-byte (or wider) hops. 3: require 3-byte hops.
# If the packet path is shorter, the bot still responds with Path: … (hex) plus a short tip (no lookup).
# Independent of require_path_bytes_greater_or_equal_to above, which can reject the command entirely.
minimum_path_bytes = 0
# Path Selection Preset
# Choose a preset that configures multiple related settings:
# - balanced: Balanced approach using both graph evidence and geographic proximity (default)
+6
View File
@@ -320,6 +320,12 @@ enabled = true
# false: Only respond to "path", "decode", or "route" keywords (default)
enable_p_shortcut = true
# Optional prefix on path replies: {sender}, {connection_info}, {path}, {timestamp}, {snr}, {rssi}
# reply_prefix = @[{sender}]
# Bytes per hop before DB repeater lookup (0/1 = off; 2 or 3 = gate weak paths to hex-only + tip)
# minimum_path_bytes = 0
# Geographic proximity calculation method
# simple: Use proximity to bot location (default)
# path: Use proximity to previous/next nodes in the path for more realistic routing
+4
View File
@@ -63,6 +63,10 @@ default_country = US
# Path command - "p" shortcut for quick path decoding
[Path_Command]
enable_p_shortcut = true
# Optional reply prefix (same placeholders as Keywords: sender, connection_info, path, timestamp, snr, rssi)
# reply_prefix = @[{sender}]
# Require 2+ or 3 bytes per hop before naming repeaters; else hex + tip only (0/1 = always name)
# minimum_path_bytes = 0
[Version_Command]
enabled = true
+15
View File
@@ -11,6 +11,21 @@ The path command supports **1-, 2-, and 3-byte-per-hop** paths (2, 4, or 6 hex c
- **Comma-separated** (e.g. `path 0102,5f7e`): Hop size is inferred from token length. All tokens must be the same length (2, 4, or 6 hex chars). Example: `0102,5f7e` → two 2-byte hops.
- **Continuous hex** (e.g. `path 01025f7e`): The bot’s [Bot] **`prefix_bytes`** is used (2 hex chars = 1 byte, 4 = 2 bytes, 6 = 3 bytes). Use comma-separated input to force a multi-byte interpretation when the bot is in 1-byte mode.
## Reply prefix and repeater name gating
These options only affect the **path** command’s reply text and whether repeater names are resolved from the database.
**`reply_prefix`** (string, default empty)
- Prepended as the first line of path command RF replies (only the **first** chunk when the reply is split for length).
- Uses Python `str.format` on the **triggering** message. Placeholders: `{sender}`, `{connection_info}`, `{path}`, `{timestamp}`, `{snr}`, `{rssi}` (same idea as `[Multitest_Command]` `response_format`).
**`minimum_path_bytes`** (integer `0`–`3`, default `0`)
- **`0` or `1`**: Always resolve repeater names when decoding a path (legacy behavior).
- **`2` or `3`**: Resolve names only when the packet path uses at least that many **bytes per hop** (from routing metadata or inferred from comma-separated hex width). Otherwise the bot replies with `Path: …` (hex) and a short tip, without a DB lookup.
- **Not** the same as `require_path_bytes_greater_or_equal_to`: that setting can block the path command entirely; `minimum_path_bytes` only gates naming.
## Quick Start: Presets
The Path Command supports three presets that configure multiple related settings:
+102 -51
View File
@@ -139,6 +139,18 @@ class PathCommand(BaseCommand):
if "p" not in self.keywords:
self.keywords.append("p")
reply_prefix_raw = bot.config.get('Path_Command', 'reply_prefix', fallback='')
self.path_reply_prefix = self._strip_quotes_from_config(reply_prefix_raw).strip()
minimum_path_bytes_raw = bot.config.getint('Path_Command', 'minimum_path_bytes', fallback=0)
if minimum_path_bytes_raw not in (0, 1, 2, 3):
self.logger.warning(
f"Invalid Path_Command.minimum_path_bytes={minimum_path_bytes_raw}; defaulting to 0"
)
self.minimum_path_bytes = 0
else:
self.minimum_path_bytes = minimum_path_bytes_raw
try:
# Try to get location from Bot section
if bot.config.has_section('Bot'):
@@ -166,6 +178,52 @@ class PathCommand(BaseCommand):
except Exception as e:
self.logger.warning(f"Error reading bot location from config: {e} - geographic proximity guessing disabled")
def _bytes_per_hop_from_nodes_and_routing(
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)
def _should_resolve_repeater_names(
self, node_ids: list[str], routing_info: Optional[dict[str, Any]]
) -> bool:
if self.minimum_path_bytes in (0, 1):
return True
bph = self._bytes_per_hop_from_nodes_and_routing(node_ids, routing_info)
return bph >= self.minimum_path_bytes
def _format_path_reply_prefix(self, message: MeshMessage) -> str:
if not self.path_reply_prefix:
return ''
formatted = self.format_response(message, self.path_reply_prefix).rstrip()
if not formatted:
return ''
return formatted + '\n'
def _format_repeater_resolution_deferred(self, node_ids: list[str]) -> str:
path_display = ','.join(node_ids)
return self.translate(
'commands.path.repeater_resolution_deferred',
path=path_display,
minimum_path_bytes=self.minimum_path_bytes,
)
async def _decode_node_ids(
self, node_ids: list[str], routing_info: Optional[dict[str, Any]] = None
) -> str:
self.logger.info(f"Decoding path with {len(node_ids)} nodes: {','.join(node_ids)}")
if not self._should_resolve_repeater_names(node_ids, routing_info):
return self._format_repeater_resolution_deferred(node_ids)
repeater_info = await self._lookup_repeater_names(node_ids)
return self._format_path_response(node_ids, repeater_info)
def can_execute(self, message: MeshMessage, skip_channel_check: bool = False) -> bool:
"""Check if this command can be executed with the given message.
@@ -219,7 +277,9 @@ class PathCommand(BaseCommand):
await self._send_path_response(message, response)
return True
async def _decode_path(self, path_input: str) -> str:
async def _decode_path(
self, path_input: str, routing_info: Optional[dict[str, Any]] = None
) -> str:
"""Decode hex path data to repeater names.
Comma-separated tokens infer hop size (2, 4, or 6 hex chars per node).
Otherwise uses bot.prefix_hex_chars via parse_path_string().
@@ -249,9 +309,7 @@ class PathCommand(BaseCommand):
if not node_ids:
return self.translate('commands.path.no_valid_hex')
self.logger.info(f"Decoding path with {len(node_ids)} nodes: {','.join(node_ids)}")
repeater_info = await self._lookup_repeater_names(node_ids)
return self._format_path_response(node_ids, repeater_info)
return await self._decode_node_ids(node_ids, routing_info)
except Exception as e:
self.logger.error(f"Error decoding path: {e}")
@@ -1663,55 +1721,50 @@ class PathCommand(BaseCommand):
async def _send_path_response(self, message: MeshMessage, response: str):
"""Send path response, splitting into multiple messages if necessary"""
# Store the complete response for web viewer integration BEFORE splitting
# command_manager will prioritize command.last_response over _last_response
# This ensures capture_command gets the full response, not just the last split message
self.last_response = response
prefix = self._format_path_reply_prefix(message)
self.last_response = prefix + response if prefix else response
# Get dynamic max message length based on message type and bot username
max_length = self.get_max_message_length(message)
prefix_len = self._count_byte_length(prefix)
first_segment_max = max_length - prefix_len
if first_segment_max < 1:
first_segment_max = 1
if self._count_byte_length(response) <= max_length:
# Single message is fine
await self.send_response(message, response)
else:
# Split into multiple messages for over-the-air transmission
# But keep the full response in last_response for web viewer
lines = response.split('\n')
current_message = ""
message_count = 0
if self._count_byte_length(response) + prefix_len <= max_length:
await self.send_response(message, prefix + response)
return
for i, line in enumerate(lines):
# Check if adding this line would exceed max_length display width
if self._count_byte_length(current_message) + self._count_byte_length(line) + 1 > max_length: # +1 for newline
# Send current message and start new one
if current_message:
# Add ellipsis on new line to end of continued message (if not the last message)
if i < len(lines):
current_message += self.translate('commands.path.continuation_end')
# Per-user rate limit applies only to first message (trigger); skip for continuations
await self.send_response(
message, current_message.rstrip(),
skip_user_rate_limit=(message_count > 0)
)
await asyncio.sleep(3.0) # Delay between messages (same as other commands)
message_count += 1
lines = response.split('\n')
current_message = ""
message_count = 0
# Start new message with ellipsis on new line at beginning (if not first message)
if message_count > 0:
current_message = self.translate('commands.path.continuation_start', line=line)
else:
current_message = line
for i, line in enumerate(lines):
body_budget = first_segment_max if message_count == 0 else max_length
if self._count_byte_length(current_message) + self._count_byte_length(line) + 1 > body_budget:
if current_message:
if i < len(lines):
current_message += self.translate('commands.path.continuation_end')
out = (prefix + current_message.rstrip()) if message_count == 0 else current_message.rstrip()
await self.send_response(
message, out,
skip_user_rate_limit=(message_count > 0)
)
await asyncio.sleep(3.0)
message_count += 1
if message_count > 0:
current_message = self.translate('commands.path.continuation_start', line=line)
else:
# Add line to current message
if current_message:
current_message += f"\n{line}"
else:
current_message = line
current_message = line
else:
if current_message:
current_message += f"\n{line}"
else:
current_message = line
# Send the last message if there's content (continuation; skip per-user rate limit)
if current_message:
await self.send_response(message, current_message, skip_user_rate_limit=True)
if current_message:
out = (prefix + current_message.rstrip()) if message_count == 0 else current_message.rstrip()
await self.send_response(message, out, skip_user_rate_limit=True)
async def _extract_path_from_recent_messages(self) -> str:
"""Extract path from the current message's path information (same as test command).
@@ -1732,9 +1785,7 @@ class PathCommand(BaseCommand):
path_nodes = routing_info.get('path_nodes', [])
if path_nodes:
node_ids = [n.upper() for n in path_nodes]
self.logger.info(f"Decoding path from routing_info with {len(node_ids)} nodes: {','.join(node_ids)}")
repeater_info = await self._lookup_repeater_names(node_ids)
return self._format_path_response(node_ids, repeater_info)
return await self._decode_node_ids(node_ids, routing_info)
# Fallback: parse message.path string (e.g. no routing_info or legacy path)
if not msg.path:
@@ -1747,10 +1798,10 @@ class PathCommand(BaseCommand):
path_part = path_string.split(" via ROUTE_TYPE_")[0] if " via ROUTE_TYPE_" in path_string else path_string
if ',' in path_part:
return await self._decode_path(path_part)
return await self._decode_path(path_part, routing_info)
hex_pattern = rf'[0-9a-fA-F]{{{getattr(self.bot, "prefix_hex_chars", 2)}}}'
if re.search(hex_pattern, path_part):
return await self._decode_path(path_part)
return await self._decode_path(path_part, routing_info)
return self.translate('commands.path.path_prefix', path_string=path_string)
except Exception as e:
+4
View File
@@ -918,6 +918,10 @@ wind_speed_unit = mph
precipitation_unit = inch
[Path_Command]
# Optional prefix on path command replies: {sender}, {connection_info}, {path}, {timestamp}, {snr}, {rssi}
# reply_prefix =
# Bytes per hop before repeater name lookup (0/1 = always; 2/3 = gate to hex + tip if shorter)
# minimum_path_bytes = 0
# Geographic proximity calculation method
# simple: Use proximity to bot location (default)
# path: Use proximity to previous/next nodes in the path for more realistic routing
+77
View File
@@ -4,6 +4,8 @@ Unit tests for PathCommand multi-byte path support: routing_info usage and comma
"""
from unittest.mock import MagicMock, Mock
import pytest
from modules.commands.path_command import PathCommand
@@ -145,3 +147,78 @@ class TestPathCommandExtractPathUsesRoutingInfo:
path_command.translate = lambda key, **kwargs: "Direct connection" if "direct" in key.lower() else key
result = await path_command._extract_path_from_recent_messages()
assert "direct" in result.lower()
@pytest.mark.unit
class TestPathCommandMinimumPathBytes:
"""minimum_path_bytes gates repeater lookup (not command execution)."""
@pytest.fixture
def path_command(self, mock_bot):
mock_bot.translator = MagicMock()
def _translate(key, **kwargs):
if key == "commands.path.repeater_resolution_deferred":
return f"Path: {kwargs['path']}\n\nTip:{kwargs['minimum_path_bytes']}"
return key
mock_bot.translator.translate = Mock(side_effect=_translate)
return PathCommand(mock_bot)
@pytest.mark.asyncio
async def test_skips_lookup_when_hops_shorter_than_minimum(self, path_command, mock_bot):
path_command.minimum_path_bytes = 2
called = []
async def capture_lookup(node_ids, lookup_func=None):
called.append(node_ids)
return {}
path_command._lookup_repeater_names = capture_lookup
out = await path_command._decode_path("01,AB")
assert called == []
assert "01,AB" in out
assert "Tip:2" in out
@pytest.mark.asyncio
async def test_runs_lookup_when_hops_meet_minimum(self, path_command, mock_bot):
path_command.minimum_path_bytes = 2
called = []
async def capture_lookup(node_ids, lookup_func=None):
called.append(list(node_ids))
return {nid: {"found": True, "name": "R"} for nid in node_ids}
path_command._lookup_repeater_names = capture_lookup
mock_bot.translator.translate = Mock(
side_effect=lambda key, **kwargs: (
f"{kwargs['node_id']}: {kwargs['name']}" if key == "commands.path.node_format" else key
)
)
out = await path_command._decode_path("0102,ABCD")
assert called == [["0102", "ABCD"]]
assert "0102" in out
@pytest.mark.asyncio
async def test_routing_info_bytes_per_hop_authoritative(self, path_command, mock_bot):
"""bytes_per_hop=1 in packet forces deferred even if node strings are wide (should not happen on wire)."""
path_command.minimum_path_bytes = 2
called = []
async def capture_lookup(node_ids, lookup_func=None):
called.append(True)
return {}
path_command._lookup_repeater_names = capture_lookup
path_command._current_message = MeshMessage(
content="path",
routing_info={
"path_length": 1,
"path_nodes": ["0102"],
"bytes_per_hop": 1,
},
)
out = await path_command._extract_path_from_recent_messages()
assert called == []
assert "0102" in out
assert "Tip:2" in out
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""Unit tests for PathCommand UTF-8 byte truncation and multi-message splitting (PR #128)."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
@@ -96,6 +96,45 @@ class TestPathCommandSendPathResponseByteSplitting:
path_cmd.send_response.assert_awaited_once()
@pytest.mark.unit
class TestPathCommandReplyPrefix:
"""reply_prefix prepended to first RF payload only; last_response includes prefix."""
@pytest.fixture
def path_cmd(self, mock_bot):
mock_bot.translator = MagicMock()
mock_bot.translator.translate = Mock(side_effect=lambda key, **kwargs: key)
cmd = PathCommand(mock_bot)
cmd.send_response = AsyncMock(return_value=True)
return cmd
@pytest.mark.asyncio
async def test_prepends_prefix_to_single_send(self, path_cmd, mock_bot):
path_cmd.path_reply_prefix = "[{sender}]"
path_cmd.get_max_message_length = lambda _msg: 200
msg = MeshMessage(content="path", channel="general", is_dm=False, sender_id="alice")
await path_cmd._send_path_response(msg, "line1")
path_cmd.send_response.assert_awaited_once()
payload = path_cmd.send_response.call_args[0][1]
assert payload == "[alice]\nline1"
assert path_cmd.last_response == "[alice]\nline1"
@pytest.mark.asyncio
async def test_prefix_only_on_first_split_message(self, path_cmd, mock_bot):
path_cmd.path_reply_prefix = "P:"
path_cmd.get_max_message_length = lambda _msg: 25
path_cmd.translate = MockTranslateForSend()
msg = MeshMessage(content="path", channel="general", is_dm=False)
response = "a" * 12 + "\n" + "b" * 13
with patch("modules.commands.path_command.asyncio.sleep", new_callable=AsyncMock):
await path_cmd._send_path_response(msg, response)
assert path_cmd.send_response.await_count >= 2
first = path_cmd.send_response.call_args_list[0][0][1]
assert first.startswith("P:\n")
second = path_cmd.send_response.call_args_list[1][0][1]
assert not second.startswith("P:\n")
class MockTranslateForSend:
def __call__(self, key: str, **kwargs):
if key == "commands.path.continuation_end":
+2 -1
View File
@@ -357,7 +357,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Unbekannt"
"unknown_name": "Unbekannt",
"repeater_resolution_deferred": "Pfad: {path}\n\n💡 Hinweis: Repeater-Erkennung erfordert Pfade mit {minimum_path_bytes} Byte(s) pro Hop."
},
"prefix": {
"description": "Repeater nach zweistelligem Präfix suchen (z.B. 'prefix 1A')",
+2 -1
View File
@@ -337,7 +337,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Unknown"
"unknown_name": "Unknown",
"repeater_resolution_deferred": "Path: {path}\n\n💡 Tip: Repeater identification requires {minimum_path_bytes}-byte paths."
},
"prefix": {
"description": "Look up repeaters by two-character prefix (e.g., 'prefix 1A')",
+2 -1
View File
@@ -271,7 +271,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Unknown"
"unknown_name": "Unknown",
"repeater_resolution_deferred": "Path: {path}\n\n💡 Tip: Repeater identification requires {minimum_path_bytes}-byte paths."
},
"multitest": {
"description": "Listens for 6 seconds and collects all unique paths from incoming messages with the same packet hash",
+2 -1
View File
@@ -213,7 +213,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Desconocido"
"unknown_name": "Desconocido",
"repeater_resolution_deferred": "Ruta: {path}\n\n💡 Consejo: la identificación de repetidores requiere rutas de {minimum_path_bytes} byte(s) por salto."
},
"prefix": {
"description": "Buscar repetidores por prefijo de dos caracteres (ej., 'prefix 1A')",
+2 -1
View File
@@ -409,7 +409,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Inconnu"
"unknown_name": "Inconnu",
"repeater_resolution_deferred": "Chemin : {path}\n\n💡 Astuce : l'identification des répéteurs exige des chemins de {minimum_path_bytes} octet(s) par saut."
},
"prefix": {
"description": "Rechercher des répéteurs par préfixe de deux caractères (ex : 'prefix 1A')",
+2 -1
View File
@@ -337,7 +337,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Inconnu"
"unknown_name": "Inconnu",
"repeater_resolution_deferred": "Chemin : {path}\n\n💡 Astuce : l'identification des répéteurs exige des chemins de {minimum_path_bytes} octet(s) par saut."
},
"prefix": {
"description": "Recherche répéteurs par préfixe 2 caractères (ex: 'prefix 1A')",
+2 -1
View File
@@ -337,7 +337,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Onbekend"
"unknown_name": "Onbekend",
"repeater_resolution_deferred": "Pad: {path}\n\n💡 Tip: repeateridentificatie vereist paden met {minimum_path_bytes} byte(s) per hop."
},
"prefix": {
"description": "Zoek repeaters op twee-karakter prefix (bijv. 'prefix 1A')",
+2 -1
View File
@@ -214,7 +214,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Nieznana"
"unknown_name": "Nieznana",
"repeater_resolution_deferred": "Ścieżka: {path}\n\n💡 Wskazówka: identyfikacja repeaterów wymaga ścieżek z {minimum_path_bytes} bajt(ami) na skok."
},
"multitest": {
"description": "Nasłuchuje 6 sekund aby zebrać unikalne ścieżki z przychodzących wiadomości z tym samym hash'em pakietu",
+2 -1
View File
@@ -334,7 +334,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Desconhecido"
"unknown_name": "Desconhecido",
"repeater_resolution_deferred": "Caminho: {path}\n\n💡 Dica: a identificação de repetidores exige caminhos com {minimum_path_bytes} byte(s) por salto."
},
"prefix": {
"description": "Procurar repetidores por prefixo de dois caracteres (ex: 'prefixo 1A')",
+2 -1
View File
@@ -333,7 +333,8 @@
"continuation_start": "...\n{line}",
"continuation_end": "\n...",
"truncation": "...",
"unknown_name": "Desconhecido"
"unknown_name": "Desconhecido",
"repeater_resolution_deferred": "Caminho: {path}\n\n💡 Dica: a identificação de repetidores exige caminhos com {minimum_path_bytes} byte(s) por salto."
},
"prefix": {
"description": "Pesquisar repetidores por prefixo de dois caracteres (ex: 'prefixo 1A')",