diff options
Diffstat (limited to 'sb_builder/mainwindow.py')
| -rw-r--r-- | sb_builder/mainwindow.py | 268 |
1 files changed, 268 insertions, 0 deletions
diff --git a/sb_builder/mainwindow.py b/sb_builder/mainwindow.py new file mode 100644 index 0000000..734811b --- /dev/null +++ b/sb_builder/mainwindow.py @@ -0,0 +1,268 @@ +# sb-builder - main window +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +"""Main window: tabs, status bar, connection polling, VM control, notifications.""" + +import time + +from PyQt6.QtCore import QMetaType, QObject, QTimer, pyqtSignal, pyqtSlot +from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage +from PyQt6.QtWidgets import QLabel, QMainWindow, QMessageBox, QPushButton, QTabBar, QTabWidget + +from . import VERSION, config +from .dashboard import Dashboard +from .fmt import job_title +from .jobs import JobsTab +from .remote import DOWN, MISSING + +NOTIFY = "org.freedesktop.Notifications" +NOTIFY_PATH = "/org/freedesktop/Notifications" +END_EVENT = {"ok": "finished", "failed": "failed", "killed": "killed", "lost": "lost"} +MAX_RETRY_MS = 30000 + + +class Notifier(QObject): + """freedesktop notifications over D-Bus; clicked(job id) on the default action.""" + + clicked = pyqtSignal(str) + + def __init__(self, parent=None): + super().__init__(parent) + bus = QDBusConnection.sessionBus() + self.iface = QDBusInterface(NOTIFY, NOTIFY_PATH, NOTIFY, bus) + bus.connect(NOTIFY, NOTIFY_PATH, NOTIFY, "ActionInvoked", self._action) + self.ids = {} # notification id -> job id + + def send(self, title, body, jid=""): + reply = self.iface.call( + "Notify", "sb-builder", QDBusArgument(0, QMetaType.Type.UInt.value), "", title, body, + QDBusArgument(["default", "Open"], QMetaType.Type.QStringList.value), {}, -1) + if reply.type() == QDBusMessage.MessageType.ReplyMessage: + self.ids[reply.arguments()[0]] = jid + + @pyqtSlot("uint", "QString") + def _action(self, nid, _key): + if nid in self.ids: + self.clicked.emit(self.ids.pop(nid)) + + +class MainWindow(QMainWindow): + def __init__(self, cfg, remote, notifier=None): + super().__init__() + self.cfg, self.remote = cfg, remote + self.notifier = notifier or Notifier(self) + self.notifier.clicked.connect(self.open_job) + self.jobs_tab = None + self.seen = None # job id -> status at the last poll; None before the first + self.online = None # None until the first answer + self.overlays = [] + self.vm_action = None # "start" or "resume" offered on the VM button + self.vm_woken = False # we started the VM: notify once ssh answers + self.running_id = None + self.setWindowTitle("sb-builder") + + self.tabs = QTabWidget(tabsClosable=True) + self.tabs.tabCloseRequested.connect(self._close_tab) + self.tabs.currentChanged.connect(lambda _i: self.poll_stats()) + self.dashboard = Dashboard() + self.dashboard.refresh_btn.clicked.connect(self.poll_stats) + self.tabs.addTab(self.dashboard, "Dashboard") + self.tabs.tabBar().setTabButton(0, QTabBar.ButtonPosition.RightSide, None) + self.setCentralWidget(self.tabs) + + file_menu = self.menuBar().addMenu("&File") + file_menu.addAction("&Reload config", self.reload_config) + file_menu.addAction("&Quit", self.close).setShortcut("Ctrl+Q") + self.menuBar().addMenu("&Open").addAction("&Jobs", self.open_jobs) + + self.conn = QLabel("● connecting") + self.vm_btn = QPushButton() + self.vm_btn.clicked.connect(lambda: self._vm_do(self.vm_action)) + self.vm_btn.hide() + self.job_btn = QPushButton(flat=True) + self.job_btn.clicked.connect(lambda: self.open_job(self.running_id)) + self.job_btn.hide() + for w in (self.job_btn, self.vm_btn, self.conn): + self.statusBar().addPermanentWidget(w) + + self.jobs_timer = QTimer(self, timeout=self.poll_jobs) + self.stats_timer = QTimer(self, timeout=self.poll_stats) + self._start_timers() + QTimer.singleShot(0, self.poll_jobs) + + # ── polling ───────────────────────────────────────────────────────────── + + def _start_timers(self): + self.jobs_timer.start(self.cfg["refresh"]["jobs"] * 1000) + self.stats_timer.start(self.cfg["refresh"]["dashboard"] * 1000) + + def poll_jobs(self): + self.remote.call(self._got_jobs, "jobs") + + def poll_stats(self): + if self.online and self.tabs.currentWidget() is self.dashboard: + self.remote.call(self._got_stats, "stats") + + def _got_jobs(self, data, err): + if err: + return self._failed(err) + self._up() + jobs = data["jobs"] + self._notify_ends(jobs) + self.dashboard.show_jobs(jobs) + if self.jobs_tab: + self.jobs_tab.set_jobs(jobs) + run = next((j for j in jobs if j["status"] == "running"), None) + self.running_id = run and run["id"] + self.job_btn.setVisible(run is not None) + if run: + self.job_btn.setText(f"▶ {job_title(run)}") + + def _got_stats(self, data, err): + if err: + return self._failed(err) + self.dashboard.show_stats(data) + + def _got_version(self, data, err): + if err: + return self._failed(err) + self.overlays = data.get("overlays", []) + if self.jobs_tab: + self.jobs_tab.set_overlays(self.overlays) + if data["version"] != VERSION: + self.conn.setText(f"● helper {data['version']}, GUI {VERSION}: run ./install.sh --remote") + + def _up(self): + if self.online: + return + self.online = True + self.conn.setText("● connected") + self.vm_btn.hide() + self.dashboard.set_stale(None) + self.jobs_timer.setInterval(self.cfg["refresh"]["jobs"] * 1000) + self.remote.call(self._got_version, "version") + self.poll_stats() + if self.vm_woken: + self.vm_woken = False + self.notify("vm-up", f"{self.cfg['vm']['domain']} is up") + + def _failed(self, err): + if err == MISSING: + self.conn.setText("● sbb-remote missing: run ./install.sh --remote") + return + if err != DOWN: + self.statusBar().showMessage(err, 10000) + return + if self.online is not False: + self.online = False + self.dashboard.set_stale(time.time()) + base = self.cfg["refresh"]["jobs"] * 1000 + self.jobs_timer.setInterval(min(self.jobs_timer.interval() * 2, max(MAX_RETRY_MS, base))) + self._check_vm() + + # ── VM control ────────────────────────────────────────────────────────── + + def _check_vm(self): + if not self.cfg["vm"]["domain"]: + self.conn.setText("● down") + return + self.remote.virsh(self._got_domstate, "domstate", self.cfg["vm"]["domain"]) + + def _got_domstate(self, rc, state): + if self.online: + return # ssh came back while virsh ran + self.vm_btn.hide() + if rc != 0: + self.conn.setText("● down") + elif state == "shut off": + self.conn.setText("● VM off") + if self.cfg["vm"]["autostart"]: + self._vm_do("start") + else: + self._offer("Start VM", "start") + elif state == "paused": + self.conn.setText("● VM paused") + self._offer("Resume VM", "resume") + elif state == "running": + self.conn.setText("● VM running, ssh unreachable") + else: + self.conn.setText(f"● VM {state}") + + def _offer(self, label, action): + self.vm_action = action + self.vm_btn.setText(label) + self.vm_btn.show() + + def _vm_do(self, action): + self.vm_btn.hide() + self.vm_woken = True + self.conn.setText(f"● VM {action} requested") + self.remote.virsh( + lambda rc, out: rc and self.statusBar().showMessage(f"virsh {action}: {out}", 10000), + action, self.cfg["vm"]["domain"]) + + # ── notifications ─────────────────────────────────────────────────────── + + def notify(self, event, body, jid=""): + n = self.cfg["notify"] + if n["enabled"] and event in n["events"]: + self.notifier.send(f"sb-builder: {event}", body, jid) + + def _notify_ends(self, jobs): + prev, self.seen = self.seen, {j["id"]: j["status"] for j in jobs} + if prev is None: + return # first poll: old jobs are history, not news + for j in jobs: + event = END_EVENT.get(j["status"]) + if event and prev.get(j["id"], "running") != j["status"]: + self.notify(event, job_title(j), j["id"]) + + # ── tabs ──────────────────────────────────────────────────────────────── + + def open_jobs(self): + if self.jobs_tab is None: + self.jobs_tab = JobsTab(self.remote) + self.jobs_tab.refresh_requested.connect(self.poll_jobs) + self.jobs_tab.set_overlays(self.overlays) + self.tabs.addTab(self.jobs_tab, "Jobs") + self.poll_jobs() + self.tabs.setCurrentWidget(self.jobs_tab) + return self.jobs_tab + + def open_job(self, jid): + self.show() + self.raise_() + self.activateWindow() + if jid: + self.open_jobs().select(jid) + + def _close_tab(self, i): + w = self.tabs.widget(i) + if w is self.dashboard: + return + self.tabs.removeTab(i) + if w is self.jobs_tab: + self.jobs_tab = None + w.stop() + w.deleteLater() + + def reload_config(self): + cfg, err = config.load() + if err: + QMessageBox.warning(self, "Config error", f"{err}\nUsing defaults.") + self.cfg.clear() + self.cfg.update(cfg) # Remote holds the same dict + self._start_timers() |