# 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 os import time from PyQt6.QtCore import QSettings, 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 (QHeaderView, QLabel, QMainWindow, QMessageBox, QPushButton, QSplitter, QStyle, QTabWidget, QToolBar, QToolButton) 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, NextStep, is_preview from .packages import PackagesTab from .results import ResultsTab from .remote import DOWN, MISSING NOTIFY = "org.freedesktop.Notifications" NOTIFY_PATH = "/org/freedesktop/Notifications" # Jobs tab icon: running, then the end status until the tab is shown BADGE = {"running": ("system-run", QStyle.StandardPixmap.SP_MediaPlay), "ok": ("emblem-ok", QStyle.StandardPixmap.SP_DialogApplyButton), "bad": ("dialog-error", QStyle.StandardPixmap.SP_MessageBoxCritical)} # 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, state=None): super().__init__() # window size, column widths, splitters: kept across runs, apart from config.toml self.state = QSettings(state or os.path.join(os.path.dirname(config.path()), "state.conf"), QSettings.Format.IniFormat) self.cfg, self.remote = cfg, remote self.notifier = notifier or Notifier(self) self.notifier.clicked.connect(self.open_job) self.hint_dialogs = [] # open HintResultDialogs, kept alive while shown self.next_steps = {} # job id -> its open NextStep dialog self.mine = set() # ids of the jobs this session started: only they get next steps self.badge = None # Jobs tab icon state: None, "running", "ok", "bad" 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() self.tabs.setStyleSheet(TAB_STYLE) self.dashboard = Dashboard() self.dashboard.refresh.triggered.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.jobs_tab = JobsTab(self.remote) self.jobs_tab.refresh_requested.connect(self.poll_jobs) self.jobs_tab.started.connect(self.mine.add) self.jobs_tab.preview_requested.connect(self.show_next_step) self.packages_tab = PackagesTab(self.remote) self.packages_tab.start_requested.connect(lambda args: self.jobs_tab.start(*args)) self.packages_tab.hints_requested.connect(self.open_hints) self.hints_tab = HintsTab(self.remote, self.cfg) self.hints_tab.start_requested.connect(lambda args: self.jobs_tab.start(*args)) self.hints_tab.result_ready.connect(self.show_hint_result) self.results_tab = ResultsTab(self.remote) # permanent tabs (UI guidelines); each loads on first show for tab, label in ((self.dashboard, "Dashboard"), (self.packages_tab, "Packages"), (self.hints_tab, "Hints"), (self.results_tab, "Results"), (self.jobs_tab, "Jobs")): self.tabs.addTab(tab, label) 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)) # main ribbon: the current tab's actions; small ribbon under it: its view controls self.ribbon = QToolBar("Actions", movable=False, floatable=False) self.ribbon.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextUnderIcon) self.ribbon.setIconSize(QSize(32, 32)) self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.ribbon) self.addToolBarBreak() self.view_bar = QToolBar("View", movable=False, floatable=False) self.view_bar.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextBesideIcon) self.view_bar.setIconSize(QSize(16, 16)) self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.view_bar) self.tabs.currentChanged.connect(self._tab_changed) self._tab_changed() 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) self.resize(1100, 700) self._restore_view() # ── 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) self.jobs_tab.set_jobs(jobs) run = next((j for j in jobs if j["status"] == "running"), None) if run: self._badge("running") elif self.badge == "running": self._badge(None) # ended unseen (first poll, reconnect): nothing to report 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", []) 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 self.tabs.currentWidget() is not self.jobs_tab: self._badge("ok" if j["status"] == "ok" else "bad") 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) if j["id"] not in self.mine: continue # started elsewhere or before this session: news, not a next step if j["kind"] in HINT_JOBS and j["status"] in ("ok", "failed"): self.remote.call(lambda d, err: d and self.show_hint_result(d), "job-result", j["id"]) elif is_preview(j): self.show_next_step(j, auto=True) @staticmethod def _reload(*tabs): for tab in tabs: if tab.loaded: # a tab not shown yet loads fresh when it is tab.reload() def show_next_step(self, job, auto=False): """The preview's next step as a dialog; auto: a shadowing preview without a plan shows none.""" if job["id"] in self.next_steps: self.next_steps[job["id"]].raise_() return d = NextStep(self.remote, job, self.cfg["builds"]["publish"], self) d.run.connect(lambda args: (self.jobs_tab.start_signed if args[0] == "slackrepo" else self.jobs_tab.start)(*args)) d.finished.connect(lambda _r: self.next_steps.pop(job["id"], None)) self.next_steps[job["id"]] = d if d.shadow and auto: d.loaded.connect(lambda plan: d.show() if plan else d.reject()) else: d.show() def _badge(self, state): self.badge = state i = self.tabs.indexOf(self.jobs_tab) # in the text, not a color: TAB_STYLE's tab color would override setTabTextColor self.tabs.setTabText(i, "Jobs (running)" if state == "running" else "Jobs") if state is None: self.tabs.setTabIcon(i, QIcon()) return name, fallback = BADGE[state] self.tabs.setTabIcon(i, QIcon.fromTheme(name, self.style().standardIcon(fallback))) 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.jobs_tab.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 _tab_changed(self, _i=None): tab = self.tabs.currentWidget() for bar, acts in ((self.ribbon, tab.actions), (self.view_bar, tab.view_actions)): bar.clear() bar.addActions(acts) for a in acts: if a.menu(): # a menu action opens its menu on a plain click bar.widgetForAction(a).setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup) self.view_bar.setVisible(bool(tab.view_actions)) if not getattr(tab, "loaded", True): tab.reload() if tab is self.jobs_tab and self.badge != "running": self._badge(None) self.poll_stats() def _cycle_tab(self, step): self.tabs.setCurrentIndex((self.tabs.currentIndex() + step) % self.tabs.count()) def open_jobs(self): self.tabs.setCurrentWidget(self.jobs_tab) return self.jobs_tab def open_packages(self): self.tabs.setCurrentWidget(self.packages_tab) return self.packages_tab def open_hints(self, name=None): self.tabs.setCurrentWidget(self.hints_tab) if name: self.hints_tab.select(name) return self.hints_tab def open_results(self, result=None): self.tabs.setCurrentWidget(self.results_tab) if result: self.results_tab.filter_to(result) return self.results_tab def open_job(self, jid): self.show() self.raise_() self.activateWindow() if jid: self.open_jobs().select(jid) # ── view state ────────────────────────────────────────────────────────── def _views(self): """(key, widget) of every table header and splitter in the tabs.""" # ponytail: keyed by tab class and creation order; a reordered layout # restores the wrong widths once, then saves the right ones for tab in (self.dashboard, self.packages_tab, self.hints_tab, self.results_tab, self.jobs_tab): for kind in (QHeaderView, QSplitter): for n, w in enumerate(tab.findChildren(kind)): yield f"{type(tab).__name__}/{kind.__name__}{n}", w def _restore_view(self): if self.state.value("geometry"): self.restoreGeometry(self.state.value("geometry")) for key, w in self._views(): if self.state.value(key): w.restoreState(self.state.value(key)) def closeEvent(self, event): self.state.setValue("geometry", self.saveGeometry()) for key, w in self._views(): self.state.setValue(key, w.saveState()) self.state.sync() super().closeEvent(event) 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()