feat(hyprdrive/beacon): holographic notification daemon replacing dunst

beacon is a GTK4 layer-shell notification daemon in the Cosmonaut Shell idiom.
It owns org.freedesktop.Notifications and renders each notification as a
holographic card: scanline/sweep/noise overlay (shared lib/hologram.py) clipped
to a rounded violet holo-glass card, emitted-magenta text, glow-violet frame
(accent + squiggle for critical), materialise-out-of-static intro / dissolve
outro, top-centre stack.

Implements the interactive spec subset: body markup, urgency, replaces_id,
per-urgency timeouts, action pills, icons (name/path/raw), click-to-dismiss, and
the NotificationClosed / ActionInvoked signals.

- scripts/beacon-start.sh: LD_PRELOAD launcher
- autostart.lua: dunst -> beacon
- windowrules.lua: blur layer-rule for the `beacon` namespace (no_anim; it has
  its own hologram intro)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
main
Amir Alexander Abdelbaki 2026-07-19 16:22:01 +02:00
parent 178c5e3e91
commit 838f954cad
12 changed files with 989 additions and 1 deletions

View File

@ -0,0 +1,278 @@
"""Holographic scanline/sweep/noise overlay — the same treatment and tuning as
the rest of the Cosmonaut Shell suite (astro-menu / station-bar / orbit-menu /
horizon-dock lib/hologram.py), reused here so notification cards read as one more
orbit of the same look: scanline grid + a slow vertical sweep + drifting noise
specks, and a 'materialise out of static' intro when a card first appears.
Each notification card owns its own overlay (see notification.py); the stack
window (window.py) runs a single frame-clock tick and feeds every visible card's
overlay via .tick(dt).
"""
from __future__ import annotations
import math
import random
import cairo
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib, Gtk # noqa: E402
# Same CyberQueer violet/magenta/red combo as the rest of the suite's hologram.
_VIOLET = (0x50 / 255, 0x18 / 255, 0xDD / 255)
_MAGENTA = (0.92, 0.0, 0.65)
_ACCENT = (0xE4 / 255, 0x00 / 255, 0x46 / 255)
class HologramOverlay:
SCANLINE_GAP = 4.0
SCANLINE_ALPHA = 0.18 # a touch stronger than the bar's — cards are the focus
SWEEP_PERIOD = 3.2 # seconds for one top-to-bottom pass
SWEEP_HALF_HEIGHT = 34.0 # card-sized panel, taller than the thin bar strip
NOISE_COUNT = 26
NOISE_COLORS = [_MAGENTA, _MAGENTA, _ACCENT] # magenta-biased specks
NOISE_LIFETIME = (0.5, 1.4)
NOISE_FADE_IN = 0.2
NOISE_FADE_OUT = 0.35
NOISE_ALPHA_RANGE = (0.10, 0.34)
EDGE_FADE_X = 40.0 # smooth horizontal fade-out of the scanline field
EDGE_FADE_Y = 34.0 # smooth vertical fade-out
INTRO_DURATION = 0.9 # brisk 'materialise out of static' when a card pops in
INTRO_STATIC = 900 # static specks at the very start of the intro
OUTRO_DURATION = 0.4 # quick reverse dissolve back into static on dismiss
def __init__(self, enabled: bool = True, clip_func=None, fade_widget=None,
intro_duration: float | None = None) -> None:
self.enabled = enabled
self._clip_func = clip_func # optional path-setter to clip the holo to the card shape
# widget whose opacity is ramped 0->1 during the intro so the card content
# genuinely fades in, rather than a solid haze block popping on
self._fade_widget = fade_widget
if intro_duration is not None:
self.INTRO_DURATION = intro_duration
self._sat_time = 0.0
self._particles: list[dict] = []
self._intro_t: float | None = None
self._outro_t: float | None = None
self._outro_done = None
# Wall-clock safety net: the frame-clock tick only advances while the
# compositor sends frame callbacks; if those stall the intro could freeze
# with content stuck at opacity 0. This timeout force-resolves it anyway.
self._intro_deadline_id: int | None = None
self._mask_cache: tuple | None = None
self.widget = Gtk.DrawingArea()
self.widget.set_can_target(False) # never steals clicks from the card underneath
self.widget.add_css_class("beacon-hologram")
self.widget.set_hexpand(True)
self.widget.set_vexpand(True)
self.widget.set_halign(Gtk.Align.FILL)
self.widget.set_valign(Gtk.Align.FILL)
self.widget.set_draw_func(self._draw_frame)
def tick(self, dt: float) -> None:
if not self.enabled:
return
self._sat_time += dt
if self._intro_t is not None:
self._intro_t += dt
if self._intro_t >= self.INTRO_DURATION:
self._finish_intro()
elif self._fade_widget is not None:
p = self._intro_t / self.INTRO_DURATION
self._fade_widget.set_opacity(p * p * (3 - 2 * p))
if self._outro_t is not None:
self._outro_t += dt
po = min(1.0, self._outro_t / self.OUTRO_DURATION)
if self._fade_widget is not None:
self._fade_widget.set_opacity(1.0 - po * po * (3 - 2 * po))
if self._outro_t >= self.OUTRO_DURATION:
done = self._outro_done
self._outro_t = None
self._outro_done = None
if done is not None:
done()
self.widget.queue_draw()
def _finish_intro(self) -> None:
self._intro_t = None
if self._intro_deadline_id is not None:
GLib.source_remove(self._intro_deadline_id)
self._intro_deadline_id = None
if self._fade_widget is not None:
self._fade_widget.set_opacity(1.0)
self.widget.queue_draw()
def start_intro(self) -> None:
"""Kick off the 'card materialising out of static' opening effect."""
if self.enabled:
self._outro_t = None
self._outro_done = None
self._intro_t = 0.0
if self._fade_widget is not None:
self._fade_widget.set_opacity(0.0)
if self._intro_deadline_id is not None:
GLib.source_remove(self._intro_deadline_id)
self._intro_deadline_id = GLib.timeout_add(
int(self.INTRO_DURATION * 1000) + 150, self._on_intro_deadline)
def _on_intro_deadline(self) -> bool:
self._intro_deadline_id = None
if self._intro_t is not None:
self._finish_intro()
return False # one-shot
def start_outro(self, on_done) -> None:
"""Reverse of the intro (card dissolving back into static), then on_done."""
if not self.enabled:
on_done()
return
self._intro_t = None
if self._intro_deadline_id is not None:
GLib.source_remove(self._intro_deadline_id)
self._intro_deadline_id = None
self._outro_t = 0.0
self._outro_done = on_done
# -- drawing --------------------------------------------------------------
def _draw_frame(self, _area, cr, width: float, height: float) -> None:
if not self.enabled or width <= 0 or height <= 0:
return
if self._clip_func is not None:
cr.save()
self._clip_func(cr, width, height) # clip the scanlines to the card's rounded shape
cr.clip()
cr.push_group()
self._draw_content(cr, width, height)
cr.pop_group_to_source()
cr.mask(self._edge_fade_mask(width, height))
if self._intro_t is not None:
self._draw_intro(cr, width, height)
elif self._outro_t is not None:
self._draw_outro(cr, width, height)
if self._clip_func is not None:
cr.restore()
def _edge_fade_mask(self, width: float, height: float):
key = (int(width), int(height))
if self._mask_cache is not None and self._mask_cache[0] == key:
return self._mask_cache[1]
w, h = max(1, key[0]), max(1, key[1])
surf = cairo.ImageSurface(cairo.FORMAT_A8, w, h)
m = cairo.Context(surf)
m.set_source_rgba(0, 0, 0, 1)
m.paint()
m.set_operator(cairo.OPERATOR_DEST_OUT)
fx = min(self.EDGE_FADE_X, w / 2)
fy = min(self.EDGE_FADE_Y, h / 2)
def band(x0, y0, x1, y1, rx, ry, rw, rh):
gr = cairo.LinearGradient(x0, y0, x1, y1)
gr.add_color_stop_rgba(0.0, 0, 0, 0, 1)
gr.add_color_stop_rgba(1.0, 0, 0, 0, 0)
m.set_source(gr)
m.rectangle(rx, ry, rw, rh)
m.fill()
band(0, 0, fx, 0, 0, 0, fx, h) # left
band(w, 0, w - fx, 0, w - fx, 0, fx, h) # right
band(0, 0, 0, fy, 0, 0, w, fy) # top
band(0, h, 0, h - fy, 0, h - fy, w, fy) # bottom
pattern = cairo.SurfacePattern(surf)
self._mask_cache = (key, pattern)
return pattern
def _draw_intro(self, cr, width: float, height: float) -> None:
p = min(1.0, max(0.0, (self._intro_t or 0.0) / self.INTRO_DURATION))
strength = 1.0 - p
colors = [_MAGENTA, _MAGENTA, _ACCENT, (0.85, 0.85, 0.95)]
count = int(self.INTRO_STATIC * (strength ** 0.5))
for _ in range(count):
x = random.uniform(0, width)
y = random.uniform(0, height)
col = random.choice(colors)
cr.set_source_rgba(*col, random.uniform(0.25, 0.85) * (0.35 + 0.65 * strength))
sz = random.uniform(1.0, 2.8)
cr.rectangle(x, y, sz, sz)
cr.fill()
band = p * height
cr.set_source_rgba(*_ACCENT, 0.5 * (1.0 - abs(2 * p - 1)))
cr.rectangle(0, band - 2.0, width, 4.0)
cr.fill()
def _draw_outro(self, cr, width: float, height: float) -> None:
po = min(1.0, max(0.0, (self._outro_t or 0.0) / self.OUTRO_DURATION))
colors = [_MAGENTA, _MAGENTA, _ACCENT, (0.85, 0.85, 0.95)]
count = int(self.INTRO_STATIC * (po ** 0.5))
for _ in range(count):
x = random.uniform(0, width)
y = random.uniform(0, height)
col = random.choice(colors)
cr.set_source_rgba(*col, random.uniform(0.25, 0.85) * (0.35 + 0.65 * po))
sz = random.uniform(1.0, 2.8)
cr.rectangle(x, y, sz, sz)
cr.fill()
band = (1.0 - po) * height
cr.set_source_rgba(*_ACCENT, 0.5 * (1.0 - abs(2 * po - 1)))
cr.rectangle(0, band - 2.0, width, 4.0)
cr.fill()
def _draw_content(self, cr, width: float, height: float) -> None:
r, g, b = _VIOLET
cr.save()
cr.set_source_rgba(r, g, b, self.SCANLINE_ALPHA)
cr.set_line_width(1.0)
y = 0.0
while y < height:
cr.move_to(0, y)
cr.line_to(width, y)
y += self.SCANLINE_GAP
cr.stroke()
cr.restore()
phase = (self._sat_time % self.SWEEP_PERIOD) / self.SWEEP_PERIOD
sweep_y = phase * height
hh = self.SWEEP_HALF_HEIGHT
grad = cairo.LinearGradient(0, sweep_y - hh, 0, sweep_y + hh)
grad.add_color_stop_rgba(0.0, r, g, b, 0.0)
grad.add_color_stop_rgba(0.5, r, g, b, 0.09)
grad.add_color_stop_rgba(1.0, r, g, b, 0.0)
cr.set_source(grad)
cr.rectangle(0, sweep_y - hh, width, hh * 2)
cr.fill()
flicker = 0.012 + 0.007 * math.sin(self._sat_time * 11.0)
cr.set_source_rgba(r, g, b, max(0.0, flicker))
cr.paint()
self._draw_noise(cr, width, height)
def _draw_noise(self, cr, width: float, height: float) -> None:
now = self._sat_time
self._particles = [p for p in self._particles if now - p["birth"] < p["life"]]
while len(self._particles) < self.NOISE_COUNT:
self._particles.append({
"x": random.uniform(0, width),
"y": random.uniform(0, height),
"w": random.uniform(1.0, 2.6),
"h": random.uniform(1.0, 2.0),
"color": random.choice(self.NOISE_COLORS),
"peak_alpha": random.uniform(*self.NOISE_ALPHA_RANGE),
"birth": now,
"life": random.uniform(*self.NOISE_LIFETIME),
})
for p in self._particles:
t = (now - p["birth"]) / p["life"]
if t < self.NOISE_FADE_IN:
envelope = t / self.NOISE_FADE_IN
elif t > 1.0 - self.NOISE_FADE_OUT:
envelope = max(0.0, (1.0 - t) / self.NOISE_FADE_OUT)
else:
envelope = 1.0
r, g, b = p["color"]
cr.set_source_rgba(r, g, b, p["peak_alpha"] * envelope)
cr.rectangle(p["x"], p["y"], p["w"], p["h"])
cr.fill()

View File

@ -0,0 +1,82 @@
#!/usr/bin/env python3
"""beacon — the Cosmonaut Shell notification daemon for hyprdrive.
Replaces dunst: owns org.freedesktop.Notifications and renders each notification
as a holographic card (scanline/sweep/noise overlay, emitted-magenta text, violet
holo-glass, radio-squiggle divider) in a top-centre layer-shell stack, so
notifications read as one more orbit of the astro-menu / orbit-menu / station-bar
look instead of a plain popup.
main.py run the resident daemon (owns the notification bus name, stays hidden
until a notification arrives)
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
# beacon-start.sh LD_PRELOADs libgtk4-layer-shell (load-ordering requirement ahead
# of libwayland-client). Drop it once resident so it isn't inherited by anything
# this process launches (same rationale as the rest of the suite's main.py).
os.environ.pop("LD_PRELOAD", None)
sys.path.insert(0, str(Path(__file__).resolve().parent))
import gi # noqa: E402
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib, Gtk # noqa: E402
import theme # noqa: E402
from paths import APP_ID, FDN_NAME # noqa: E402
from server import NotificationServer # noqa: E402
from window import BeaconWindow # noqa: E402
class BeaconApp(Gtk.Application):
def __init__(self) -> None:
super().__init__(application_id=APP_ID,
flags=Gio.ApplicationFlags.DEFAULT_FLAGS)
self.window: BeaconWindow | None = None
self._server: NotificationServer | None = None
self._name_id = 0
def do_startup(self) -> None:
Gtk.Application.do_startup(self)
theme.load_css()
self.window = BeaconWindow(self, on_closed=self._on_closed)
# Own the freedesktop notification name; the server is created once the bus
# connection is in hand.
self._name_id = Gio.bus_own_name(
Gio.BusType.SESSION, FDN_NAME, Gio.BusNameOwnerFlags.NONE,
self._on_bus_acquired, None, self._on_name_lost)
self.hold() # stay alive with no visible window
def _on_bus_acquired(self, connection: Gio.DBusConnection, _name: str) -> None:
assert self.window is not None
self._server = NotificationServer(connection, self.window)
def _on_closed(self, nid: int, reason: int) -> None:
if self._server is not None:
self._server.emit_closed(nid, reason)
def _on_name_lost(self, _connection, _name: str) -> None:
# Another notification daemon (e.g. a still-running dunst) already owns it.
sys.stderr.write(
"beacon: could not acquire org.freedesktop.Notifications "
"(another notification daemon is running); exiting.\n")
self.quit()
def do_activate(self) -> None:
pass # resident daemon: nothing to do on activate
def main() -> int:
GLib.set_prgname("beacon")
return BeaconApp().run(sys.argv)
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -0,0 +1,188 @@
"""A single holographic notification card.
Layout: [app icon] [ summary (emitted-magenta, letter-spaced) / radio-squiggle
divider / body ] with an optional row of action-pill buttons, all under a
scanline/sweep/noise HologramOverlay (lib/hologram.py) clipped to the card's
rounded rectangle. The card materialises out of static (holo intro) when it
appears and dissolves back into static when dismissed.
"""
from __future__ import annotations
import math
from typing import Callable, Optional
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0")
gi.require_version("GdkPixbuf", "2.0")
from gi.repository import Gdk, GdkPixbuf, GLib, Gtk # noqa: E402
from lib.hologram import HologramOverlay
_SQUIGGLE = "" * 16 # ∿ sine-wave "tuned signal" divider
_CARD_RADIUS = 16.0
def _rounded_path(cr, w: float, h: float, r: float = _CARD_RADIUS) -> None:
r = min(r, w / 2, h / 2)
cr.new_sub_path()
cr.arc(w - r, r, r, -math.pi / 2, 0)
cr.arc(w - r, h - r, r, 0, math.pi / 2)
cr.arc(r, h - r, r, math.pi / 2, math.pi)
cr.arc(r, r, r, math.pi, 3 * math.pi / 2)
cr.close_path()
class NotificationCard:
def __init__(self, nid: int, summary: str, body: str, urgency: int,
icon: str, image_data, actions: list[str],
on_action: Callable[[int, str], None],
on_dismiss: Callable[[int], None]) -> None:
self.nid = nid
self._on_action = on_action
self._on_dismiss = on_dismiss
self._dismissing = False
critical = urgency >= 2
# -- content -----------------------------------------------------------
content = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=12)
content.add_css_class("beacon-content")
img = self._build_icon(icon, image_data)
if img is not None:
content.append(img)
textcol = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
textcol.set_hexpand(True)
textcol.set_valign(Gtk.Align.CENTER)
if summary:
lbl = Gtk.Label(label=summary, xalign=0.0)
lbl.add_css_class("beacon-summary")
lbl.set_wrap(True)
lbl.set_wrap_mode(2) # WORD_CHAR
lbl.set_max_width_chars(30)
textcol.append(lbl)
squig = Gtk.Label(label=_SQUIGGLE, xalign=0.0)
squig.add_css_class("beacon-squiggle")
textcol.append(squig)
if body:
blbl = Gtk.Label(xalign=0.0)
blbl.add_css_class("beacon-body")
blbl.set_wrap(True)
blbl.set_wrap_mode(2)
blbl.set_max_width_chars(34)
# bodies may carry a small Pango-markup subset (<b>/<i>/<u>/<a>…);
# fall back to plain text if the sender's markup won't parse.
try:
blbl.set_markup(body)
except GLib.GError:
blbl.set_text(body)
textcol.append(blbl)
content.append(textcol)
if actions:
row = self._build_actions(actions)
if row is not None:
textcol.append(row)
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
card.add_css_class("beacon-card")
if critical:
card.add_css_class("critical")
card.append(content)
card.set_size_request(340, -1)
# -- hologram overlay --------------------------------------------------
self._holo = HologramOverlay(enabled=True, clip_func=_rounded_path,
fade_widget=content)
overlay = Gtk.Overlay()
overlay.set_child(card)
overlay.add_overlay(self._holo.widget)
overlay.set_measure_overlay(self._holo.widget, False)
# left-click anywhere on the card = invoke default action if present, else
# dismiss; right-click always dismisses.
self._default_action = "default" if "default" in actions else None
left = Gtk.GestureClick(button=1)
left.connect("released", self._on_left_click)
overlay.add_controller(left)
right = Gtk.GestureClick(button=3)
right.connect("released", lambda *_a: self.dismiss())
overlay.add_controller(right)
self.widget = overlay
self._holo.start_intro()
# -- public ---------------------------------------------------------------
def tick(self, dt: float) -> None:
self._holo.tick(dt)
def dismiss(self) -> None:
"""Play the dissolve, then hand the id back to the window for removal."""
if self._dismissing:
return
self._dismissing = True
self._holo.start_outro(lambda: self._on_dismiss(self.nid))
# -- internals ------------------------------------------------------------
def _on_left_click(self, *_a) -> None:
if self._default_action is not None:
self._on_action(self.nid, self._default_action)
else:
self.dismiss()
def _build_actions(self, actions: list[str]) -> Optional[Gtk.Box]:
# actions is a flat [key, label, key, label, …] list; "default" is the
# implicit click action and isn't shown as a button.
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
row.add_css_class("beacon-actions")
row.set_margin_top(6)
shown = 0
for i in range(0, len(actions) - 1, 2):
key, label = actions[i], actions[i + 1]
if key == "default":
continue
btn = Gtk.Button(label=label or key)
btn.add_css_class("beacon-action")
btn.connect("clicked", lambda _b, k=key: self._on_action(self.nid, k))
row.append(btn)
shown += 1
return row if shown else None
def _build_icon(self, icon: str, image_data) -> Optional[Gtk.Image]:
img: Optional[Gtk.Image] = None
if image_data is not None:
pb = self._pixbuf_from_hint(image_data)
if pb is not None:
img = Gtk.Image.new_from_paintable(Gdk.Texture.new_for_pixbuf(pb))
if img is None and icon:
if icon.startswith("file://"):
icon = icon[len("file://"):]
if icon.startswith("/"):
img = Gtk.Image.new_from_file(icon)
else:
img = Gtk.Image.new_from_icon_name(icon)
if img is None:
return None
img.add_css_class("beacon-icon")
img.set_pixel_size(44)
img.set_valign(Gtk.Align.START)
return img
@staticmethod
def _pixbuf_from_hint(data) -> Optional[GdkPixbuf.Pixbuf]:
# Spec "image-data": (width, height, rowstride, has_alpha, bits, channels, bytes)
try:
w, h, rowstride, has_alpha, bits, channels, raw = data
return GdkPixbuf.Pixbuf.new_from_bytes(
GLib.Bytes.new(bytes(raw)), GdkPixbuf.Colorspace.RGB,
has_alpha, bits, w, h, rowstride)
except (ValueError, TypeError):
return None

View File

@ -0,0 +1,17 @@
"""Shared filesystem locations. Works whether run from the repo or ~/.config."""
from __future__ import annotations
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent
STYLE_DIR = BASE_DIR / "style"
# GApplication single-instance id (our own process). NOT the freedesktop
# notification name — that well-known name (org.freedesktop.Notifications) is
# owned separately in main.py, the same one dunst used to hold.
APP_ID = "eu.abdelbaki.beacon"
FDN_NAME = "org.freedesktop.Notifications"
FDN_PATH = "/org/freedesktop/Notifications"
FDN_IFACE = "org.freedesktop.Notifications"

View File

@ -0,0 +1,146 @@
"""org.freedesktop.Notifications D-Bus service — the well-known name dunst used
to own. Parses each Notify call and hands it to the stack window (window.py) to
render as a holographic card; owns per-notification expiry timers and emits the
spec's NotificationClosed / ActionInvoked signals.
Deliberately small: the visible/interactive spec subset (body markup, actions,
icons, urgency, replaces_id, timeouts, persistence) no sound/markup-hint
gymnastics.
"""
from __future__ import annotations
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib # noqa: E402
from paths import FDN_IFACE, FDN_PATH
# reasons per the spec: 1 expired, 2 dismissed by user, 3 closed by call, 4 undefined
_INTROSPECTION_XML = """
<node>
<interface name="org.freedesktop.Notifications">
<method name="GetCapabilities">
<arg type="as" name="capabilities" direction="out"/>
</method>
<method name="Notify">
<arg type="s" name="app_name" direction="in"/>
<arg type="u" name="replaces_id" direction="in"/>
<arg type="s" name="app_icon" direction="in"/>
<arg type="s" name="summary" direction="in"/>
<arg type="s" name="body" direction="in"/>
<arg type="as" name="actions" direction="in"/>
<arg type="a{sv}" name="hints" direction="in"/>
<arg type="i" name="expire_timeout" direction="in"/>
<arg type="u" name="id" direction="out"/>
</method>
<method name="CloseNotification">
<arg type="u" name="id" direction="in"/>
</method>
<method name="GetServerInformation">
<arg type="s" name="name" direction="out"/>
<arg type="s" name="vendor" direction="out"/>
<arg type="s" name="version" direction="out"/>
<arg type="s" name="spec_version" direction="out"/>
</method>
<signal name="NotificationClosed">
<arg type="u" name="id"/>
<arg type="u" name="reason"/>
</signal>
<signal name="ActionInvoked">
<arg type="u" name="id"/>
<arg type="s" name="action_key"/>
</signal>
</interface>
</node>
"""
# server default timeouts (ms) by urgency when the client passes -1
_DEFAULT_TIMEOUT = {0: 5000, 1: 8000, 2: 0} # low / normal / critical(never)
class NotificationServer:
def __init__(self, connection: Gio.DBusConnection, window) -> None:
self._conn = connection
self._window = window
self._next_id = 1
self._timers: dict[int, int] = {}
node = Gio.DBusNodeInfo.new_for_xml(_INTROSPECTION_XML)
self._iface = node.interfaces[0]
connection.register_object(
FDN_PATH, self._iface, self._on_method_call, None, None)
# -- D-Bus dispatch -------------------------------------------------------
def _on_method_call(self, _conn, _sender, _path, _iface, method, params, invocation):
if method == "Notify":
invocation.return_value(GLib.Variant("(u)", (self._notify(params),)))
elif method == "CloseNotification":
(nid,) = params.unpack()
self._window.close_notification(int(nid), reason=3)
self._cancel_timer(int(nid))
invocation.return_value(None)
elif method == "GetCapabilities":
invocation.return_value(GLib.Variant("(as)", (
["body", "body-markup", "icon-static", "actions", "persistence"],)))
elif method == "GetServerInformation":
invocation.return_value(GLib.Variant("(ssss)", (
"beacon", "abdelbaki.eu", "1.0", "1.2")))
else:
invocation.return_error_literal(
Gio.dbus_error_quark(), Gio.DBusError.UNKNOWN_METHOD, method)
def _notify(self, params) -> int:
(app_name, replaces_id, app_icon, summary, body,
actions, hints, expire_timeout) = params.unpack()
nid = int(replaces_id) if replaces_id else self._next_id
if not replaces_id:
self._next_id += 1
urgency = int(hints.get("urgency", 1))
image_data = (hints.get("image-data") or hints.get("image_data")
or hints.get("icon_data"))
image_path = hints.get("image-path") or hints.get("image_path")
icon = image_path or app_icon or ""
self._window.show_notification(
nid, summary, body, urgency, icon, image_data, list(actions),
self._emit_action)
self._arm_timer(nid, int(expire_timeout), urgency)
return nid
# -- expiry timers --------------------------------------------------------
def _arm_timer(self, nid: int, expire_timeout: int, urgency: int) -> None:
self._cancel_timer(nid)
if expire_timeout < 0:
ms = _DEFAULT_TIMEOUT.get(urgency, 8000)
else:
ms = expire_timeout
if ms <= 0:
return # 0 = never expire (or critical default)
self._timers[nid] = GLib.timeout_add(ms, self._on_expire, nid)
def _on_expire(self, nid: int) -> bool:
self._timers.pop(nid, None)
self._window.close_notification(nid, reason=1) # 1 = expired
return False
def _cancel_timer(self, nid: int) -> None:
tid = self._timers.pop(nid, None)
if tid is not None:
GLib.source_remove(tid)
# -- signals (also the window's on_closed / on_action callbacks) -----------
def emit_closed(self, nid: int, reason: int) -> None:
self._cancel_timer(nid)
self._conn.emit_signal(None, FDN_PATH, FDN_IFACE, "NotificationClosed",
GLib.Variant("(uu)", (nid, reason)))
def _emit_action(self, nid: int, key: str) -> None:
self._conn.emit_signal(None, FDN_PATH, FDN_IFACE, "ActionInvoked",
GLib.Variant("(us)", (nid, key)))
# dismiss the card once its action fired, matching typical daemon behaviour
self._window.close_notification(nid, reason=2)

View File

@ -0,0 +1,8 @@
/* Generated from ~/Dotfiles/colors.conf by apply-theme.sh do not hand-edit the
* hex values; edit colors.conf and re-run apply-theme.sh. (Kept in sync via the
* sed pipeline in USER_FILES; these defaults are the CyberQueer palette.) */
@define-color text #D6ABAB;
@define-color bg #1A1A1A;
@define-color accent #E40046;
@define-color violet #5018DD;
@define-color danger #F50505;

View File

@ -0,0 +1,90 @@
/* beacon holographic notification cards. Matches the astro-menu idiom
* (astro-menu/style/style.css): emitted-magenta text, violet holo-glass fills,
* glow-violet / accent frames, Agave Nerd Font Mono, rounded cards. The compositor
* blurs behind the translucent card fills (see the `beacon` layer-rule in
* hypr/usr/windowrules.lua); the scanline/sweep/noise depth is Cairo-drawn on top
* by lib/hologram.py. */
/* Emitted magenta reads as projected light on the blurred glass, exactly as the
* astro panel overrides its @text. */
@define-color text #EB00A6;
* {
font-family: "Agave Nerd Font Mono", monospace;
font-size: 12pt;
}
/* The CyberQueer GTK theme paints `* { background-color:#1a1a1a }` on every node,
* which would fill the surface and the gaps between cards with an opaque slab.
* Blank the structural nodes; the card asserts its own glass fill below. */
window,
window.background,
.beacon-window,
.beacon-stack,
.beacon-content,
box,
overlay,
label,
image,
drawingarea {
background: transparent;
background-color: transparent;
}
/* the holo-glass card */
.beacon-card {
background-color: alpha(@violet, 0.40);
border: 2px solid #8A5CFF; /* glow_violet — emitted-light frame */
border-radius: 16px;
padding: 12px 14px;
box-shadow: 0 0 16px 1px alpha(#8A5CFF, 0.35);
}
.beacon-card.critical {
border-color: @accent;
background-color: alpha(@accent, 0.14);
box-shadow: 0 0 18px 1px alpha(@accent, 0.40);
}
/* summary = emitted magenta, letter-spaced like the astro HUD headlines */
.beacon-summary {
color: @text;
font-weight: bold;
font-size: 12pt;
letter-spacing: 1px;
}
.beacon-card.critical .beacon-summary { color: @accent; }
/* radio-squiggle divider (the tuned-signal line) */
.beacon-squiggle {
color: #8A5CFF;
font-size: 9pt;
margin: 1px 0;
opacity: 0.9;
}
.beacon-card.critical .beacon-squiggle { color: @accent; }
.beacon-body {
color: @text;
font-size: 11pt;
opacity: 0.95;
}
.beacon-icon { margin-right: 2px; }
/* action pills — astro-menu's .quad-action idiom */
.beacon-action {
color: @text;
background-color: alpha(@violet, 0.4);
border: 2px solid #8A5CFF;
border-radius: 20px;
padding: 2px 12px;
min-height: 22px;
transition: border-color 180ms ease, color 180ms ease, box-shadow 220ms ease, background 180ms ease;
}
.beacon-action:hover {
border-color: @accent;
color: @accent;
box-shadow: 0 0 12px 1px alpha(@accent, 0.55);
}
.beacon-hologram { background: transparent; }

View File

@ -0,0 +1,30 @@
"""Load the two stylesheets as ordered CSS providers — same scheme as the rest
of the Cosmonaut Shell suite (orbit-menu, horizon-dock, astro-menu, station-bar).
_colors.css defines the CyberQueer @define-color names; style.css consumes them.
Priority is USER+1 for the same reason as the others: the CyberQueer GTK theme
at ~/.config/gtk-4.0/gtk.css loads at PRIORITY_USER (800), above APPLICATION
(600), and would beat our transparent structural containers otherwise.
"""
from __future__ import annotations
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gdk, Gtk # noqa: E402
from paths import STYLE_DIR
def load_css() -> None:
display = Gdk.Display.get_default()
for name in ("_colors.css", "style.css"):
path = STYLE_DIR / name
if not path.exists():
continue
provider = Gtk.CssProvider()
provider.load_from_path(str(path))
Gtk.StyleContext.add_provider_for_display(
display, provider, Gtk.STYLE_PROVIDER_PRIORITY_USER + 1
)

View File

@ -0,0 +1,128 @@
"""The notification stack: a top-centre layer-shell surface that holds the live
notification cards and drives one frame-clock tick for all their holograms.
Sized to its content (anchored TOP only, so it floats centred like dunst did),
on the OVERLAY layer above normal windows. Newest card on top. The surface is
hidden whenever no cards are showing, so it never eats clicks on an empty screen.
"""
from __future__ import annotations
from typing import Callable, Optional
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Gtk4LayerShell", "1.0")
from gi.repository import Gtk # noqa: E402
from gi.repository import Gtk4LayerShell as LayerShell # noqa: E402
from notification import NotificationCard
MARGIN_TOP = 46
MAX_VISIBLE = 6
class BeaconWindow(Gtk.ApplicationWindow):
def __init__(self, app: Gtk.Application,
on_closed: Callable[[int, int], None]) -> None:
super().__init__(application=app)
self._on_closed = on_closed # (id, reason) -> emit NotificationClosed
self._cards: dict[int, NotificationCard] = {}
self._order: list[int] = [] # newest first
self._tick_id: Optional[int] = None
self._last_tick: Optional[float] = None
self.set_decorated(False)
self.add_css_class("beacon-window")
self._stack = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
self._stack.add_css_class("beacon-stack")
self._stack.set_halign(Gtk.Align.CENTER)
self._stack.set_valign(Gtk.Align.START)
self.set_child(self._stack)
self._init_layer_shell()
self.set_visible(False)
def _init_layer_shell(self) -> None:
LayerShell.init_for_window(self)
LayerShell.set_layer(self, LayerShell.Layer.OVERLAY)
LayerShell.set_namespace(self, "beacon")
LayerShell.set_keyboard_mode(self, LayerShell.KeyboardMode.NONE)
LayerShell.set_anchor(self, LayerShell.Edge.TOP, True)
LayerShell.set_margin(self, LayerShell.Edge.TOP, MARGIN_TOP)
LayerShell.set_exclusive_zone(self, 0)
# -- public: driven by the D-Bus server -----------------------------------
def show_notification(self, nid: int, summary: str, body: str, urgency: int,
icon: str, image_data, actions: list[str],
on_action: Callable[[int, str], None]) -> None:
if nid in self._cards: # replaces_id: swap in place
self._drop_card(nid)
card = NotificationCard(nid, summary, body, urgency, icon, image_data,
actions, on_action, self._card_dismissed)
self._cards[nid] = card
self._order.insert(0, nid)
self._stack.prepend(card.widget)
# cap the stack: quietly expire the oldest beyond the limit
while len(self._order) > MAX_VISIBLE:
old = self._order[-1]
self._card_dismissed(old, reason=4) # 4 = undefined/expired-by-limit
self.set_visible(True)
self._ensure_tick()
def close_notification(self, nid: int, reason: int = 3) -> bool:
"""Server-/user-requested close. reason 3 = closed by CloseNotification."""
if nid not in self._cards:
return False
card = self._cards[nid]
# play the dissolve; removal + the NotificationClosed signal follow
card.dismiss()
self._pending_reason[nid] = reason
return True
# -- card lifecycle -------------------------------------------------------
_pending_reason: dict[int, int] = {}
def _card_dismissed(self, nid: int, reason: int = 2) -> None:
"""Called after a card's dissolve finishes (reason 2 = dismissed by user),
or directly for cap-expiry. Removes it and signals the closure."""
if nid not in self._cards:
return
reason = self._pending_reason.pop(nid, reason)
self._drop_card(nid)
self._on_closed(nid, reason)
if not self._cards:
self.set_visible(False)
self._stop_tick()
def _drop_card(self, nid: int) -> None:
card = self._cards.pop(nid, None)
if card is not None:
self._stack.remove(card.widget)
if nid in self._order:
self._order.remove(nid)
# -- animation tick -------------------------------------------------------
def _on_tick(self, _widget, frame_clock) -> bool:
now = frame_clock.get_frame_time() / 1_000_000
dt = 0.0 if self._last_tick is None else max(0.0, now - self._last_tick)
self._last_tick = now
for card in list(self._cards.values()):
card.tick(dt)
return True
def _ensure_tick(self) -> None:
if self._tick_id is None:
self._last_tick = None
self._tick_id = self.add_tick_callback(self._on_tick)
def _stop_tick(self) -> None:
if self._tick_id is not None:
self.remove_tick_callback(self._tick_id)
self._tick_id = None
self._last_tick = None

View File

@ -9,7 +9,7 @@ hl.on("hyprland.start", function()
hl.exec_cmd("systemctl --user start hyprpolkitagent")
hl.exec_cmd("hyprsunset")
hl.exec_cmd("nm-applet")
hl.exec_cmd("dunst")
hl.exec_cmd("bash ~/Dotfiles/desktopenvs/hyprdrive/scripts/beacon-start.sh")
hl.exec_cmd("[workspace special:magic silent] kitty")
hl.exec_cmd("hyprctl setcursor Nordzy-cursors-lefthand 50")
hl.exec_cmd("hyprpaper")

View File

@ -127,6 +127,14 @@ for _, ns in ipairs({ "astro-menu", "orbit-menu", "horizon-dock", "station-bar"
blur = true, ignore_alpha = 0.1, no_anim = true })
end
-- beacon (our notification daemon, namespace "beacon") renders holographic
-- notification cards. Blur what shows through their translucent violet glass so
-- they read as the same frosted holo panels as the menus. no_anim: each card
-- materialises out of static via its own in-app hologram intro, so the compositor
-- should map the surface in place rather than slide it.
hl.layer_rule({ name = "cosmoshell-blur-beacon", match = { namespace = "beacon" },
blur = true, ignore_alpha = 0.1, no_anim = true })
-- smart gaps
hl.workspace_rule({ workspace = "w[tv1]s[false]", gaps_out = 0, gaps_in = 0 })
hl.workspace_rule({ workspace = "f[1]s[false]", gaps_out = 0, gaps_in = 0 })

View File

@ -0,0 +1,13 @@
#!/usr/bin/env bash
# Resident launcher for beacon, the Cosmonaut Shell notification daemon (replaces
# dunst). Same LD_PRELOAD requirement and rationale as the rest of the suite's
# start scripts: gtk4-layer-shell must load before libwayland-client, which isn't
# guaranteed under PyGObject.
APP="${HOME}/.config/beacon/main.py"
SO="$(ldconfig -p 2>/dev/null | awk '/libgtk4-layer-shell\.so/ {print $NF; exit}')"
if [[ -n "${SO:-}" ]]; then
export LD_PRELOAD="${SO}${LD_PRELOAD:+:${LD_PRELOAD}}"
fi
exec python3 "$APP" "$@"