add new app for systemd monitoring

app/sovran_systemsos_hub/__init__.py

@@ -0,0 +1,50 @@
+"""Thin wrapper around the systemctl CLI for Sovran_SystemsOS_Hub."""
+
+from __future__ import annotations
+
+import subprocess
+from typing import Literal
+
+
+def _run(cmd: list[str]) -> str:
+    """Run a command and return stripped stdout (empty string on failure)."""
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        return result.stdout.strip()
+    except Exception:
+        return ""
+
+
+def is_active(unit: str, scope: Literal["system", "user"] = "system") -> str:
+    """Return the ActiveState string (active / inactive / failed / …)."""
+    return _run(["systemctl", f"--{scope}", "is-active", unit]) or "unknown"
+
+
+def is_enabled(unit: str, scope: Literal["system", "user"] = "system") -> str:
+    """Return the UnitFileState string (enabled / disabled / masked / …)."""
+    return _run(["systemctl", f"--{scope}", "is-enabled", unit]) or "unknown"
+
+
+def run_action(
+    action: str,
+    unit: str,
+    scope: Literal["system", "user"] = "system",
+    method: str = "systemctl",
+) -> bool:
+    """Start / stop / restart a unit.
+
+    For *system* units the command is elevated via *method* (pkexec or sudo).
+    Returns True on apparent success.
+    """
+    base_cmd = ["systemctl", f"--{scope}", action, unit]
+
+    if scope == "system" and method == "pkexec":
+        cmd = ["pkexec", "--user", "root"] + base_cmd
+    else:
+        cmd = base_cmd
+
+    try:
+        subprocess.Popen(cmd)
+        return True
+    except Exception:
+        return False

app/sovran_systemsos_hub/application.py

@@ -0,0 +1,97 @@
+"""Sovran_SystemsOS_Hub — Main GTK4 Application."""
+
+from __future__ import annotations
+
+import gi
+
+gi.require_version("Gtk", "4.0")
+gi.require_version("Adw", "1")
+
+from gi.repository import Adw, Gio, GLib, Gtk  # noqa: E402
+
+from .config import load_config  # noqa: E402
+from .service_row import ServiceRow  # noqa: E402
+
+APP_ID = "com.sovransystems.hub"
+
+
+class SovranHubWindow(Adw.ApplicationWindow):
+    """Primary window: a list of service rows with auto-refresh."""
+
+    def __init__(self, app: Adw.Application, config: dict):
+        super().__init__(
+            application=app,
+            title="Sovran_SystemsOS Hub",
+            default_width=560,
+            default_height=620,
+        )
+        self._config = config
+        self._rows: list[ServiceRow] = []
+
+        # ── Header bar ──
+        header = Adw.HeaderBar()
+
+        refresh_btn = Gtk.Button(
+            icon_name="view-refresh-symbolic",
+            tooltip_text="Refresh now",
+        )
+        refresh_btn.connect("clicked", lambda _b: self._refresh_all())
+        header.pack_end(refresh_btn)
+
+        # ── Content ──
+        self._group = Adw.PreferencesGroup(title="Services")
+
+        page = Adw.PreferencesPage()
+        page.add(self._group)
+
+        toolbar_view = Adw.ToolbarView()
+        toolbar_view.add_top_bar(header)
+        toolbar_view.set_content(page)
+
+        self.set_content(toolbar_view)
+
+        # ── Populate ──
+        self._build_rows()
+        self._start_auto_refresh()
+
+    def _build_rows(self):
+        """Create ServiceRow widgets from config."""
+        for entry in self._config.get("services", []):
+            row = ServiceRow(
+                name=entry.get("name", entry["unit"]),
+                unit=entry["unit"],
+                scope=entry.get("type", "system"),
+                method=self._config.get("command_method", "systemctl"),
+            )
+            self._group.add(row)
+            self._rows.append(row)
+
+    def _refresh_all(self):
+        for row in self._rows:
+            row.refresh()
+
+    def _start_auto_refresh(self):
+        interval = self._config.get("refresh_interval", 5)
+        if interval and interval > 0:
+            GLib.timeout_add_seconds(interval, self._auto_refresh_cb)
+
+    def _auto_refresh_cb(self) -> bool:
+        self._refresh_all()
+        return True  # keep the timer alive
+
+
+class SovranHubApp(Adw.Application):
+    """Sovran_SystemsOS_Hub Adw.Application."""
+
+    def __init__(self):
+        super().__init__(
+            application_id=APP_ID,
+            flags=Gio.ApplicationFlags.DEFAULT_FLAGS,
+        )
+        self._config = load_config()
+
+    def do_activate(self):
+        win = self.get_active_window()
+        if not win:
+            win = SovranHubWindow(self, self._config)
+        win.present()

app/sovran_systemsos_hub/config.py

@@ -0,0 +1,20 @@
+"""Load the Nix-generated config for Sovran_SystemsOS_Hub."""
+
+import json
+import os
+
+
+def load_config() -> dict:
+    """Read config from the path injected by the Nix derivation."""
+    path = os.environ.get(
+        "SOVRAN_HUB_CONFIG",
+        os.path.join(
+            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            "config.json",
+        ),
+    )
+    try:
+        with open(path, "r") as fh:
+            return json.load(fh)
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {"refresh_interval": 5, "command_method": "systemctl", "services": []}

app/sovran_systemsos_hub/service_row.py

@@ -0,0 +1,104 @@
+"""Adw.ActionRow subclass representing a single systemd unit."""
+
+from __future__ import annotations
+
+import gi
+
+gi.require_version("Gtk", "4.0")
+gi.require_version("Adw", "1")
+
+from gi.repository import Adw, GLib, Gtk  # noqa: E402
+
+from . import systemctl  # noqa: E402
+
+LOADING_STATES = {"reloading", "activating", "deactivating", "maintenance"}
+
+
+class ServiceRow(Adw.ActionRow):
+    """A row showing one systemd unit with a toggle switch and action buttons."""
+
+    def __init__(
+        self,
+        name: str,
+        unit: str,
+        scope: str = "system",
+        method: str = "systemctl",
+        **kwargs,
+    ):
+        super().__init__(title=name, subtitle=unit, **kwargs)
+        self._unit = unit
+        self._scope = scope
+        self._method = method
+
+        # ── Active / Inactive switch ──
+        self._switch = Gtk.Switch(valign=Gtk.Align.CENTER)
+        self._switch.connect("state-set", self._on_toggled)
+        self.add_suffix(self._switch)
+        self.set_activatable_widget(self._switch)
+
+        # ── Restart button ──
+        restart_btn = Gtk.Button(
+            icon_name="view-refresh-symbolic",
+            valign=Gtk.Align.CENTER,
+            tooltip_text="Restart",
+            css_classes=["flat"],
+        )
+        restart_btn.connect("clicked", self._on_restart)
+        self.add_suffix(restart_btn)
+
+        # ── Status pill ──
+        self._status_label = Gtk.Label(
+            css_classes=["caption", "dim-label"],
+            valign=Gtk.Align.CENTER,
+            margin_end=4,
+        )
+        self.add_suffix(self._status_label)
+
+        # Initial state
+        self.refresh()
+
+    # ── public API ──
+
+    def refresh(self):
+        """Poll systemctl and update the row."""
+        active_state = systemctl.is_active(self._unit, self._scope)
+        enabled_state = systemctl.is_enabled(self._unit, self._scope)
+
+        is_active = active_state == "active"
+        is_loading = active_state in LOADING_STATES
+        is_failed = active_state == "failed"
+
+        # Block the handler so we don't trigger a start/stop when we
+        # programmatically flip the switch.
+        self._switch.handler_block_by_func(self._on_toggled)
+        self._switch.set_active(is_active)
+        self._switch.handler_unblock_by_func(self._on_toggled)
+
+        self._switch.set_sensitive(not is_loading)
+
+        # Status text
+        label = enabled_state
+        if is_failed:
+            label = "failed"
+        elif is_loading:
+            label = active_state
+        self._status_label.set_label(label)
+
+        # Visual cue for failures
+        if is_failed:
+            self.add_css_class("error")
+        else:
+            self.remove_css_class("error")
+
+    # ── signal handlers ──
+
+    def _on_toggled(self, switch: Gtk.Switch, state: bool) -> bool:
+        action = "start" if state else "stop"
+        systemctl.run_action(action, self._unit, self._scope, self._method)
+        # Delay refresh so systemd has a moment to change state
+        GLib.timeout_add(1500, self.refresh)
+        return False  # let GTK update the switch position
+
+    def _on_restart(self, _btn: Gtk.Button):
+        systemctl.run_action("restart", self._unit, self._scope, self._method)
+        GLib.timeout_add(1500, self.refresh)

app/sovran_systemsos_hub/systemctl.py

@@ -0,0 +1,50 @@
+"""Thin wrapper around the systemctl CLI for Sovran_SystemsOS_Hub."""
+
+from __future__ import annotations
+
+import subprocess
+from typing import Literal
+
+
+def _run(cmd: list[str]) -> str:
+    """Run a command and return stripped stdout (empty string on failure)."""
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        return result.stdout.strip()
+    except Exception:
+        return ""
+
+
+def is_active(unit: str, scope: Literal["system", "user"] = "system") -> str:
+    """Return the ActiveState string (active / inactive / failed / …)."""
+    return _run(["systemctl", f"--{scope}", "is-active", unit]) or "unknown"
+
+
+def is_enabled(unit: str, scope: Literal["system", "user"] = "system") -> str:
+    """Return the UnitFileState string (enabled / disabled / masked / …)."""
+    return _run(["systemctl", f"--{scope}", "is-enabled", unit]) or "unknown"
+
+
+def run_action(
+    action: str,
+    unit: str,
+    scope: Literal["system", "user"] = "system",
+    method: str = "systemctl",
+) -> bool:
+    """Start / stop / restart a unit.
+
+    For *system* units the command is elevated via *method* (pkexec or sudo).
+    Returns True on apparent success.
+    """
+    base_cmd = ["systemctl", f"--{scope}", action, unit]
+
+    if scope == "system" and method == "pkexec":
+        cmd = ["pkexec", "--user", "root"] + base_cmd
+    else:
+        cmd = base_cmd
+
+    try:
+        subprocess.Popen(cmd)
+        return True
+    except Exception:
+        return False