auto respond bot

Added an optional auto-responder bot that listens for configurable trigger words (from localisation) and sends a response with available link metadata (like SNR/RSSI/hops/transport) The feature is toggleable from the UI shortcut ctl+B
This commit is contained in:
Kelly
2026-03-28 19:57:22 -07:00
parent bc69c709ff
commit 0760d6d0ca
8 changed files with 129 additions and 1 deletions

View File

@@ -60,6 +60,7 @@ For smaller displays you may wish to enable `single_pane_mode`:
- `CTRL` + `t` or `F4` = With the Node List highlighted, send a traceroute to the selected node
- `F5` = Display a node's info
- `CTRL` + `f` = With the Node List highlighted, favorite the selected node
- `CTRL` + `b` = Enable/Disable Autoresponder Bot (ping / pong)
- `CTRL` + `g` = With the Node List highlighted, ignore the selected node
- `CTRL` + `d` = With the Channel List hightlighted, archive a chat to reduce UI clutter. Messages will be saved in the db and repopulate if you send or receive a DM from this user.
- `CTRL` + `d` = With the Note List highlghted, remove a node from your nodedb.

View File

@@ -90,6 +90,8 @@ confirm.remove_ignored, "Remove {name} from Ignored?", ""
confirm.region_unset, "Your region is UNSET. Set it now?", ""
dialog.resize_title, "Resize Terminal", ""
dialog.resize_body, "Please resize the terminal to at least {rows} rows.", ""
bot.catch_words, "ping; test", "Semicolon-separated bot trigger words."
bot.response.word, "Pong!", "Bot response word."
[User Settings]
user, "User"

View File

@@ -82,6 +82,8 @@ help.ignore, "Ctrl+G = Ignorer", ""
help.search, "Ctrl+/ ou / = Rechercher", ""
help.help, "Ctrl+K = Aide", ""
help.no_help, "Aucune aide disponible.", ""
bot.catch_words, "ping", "Mots déclencheurs du bot séparés par des virgules."
bot.response.word, "Pong!", "Mot de reponse du bot (orthographe preferee)."
[User Settings]
user, "Utilisateur", ""

View File

@@ -90,6 +90,8 @@ confirm.remove_ignored, "Убрать {name} из игнорируемых?", ""
confirm.region_unset, "Ваш регион НЕ ЗАДАН. Установить сейчас?", ""
dialog.resize_title, "Увеличьте окно", ""
dialog.resize_body, "Пожалуйста, увеличьте окно до {rows} строк.", ""
bot.catch_words, "ping; пинг", "Слова для активации бота, разделенные точкой с запятой."
bot.response.word, "Понг!", "Ответное слово бота (предпочтительное написание)."
[User Settings]
user, "Пользователь"

View File

@@ -0,0 +1,87 @@
# A basic auto-responder bot that replies to specific messages when bot mode is enabled.
import logging
import threading
import time
from typing import Any, Dict
from contact.utilities.singleton import app_state, interface_state, ui_state
from contact.message_handlers.tx_handler import send_message
from contact.utilities.i18n import t
BOT_RESPONSE_DELAY_SECONDS = 2.3
def _get_bot_catch_words() -> set[str]:
"""Return normalized bot trigger words from localisation settings."""
raw_words = t("ui.bot.catch_words", default="ping")
words = {
word.strip().casefold()
for word in raw_words.replace(";", ",").split(",")
if word.strip()
}
return words or {"ping"}
def is_bot_message(message: str) -> bool:
"""Return True when the incoming message should trigger an automatic response."""
return message.strip().casefold() in _get_bot_catch_words()
def bot_respond(packet: Dict[str, Any], message: str, send_channel: int) -> bool:
"""Send a basic response when bot mode is enabled."""
if not ui_state.bot_mode_enabled:
return False
if not is_bot_message(message):
""" Only respond to specific messages. """
return False
from_node = packet.get("from")
if from_node is None:
return False
if from_node == interface_state.myNodeNum:
return False
snr = packet.get('rxSnr', -128)
rssi = packet.get('rxRssi', -128)
replyIDset = packet.get('replyId', False)
hop_start = packet.get('hopStart', 0)
hop_limit = packet.get('hopLimit', 0)
transport_type = packet.get('transportMechanism', None)
hops = hop_start - hop_limit
details = []
if snr != -128:
details.append(f"SNR: {snr}")
if rssi != -128:
details.append(f"RSSI: {rssi}")
if hops != 0:
details.append(f"Hops: {hops}")
if replyIDset:
details.append(f"Relay: {replyIDset}")
transport_text = str(transport_type).upper() if transport_type is not None else ""
for transport_name in ("UDP", "MQTT"):
if transport_name in transport_text:
details.append(f"Via: {transport_name}")
response_data_string = t("ui.bot.response.word", default="Pong!")
if details:
response_data_string += f" {', '.join(details)}"
def send_response_delayed() -> None:
try:
time.sleep(BOT_RESPONSE_DELAY_SECONDS)
with app_state.lock:
if not ui_state.bot_mode_enabled:
return
send_message(response_data_string,channel=send_channel)
# Import locally to avoid circular import at module import time.
from contact.ui.contact_ui import request_ui_redraw
request_ui_redraw(channels=True, messages=True, scroll_messages_to_bottom=True)
logging.info("Bot response sent to %s on channel index %s", from_node, send_channel)
except Exception:
logging.exception("Bot response send failed for destination %s", from_node)
threading.Thread(target=send_response_delayed, name="bot-response", daemon=True).start()
return True

View File

@@ -60,6 +60,7 @@ from contact.utilities.db_handler import (
import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state, app_state, menu_state
from contact.message_handlers.bot_handler import bot_respond
def play_sound():
@@ -168,6 +169,8 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
channel_number = ui_state.channel_list.index(packet["from"])
bot_respond(packet, message_string, channel_number)
channel_id = ui_state.channel_list[channel_number]
if channel_id != ui_state.channel_list[ui_state.selected_channel]:

View File

@@ -4,7 +4,7 @@ import time
import traceback
from typing import Union
from contact.utilities.utils import get_channels, get_readable_duration, get_time_ago, refresh_node_list
from contact.utilities.utils import get_channels, get_readable_duration, get_time_ago, refresh_node_list, add_new_message
from contact.settings import settings_menu
from contact.message_handlers.tx_handler import send_message, send_traceroute
from contact.utilities.utils import parse_protobuf
@@ -445,6 +445,9 @@ def main_ui(stdscr: curses.window) -> None:
elif char == chr(6): # Ctrl + F to toggle favorite
handle_ctrl_f(stdscr)
elif char == chr(2): # Ctrl + B to toggle bot responder
handle_ctrl_b(stdscr)
elif char == chr(7): # Ctrl + G to toggle ignored
handle_ctlr_g(stdscr)
@@ -919,6 +922,7 @@ def handle_ctrl_k(stdscr: curses.window) -> None:
t("ui.help.node_info", default="F5 = Full node info"),
t("ui.help.archive_chat", default="Ctrl+D = Archive chat / remove node"),
t("ui.help.favorite", default="Ctrl+F = Favorite"),
t("ui.help.bot_responder", default="Ctrl+B = Toggle Bot Responder"),
t("ui.help.ignore", default="Ctrl+G = Ignore"),
t("ui.help.search", default="Ctrl+/ = Search"),
t("ui.help.help", default="Ctrl+K = Help"),
@@ -930,6 +934,32 @@ def handle_ctrl_k(stdscr: curses.window) -> None:
handle_resize(stdscr, False)
def handle_ctrl_b(stdscr: curses.window) -> None:
"""Handle Ctrl + B key events to toggle automatic bot responses."""
ui_state.bot_mode_enabled = not ui_state.bot_mode_enabled
status = t("ui.status.enabled", default="Enabled") if ui_state.bot_mode_enabled else t(
"ui.status.disabled", default="Disabled"
)
curses.curs_set(0)
contact.ui.dialog.dialog(
t("ui.dialog.bot_responder_title", default="Bot Responder"),
t("ui.dialog.bot_responder_body", default="Bot responder is now {status}.", status=status),
)
if ui_state.channel_list:
channel_id = ui_state.channel_list[ui_state.selected_channel]
add_new_message(
channel_id,
f"{config.message_prefix} Info: ",
t("ui.status.bot_mode", default="Bot responder is now {status}.", status=status.lower()),
)
draw_messages_window(True)
curses.curs_set(1)
handle_resize(stdscr, False)
def handle_ctrl_d() -> None:
if ui_state.current_window == 0:
if isinstance(ui_state.channel_list[ui_state.selected_channel], int):

View File

@@ -27,6 +27,7 @@ class ChatUIState:
current_window: int = 0
last_sent_time: float = 0.0
last_traceroute_time: float = 0.0
bot_mode_enabled: bool = False
selected_index: int = 0
start_index: List[int] = field(default_factory=lambda: [0, 0, 0])