Add custom trigger words and user statistics/leaderboard feature

Co-authored-by: SpudGunMan <12676665+SpudGunMan@users.noreply.github.com>

config.template

@@ -41,6 +41,14 @@ explicitCmd = True
 # list of favorite nodes numbers ex: 2813308004,4258675309 used by script/addFav.py
 favoriteNodeList = 
 
+# Custom trigger words for localization (comma-separated)
+# These words will trigger the same response as 'ping' command
+customPingWords = 
+# These words will trigger the same response as 'test' command
+customTestWords = 
+# Enable user statistics tracking for leaderboard/top command
+enableStatsTracking = True
+
 # motd is reset to this value on boot
 motd = Thanks for using MeshBOT! Have a good day!
 welcome_message = MeshBot, here for you like a friend who is not. Try sending: ping @foo or, cmd

data/trigger_words.json

@@ -0,0 +1,16 @@
+{
+  "ping_aliases": {
+    "description": "Custom trigger words that work like 'ping'",
+    "enabled": true,
+    "words": []
+  },
+  "test_aliases": {
+    "description": "Custom trigger words that work like 'test'",
+    "enabled": true,
+    "words": []
+  },
+  "examples": {
+    "ping_aliases": ["привет", "hola", "bonjour", "ciao"],
+    "test_aliases": ["prueba", "テスト", "testen", "测试"]
+  }
+}

mesh_bot.py

@@ -93,6 +93,8 @@ def auto_response(message, snr, rssi, hop, pkiStatus, message_from_id, channel_n
     "tictactoe": lambda: handleTicTacToe(message, message_from_id, deviceID),
     "tic-tac-toe": lambda: handleTicTacToe(message, message_from_id, deviceID),
     "tide": lambda: handle_tide(message_from_id, deviceID, channel_number),
+    "top": lambda: handle_top(message, message_from_id, deviceID, isDM),
+    "leaderboard": lambda: handle_top(message, message_from_id, deviceID, isDM),
     "valert": lambda: get_volcano_usgs(),
     "videopoker": lambda: handleVideoPoker(message, message_from_id, deviceID),
     "whereami": lambda: handle_whereami(message_from_id, deviceID, channel_number),
@@ -120,6 +122,23 @@ def auto_response(message, snr, rssi, hop, pkiStatus, message_from_id, channel_n
 
     # set the command handler
     command_handler = default_commands
+    
+    # Add custom ping words to command handler dynamically
+    if customPingWords and customPingWords[0]:
+        for word in customPingWords:
+            if word.strip():
+                word_lower = word.strip().lower()
+                if word_lower not in command_handler:
+                    command_handler[word_lower] = lambda: handle_ping(message_from_id, deviceID, message, hop, snr, rssi, isDM, channel_number)
+    
+    # Add custom test words to command handler dynamically
+    if customTestWords and customTestWords[0]:
+        for word in customTestWords:
+            if word.strip():
+                word_lower = word.strip().lower()
+                if word_lower not in command_handler:
+                    command_handler[word_lower] = lambda: handle_ping(message_from_id, deviceID, message, hop, snr, rssi, isDM, channel_number)
+    
     cmds = [] # list to hold the commands found in the message
     # check the message for commands words list, processed after system.messageTrap
     for key in command_handler:
@@ -146,6 +165,11 @@ def auto_response(message, snr, rssi, hop, pkiStatus, message_from_id, channel_n
         else:
             # run the first command after sorting
             bot_response = command_handler[cmds[0]['cmd']]()
+            
+            # Track command usage stats
+            if enableStatsTracking:
+                update_user_stat(message_from_id, 'commands', 1, deviceID)
+            
             # append the command to the cmdHistory list for lheard and history
             if len(cmdHistory) > 50:
                 cmdHistory.pop(0)
@@ -170,20 +194,37 @@ def handle_ping(message_from_id, deviceID,  message, hop, snr, rssi, isDM, chann
     
     msg = ""
     type = ''
+    message_lower = message.lower()
 
-    if "ping" in message.lower():
+    # Check for custom ping words from config
+    custom_ping_match = False
+    custom_test_match = False
+    
+    if customPingWords and customPingWords[0]:
+        for word in customPingWords:
+            if word.strip() and word.strip().lower() in message_lower:
+                custom_ping_match = True
+                break
+    
+    if customTestWords and customTestWords[0]:
+        for word in customTestWords:
+            if word.strip() and word.strip().lower() in message_lower:
+                custom_test_match = True
+                break
+
+    if "ping" in message_lower or custom_ping_match:
         msg = "🏓PONG\n"
         type = "🏓PING"
-    elif "test" in message.lower() or "testing" in message.lower():
+    elif "test" in message_lower or "testing" in message_lower or custom_test_match:
         msg = random.choice(["🎙Testing 1,2,3\n", "🎙Testing\n",\
                              "🎙Testing, testing\n",\
                              "🎙Ah-wun, ah-two...\n", "🎙Is this thing on?\n",\
                              "🎙Roger that!\n",])
         type = "🎙TEST"
-    elif "ack" in message.lower():
+    elif "ack" in message_lower:
         msg = random.choice(["✋ACK-ACK!\n", "✋Ack to you!\n"])
         type = "✋ACK"
-    elif "cqcq" in message.lower() or "cq" in message.lower() or "cqcqcq" in message.lower():
+    elif "cqcq" in message_lower or "cq" in message_lower or "cqcqcq" in message_lower:
         myname = get_name_from_number(myNodeNum, 'short', deviceID)
         msg = f"QSP QSL OM DE  {myname}   K\n"
     else:
@@ -318,6 +359,55 @@ def handle_echo(message, message_from_id, deviceID, isDM, channel_number):
     else:
         return "Please provide a message to echo back to you. Example:echo Hello World"
 
+def handle_top(message, message_from_id, deviceID, isDM):
+    """
+    Handle the top/leaderboard command for user statistics
+    Usage: 
+        top or leaderboard - shows top message senders
+        top messages - top message senders
+        top commands - top command users
+        top battery - lowest battery users
+        top online - most recently active users
+    """
+    if not enableStatsTracking:
+        return "Statistics tracking is disabled."
+    
+    if "?" in message:
+        return ("Top/Leaderboard commands:\n"
+                "top or top messages - Most messages sent\n"
+                "top commands - Most commands used\n"
+                "top battery - Lowest battery levels\n"
+                "top online - Most recently active\n"
+                "Add number for limit (e.g., top 5)")
+    
+    message_lower = message.lower()
+    stat_type = 'messages'  # default
+    limit = 10  # default
+    
+    # Parse the command to get stat type and limit
+    parts = message_lower.split()
+    for part in parts:
+        if part.isdigit():
+            limit = min(int(part), 20)  # Cap at 20
+        elif part in ['messages', 'commands', 'battery', 'online']:
+            stat_type = part
+    
+    # Generate and format the leaderboard
+    msg = format_leaderboard(stat_type, limit)
+    
+    # Replace node IDs with short names for better readability
+    if "🏆" in msg:
+        for part in msg.split():
+            part_clean = part.rstrip(':,%')
+            if len(part_clean) == 10 and part_clean.isdigit():
+                try:
+                    short_name = get_name_from_number(int(part_clean), 'short', deviceID)
+                    msg = msg.replace(part_clean, short_name)
+                except:
+                    pass
+    
+    return msg
+
 def handle_wxalert(message_from_id, deviceID, message):
     if use_meteo_wxApi:
         return "wxalert is not supported"
@@ -1415,6 +1505,12 @@ def onReceive(packet, interface):
     # if message_from_id is not in the seenNodes list add it
     if not any(node['nodeID'] == message_from_id for node in seenNodes):
         seenNodes.append({'nodeID': message_from_id, 'rxInterface': rxNode, 'channel': channel_number, 'welcome': False, 'lastSeen': time.time()})
+    else:
+        # Update last seen time for existing node
+        for node in seenNodes:
+            if node['nodeID'] == message_from_id:
+                node['lastSeen'] = time.time()
+                break
 
     # BBS DM MAIL CHECKER
     if bbs_enabled and 'decoded' in packet:
@@ -1435,6 +1531,11 @@ def onReceive(packet, interface):
             message_string = message_bytes.decode('utf-8')
             via_mqtt = packet['decoded'].get('viaMqtt', False)
             rx_time = packet['decoded'].get('rxTime', time.time())
+            
+            # Track message stats
+            if enableStatsTracking:
+                update_user_stat(message_from_id, 'messages', 1, rxNode)
+                update_user_stat(message_from_id, 'last_seen', time.time(), rxNode)
 
             # check if the packet is from us
             if message_from_id in [myNodeNum1, myNodeNum2, myNodeNum3, myNodeNum4, myNodeNum5, myNodeNum6, myNodeNum7, myNodeNum8, myNodeNum9]:

modules/settings.py

@@ -236,6 +236,9 @@ try:
     favoriteNodeList = config['general'].get('favoriteNodeList', '').split(',')
     enableEcho = config['general'].getboolean('enableEcho', False) # default False
     echoChannel = config['general'].getint('echoChannel', '9') # default 9, empty string to ignore
+    customPingWords = config['general'].get('customPingWords', '').split(',') # custom trigger words for ping
+    customTestWords = config['general'].get('customTestWords', '').split(',') # custom trigger words for test
+    enableStatsTracking = config['general'].getboolean('enableStatsTracking', True) # default True
 
     # emergency response
     emergency_responder_enabled = config['emergencyHandler'].getboolean('enabled', False)

modules/stats.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+# User Statistics Tracking Module for Leaderboard Feature
+# K7MHI Kelly Keeton 2025
+
+import time
+import json
+import os
+from modules.log import *
+
+# File to store user statistics
+STATS_FILE = "data/user_stats.json"
+
+# Global statistics dictionary
+user_stats = {}
+
+def load_stats():
+    """Load user statistics from file"""
+    global user_stats
+    try:
+        if os.path.exists(STATS_FILE):
+            with open(STATS_FILE, 'r') as f:
+                user_stats = json.load(f)
+                logger.debug(f"Loaded user stats: {len(user_stats)} users")
+        else:
+            user_stats = {}
+            logger.info(f"No existing stats file found, starting fresh")
+    except Exception as e:
+        logger.error(f"Error loading user stats: {e}")
+        user_stats = {}
+
+def save_stats():
+    """Save user statistics to file"""
+    try:
+        os.makedirs(os.path.dirname(STATS_FILE), exist_ok=True)
+        with open(STATS_FILE, 'w') as f:
+            json.dump(user_stats, f, indent=2)
+    except Exception as e:
+        logger.error(f"Error saving user stats: {e}")
+
+def update_user_stat(user_id, stat_type, value=1, device_id=1):
+    """
+    Update a specific statistic for a user
+    
+    Args:
+        user_id: User's node ID
+        stat_type: Type of stat ('messages', 'commands', 'last_seen', 'battery', etc.)
+        value: Value to add or set (default 1 for counters)
+        device_id: Device interface ID
+    """
+    global user_stats
+    
+    user_id_str = str(user_id)
+    
+    if user_id_str not in user_stats:
+        user_stats[user_id_str] = {
+            'messages': 0,
+            'commands': 0,
+            'first_seen': time.time(),
+            'last_seen': time.time(),
+            'battery': 100,
+            'uptime': 0,
+            'device_id': device_id
+        }
+    
+    # Update the specific stat
+    if stat_type in ['messages', 'commands']:
+        user_stats[user_id_str][stat_type] += value
+    else:
+        user_stats[user_id_str][stat_type] = value
+    
+    # Always update last_seen when any stat is updated
+    user_stats[user_id_str]['last_seen'] = time.time()
+
+def get_top_users(stat_type='messages', limit=10, timeframe=None):
+    """
+    Get top users for a specific statistic
+    
+    Args:
+        stat_type: Type of stat to sort by
+        limit: Number of top users to return
+        timeframe: Optional time range in seconds (e.g., 86400 for last 24 hours)
+    
+    Returns:
+        List of tuples: [(user_id, stat_value), ...]
+    """
+    global user_stats
+    current_time = time.time()
+    
+    # Filter by timeframe if specified
+    filtered_stats = user_stats
+    if timeframe:
+        filtered_stats = {
+            uid: stats for uid, stats in user_stats.items()
+            if current_time - stats.get('last_seen', 0) <= timeframe
+        }
+    
+    # Sort by the requested stat
+    if stat_type in ['messages', 'commands', 'uptime']:
+        sorted_users = sorted(
+            filtered_stats.items(),
+            key=lambda x: x[1].get(stat_type, 0),
+            reverse=True
+        )
+    elif stat_type == 'battery':
+        # For battery, lower is "better" for the depleted leaderboard
+        sorted_users = sorted(
+            filtered_stats.items(),
+            key=lambda x: x[1].get(stat_type, 100),
+            reverse=False
+        )
+    elif stat_type == 'online':
+        # Most recently seen
+        sorted_users = sorted(
+            filtered_stats.items(),
+            key=lambda x: x[1].get('last_seen', 0),
+            reverse=True
+        )
+    else:
+        sorted_users = sorted(
+            filtered_stats.items(),
+            key=lambda x: x[1].get(stat_type, 0),
+            reverse=True
+        )
+    
+    return [(uid, stats.get(stat_type, 0)) for uid, stats in sorted_users[:limit]]
+
+def format_leaderboard(stat_type, limit=10, timeframe=None):
+    """
+    Format a leaderboard message for display
+    
+    Args:
+        stat_type: Type of stat to display
+        limit: Number of users to show
+        timeframe: Optional timeframe in seconds
+    
+    Returns:
+        Formatted string for display
+    """
+    top_users = get_top_users(stat_type, limit, timeframe)
+    
+    if not top_users:
+        return "No statistics available yet."
+    
+    # Stat type display names
+    stat_names = {
+        'messages': 'Most Messages',
+        'commands': 'Most Commands',
+        'battery': 'Lowest Battery',
+        'online': 'Most Recently Active',
+        'uptime': 'Highest Uptime'
+    }
+    
+    title = stat_names.get(stat_type, f'Top {stat_type.title()}')
+    timeframe_str = ""
+    if timeframe == 86400:
+        timeframe_str = " (24h)"
+    elif timeframe == 604800:
+        timeframe_str = " (7d)"
+    
+    msg = f"🏆 {title}{timeframe_str}:\n"
+    
+    for idx, (uid, value) in enumerate(top_users, start=1):
+        if stat_type == 'battery':
+            msg += f"{idx}. {uid}: {value}%\n"
+        elif stat_type == 'online':
+            # Show how long ago they were seen
+            elapsed = time.time() - value
+            if elapsed < 60:
+                time_str = f"{int(elapsed)}s ago"
+            elif elapsed < 3600:
+                time_str = f"{int(elapsed/60)}m ago"
+            else:
+                time_str = f"{int(elapsed/3600)}h ago"
+            msg += f"{idx}. {uid}: {time_str}\n"
+        elif stat_type == 'uptime':
+            # Convert seconds to hours
+            hours = int(value / 3600)
+            msg += f"{idx}. {uid}: {hours}h\n"
+        else:
+            msg += f"{idx}. {uid}: {value}\n"
+    
+    return msg
+
+# Load stats on module import
+load_stats()

modules/system.py

@@ -56,6 +56,11 @@ def cleanup_memory():
                               if ping.get('message_from_id', 0) != 0 and 
                               ping.get('count', 0) > 0]
         
+        # Save user statistics if stats tracking is enabled
+        if enableStatsTracking:
+            save_stats()
+            logger.debug(f"System: Saved user statistics")
+        
     except Exception as e:
         logger.error(f"System: Error during memory cleanup: {e}")
 
@@ -89,8 +94,21 @@ def cleanup_game_trackers(current_time):
 # Ping Configuration
 if ping_enabled:
     # ping, pinging, ack, testing, test, pong
-    trap_list_ping = ("ping", "pinging", "ack", "testing", "test", "pong", "🔔", "cq","cqcq", "cqcqcq")
-    trap_list = trap_list + trap_list_ping
+    trap_list_ping = ["ping", "pinging", "ack", "testing", "test", "pong", "🔔", "cq","cqcq", "cqcqcq"]
+    
+    # Add custom ping words from config
+    if customPingWords and customPingWords[0]:  # Check if list is not empty and first item is not empty string
+        custom_ping = [word.strip().lower() for word in customPingWords if word.strip()]
+        trap_list_ping.extend(custom_ping)
+        logger.info(f"System: Added custom ping words: {custom_ping}")
+    
+    # Add custom test words from config
+    if customTestWords and customTestWords[0]:  # Check if list is not empty and first item is not empty string
+        custom_test = [word.strip().lower() for word in customTestWords if word.strip()]
+        trap_list_ping.extend(custom_test)
+        logger.info(f"System: Added custom test words: {custom_test}")
+    
+    trap_list = trap_list + tuple(trap_list_ping)
     help_message = help_message + "ping"
 
 # Echo Configuration
@@ -128,6 +146,12 @@ if whoami_enabled:
     trap_list = trap_list + trap_list_whoami
     help_message = help_message + ", whoami"
 
+# Stats Tracking Configuration
+if enableStatsTracking:
+    from modules.stats import * # from the spudgunman/meshing-around repo
+    trap_list = trap_list + ("top", "leaderboard")
+    help_message = help_message + ", top"
+
 # Solar Conditions Configuration
 if solar_conditions_enabled:
     from modules.space import * # from the spudgunman/meshing-around repo
@@ -1099,6 +1123,14 @@ def consumeMetadata(packet, rxNode=0, channel=-1):
             telemetry_packet = packet['decoded']['telemetry']
             if telemetry_packet.get('deviceMetrics'):
                 deviceMetrics = telemetry_packet['deviceMetrics']
+                
+                # Track battery and uptime stats
+                if enableStatsTracking:
+                    if deviceMetrics.get('batteryLevel') is not None:
+                        update_user_stat(nodeID, 'battery', deviceMetrics['batteryLevel'], rxNode)
+                    if deviceMetrics.get('uptimeSeconds') is not None:
+                        update_user_stat(nodeID, 'uptime', deviceMetrics['uptimeSeconds'], rxNode)
+                
                 #if uptime is in deviceMetrics and uptime is not 0 set uptime
                 # if deviceMetrics.get('uptimeSeconds') is not None and deviceMetrics['uptimeSeconds'] != 0:
                 #     if highestUptime < deviceMetrics['uptimeSeconds']:

pong_bot.py

@@ -39,6 +39,23 @@ def auto_response(message, snr, rssi, hop, pkiStatus, message_from_id, channel_n
         "test": lambda: handle_ping(message_from_id, deviceID, message, hop, snr, rssi, isDM, channel_number),
         "testing": lambda: handle_ping(message_from_id, deviceID, message, hop, snr, rssi, isDM, channel_number),
     }
+    
+    # Add custom ping words to command handler dynamically
+    if customPingWords and customPingWords[0]:
+        for word in customPingWords:
+            if word.strip():
+                word_lower = word.strip().lower()
+                if word_lower not in command_handler:
+                    command_handler[word_lower] = lambda: handle_ping(message_from_id, deviceID, message, hop, snr, rssi, isDM, channel_number)
+    
+    # Add custom test words to command handler dynamically
+    if customTestWords and customTestWords[0]:
+        for word in customTestWords:
+            if word.strip():
+                word_lower = word.strip().lower()
+                if word_lower not in command_handler:
+                    command_handler[word_lower] = lambda: handle_ping(message_from_id, deviceID, message, hop, snr, rssi, isDM, channel_number)
+    
     cmds = [] # list to hold the commands found in the message
     for key in command_handler:
         if key in message_lower.split(' '):
@@ -70,11 +87,28 @@ def handle_ping(message_from_id, deviceID,  message, hop, snr, rssi, isDM, chann
     
     msg = ""
     type = ''
+    message_lower = message.lower()
+    
+    # Check for custom ping words from config
+    custom_ping_match = False
+    custom_test_match = False
+    
+    if customPingWords and customPingWords[0]:
+        for word in customPingWords:
+            if word.strip() and word.strip().lower() in message_lower:
+                custom_ping_match = True
+                break
+    
+    if customTestWords and customTestWords[0]:
+        for word in customTestWords:
+            if word.strip() and word.strip().lower() in message_lower:
+                custom_test_match = True
+                break
 
-    if "ping" in message.lower():
+    if "ping" in message_lower or custom_ping_match:
         msg = "🏓PONG\n"
         type = "🏓PING"
-    elif "test" in message.lower() or "testing" in message.lower():
+    elif "test" in message_lower or "testing" in message_lower or custom_test_match:
         msg = random.choice(["🎙Testing 1,2,3\n", "🎙Testing\n",\
                              "🎙Testing, testing\n",\
                              "🎙Ah-wun, ah-two...\n", "🎙Is this thing on?\n",\