# sb-builder - main window # Copyright (C) 2026 Danilo M. # # 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, QSize, Qt, QTimer, pyqtSignal, pyqtSlot from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage from PyQt6.QtGui import QIcon, QKeySequence, QShortcut from PyQt6.QtWidgets import (QLabel, QMainWindow, QMessageBox, QPushButton, QStyle, QTabBar, QTabWidget, QToolBar) from . import VERSION, config from .dashboard import Dashboard from .fmt import job_title from .hints_tab import HINT_JOBS, HintResultDialog, HintsTab from .jobs import JobsTab from .packages import PackagesTab from .results import ResultsTab from .remote import DOWN, MISSING NOTIFY = "org.freedesktop.Notifications" NOTIFY_PATH = "/org/freedesktop/Notifications" # ribbon entries: label, freedesktop icon name, Qt fallback icon RIBBON = (("Dashboard", "go-home", QStyle.StandardPixmap.SP_DirHomeIcon), ("Packages", "package-x-generic", QStyle.StandardPixmap.SP_FileDialogDetailedView), ("Hints", "document-edit", QStyle.StandardPixmap.SP_FileDialogContentsView), ("Results", "view-list-text", QStyle.StandardPixmap.SP_MessageBoxWarning), ("Jobs", "system-run", QStyle.StandardPixmap.SP_MediaPlay)) # tabs as a distinct, centered row; colors from the palette so the theme applies TAB_STYLE = """ QTabWidget::tab-bar { alignment: center; } QTabWidget::pane { border-top: 2px solid palette(highlight); top: -2px; } QTabBar::tab { padding: 6px 20px; margin: 0 3px; border: 1px solid palette(mid); border-bottom: none; border-top-left-radius: 8px; border-top-right-radius: 8px; background: palette(button); color: palette(button-text); } QTabBar::tab:selected { background: palette(highlight); color: palette(highlighted-text); } QTabBar::tab:hover:!selected { background: palette(midlight); } """ 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.packages_tab = None self.results_tab = None self.hints_tab = None self.hint_dialogs = [] # open HintResultDialogs, kept alive while shown 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.setStyleSheet(TAB_STYLE) 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.dashboard.tiles["Packages"].link(self.open_packages) self.dashboard.jobs.link(self.open_jobs) self.dashboard.result_clicked.connect(self.open_results) 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") # window-wide: QTabWidget's own Ctrl+PgUp/PgDn needs focus on the tabs self.tab_next = QShortcut(QKeySequence("Ctrl+PgDown"), self, lambda: self._cycle_tab(1)) self.tab_prev = QShortcut(QKeySequence("Ctrl+PgUp"), self, lambda: self._cycle_tab(-1)) self.ribbon = QToolBar("Screens", movable=False, floatable=False) self.ribbon.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextUnderIcon) self.ribbon.setIconSize(QSize(32, 32)) self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.ribbon) slots = {"Dashboard": lambda: self.tabs.setCurrentWidget(self.dashboard), "Packages": self.open_packages, "Hints": lambda: self.open_hints(), "Results": lambda: self.open_results(), "Jobs": self.open_jobs} for label, name, fallback in RIBBON: icon = QIcon.fromTheme(name, self.style().standardIcon(fallback)) self.ribbon.addAction(icon, label, slots[label]) self.conn = QLabel("● connecting") self.key = QLabel() self.key.hide() 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.key, 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_slow) 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_slow(self): self.poll_stats() if self.online: self.remote.call(self._got_key, "gpg-status") def _got_key(self, data, err): if err: return self._failed(err) self.key.setVisible(data["unlocked"] is not None) self.key.setText("🔓 key unlocked" if data["unlocked"] else "🔒 key locked") self.key.setToolTip(f"package signing key {data['key']}") 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_slow() 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"]) if j["kind"] in ("slackrepo", "publish"): # build state changed self._reload(self.packages_tab, self.results_tab) if j["kind"] in HINT_JOBS and j["status"] in ("ok", "failed"): self._reload(self.packages_tab, self.hints_tab) self.remote.call(lambda d, err: d and self.show_hint_result(d), "job-result", j["id"]) @staticmethod def _reload(*tabs): for tab in tabs: if tab: tab.reload() def show_hint_result(self, result): if not result["ok"] and not result["failed"]: return # nothing changed, nothing failed: no news d = HintResultDialog(self.remote, result, self) d.build_requested.connect( lambda pkgs: self.open_jobs().start("slackrepo", "update", "--preview", *pkgs)) d.changed.connect(lambda: self._reload(self.packages_tab, self.hints_tab)) d.finished.connect(lambda _r: self.hint_dialogs.remove(d)) self.hint_dialogs.append(d) d.show() # ── tabs ──────────────────────────────────────────────────────────────── def open_jobs(self): if self.jobs_tab is None: self.jobs_tab = JobsTab(self.remote, self.cfg["builds"]["publish"]) 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 _cycle_tab(self, step): self.tabs.setCurrentIndex((self.tabs.currentIndex() + step) % self.tabs.count()) def open_packages(self): if self.packages_tab is None: self.packages_tab = PackagesTab(self.remote) self.packages_tab.start_requested.connect(lambda args: self.open_jobs().start(*args)) self.packages_tab.hints_requested.connect(self.open_hints) self.tabs.addTab(self.packages_tab, "Packages") self.packages_tab.reload() self.tabs.setCurrentWidget(self.packages_tab) return self.packages_tab def open_hints(self, name=None): if self.hints_tab is None: self.hints_tab = HintsTab(self.remote, self.cfg) self.hints_tab.start_requested.connect(lambda args: self.open_jobs().start(*args)) self.tabs.addTab(self.hints_tab, "Hints") self.hints_tab.reload() if name: self.hints_tab.select(name) self.tabs.setCurrentWidget(self.hints_tab) return self.hints_tab def open_results(self, result=None): if self.results_tab is None: self.results_tab = ResultsTab(self.remote) self.tabs.addTab(self.results_tab, "Results") self.results_tab.reload() if result: self.results_tab.filter_to(result) self.tabs.setCurrentWidget(self.results_tab) return self.results_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() if w is self.packages_tab: self.packages_tab = None if w is self.results_tab: self.results_tab = None if w is self.hints_tab: self.hints_tab = None 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()