Files
meshcore-bot/modules/template_reference.py

535 lines
23 KiB
Python

"""Placeholder and filter reference for template settings, and their preview.
A plugin's ``settings_schema`` field marks itself as a template with
``"template": template_spec(...)``. The web viewer's Plugins page reads that to
show which placeholders the field takes (and, for piped templates, the filters),
and the preview endpoint renders a piped template against sample messages with
:func:`render_preview`.
Three template syntaxes exist, and a field must say which one it uses, because a
reference that offered filters to a ``str.format`` field would be wrong:
- ``piped``: :mod:`modules.response_template` (``{field|filter:arg}``, quoted
literals, conditionals). Used by the test reply and the path reply prefix.
- ``format``: Python ``str.format``. Placeholders only, no filters. What an
unknown name does depends on the plugin (the greeter sends nothing, multitest
falls back to its default), so each such field names a renderer in
:data:`FORMAT_PREVIEWS` that reproduces its plugin's handling.
The catalog here is data for the UI. The fields themselves come from the
commands that render them; ``tests/test_template_reference.py`` keeps the
filter list in step with :data:`~modules.response_template.RESPONSE_TEMPLATE_FILTERS`.
"""
from __future__ import annotations
import configparser
import dataclasses
import json
import re
from pathlib import Path
from typing import Any, Callable, Optional
from .models import MeshMessage
from .response_template import format_piped_template
from .utils import decode_escape_sequences
# Sample messages the preview renders against. Each differs in what the piped
# filters test (hop count and bytes per hop), so a conditional shows its effect.
SCENARIOS: list[dict[str, Any]] = [
{
"id": "multibyte",
"label": "3 hops, 2-byte path",
"nodes": ["a1b2", "c3d4", "e5f6"],
"bytes_per_hop": 2,
},
{
"id": "onebyte",
"label": "3 hops, 1-byte path",
"nodes": ["a1", "c3", "e5"],
"bytes_per_hop": 1,
},
{
"id": "direct",
"label": "Direct (0 hops)",
"nodes": [],
"bytes_per_hop": None,
},
]
# name -> (description, sample value or {scenario_id: value}).
_MESSAGE_PLACEHOLDERS: dict[str, tuple[str, Any]] = {
"sender": ("Sender's name", "Alice"),
"connection_info": ("Path, SNR and RSSI together", {
"multibyte": "a1b2,c3d4,e5f6 (3 hops) | SNR: 9.5 dB | RSSI: -98 dBm",
"onebyte": "a1,c3,e5 (3 hops) | SNR: 9.5 dB | RSSI: -98 dBm",
"direct": "Direct | SNR: 9.5 dB | RSSI: -98 dBm",
}),
"path": ("Route the message took", {
"multibyte": "a1b2,c3d4,e5f6 (3 hops)",
"onebyte": "a1,c3,e5 (3 hops)",
"direct": "Direct",
}),
"hops": ("Hop count; ? when unknown", {"multibyte": "3", "onebyte": "3", "direct": "0"}),
"hops_label": ("Hop count with the word: 1 hop, 3 hops", {
"multibyte": "3 hops", "onebyte": "3 hops", "direct": "0 hops",
}),
"timestamp": ("Bot's time when the message arrived (HH:MM:SS)", "14:32:07"),
"snr": ("Signal-to-noise ratio in dB", "9.5"),
"rssi": ("Signal strength in dBm", "-98"),
"packet_hash": ("16-character packet hash; empty without RF data", "3F2A9C1B7E4D0A65"),
}
_EXTRA_PLACEHOLDERS: dict[str, tuple[str, Any]] = {
"phrase": ("Text after the trigger word (test radio check → radio check)", "radio check"),
"phrase_part": ("': <phrase>', or empty when there is none", ": radio check"),
"elapsed": ("Time the message took to arrive, e.g. 842ms", "842ms"),
"path_distance": (
"Distance travelled, sender → hops → bot; empty unless every node has a location",
{"multibyte": "12.4km", "onebyte": "12.4km", "direct": ""},
),
"firstlast_distance": (
"Distance between the first and last repeater; empty if either has no location",
{"multibyte": "8.1km", "onebyte": "8.1km", "direct": ""},
),
}
ALL_PLACEHOLDERS = {**_MESSAGE_PLACEHOLDERS, **_EXTRA_PLACEHOLDERS}
MESSAGE_PLACEHOLDERS = tuple(_MESSAGE_PLACEHOLDERS)
# Filters for piped templates, in the order the reference lists them. Aliases
# resolve to the same filter and are listed with it rather than on their own.
PIPED_FILTERS: list[dict[str, Any]] = [
{
"name": "hops_min", "arg": "N", "aliases": [],
"help": "Clear the value unless the message travelled at least N hops. "
"hops_min:1 drops it on a direct message.",
"example": "{firstlast_distance|hops_min:1}",
},
{
"name": "pathbytes_min", "arg": "N", "aliases": ["pathbytes"],
"help": "Clear the value unless the path uses at least N bytes per hop (1-3).",
"example": "{path_distance|pathbytes_min:2}",
},
{
"name": "prefix_if_nonempty", "arg": "text", "aliases": [],
"help": "Put text in front of the value, only when the value is not empty. "
"Unquoted, the text runs to the closing } (it may contain |), so it must "
"come last; quote it to add filters after it.",
"example": "{path_distance|prefix_if_nonempty: | Dist: }",
},
{
"name": "if_nonempty", "arg": "text", "aliases": ["if_notempty"],
"help": "Replace the value with text when the value is not empty.",
"example": "{packet_hash|if_nonempty:#}",
},
{
"name": "shorten", "arg": None, "aliases": ["shorten_url"],
"help": "Shorten a URL with the [External_Data] shortener; the long URL is kept "
"if that fails. Not applied in the preview.",
"example": "{\"https://example.com/a/long/link\"|shorten}",
},
]
PIPED_SYNTAX_NOTES = [
"A filter argument in double quotes ends at the closing quote, so more filters can "
"follow: {path_distance|prefix_if_nonempty:\" Dist \"|hops_min:1}",
"A quoted literal can hold placeholders: {\"via {path}\"|hops_min:1}",
"An unknown placeholder renders as nothing.",
]
def template_spec(
syntax: str,
placeholders: tuple[str, ...] | list[str] | dict[str, str],
*,
notes: Optional[list[str]] = None,
escapes: bool = True,
samples: Optional[dict[str, Any]] = None,
blank_default: Optional[str] = None,
blank_config: Optional[tuple[str, str]] = None,
preview: Optional[str] = None,
) -> dict[str, Any]:
"""Build the ``"template"`` attribute of a settings_schema field.
``placeholders`` is a sequence of names from :data:`ALL_PLACEHOLDERS`, or a
``{name: description}`` mapping for fields with their own set (greeter mesh
info, MQTT weather). ``escapes`` says whether ``\\n`` in the value becomes a
newline. ``samples`` overrides preview values for this field, as
``{name: value_or_{scenario_id: value}}``. ``blank_default`` is the
template the plugin uses when the field is blank, which the preview renders
in that case; ``blank_config`` is a ``(section, key)`` the plugin tries
first. ``preview`` names the :data:`FORMAT_PREVIEWS` renderer for a
``format`` field; piped fields always preview.
"""
if preview is not None and preview not in FORMAT_PREVIEWS:
raise ValueError(f"unknown format preview {preview!r}")
if syntax not in ("piped", "format"):
raise ValueError(f"unknown template syntax {syntax!r}")
if isinstance(placeholders, dict):
entries = [{"name": n, "help": h} for n, h in placeholders.items()]
else:
entries = [{"name": n, "help": ALL_PLACEHOLDERS[n][0]} for n in placeholders]
spec: dict[str, Any] = {
"syntax": syntax,
"placeholders": entries,
"notes": list(notes or []),
"escapes": escapes,
}
if samples:
spec["samples"] = samples
if blank_default is not None:
spec["blank_default"] = blank_default
if blank_config is not None:
spec["blank_config"] = list(blank_config)
if syntax == "piped":
spec["filters"] = PIPED_FILTERS
spec["syntax_notes"] = PIPED_SYNTAX_NOTES
spec["previewable"] = True
spec["sample_note"] = "Sample values from a message by Alice."
elif preview:
spec["preview"] = preview
spec["previewable"] = True
spec["sample_note"] = FORMAT_PREVIEWS[preview][1]
return spec
def _scenario_message(scenario: dict[str, Any]) -> MeshMessage:
nodes = scenario["nodes"]
routing_info: dict[str, Any] = {"path_length": len(nodes), "path_nodes": list(nodes)}
if scenario["bytes_per_hop"]:
routing_info["bytes_per_hop"] = scenario["bytes_per_hop"]
return MeshMessage(
content="test radio check",
sender_id="Alice",
hops=len(nodes),
snr=9.5,
rssi=-98,
routing_info=routing_info,
)
def _sample(value: Any, scenario_id: str) -> str:
if isinstance(value, dict):
return str(value.get(scenario_id, ""))
return str(value)
def _row(row_id: str, label: str, output: str, error: Optional[str] = None) -> dict[str, Any]:
row: dict[str, Any] = {
"id": row_id,
"label": label,
"output": output,
"bytes": len(output.encode("utf-8")),
}
if error:
row["error"] = error
return row
def _strip_quotes(text: str) -> str:
"""BaseCommand._strip_quotes_from_config: drop one pair of surrounding double quotes."""
if text and text.startswith('"') and text.endswith('"'):
return text[1:-1]
return text
# A preview renders str.format specs for real, and `{sender:999999999}` would build
# a gigabyte string; the endpoint is admin-only but still shouldn't be a memory lever.
_MAX_FORMAT_WIDTH = 500
_FORMAT_SPEC_RE = re.compile(r"\{[^{}:]*:([^{}]*)\}")
def _oversized_format_spec(text: str) -> bool:
return any(
int(number) > _MAX_FORMAT_WIDTH
for spec in _FORMAT_SPEC_RE.findall(text)
for number in re.findall(r"\d+", spec)
)
Preview = dict[str, Any]
Renderer = Callable[[dict[str, Any], str, Optional[configparser.ConfigParser]], Preview]
def _preview(rows: list[dict[str, Any]], note: Optional[str] = None) -> Preview:
return {"scenarios": rows, "note": note}
def render_preview(
spec: dict[str, Any],
template: str,
config: Optional[configparser.ConfigParser] = None,
) -> Preview:
"""Render a template field as its plugin would, against sample data.
Returns ``{"scenarios": rows, "note": text_or_None}``. Each row is
``{id, label, output, bytes}``, plus ``error`` when the plugin would hit a
problem; ``output`` is then what the plugin sends instead. ``note`` says when
the preview shows a fallback (a blank field) rather than the template.
``config`` is the bot's current configuration, for the settings beyond the
template that change the output (a [Keywords] fallback, the multitest path
layout, the MQTT length limit); without it the defaults apply.
"""
if spec.get("syntax") == "piped":
return _preview_piped(spec, template, config)
renderer = FORMAT_PREVIEWS.get(spec.get("preview") or "")
if renderer is None:
raise ValueError("this template field has no preview")
if _oversized_format_spec(template):
return _preview([_row("template", "Template", "",
f"A format width or precision over {_MAX_FORMAT_WIDTH} isn't previewed.")])
return renderer[0](spec, template, config)
def _config_get(config: Optional[configparser.ConfigParser], section: str, key: str) -> Optional[str]:
if config is None or not config.has_section(section) or not config.has_option(section, key):
return None
return config.get(section, key, raw=True)
def _preview_piped(spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]) -> Preview:
"""Render a piped template against every sample scenario.
Mirrors what the commands do before rendering: surrounding double quotes are
stripped, a blank result falls back (to ``blank_config`` in the bot's config,
then ``blank_default``) and, when the field takes them, escape sequences are
decoded. The ``shorten`` filter gets no config, so it passes the URL through
instead of calling the shortener.
"""
text = _strip_quotes(template.strip()).strip()
note = None
if not text:
fallback_key = spec.get("blank_config")
configured = _config_get(config, *fallback_key) if fallback_key else None
if configured and fallback_key:
text = _strip_quotes(configured)
note = f"The field is blank, so the bot uses [{fallback_key[0]}] {fallback_key[1]}."
elif spec.get("blank_default"):
text = spec["blank_default"]
note = "The field is blank, so this is the built-in default."
if spec.get("escapes", True):
text = decode_escape_sequences(text)
overrides = spec.get("samples") or {}
names = [p["name"] for p in spec.get("placeholders", [])]
results = []
for scenario in SCENARIOS:
sid = scenario["id"]
fields = {}
for name in names:
override = overrides.get(name)
if override is not None and (not isinstance(override, dict) or sid in override):
fields[name] = _sample(override, sid)
elif name in ALL_PLACEHOLDERS:
fields[name] = _sample(ALL_PLACEHOLDERS[name][1], sid)
output = format_piped_template(text, fields, message=_scenario_message(scenario))
results.append(_row(sid, scenario["label"], output))
return _preview(results, note)
# --- str.format fields ------------------------------------------------------
_SAVES_DEFAULT = "The field is blank; saving it stores the default, shown here."
def _blank_or(spec: dict[str, Any], template: str) -> tuple[str, Optional[str]]:
"""The template a blank field saves as (the schema default), with a note saying so."""
text = template.strip()
if text or not spec.get("blank_default"):
return text, None
return spec["blank_default"], _SAVES_DEFAULT
def _describe_format_error(exc: Exception) -> str:
"""Say what is wrong with a str.format template, in the operator's terms."""
if isinstance(exc, KeyError):
return f"Unknown placeholder {{{exc.args[0]}}}"
if isinstance(exc, IndexError):
return "Numbered placeholders such as {0} aren't supported"
if isinstance(exc, ValueError):
return f"Unbalanced braces ({exc}); write {{{{ or }}}} for a literal brace"
return f"Template error: {exc}"
def _greeting_rows(text: str, row_id: str, label: str) -> list[dict[str, Any]]:
"""Rows for one greeting template, as GreeterCommand formats and sends it."""
from .commands.greeter_command import split_greeting_parts
parts = split_greeting_parts(text)
try:
formatted = [part.format(sender="Alice") for part in parts]
except Exception as exc: # noqa: BLE001 - mirrors the greeter's catch-all
return [_row(row_id, label, "", f"{_describe_format_error(exc)}. The greeter sends no greeting.")]
if len(formatted) == 1:
return [_row(row_id, label, formatted[0])]
return [
_row(f"{row_id}-{i}", f"{label}, message {i} of {len(formatted)}", part)
for i, part in enumerate(formatted, 1)
]
def _preview_greeting(spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]) -> Preview:
text, note = _blank_or(spec, template)
return _preview(_greeting_rows(decode_escape_sequences(text), "greeting", "Greeting"), note)
def _preview_channel_greetings(
spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]
) -> Preview:
from .commands.greeter_command import parse_channel_greetings
greetings, ignored = parse_channel_greetings(template.strip())
rows = [
_row(f"ignored-{i}", "Ignored", fragment.strip(),
"Has no channel: in front and no entry before it to join, so the greeter skips it.")
for i, fragment in enumerate(ignored)
]
for key, entry in greetings.items():
rows.extend(_greeting_rows(entry["greeting"], f"ch-{key}", entry["channel"]))
note = None if rows else "No channel greetings: every channel gets the default greeting."
return _preview(rows, note)
_MESH_INFO_SAMPLE = {"total_contacts": 412, "repeaters": 57, "companions": 331, "recent_activity_24h": 38}
def _preview_mesh_info(spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]) -> Preview:
text, note = _blank_or(spec, template)
text = decode_escape_sequences(text)
try:
return _preview([_row("mesh_info", "Appended", text.format(**_MESH_INFO_SAMPLE))], note)
except Exception as exc: # noqa: BLE001 - the greeter catches everything here
return _preview([_row("mesh_info", "Appended", "",
f"{_describe_format_error(exc)}. The greeting goes out without mesh info.")], note)
_MULTITEST_PATHS = ["a1,c3,e5", "a1,b7,e5", "a1,c3,f2"]
def _preview_multitest(spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]) -> Preview:
"""As MultitestCommand._load_config and its reply: layout, decoding, default on error."""
from .commands.multitest_command import _condense_path_lines, _parse_condense_paths_mode
mode = _parse_condense_paths_mode(_config_get(config, "Multitest_Command", "condense_paths") or "true")
paths = sorted(_MULTITEST_PATHS)
paths_text = _condense_path_lines(paths, mode) if mode != "off" else "\n".join(paths)
sample = {"sender": "Alice", "path_count": len(paths), "paths": paths_text, "listening_duration": 6}
default = f"Paths({len(paths)}):\n{paths_text}"
text = _strip_quotes(template.strip()).strip()
if not text:
return _preview([_row("reply", "Reply", default)], "The field is blank, so multitest uses its default format.")
try:
text = text.encode("latin-1").decode("unicode_escape")
except (UnicodeDecodeError, UnicodeEncodeError):
pass
try:
return _preview([_row("reply", "Reply", text.format(**sample))])
except (KeyError, ValueError) as exc:
return _preview([_row("reply", "Reply", default,
f"{_describe_format_error(exc)}. Multitest sends its default format instead.")])
except Exception as exc: # noqa: BLE001 - shows what reaches the command's caller
return _preview([_row("reply", "Reply", "", f"{_describe_format_error(exc)}. The multitest reply fails.")])
_MQTT_SAMPLES = [
("full", "Full station reading", {
"time": "2026-10-04 14:30:00", "device": "WS-2902", "temperature_F": 61.3, "humidity": 72,
"dewpoint_f": 52.1, "pressure_hpa": 1016.4, "wind_current_kmh": 9.7, "wind_peak_kmh": 22.5,
"wind_direction_deg": 225, "rain_today_in": 0.12, "battery_voltage_v": 3.1,
}),
("partial", "Reading without humidity", {
"time": "2026-10-04 14:30:00", "device": "WS-2902", "temperature_F": 61.3,
}),
]
def _mqtt_reply_for_error(err: str) -> str:
"""The reply wx sends for a payload error (WeatherCommon._mqtt_weather_error_key, in English)."""
try:
path = Path(__file__).resolve().parent.parent / "translations" / "en.json"
wx = json.loads(path.read_text(encoding="utf-8"))["commands"]["wx"]
except (OSError, ValueError, KeyError):
wx = {}
if err in ("no_data", "empty_payload", "empty_after_sanitize"):
return wx.get("mqtt_weather_no_data", err)
return wx.get("mqtt_weather_payload_error", "{detail}").format(detail=err.replace("_", " "))
def _preview_mqtt_json(spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]) -> Preview:
"""Through format_mqtt_weather_payload, with the configured length limit."""
from .clients.mqtt_weather import format_mqtt_weather_payload, load_mqtt_weather_format_config
text, note = _blank_or(spec, template)
fmt = dataclasses.replace(
load_mqtt_weather_format_config(config or configparser.ConfigParser()),
output_mode="json_template",
json_template=text,
json_device_key="",
json_device_value="",
)
rows = []
for row_id, label, data in _MQTT_SAMPLES:
out, err = format_mqtt_weather_payload(json.dumps(data).encode(), fmt)
if err:
rows.append(_row(row_id, label, _mqtt_reply_for_error(err), f"The template failed ({err}); wx sends this instead."))
else:
rows.append(_row(row_id, label, out or ""))
return _preview(rows, note)
class _WarningCatcher:
"""Logger stand-in that keeps the warnings a renderer logs."""
def __init__(self) -> None:
self.warnings: list[str] = []
def warning(self, msg: str, *args: Any) -> None:
self.warnings.append(msg % args if args else msg)
def _temperature_preview(key: str, high: Optional[int], low: Optional[int], label: str) -> Renderer:
"""Preview one [Weather] temperature_*_format through utils.format_temperature_high_low."""
def render(spec: dict[str, Any], template: str, config: Optional[configparser.ConfigParser]) -> Preview:
from .utils import format_temperature_high_low
text, note = _blank_or(spec, template)
weather = configparser.ConfigParser(interpolation=None)
weather.add_section("Weather")
weather.set("Weather", key, text)
catcher = _WarningCatcher()
try:
out = format_temperature_high_low(weather, high, low, "°F", logger=catcher)
except Exception as exc: # noqa: BLE001 - what escapes reaches the weather reply
return _preview([_row(key, label, "", f"{_describe_format_error(exc)}. The weather reply fails.")], note)
if catcher.warnings:
# The template failed and the default went out; say why in the template's terms.
try:
text.format(high=high, low=low, units="°F", high_label="H", low_label="L")
reason = "Template error"
except Exception as exc: # noqa: BLE001
reason = _describe_format_error(exc)
return _preview([_row(key, label, out, f"{reason}. The bot uses the default format instead.")], note)
return _preview([_row(key, label, out)], note)
return render
# name -> (renderer, description of the sample data shown under the preview)
FORMAT_PREVIEWS: dict[str, tuple[Renderer, str]] = {
"greeting": (_preview_greeting, "Greeting a user named Alice. Mesh info, when on, is added to the last message."),
"channel_greetings": (_preview_channel_greetings, "Greeting a user named Alice on each listed channel."),
"mesh_info": (_preview_mesh_info, "Sample counts: 412 contacts, 57 repeaters, 331 companions, 38 active."),
"multitest": (_preview_multitest, "Alice heard over 3 paths in 6 seconds, laid out per the saved Path layout."),
"mqtt_json": (_preview_mqtt_json, "Two sample station payloads; the second has no humidity. "
"Uses the saved passthrough length limit."),
"temp_pair": (_temperature_preview("temperature_high_low_format", 68, 51, "High 68, low 51"),
"A day with a high of 68 and a low of 51 °F."),
"temp_high": (_temperature_preview("temperature_high_only_format", 68, None, "High 68 only"),
"A day with only a high of 68 °F."),
"temp_low": (_temperature_preview("temperature_low_only_format", None, 51, "Low 51 only"),
"A night with only a low of 51 °F."),
}