tlu/tlumon/tlumon

#!/usr/bin/env python3
import gi

gi.require_version("Gtk", "3.0")

from gi.repository import Gtk
from gi.repository import Gio
from gi.repository import GLib
import os
import time
import threading
import logging
import subprocess
import psutil
import socket
import yaml

LOGLEVEL=logging.WARNING

SLEEPTIMER=5

class Main:
    def __init__(self):
        scriptpath = os.path.dirname(os.path.realpath(__file__))
        logging.debug("Executing from %s", scriptpath)
        self.builder = Gtk.Builder()
        self.builder.add_from_file(f"{scriptpath}/main.glade")
        self.builder.connect_signals(self)

        self.windowMain = self.builder.get_object("MainWindow")
        self.DialogChangeNIC = self.builder.get_object("dlg_nic_name")
        #load nics to combobox
        nics = Gtk.ListStore(str)
        for interface in psutil.net_if_addrs().keys():
            if interface != "lo":
                nics.append([interface])
        self.combobox = self.builder.get_object("cbox_nics")
        self.combobox.set_model(nics)
        renderer_text = Gtk.CellRendererText()
        self.combobox.pack_start(renderer_text, True)
        self.combobox.add_attribute(renderer_text, "text", 0)

        self.windowMain.connect("delete-event", Gtk.main_quit)
        self.windowMain.show()

        self.icon_ok = Gio.ThemedIcon(name="gtk-yes")
        self.icon_not_ok = Gio.ThemedIcon(name="gtk-no")

        yamlpath = scriptpath.replace("tlumon","salt/pillars")
        try:
            with open(f"{yamlpath}/network.sls", "r") as f:
                logging.debug("Reading global config %s", f"{yamlpath}/network.sls")
                config = yaml.safe_load(f)
        except FileNotFoundError as e:
            logging.error("Couldn't read global config: %s", e)
        try:
            with open(f"{yamlpath}/local.sls", "r") as f:
                localconfig = yaml.safe_load(f)
        except FileNotFoundError as e:
            logging.debug("Couldn't read local config: %s", e)
            localconfig = {}

        print(config)
        try:
            self.lan_nic = localconfig["network"]["interface"]["internal"]
        except KeyError:
            self.lan_nic = config["network"]["interface"]["internal"]
        try:
            self.wan_nic = localconfig["network"]["interface"]["external"]
        except KeyError:
            self.wan_nic = config["network"]["interface"]["external"]

        self.builder.get_object("lbl_wan").set_text(f"wan ({self.wan_nic})")
        self.builder.get_object("lbl_lan").set_text(f"lan ({self.lan_nic})")

        self.service_checks = [
                    "nginx",
                    "rmt-server",
                    "chronyd",
                    "rmt-server-sync.timer",
                    "rmt-server-mirror.timer",
                    "rmt-server-systems-scc-sync.timer",
                    "atftpd.socket",
                    "dnsmasq",
                    "mariadb"
                    ]

        timer_thread = threading.Thread(target=self.update)
        timer_thread.daemon = True
        timer_thread.start()


    def update(self):
        while True:
            logging.debug("Update triggered")
            GLib.idle_add(self.check_service)
            GLib.idle_add(self.check_ip)
            time.sleep(SLEEPTIMER)


    def check_service(self):
        for service in self.service_checks:
            logging.debug("Checking status of service %s", service)
            image = self.builder.get_object("img_service_"+service)
            result = subprocess.run(["systemctl", "is-active", service, "--quiet"])
            if result.returncode == 0:
                logging.debug("%s is running", service)
                image.set_from_gicon(self.icon_ok, Gtk.IconSize.BUTTON)
            else:
                logging.debug("%s is not running, exit code %i",service, result.returncode)
                image.set_from_gicon(self.icon_not_ok, Gtk.IconSize.BUTTON)

    def check_ip(self):
        #checking LAN network
        logging.debug("Checking IP for LAN: %s", self.lan_nic)
        image = self.builder.get_object("img_ip_lan")
        text = self.builder.get_object("lbl_lan_status")
        ip = list(self.get_ip_address(self.lan_nic))
        if ip:
            logging.debug("found : %s", " ".join(ip))
            image.set_from_gicon(self.icon_ok, Gtk.IconSize.DIALOG)
            text.set_text(" ".join(ip))
        else:
            logging.debug("no IP found")
            image.set_from_gicon(self.icon_not_ok, Gtk.IconSize.DIALOG)
            text.set_text("N/A")

        logging.debug("Checking IP for WAN: %s", self.wan_nic)
        ip = list(self.get_ip_address(self.wan_nic))
        image = self.builder.get_object("img_ip_wan")
        text = self.builder.get_object("lbl_wan_status")
        ip = list(self.get_ip_address(self.wan_nic))
        if ip:
            logging.debug("found : %s", " ".join(ip))
            image.set_from_gicon(self.icon_ok, Gtk.IconSize.DIALOG)
            text.set_text(" ".join(ip))
        else:
            logging.debug("no IP found")
            image.set_from_gicon(self.icon_not_ok, Gtk.IconSize.DIALOG)
            text.set_text("N/A")

    def get_ip_address(self, interface):
        for snic in psutil.net_if_addrs()[interface]:
            if snic.family == socket.AF_INET:
                yield (snic.address)

    def change_nic(self, widget):
        button = Gtk.Buildable.get_name(widget)
        logging.debug("pressed %s", button)
        response = self.DialogChangeNIC.run()
        self.DialogChangeNIC.hide()
        if response == Gtk.ResponseType.OK:
             logging.debug("Ok was pressed")
             tree_iter = self.combobox.get_active_iter()
             if tree_iter is not None:
                 model = self.combobox.get_model()
                 name = model[tree_iter][:2][0]
                 if button == "btn_change_wan_nic":
                     self.wan_nic = namn
                     self.builder.get_object("lbl_wan").set_text(f"wan ({self.wan_nic})")
                 if button == "btn_change_lan_nic":
                     self.lan_nic = name
                     self.builder.get_object("lbl_lan").set_text(f"lan ({self.lan_nic})")
        self.check_ip()

if __name__ == '__main__':
    logging.basicConfig(format='%(levelname)s:\t%(message)s', level=LOGLEVEL)
    main = Main()
    Gtk.main()