monitor.py

import argparse

import config


def positive_int(value):
    """Validator argparse: pastikan nilai bilangan bulat positif."""
    try:
        value = int(value)
    except ValueError:
        raise argparse.ArgumentTypeError("harus bilangan bulat")
    if value <= 0:
        raise argparse.ArgumentTypeError("harus lebih besar dari 0")
    return value


def parse_args(argv=None):
    parser = argparse.ArgumentParser(
        prog="monitor",
        description="Server Monitor - TUI untuk monitoring Linux server.",
    )
    parser.add_argument(
        "--refresh",
        type=positive_int,
        default=config.REFRESH_RATE,
        metavar="DETIK",
        help=(
            "Interval refresh widget System/Network dalam detik "
            f"(default: {config.REFRESH_RATE})"
        ),
    )
    parser.add_argument(
        "--slow-refresh",
        type=positive_int,
        default=config.SLOW_REFRESH_RATE,
        metavar="DETIK",
        help=(
            "Interval refresh Docker/Kubernetes/Services dalam detik "
            f"(default: {config.SLOW_REFRESH_RATE})"
        ),
    )
    parser.add_argument(
        "--interface",
        default=config.NETWORK_INTERFACE,
        metavar="NAMA",
        help="Interface jaringan yang dipantau (default: deteksi otomatis)",
    )
    parser.add_argument(
        "--gateway",
        default=config.GATEWAY,
        metavar="IP",
        help="Alamat gateway (default: dari config.py)",
    )
    parser.add_argument(
        "--kubeconfig",
        default=config.KUBECONFIG_PATH,
        metavar="PATH",
        help=(
            "Path kubeconfig untuk memantau Kubernetes/K3s "
            "(default: ~/.kube/config atau variabel KUBECONFIG)"
        ),
    )
    parser.add_argument(
        "--no-docker",
        action="store_true",
        help="Sembunyikan widget Docker",
    )
    parser.add_argument(
        "--no-kubernetes",
        action="store_true",
        help="Sembunyikan widget Kubernetes",
    )
    parser.add_argument(
        "--no-services",
        action="store_true",
        help="Sembunyikan widget Services",
    )
    return parser.parse_args(argv)


def apply_args(args):
    """Terapkan argumen CLI ke konfigurasi runtime.

    Dipanggil SEBELUM import widget, karena widget membaca nilai config
    saat module-nya diimpor.
    """
    config.REFRESH_RATE = args.refresh
    config.SLOW_REFRESH_RATE = args.slow_refresh
    config.NETWORK_INTERFACE = args.interface
    config.GATEWAY = args.gateway
    config.KUBECONFIG_PATH = args.kubeconfig
    config.SHOW_DOCKER = not args.no_docker
    config.SHOW_KUBERNETES = not args.no_kubernetes
    config.SHOW_SERVICES = not args.no_services


apply_args(parse_args())

from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical
from textual.widget import Widget
from textual.widgets import Header, Footer

from widgets.banner import BannerWidget
from widgets.system import SystemWidget
from widgets.network import NetworkWidget
from widgets.docker import DockerWidget
from widgets.kubernetes import KubernetesWidget
from widgets.services import ServicesWidget

from config import (
    SHOW_DOCKER,
    SHOW_KUBERNETES,
    SHOW_SERVICES,
)


class ServerMonitor(App):
    TITLE = "Server Monitor"

    CSS_PATH = "monitor.css"

    BINDINGS = [
        ("q", "quit", "Quit"),
        ("r", "refresh", "Refresh"),
    ]

    def compose(self) -> ComposeResult:
        yield Header()
        yield BannerWidget()

        with Container(id="dashboard"):

            with Horizontal(id="top-row"):
                yield SystemWidget()

                with Vertical(id="right-column"):
                    if SHOW_SERVICES:
                        yield ServicesWidget()
                    yield NetworkWidget()

            if SHOW_KUBERNETES:
                yield KubernetesWidget()

            if SHOW_DOCKER:
                yield DockerWidget()

        yield Footer()

    def action_refresh(self):
        # Paksa semua widget mengambil data baru, bukan sekadar refresh layout.
        for widget in self.query(Widget):
            refresh = getattr(widget, "refresh_data", None)
            if callable(refresh):
                refresh()


if __name__ == "__main__":
    ServerMonitor().run()