piermesh/src/tlog.py

# TUI library
import curses
from curses import wrapper

# Generic imports
import os
import logging
import queue
import shutil
import time

# System monitoring library
import psutil

# TODO: Multi node mode
# TODO: Tab per node

class VHandler(logging.Handler):
    """
    Custom log handler to push logs into a thread-safe queue so the TUI can read them

    `🔗 Source <https://git.utopic.work/PierMesh/piermesh/src/branch/main/src/tlog.py>`_

    Attributes
    ----------

    tolog: Queue.queue
        Thread-safe log queue
    
    """
    tolog = queue.Queue()

    def __init__(self, level, tolog):
        super().__init__(level)
        self.tolog = tolog

    def emit(self, record):
        r = self.format(record)
        self.tolog.put([r, record.levelno])

def logUI(stdscr, tolog, nodeNickname):
    """
    TUI loop
    """
    logcache = []
    height, width = shutil.get_terminal_size((49, 27))
    if height < 28 or height < 28:
        print("Console too small or couldnt retrieve size, please switch to no tui mode, exiting...")
        exit()
    logger = logging.getLogger("__main__." + __name__)
    p = psutil.Process(os.getpid())

    curses.start_color()
    stdscr.keypad(True)

    stdscr.nodelay(True)
    curses.init_color(9, 200, 200, 200)
    grey = 9

    curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLACK)
    curses.init_pair(2, curses.COLOR_BLACK, curses.COLOR_GREEN)
    curses.init_pair(3, curses.COLOR_WHITE, grey)

    curses.init_color(curses.COLOR_WHITE, 1000, 1000, 1000)
    curses.init_color(curses.COLOR_YELLOW, 1000, 1000, 0)
    curses.init_color(curses.COLOR_GREEN, 0, 1000, 0)

    curses.init_color(logging.DEBUG, 0, 1000, 0)
    curses.init_pair(logging.DEBUG, 0, logging.DEBUG)
    curses.init_color(logging.INFO, 1000, 1000, 1000)
    curses.init_pair(logging.INFO, 0, logging.INFO)
    curses.init_pair(logging.WARNING, 0, curses.COLOR_YELLOW)
    curses.init_pair(logging.ERROR, curses.COLOR_WHITE, curses.COLOR_RED)
    curses.init_color(logging.CRITICAL, 1000, 0, 0)
    curses.init_pair(logging.CRITICAL, curses.COLOR_WHITE, logging.CRITICAL)
    icon = []
    with open("piermesh-mini.ascii", "r") as f:
        icon = f.read().split("\n")

    stdscr.bkgd(' ', curses.color_pair(1))
    stdscr.clear()

    iwin = curses.newwin(13, 50, 25, 0)
    iwin.bkgd(" ", curses.color_pair(3))
    for it, line in enumerate(icon[:-1]):
        iwin.addstr(it+1, 0, " " + line, curses.color_pair(3))
    iwin.addstr(10, 2, " Keys: q to abort, ↑ scroll up", curses.color_pair(2))
    iwin.addstr(11, 2, " ↓ scroll down ", curses.color_pair(2))

    head = curses.newwin(4, 50, 0, 0)
    head.bkgd(" ", curses.color_pair(3))
    head.addstr(1, 1, "PierMesh TUI", curses.color_pair(3))
    head.addstr(2, 1, "Logs:", curses.color_pair(3))
    head.border(' ', ' ', ' ', ' ', ' ', ' ', ' ', ' ')

    logpad = curses.newpad(1000, 300)
    logpad.bkgdset(" ", curses.color_pair(2))

    start = 0
    gen = 0
    lastr = time.time()
    while True:
        if gen == 0 or ((time.time()-lastr) > 2):
            lastr = time.time()
            stdscr.addstr(23, 0, " System usage: ", curses.color_pair(3))

            mem = round(p.memory_info().rss/(1024*1024), 2)
            cpu = p.cpu_percent(interval=0.1)
            stdscr.addstr(
                24,
                0,
                f" MEM: {mem} Mb  CPU: {cpu}% ",
                curses.color_pair(2)
            )
            if tolog.empty() != True:
                while True:
                    logcache.insert(0, tolog.get())
                    tolog.task_done()
                    if tolog.qsize() < 1:
                        break
                    if len(logcache) > 100:
                        logcache = logcache[:100]
            logpad.clear()
            nextOffset = 0
            for it, message in enumerate(logcache):
                msg = message[0]
                if len(msg) > width:
                    msgPartA = msg[:width]
                    msgPartB = msg[width:]
                    if len(msgPartB) > width:
                        msgPartB = msgPartB[:width]
                    logpad.addstr(
                        it+nextOffset, 0, " " + msgPartA + " ", curses.color_pair(message[1]))
                    logpad.addstr(
                        it+nextOffset+1, 0, " " + msgPartB + " ", curses.color_pair(message[1]))
                    nextOffset = 1
                else:
                    logpad.addstr(
                        it+nextOffset, 0, " " + msg + " ", curses.color_pair(message[1]))
                    nextOffset = 0
            logpad.refresh(start, 0, 4, 0, 22, width)
            stdscr.refresh()
        if gen < 1:
            iwin.refresh()
            head.refresh()
            gen += 1

        ch = stdscr.getch()
        if ch == ord("q"):
            curses.nocbreak()
            stdscr.keypad(False)
            curses.echo()
            curses.endwin()
            os._exit(1)
        elif ch == curses.KEY_UP:
            if start != 0:
                start -= 1
        elif ch == curses.KEY_DOWN:
            if start < tolog.qsize():
                start += 1


def runLogUI(tolog, nodeNickname):
    """
    Some required kludge
    """
    wrapper(logUI, tolog, nodeNickname)