diff --git a/config.template b/config.template index e5c2063..54157bd 100644 --- a/config.template +++ b/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 diff --git a/data/trigger_words.json b/data/trigger_words.json new file mode 100644 index 0000000..5468552 --- /dev/null +++ b/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", "测试"] + } +} diff --git a/mesh_bot.py b/mesh_bot.py index 8c2f794..fa409e6 100755 --- a/mesh_bot.py +++ b/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]: diff --git a/modules/settings.py b/modules/settings.py index e3eb75b..37a46b3 100644 --- a/modules/settings.py +++ b/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) diff --git a/modules/stats.py b/modules/stats.py new file mode 100644 index 0000000..c9b7794 --- /dev/null +++ b/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() diff --git a/modules/system.py b/modules/system.py index 71c0b46..6649322 100644 --- a/modules/system.py +++ b/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']: diff --git a/pong_bot.py b/pong_bot.py index 40a1b68..dada62e 100755 --- a/pong_bot.py +++ b/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",\