#!/usr/bin/env python3 """NexusOS service popup daemon — toggle via SIGUSR1, instant open. Mirrors network-popup.py: an autostarted daemon holding an override_redirect window. The genmon applet's re-invokes this script, which signals the running daemon (SIGUSR1) to toggle the window instead of paying Python startup on every click. """ # PyGObject (gi.repository) is dynamically generated and its API is Optional-heavy # (e.g. Gdk.Display.get_default() is typed Display|None). These are fine at runtime, # so silence the type-checker noise for this GTK desktop script. # pyright: reportMissingModuleSource=false, reportOptionalMemberAccess=false, reportArgumentType=false, reportCallIssue=false, reportAttributeAccessIssue=false import os, sys, signal, socket, threading, subprocess from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import gi gi.require_version('Gtk', '3.0') gi.require_version('Gdk', '3.0') from gi.repository import Gtk, Gdk, GLib from nexus_menu_base import apply_css, get_panel_bottom, get_mouse_position PID_FILE = "/tmp/nexus-popup.pid" VISIBLE_FILE = "/tmp/nexus-popup-visible" # repo root is three parents up: panel -> bin -> repo REPO = Path(__file__).resolve().parent.parent.parent NEXUS_CLI = str(REPO / "management" / "nexus-cli.sh") CTRL_PANEL = str(REPO / "management" / "controlpanel.py") VENV_PY = str(REPO / "Promethean" / "bin" / "python") EDGE_PROFILE = str(Path.home() / ".config" / "nexus-edge") APP_URL = "http://localhost:5173" # name, port, nexus-cli flag SERVICES = [ ("Synapse backend", 8000, "--backend"), ("Memory service", 8001, "--memory"), ("Frontend (Vite)", 5173, "--frontend"), ] WIDTH = 264 POPUP_CSS = b""" .nx-header { padding: 11px 12px 9px 12px; } .nx-title { font-weight: bold; font-size: 1.05em; } .nx-sub { color: #a8a8a8; font-size: 0.82em; } .svc-row { padding: 8px 12px; } .svc-name { font-size: 0.95em; } .svc-port { color: #a8a8a8; font-size: 0.78em; } .dot-up { color: #8cc63f; } .dot-down { color: #6b6b6b; } .master-row { padding: 9px 12px; } .master-label { font-weight: bold; } .footer-btn { padding: 9px 12px; border-radius: 0; border: none; background-color: transparent; color: #cfcfcf; } .footer-btn:hover { background-color: rgba(140,198,63,0.14); color: #f2f2f2; } switch { background-color: #3a3d41; border-radius: 14px; border: 1px solid #4d3461; min-width: 42px; min-height: 22px; } switch:checked { background-color: #8cc63f; border-color: #6ba62a; } switch slider { background-color: #f2f2f2; border-radius: 50%; min-width: 16px; min-height: 16px; margin: 2px; } """ def port_up(port, timeout=0.3): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(timeout) try: return s.connect_ex(("127.0.0.1", port)) == 0 except OSError: return False finally: s.close() class NexusPopup: def __init__(self): self.visible = False self._click_x = None # widgets keyed by port so background refreshes can update them self._switches = {} # port -> (Gtk.Switch, handler_id) self._dots = {} # port -> Gtk.Label self._master = None self._master_handler = None self._sub_lbl = None self._build_window() def _on_sig(s, f): try: self._click_x = get_mouse_position()[0] except Exception: self._click_x = None GLib.idle_add(self.toggle) signal.signal(signal.SIGUSR1, _on_sig) with open(PID_FILE, 'w') as f: f.write(str(os.getpid())) # ── window plumbing (mirrors network-popup) ─────────────────────── def _build_window(self): self.win = Gtk.Window(type=Gtk.WindowType.POPUP) self.win.set_type_hint(Gdk.WindowTypeHint.POPUP_MENU) self.win.set_decorated(False) self.win.set_skip_taskbar_hint(True) self.win.set_keep_above(True) self.win.set_default_size(WIDTH, -1) self.win.connect('key-press-event', lambda w, e: self.hide() if e.keyval == Gdk.KEY_Escape else None) def on_button_press(w, event): wx, wy = w.get_position() ww, wh = w.get_allocated_width(), w.get_allocated_height() if (int(event.x_root) < wx or int(event.x_root) >= wx + ww or int(event.y_root) < wy or int(event.y_root) >= wy + wh): self.hide() return False self.win.connect('button-press-event', on_button_press) def on_map(w, _): gdk_win = w.get_window() if gdk_win: Gdk.Display.get_default().get_default_seat().grab( gdk_win, Gdk.SeatCapabilities.ALL, True, None, None, None) return False self.win.connect('map-event', on_map) def on_unmap(w, _): Gdk.Display.get_default().get_default_seat().ungrab() self.win.connect('unmap-event', on_unmap) def toggle(self): self.hide() if self.visible else self.show() def show(self): for child in self.win.get_children(): self.win.remove(child) self._switches.clear() self._dots.clear() self.win.add(self._build_content()) self.win.show_all() self._sync_state() self._position() self.win.present() self.visible = True open(VISIBLE_FILE, 'w').close() def hide(self): self.win.hide() self.visible = False try: os.remove(VISIBLE_FILE) except FileNotFoundError: pass def _position(self): mx = self._click_x if self._click_x is not None else get_mouse_position()[0] sw = Gdk.Screen.get_default().get_width() panel_bottom = get_panel_bottom() x = max(4, min(mx - WIDTH // 2, sw - WIDTH - 4)) self.win.move(x, panel_bottom + 2) # ── content ─────────────────────────────────────────────────────── def _build_content(self): outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) # Header head = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) head.get_style_context().add_class("nx-header") title = Gtk.Label(label="NexusOS", xalign=0.0) title.get_style_context().add_class("nx-title") self._sub_lbl = Gtk.Label(label="", xalign=0.0) self._sub_lbl.get_style_context().add_class("nx-sub") head.pack_start(title, False, False, 0) head.pack_start(self._sub_lbl, False, False, 0) outer.pack_start(head, False, False, 0) outer.pack_start(Gtk.Separator(), False, False, 0) # Per-service rows for name, port, flag in SERVICES: row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) row.get_style_context().add_class("svc-row") dot = Gtk.Label(label="●") dot.get_style_context().add_class("dot-down") row.pack_start(dot, False, False, 0) self._dots[port] = dot txt = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) nm = Gtk.Label(label=name, xalign=0.0) nm.get_style_context().add_class("svc-name") pl = Gtk.Label(label=f":{port}", xalign=0.0) pl.get_style_context().add_class("svc-port") txt.pack_start(nm, False, False, 0) txt.pack_start(pl, False, False, 0) row.pack_start(txt, True, True, 0) sw = Gtk.Switch() sw.set_valign(Gtk.Align.CENTER) hid = sw.connect('state-set', self._on_service_toggle, flag, port) self._switches[port] = (sw, hid) row.pack_start(sw, False, False, 0) outer.pack_start(row, False, False, 0) outer.pack_start(Gtk.Separator(), False, False, 0) # Master "All services" row master_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) master_row.get_style_context().add_class("master-row") ml = Gtk.Label(label="All services", xalign=0.0) ml.get_style_context().add_class("master-label") master_row.pack_start(ml, True, True, 0) self._master = Gtk.Switch() self._master.set_valign(Gtk.Align.CENTER) self._master_handler = self._master.connect('state-set', self._on_master_toggle) master_row.pack_start(self._master, False, False, 0) outer.pack_start(master_row, False, False, 0) outer.pack_start(Gtk.Separator(), False, False, 0) # Footer actions btn_panel = self._footer_button("Control Panel", self._open_control_panel) btn_app = self._footer_button("Open App", self._open_app) outer.pack_start(btn_panel, False, False, 0) outer.pack_start(Gtk.Separator(), False, False, 0) outer.pack_start(btn_app, False, False, 0) return outer def _footer_button(self, label, cb): btn = Gtk.Button(label=label) btn.get_style_context().add_class("footer-btn") btn.set_relief(Gtk.ReliefStyle.NONE) btn.connect('clicked', lambda _: (self.hide(), cb())) return btn # ── state sync ───────────────────────────────────────────────────── def _sync_state(self): """Read live port state and update every switch, dot, and the header.""" up = 0 all_up = True for _, port, _flag in SERVICES: alive = port_up(port) up += 1 if alive else 0 all_up = all_up and alive self._set_switch(port, alive) dot = self._dots.get(port) if dot: ctx = dot.get_style_context() ctx.remove_class("dot-up"); ctx.remove_class("dot-down") ctx.add_class("dot-up" if alive else "dot-down") if self._master is not None: self._master.handler_block(self._master_handler) self._master.set_active(all_up) self._master.handler_unblock(self._master_handler) if self._sub_lbl is not None: n = len(SERVICES) label = ("All services running" if up == n else "Stopped" if up == 0 else f"{up}/{n} services running") self._sub_lbl.set_text(label) def _set_switch(self, port, active): entry = self._switches.get(port) if not entry: return sw, hid = entry sw.handler_block(hid) sw.set_active(active) sw.handler_unblock(hid) # ── actions ──────────────────────────────────────────────────────── def _cli(self, *args): """Run nexus-cli.sh, then resync the UI from real port state.""" def run(): try: subprocess.run(["bash", NEXUS_CLI, *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60) except Exception: pass GLib.idle_add(self._sync_state) threading.Thread(target=run, daemon=True).start() def _on_service_toggle(self, switch, state, flag, port): self._cli("start" if state else "stop", flag) return False # let GTK move the switch now; _sync_state corrects on failure def _on_master_toggle(self, switch, state): self._cli("start" if state else "stop", "all") return False def _open_control_panel(self): try: subprocess.Popen([VENV_PY, CTRL_PANEL], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception as e: subprocess.Popen(["notify-send", "NexusOS", f"Control Panel failed: {e}"]) def _open_app(self): # Raise the existing dedicated app window if it's already open, else # launch a new one. No service lifecycle ownership here (unlike # nexus-app.sh) — the switches above own start/stop. try: already = subprocess.run( ["pgrep", "-f", f"user-data-dir={EDGE_PROFILE}"], stdout=subprocess.DEVNULL).returncode == 0 cmd = ["microsoft-edge-stable", f"--user-data-dir={EDGE_PROFILE}"] if not already: cmd += ["--no-first-run", "--no-default-browser-check", "--disable-background-mode", "--class=NexusOS"] cmd += [f"--app={APP_URL}"] subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception as e: subprocess.Popen(["notify-send", "NexusOS", f"Open App failed: {e}"]) def main(): # Already running? Signal it to toggle and exit. try: with open(PID_FILE) as f: pid = int(f.read().strip()) os.kill(pid, signal.SIGUSR1) sys.exit(0) except (FileNotFoundError, ProcessLookupError, ValueError): pass provider = Gtk.CssProvider() provider.load_from_data(POPUP_CSS) Gtk.StyleContext.add_provider_for_screen( Gdk.Screen.get_default(), provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION) apply_css() NexusPopup() Gtk.main() if __name__ == "__main__": main()