stuff/home.py

#! /usr/bin/env python3

import curses
import socket
import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0)
try:
    # doesn't even have to be reachable
    s.connect(('10.254.254.254', 1))
    IP = s.getsockname()[0]
except Exception:
    IP = '127.0.0.1'
finally:
    s.close()
iptext = f"IP address: {IP}"

mclogpath = '/home/minecraft/server/logs/latest.log'

def readlog():
    with open(mclogpath, 'r') as f:
        log = []
        line = f.readline()
        while line != "":
            log.append(line)
            line = f.readline()
            line = line.rstrip()
    return log

def print_log(win2, log, halfheight):
    win2.clear()
    win2w = width-2
    linei = 0
    linex = halfheight-1
    for line in log[-linex:]:
        try:
            win2.addnstr(linei, 0, line, win2w)
        except Exception:
            pass
        linei += 1
    win2.refresh()

readlog()

class LogEventHandler(FileSystemEventHandler):
    def __init__(self, win2, halfheight):
        self.win2 = win2
        self.halfheight = halfheight
    def on_modified(self, event):
        super(LogEventHandler, self).on_modified(event)

        log = readlog()
        print_log(self.win2, log, self.halfheight)
    def on_deleted(self, event):
        super(LogEventHandler, self).on_deleted(event)
        pass
    def on_moved(self, event):
        super(LogEventHandler, self).on_moved(event)
        pass
    def on_created(self, event):
        super(LogEventHandler, self).on_created(event)
        pass


## init
stdscr = curses.initscr()
curses.noecho()
curses.cbreak()
curses.curs_set(0)
curses.start_color()

curses.init_pair(1, curses.COLOR_GREEN, curses.COLOR_BLACK)
curses.init_pair(2, curses.COLOR_RED, curses.COLOR_BLACK)

height, width = stdscr.getmaxyx()
halfheight = height // 2
halfwidth = width // 2

stdscr.clear()
stdscr.border()
stdscr.addch(halfheight-1, 0, curses.ACS_LTEE)
stdscr.addch(halfheight-1, width-1, curses.ACS_RTEE)
for i in range(1, width-1):
    stdscr.addch(halfheight-1, i, curses.ACS_HLINE)
stdscr.addstr(0, 5, " log ")
stdscr.addstr(halfheight-1, 5, " console ")
stdscr.refresh()

## win2
win2w = width-2
win2 = curses.newwin(halfheight-2, win2w, 1, 1)
log = readlog()
print_log(win2, log, halfheight)

event_handler = LogEventHandler(win2, halfheight) 
observer = Observer()
observer.schedule(event_handler, mclogpath)
observer.start()

## win3
win3 = curses.newwin(halfheight-1, width-2, halfheight, 1)
win3height, win3width = win3.getmaxyx()

## main

def print_help():
    #win3.addstr(win3height-2, 2, "exit: exit the program")
    output.append("exit:\t\t\texit the program")
    output.append("")
    output.append("service {subcommand}:\tcommands for the minecraft systemd service")
    output.append("   status:\t\tget the status of the server (active = running)")
    output.append("   start:\t\tstart the minecraft server")
    output.append("   restart:\t\tstop and start the minecraft server")
    output.append("   stop:\t\tstop the minecraft server")
    print_output()

def print_output():
    win3.clear()
    linei = 2
    linex = win3height-2
    reverseoutput = list(reversed(output))
    for line in reverseoutput:
        win3.addstr(win3height-linei, 2, line)
        linei += 1
    win3.refresh()

output = []

while True:
    win3.addch(win3height-1, 1, '>')
    for i in range(3, win3width-1):
        win3.addch(win3height-1, i, ' ')
    l = 3
    mystr = ""
    while True:
        win3.addch(win3height-1, l, curses.ACS_VLINE)
        k = win3.getch(win3height-1, 3)
        if k == ord('\n'):
            break
        elif k == 127:
            if l > 3:
                mystr = mystr[:-1]
                win3.addch(win3height-1, l, ' ')
                l -= 1
                win3.addch(win3height-1, l, ' ')
                win3.refresh()
        elif k == -1:
            continue
        else:
            mystr = mystr + chr(k)
            win3.addch(win3height-1, l, k)
            win3.refresh()
            l += 1
        #.decode(encoding="utf-8")

    if mystr == "quit" or mystr == "exit":
        break
    elif mystr[:7] == "service":
        mysplit = mystr.split()
        if mysplit[1] == "status":
            mcstatus = subprocess.run(["sudo", "systemctl", "is-active", "minecraft"], stdout=subprocess.PIPE).stdout.decode('utf-8')
            output.append(f"minecraft.service status: {mcstatus}")
            print_output()
            #win3.addstr(win3height-2, 2, f"minecraft.service status: {mcstatus}")
        elif mysplit[1] == "start":
            output.append(f"starting minecraft.service...")
            print_output()
            subprocess.run(["sudo", "systemctl", "start", "minecraft"])
            observer.stop()
            observer.join()
            time.sleep(5)
            observer = Observer()
            observer.schedule(event_handler, mclogpath)
            observer.start()
            output.append(f"started minecraft.service")
            print_output()
        elif mysplit[1] == "restart":
            output.append("restarting minecraft.service...")
            print_output()
            subprocess.run(["sudo", "systemctl", "restart", "minecraft"])
            output.append("restarted minecraft.service")
            print_output()
        elif mysplit[1] == "stop":
            output.append("stopping minecraft.service...")
            print_output()
            subprocess.run(["sudo", "systemctl", "stop", "minecraft"])
            output.append("stopped minecraft.service")
            print_output()
    elif mystr == "help":
        print_help()
    elif mystr == "clear":
        output.clear()
        print_output()


observer.stop()
curses.echo()
curses.nocbreak()
curses.curs_set(1)
observer.join()
curses.endwin()