TUI for network sync stats

shwap-tui.py

import curses
import subprocess
import time
import json
import os

def get_stats():
    try:
        result = subprocess.check_output("celestia das sampling-stats", shell=True).decode().strip()
        stats = json.loads(result)['result']
        return stats
    except subprocess.CalledProcessError:
        return {
            "head_of_sampled_chain": 0,
            "head_of_catchup": 0,
            "network_head_height": 1
        }

def get_system_stats():
    try:
        # Get CPU usage
        cpu_percent = subprocess.check_output("ps -A -o %cpu | awk '{s+=$1} END {print s}'", shell=True).decode().strip()
        
        # Get memory usage for celestia process
        mem_info = subprocess.check_output("ps -o rss= -p $(pgrep celestia)", shell=True).decode().strip()
        mem_mb = int(mem_info) / 1024  # Convert KB to MB

        # Get total system memory
        total_mem = subprocess.check_output("sysctl -n hw.memsize", shell=True).decode().strip()
        total_mem_mb = int(total_mem) / (1024 * 1024)  # Convert bytes to MB

        mem_percent = (mem_mb / total_mem_mb) * 100

        return {
            "cpu_percent": float(cpu_percent),
            "memory_percent": mem_percent,
            "memory_used": mem_mb
        }
    except subprocess.CalledProcessError:
        return None

def create_progress_bar(current, total, width=50):
    filled_width = int(width * current / total)
    bar = '█' * filled_width + '-' * (width - filled_width)
    percent = current / total * 100
    return f'[{bar}] {percent:.2f}%'

def main(stdscr):
    curses.curs_set(0)
    stdscr.nodelay(1)
    stdscr.timeout(100)

    while True:
        stdscr.clear()

        stats = get_stats()
        system_stats = get_system_stats()

        sampled_chain = int(stats['head_of_sampled_chain'])
        catchup = int(stats['head_of_catchup'])
        network_head = int(stats['network_head_height'])

        stdscr.addstr(1, 2, f"Head of Sampled Chain: {sampled_chain}")
        stdscr.addstr(3, 2, f"Head of Catchup: {catchup}")
        stdscr.addstr(5, 2, f"Network Head Height: {network_head}")
        
        progress_bar = create_progress_bar(sampled_chain, network_head)
        stdscr.addstr(7, 2, "Sync Progress:")
        stdscr.addstr(8, 2, progress_bar)

        if system_stats:
            stdscr.addstr(10, 2, f"CPU Usage (System): {system_stats['cpu_percent']:.2f}%")
            stdscr.addstr(11, 2, f"Memory Usage (Celestia): {system_stats['memory_percent']:.2f}% ({system_stats['memory_used']:.2f} MB)")
        else:
            stdscr.addstr(10, 2, "Unable to retrieve system stats")
        
        stdscr.addstr(13, 2, "Press 'q' to quit")
        stdscr.refresh()

        key = stdscr.getch()
        if key == ord('q'):
            break

        time.sleep(1)

if __name__ == "__main__":
    curses.wrapper(main)