From c74e4f99b253a041210bbb0b809b3655e68a5111 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 8 Oct 2025 16:50:46 +0000 Subject: [PATCH] Add mesh leaderboard feature to track extreme metrics Co-authored-by: SpudGunMan <12676665+SpudGunMan@users.noreply.github.com> --- mesh_bot.py | 1 + modules/system.py | 203 +++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 192 insertions(+), 12 deletions(-) diff --git a/mesh_bot.py b/mesh_bot.py index 8c2f794..d93296a 100755 --- a/mesh_bot.py +++ b/mesh_bot.py @@ -63,6 +63,7 @@ def auto_response(message, snr, rssi, hop, pkiStatus, message_from_id, channel_n "howfar": lambda: handle_howfar(message, message_from_id, deviceID, isDM), "howtall": lambda: handle_howtall(message, message_from_id, deviceID, isDM), "joke": lambda: tell_joke(message_from_id), + "leaderboard": lambda: get_mesh_leaderboard(), "lemonstand": lambda: handleLemonade(message, message_from_id, deviceID), "lheard": lambda: handle_lheard(message, message_from_id, deviceID, isDM), "mastermind": lambda: handleMmind(message, message_from_id, deviceID), diff --git a/modules/system.py b/modules/system.py index 71c0b46..5364465 100644 --- a/modules/system.py +++ b/modules/system.py @@ -101,9 +101,9 @@ if enableEcho: # Sitrep Configuration if sitrep_enabled: - trap_list_sitrep = ("sitrep", "lheard", "sysinfo") + trap_list_sitrep = ("sitrep", "lheard", "sysinfo", "leaderboard") trap_list = trap_list + trap_list_sitrep - help_message = help_message + ", sitrep, sysinfo" + help_message = help_message + ", sitrep, sysinfo, leaderboard" # MOTD Configuration if motd_enabled: @@ -1078,8 +1078,24 @@ def displayNodeTelemetry(nodeID=0, rxNode=0, userRequested=False): return dataResponse positionMetadata = {} + +# Leaderboard for tracking extreme metrics +meshLeaderboard = { + 'lowestBattery': {'nodeID': None, 'value': 101, 'timestamp': 0}, # đŸĒĢ + 'longestUptime': {'nodeID': None, 'value': 0, 'timestamp': 0}, # 🕰ī¸ + 'fastestSpeed': {'nodeID': None, 'value': 0, 'timestamp': 0}, # 🚓 + 'highestAltitude': {'nodeID': None, 'value': 0, 'timestamp': 0}, # 🚀 + 'coldestTemp': {'nodeID': None, 'value': 999, 'timestamp': 0}, # đŸĨļ + 'hottestTemp': {'nodeID': None, 'value': -999, 'timestamp': 0}, # đŸĨĩ + 'worstAirQuality': {'nodeID': None, 'value': 0, 'timestamp': 0}, # 💨 + 'adminPackets': [], # 🚨 + 'tunnelPackets': [], # 🚨 + 'audioPackets': [], # ☎ī¸ + 'simulatorPackets': [] # 🤖 +} + def consumeMetadata(packet, rxNode=0, channel=-1): - global positionMetadata, telemetryData + global positionMetadata, telemetryData, meshLeaderboard # check type of packet try: @@ -1099,11 +1115,45 @@ def consumeMetadata(packet, rxNode=0, channel=-1): telemetry_packet = packet['decoded']['telemetry'] if telemetry_packet.get('deviceMetrics'): deviceMetrics = telemetry_packet['deviceMetrics'] - #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']: - # highestUptime = deviceMetrics['uptimeSeconds'] - # highestUptimeNode = nodeID + current_time = time.time() + + # Track lowest battery đŸĒĢ + if deviceMetrics.get('batteryLevel') is not None: + battery = deviceMetrics['batteryLevel'] + if battery > 0 and battery < meshLeaderboard['lowestBattery']['value']: + meshLeaderboard['lowestBattery'] = {'nodeID': nodeID, 'value': battery, 'timestamp': current_time} + logger.info(f"System: đŸĒĢ New low battery record: {battery}% from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + + # Track longest uptime 🕰ī¸ + if deviceMetrics.get('uptimeSeconds') is not None: + uptime = deviceMetrics['uptimeSeconds'] + if uptime > meshLeaderboard['longestUptime']['value']: + meshLeaderboard['longestUptime'] = {'nodeID': nodeID, 'value': uptime, 'timestamp': current_time} + logger.info(f"System: 🕰ī¸ New uptime record: {getPrettyTime(uptime)} from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + + # Track environment metrics (temperature, air quality) + if telemetry_packet.get('environmentMetrics'): + envMetrics = telemetry_packet['environmentMetrics'] + current_time = time.time() + + # Track coldest temperature đŸĨļ + if envMetrics.get('temperature') is not None: + temp = envMetrics['temperature'] + if temp < meshLeaderboard['coldestTemp']['value']: + meshLeaderboard['coldestTemp'] = {'nodeID': nodeID, 'value': temp, 'timestamp': current_time} + logger.info(f"System: đŸĨļ New coldest temp record: {temp}°C from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + + # Track hottest temperature đŸĨĩ + if temp > meshLeaderboard['hottestTemp']['value']: + meshLeaderboard['hottestTemp'] = {'nodeID': nodeID, 'value': temp, 'timestamp': current_time} + logger.info(f"System: đŸĨĩ New hottest temp record: {temp}°C from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + + # Track worst air quality 💨 (IAQ - higher is worse) + if envMetrics.get('iaq') is not None: + iaq = envMetrics['iaq'] + if iaq > meshLeaderboard['worstAirQuality']['value']: + meshLeaderboard['worstAirQuality'] = {'nodeID': nodeID, 'value': iaq, 'timestamp': current_time} + logger.info(f"System: 💨 New worst air quality record: IAQ {iaq} from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") if telemetry_packet.get('localStats'): localStats = telemetry_packet['localStats'] @@ -1133,6 +1183,20 @@ def consumeMetadata(packet, rxNode=0, channel=-1): for key in keys: positionMetadata[nodeID][key] = position_data.get(key, 0) + + # Track fastest speed 🚓 + if position_data.get('groundSpeed') is not None: + speed = position_data['groundSpeed'] + if speed > meshLeaderboard['fastestSpeed']['value']: + meshLeaderboard['fastestSpeed'] = {'nodeID': nodeID, 'value': speed, 'timestamp': time.time()} + logger.info(f"System: 🚓 New speed record: {speed} km/h from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + + # Track highest altitude 🚀 (also log if over highfly_altitude threshold) + if position_data.get('altitude') is not None: + altitude = position_data['altitude'] + if altitude > meshLeaderboard['highestAltitude']['value']: + meshLeaderboard['highestAltitude'] = {'nodeID': nodeID, 'value': altitude, 'timestamp': time.time()} + logger.info(f"System: 🚀 New altitude record: {altitude}m from NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") # if altitude is over highfly_altitude send a log and message for high-flying nodes and not in highfly_ignoreList if position_data.get('altitude', 0) > highfly_altitude and highfly_enabled and str(nodeID) not in highfly_ignoreList: @@ -1246,9 +1310,33 @@ def consumeMetadata(packet, rxNode=0, channel=-1): except Exception as e: logger.debug(f"System: REMOTE_HARDWARE_APP decode error: Device: {rxNode} Channel: {channel} {e} packet {packet}") - # ADMIN_APP + # ADMIN_APP - Track admin packets 🚨 + if packet_type == 'ADMIN_APP': + if debugMetadata and 'ADMIN_APP' not in metadataFilter: + print(f"DEBUG ADMIN_APP: {packet}\n\n") + try: + packet_info = {'nodeID': nodeID, 'timestamp': time.time(), 'device': rxNode, 'channel': channel} + # Keep only last 10 admin packets + meshLeaderboard['adminPackets'].append(packet_info) + if len(meshLeaderboard['adminPackets']) > 10: + meshLeaderboard['adminPackets'].pop(0) + logger.info(f"System: 🚨 Admin packet detected from Device: {rxNode} Channel: {channel} NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + except Exception as e: + logger.debug(f"System: ADMIN_APP decode error: Device: {rxNode} Channel: {channel} {e} packet {packet}") - # IP_TUNNEL_APP + # IP_TUNNEL_APP - Track tunneling packets 🚨 + if packet_type == 'IP_TUNNEL_APP': + if debugMetadata and 'IP_TUNNEL_APP' not in metadataFilter: + print(f"DEBUG IP_TUNNEL_APP: {packet}\n\n") + try: + packet_info = {'nodeID': nodeID, 'timestamp': time.time(), 'device': rxNode, 'channel': channel} + # Keep only last 10 tunnel packets + meshLeaderboard['tunnelPackets'].append(packet_info) + if len(meshLeaderboard['tunnelPackets']) > 10: + meshLeaderboard['tunnelPackets'].pop(0) + logger.info(f"System: 🚨 IP Tunnel packet detected from Device: {rxNode} Channel: {channel} NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + except Exception as e: + logger.debug(f"System: IP_TUNNEL_APP decode error: Device: {rxNode} Channel: {channel} {e} packet {packet}") # SERIAL_APP @@ -1258,9 +1346,33 @@ def consumeMetadata(packet, rxNode=0, channel=-1): # COMPRESSED_TEXT_APP - # AUDIO_APP + # AUDIO_APP - Track audio/voice packets ☎ī¸ + if packet_type == 'AUDIO_APP': + if debugMetadata and 'AUDIO_APP' not in metadataFilter: + print(f"DEBUG AUDIO_APP: {packet}\n\n") + try: + packet_info = {'nodeID': nodeID, 'timestamp': time.time(), 'device': rxNode, 'channel': channel} + # Keep only last 10 audio packets + meshLeaderboard['audioPackets'].append(packet_info) + if len(meshLeaderboard['audioPackets']) > 10: + meshLeaderboard['audioPackets'].pop(0) + logger.info(f"System: ☎ī¸ Audio packet detected from Device: {rxNode} Channel: {channel} NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + except Exception as e: + logger.debug(f"System: AUDIO_APP decode error: Device: {rxNode} Channel: {channel} {e} packet {packet}") - # SIMULATOR_APP + # SIMULATOR_APP - Track simulator packets 🤖 + if packet_type == 'SIMULATOR_APP': + if debugMetadata and 'SIMULATOR_APP' not in metadataFilter: + print(f"DEBUG SIMULATOR_APP: {packet}\n\n") + try: + packet_info = {'nodeID': nodeID, 'timestamp': time.time(), 'device': rxNode, 'channel': channel} + # Keep only last 10 simulator packets + meshLeaderboard['simulatorPackets'].append(packet_info) + if len(meshLeaderboard['simulatorPackets']) > 10: + meshLeaderboard['simulatorPackets'].pop(0) + logger.info(f"System: 🤖 Simulator packet detected from Device: {rxNode} Channel: {channel} NodeID:{nodeID} ShortName:{get_name_from_number(nodeID, 'short', rxNode)}") + except Exception as e: + logger.debug(f"System: SIMULATOR_APP decode error: Device: {rxNode} Channel: {channel} {e} packet {packet}") return True def noisyTelemetryCheck(): @@ -1276,6 +1388,73 @@ def noisyTelemetryCheck(): # reset the packet count for the node positionMetadata[nodeID]['packetCount'] = 0 +def get_mesh_leaderboard(): + """Get formatted leaderboard of extreme mesh metrics""" + global meshLeaderboard + + result = "📊 Mesh Leaderboard 📊\n" + + # Lowest battery + if meshLeaderboard['lowestBattery']['nodeID']: + nodeID = meshLeaderboard['lowestBattery']['nodeID'] + value = meshLeaderboard['lowestBattery']['value'] + result += f"đŸĒĢ Low Battery: {value}% {get_name_from_number(nodeID, 'short', 1)}\n" + + # Longest uptime + if meshLeaderboard['longestUptime']['nodeID']: + nodeID = meshLeaderboard['longestUptime']['nodeID'] + value = meshLeaderboard['longestUptime']['value'] + result += f"🕰ī¸ Longest Uptime: {getPrettyTime(value)} {get_name_from_number(nodeID, 'short', 1)}\n" + + # Fastest speed + if meshLeaderboard['fastestSpeed']['nodeID']: + nodeID = meshLeaderboard['fastestSpeed']['nodeID'] + value = meshLeaderboard['fastestSpeed']['value'] + result += f"🚓 Fastest Speed: {value} km/h {get_name_from_number(nodeID, 'short', 1)}\n" + + # Highest altitude + if meshLeaderboard['highestAltitude']['nodeID']: + nodeID = meshLeaderboard['highestAltitude']['nodeID'] + value = meshLeaderboard['highestAltitude']['value'] + altFeet = round(value * 3.28084, 0) + result += f"🚀 Highest Alt: {altFeet:,.0f}ft/{value:,.0f}m {get_name_from_number(nodeID, 'short', 1)}\n" + + # Coldest temperature + if meshLeaderboard['coldestTemp']['nodeID']: + nodeID = meshLeaderboard['coldestTemp']['nodeID'] + value = meshLeaderboard['coldestTemp']['value'] + result += f"đŸĨļ Coldest: {value}°C {get_name_from_number(nodeID, 'short', 1)}\n" + + # Hottest temperature + if meshLeaderboard['hottestTemp']['nodeID']: + nodeID = meshLeaderboard['hottestTemp']['nodeID'] + value = meshLeaderboard['hottestTemp']['value'] + result += f"đŸĨĩ Hottest: {value}°C {get_name_from_number(nodeID, 'short', 1)}\n" + + # Worst air quality + if meshLeaderboard['worstAirQuality']['nodeID']: + nodeID = meshLeaderboard['worstAirQuality']['nodeID'] + value = meshLeaderboard['worstAirQuality']['value'] + result += f"💨 Worst Air: IAQ {value} {get_name_from_number(nodeID, 'short', 1)}\n" + + # Special packet detections + if len(meshLeaderboard['adminPackets']) > 0: + result += f"🚨 Admin packets: {len(meshLeaderboard['adminPackets'])}\n" + + if len(meshLeaderboard['tunnelPackets']) > 0: + result += f"🚨 Tunnel packets: {len(meshLeaderboard['tunnelPackets'])}\n" + + if len(meshLeaderboard['audioPackets']) > 0: + result += f"☎ī¸ Audio packets: {len(meshLeaderboard['audioPackets'])}\n" + + if len(meshLeaderboard['simulatorPackets']) > 0: + result += f"🤖 Simulator packets: {len(meshLeaderboard['simulatorPackets'])}\n" + + if result == "📊 Mesh Leaderboard 📊\n": + result += "No records yet! Keep meshing! 📡" + + return result + def get_sysinfo(nodeID=0, deviceID=1): # Get the system telemetry data for return on the sysinfo command sysinfo = ''