a rough script to parse and display meshtastic nodedb

node_list.py

import json
import sys
import argparse
import subprocess
from functools import reduce
from prettytable import PrettyTable
from datetime import datetime

def convert_timestamp_to_age(timestamp):
    current_time = int(datetime.now().timestamp())
    return current_time - int(timestamp)

def format_duration(seconds):
    days = seconds // (24 * 3600)
    seconds %= (24 * 3600)
    hours = seconds // 3600
    seconds %= 3600
    minutes = seconds // 60
    seconds %= 60
    parts = []
    if days > 0:
        parts.append(f"{days}d")
    if hours > 0:
        parts.append(f"{hours}h")
    if minutes > 0:
        parts.append(f"{minutes}m")
    if seconds > 0 or not parts:
        parts.append(f"{seconds}s")
    return " ".join(parts)

def flatten(lst):
    return [item for sublist in lst for item in (flatten(sublist) if isinstance(sublist, list) else [sublist])]

def remove_inactive_nodes(data, meshtastic_flags):
    for key, value in data.items():
        if ('lastHeard' in value and 'lastHeardRaw' in value and value['lastHeardRaw'] > 3600) or 'lastHeard' not in value:
            user_id = value['user']['id']
            command = flatten(["meshtastic", meshtastic_flags.split(" "), "--remove-node", user_id])
            print(f"removing node: {value['user']['longName']}")
            subprocess.run(command)

def process_meshtastic_output(output):
    lines = output.splitlines()
    start_index = None
    end_index = None

    lines = [line for line in lines if line.strip()]
    for i, line in enumerate(lines):
        if "Nodes in mesh" in line:
            start_index = i
        if "Preferences" in line:
            end_index = i - 1
            break

    if start_index is not None and end_index is not None:
        json_lines = lines[start_index:end_index + 1]
        json_lines[0] = "{"
        json_text = "\n".join(json_lines)
        return json_text
    return None

def main():
    parser = argparse.ArgumentParser(description="Replace lastHeard field with age and display JSON as table.")
    parser.add_argument('--fullname', action='store_true', help="Display user.longName column")
    parser.add_argument('--shortname', action='store_true', help="Display user.shortName column")
    parser.add_argument('--macaddr', action='store_true', help="Display user.macaddr column")
    parser.add_argument('--hwmodel', action='store_true', help="Display user.hwModel column")
    parser.add_argument('--publickey', action='store_true', help="Display user.publicKey column")
    parser.add_argument('--num', action='store_true', help="Display num column")
    parser.add_argument('--snr', action='store_true', help="Display snr column")
    parser.add_argument('--batteryLevel', action='store_true', help="Display deviceMetrics.batteryLevel column")
    parser.add_argument('--voltage', action='store_true', help="Display deviceMetrics.voltage column")
    parser.add_argument('--channelUtilization', action='store_true', help="Display deviceMetrics.channelUtilization column")
    parser.add_argument('--airUtilTx', action='store_true', help="Display deviceMetrics.airUtilTx column")
    parser.add_argument('--uptimeSeconds', action='store_true', help="Display deviceMetrics.uptimeSeconds column")
    parser.add_argument('--hopsAway', action='store_true', help="Display hopsAway column")
    parser.add_argument('--isFavorite', action='store_true', help="Display isFavorite column")
    parser.add_argument('--latitude', action='store_true', help="Display position.latitude column")
    parser.add_argument('--longitude', action='store_true', help="Display position.longitude column")
    parser.add_argument('--altitude', action='store_true', help="Display position.altitude column")
    parser.add_argument('--positionLatitudeI', action='store_true', help="Display position.latitudeI column")
    parser.add_argument('--positionLongitudeI', action='store_true', help="Display position.longitudeI column")
    parser.add_argument('--positionTime', action='store_true', help="Display position.time column")
    parser.add_argument('--locationSource', action='store_true', help="Display position.locationSource column")
    parser.add_argument('--lastheard', action='store_true', help="Display lastHeard column")
    parser.add_argument('--remove-inactive', action='store_true', help="Remove inactive nodes")
    parser.add_argument('--meshtastic', type=str, help="Meshtastic command to run")
    parser.add_argument('--json', action='store_true', help="print json instead")

    args = parser.parse_args()

    if args.meshtastic:
        meshtastic_command = ["meshtastic"]
        meshtastic_command.extend(args.meshtastic.split() + ['--info'])
        result = subprocess.run(meshtastic_command, capture_output=True, text=True)
        output = result.stdout
        import re
        myNodeNum = re.search(r'"myNodeNum": (\d+)', output).group(1)
        json_text = process_meshtastic_output(output)
        if json_text:
            data = json.loads(json_text)
        else:
            print("Failed to extract JSON from meshtastic output.")
            sys.exit(1)
    else:
        print("Please provide the --meshtastic flag with the command to run.")
        sys.exit(1)
    
    if args.json:
        print(json.dumps(data, indent=2))
        sys.exit(0)


    # Replace lastHeard with age and store raw age for filtering
    for key, value in data.items():
        if 'lastHeard' in value:
            value['lastHeardRaw'] = convert_timestamp_to_age(value['lastHeard'])
            value['lastHeard'] = format_duration(value['lastHeardRaw'])


    # Remove inactive nodes if the flag is set
    if args.remove_inactive:
        remove_inactive_nodes(data, args.meshtastic)
        # Filter out inactive nodes from the display
        data = {k: v for k, v in data.items() if 'lastHeardRaw' not in v or v['lastHeardRaw'] <= 3600}

    data = sorted(data.items(), key=lambda item: item[1].get('lastHeardRaw', float('inf')))
    if args.hopsAway:
        data = sorted(data, key=lambda item: item[1].get('hopsAway', float('inf')))

    # Prepare data for the table
    table = PrettyTable()

    columns = []

    columns.append('longName')
    columns.append('lastHeard')

    if args.shortname:
        columns.append('shortName')

    if args.macaddr:
        columns.append('macaddr')

    if args.hwmodel:
        columns.append('hwModel')

    if args.publickey:
        columns.append('publicKey')

    if args.num:
        columns.append('num')

    if args.snr:
        columns.append('snr')

    if args.batteryLevel:
        columns.append('batteryLevel')

    if args.voltage:
        columns.append('voltage')

    if args.channelUtilization:
        columns.append('channelUtilization')

    if args.airUtilTx:
        columns.append('airUtilTx')

    if args.uptimeSeconds:
        columns.append('uptimeSeconds')

    if args.hopsAway:
        columns.append('hopsAway')

    if args.isFavorite:
        columns.append('isFavorite')

    if args.latitude:
        columns.append('latitude')

    if args.longitude:
        columns.append('longitude')

    if args.altitude:
        columns.append('altitude')

    if args.positionLatitudeI:
        columns.append('latitudeI')

    if args.positionLongitudeI:
        columns.append('longitudeI')

    if args.positionTime:
        columns.append('time')

    if args.locationSource:
        columns.append('locationSource')

    table.field_names = columns

    for key, value in data:
        if int(value['num']) == int(myNodeNum):
            continue
        row = []
        row.append(value['user']['longName'])
        row.append(value.get('lastHeard', 'N/A'))
        if 'shortName' in columns:
            row.append(value['user']['shortName'])
        if 'macaddr' in columns:
            row.append(value['user']['macaddr'])
        if 'hwModel' in columns:
            row.append(value['user']['hwModel'])
        if 'publicKey' in columns:
            row.append(value['user']['publicKey'])
        if 'num' in columns:
            row.append(value['num'])
        if 'snr' in columns:
            row.append(value['snr'])
        if 'batteryLevel' in columns:
            row.append(value.get('deviceMetrics', {}).get('batteryLevel', 'N/A'))
        if 'voltage' in columns:
            row.append(value.get('deviceMetrics', {}).get('voltage', 'N/A'))
        if 'channelUtilization' in columns:
            row.append(value.get('deviceMetrics', {}).get('channelUtilization', 'N/A'))
        if 'airUtilTx' in columns:
            row.append(value.get('deviceMetrics', {}).get('airUtilTx', 'N/A'))
        if 'uptimeSeconds' in columns:
            row.append(value.get('deviceMetrics', {}).get('uptimeSeconds', 'N/A'))
        if 'hopsAway' in columns:
            row.append(value.get('hopsAway', 'N/A'))
        if 'isFavorite' in columns:
            row.append(value.get('isFavorite', 'N/A'))
        if 'latitude' in columns:
            row.append(value.get('position', {}).get('latitude', 'N/A'))
        if 'longitude' in columns:
            row.append(value.get('position', {}).get('longitude', 'N/A'))
        if 'altitude' in columns:
            row.append(value.get('position', {}).get('altitude', 'N/A'))
        if 'latitudeI' in columns:
            row.append(value.get('position', {}).get('latitudeI', 'N/A'))
        if 'longitudeI' in columns:
            row.append(value.get('position', {}).get('longitudeI', 'N/A'))
        if 'time' in columns:
            row.append(value.get('position', {}).get('time', 'N/A'))
        if 'locationSource' in columns:
            row.append(value.get('position', {}).get('locationSource', 'N/A'))

        table.add_row(row)

    if not args.remove_inactive:
      print(table)

if __name__ == "__main__":
    main()