80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
"""Async client for beacon's eu.abdelbaki.beacon.History1 D-Bus interface.
|
|
|
|
Runs on the GTK main loop: every call is async (Gio.DBusProxy.call, never
|
|
call_sync) so a slow or hung beacon never freezes the panel. If beacon isn't
|
|
running yet (or drops off the bus), calls fail — callers get an empty list
|
|
or a silent no-op rather than a crash, the same defensive stance beacon's
|
|
own controls.py/history.py took.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable, Optional
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gio, GLib # noqa: E402
|
|
|
|
# Same well-known name/path/interface beacon's own paths.py defines —
|
|
# duplicated here as plain strings rather than a shared import, matching how
|
|
# screenrec.sh independently hardcodes them too (each deployed component is
|
|
# its own ~/.config/<name> tree with no shared Python path).
|
|
FDN_NAME = "org.freedesktop.Notifications"
|
|
FDN_PATH = "/org/freedesktop/Notifications"
|
|
HISTORY_IFACE = "eu.abdelbaki.beacon.History1"
|
|
|
|
|
|
class HistoryClient:
|
|
def __init__(self,
|
|
on_added: Optional[Callable[[dict], None]] = None,
|
|
on_removed: Optional[Callable[[int], None]] = None,
|
|
on_cleared: Optional[Callable[[], None]] = None) -> None:
|
|
self._on_added = on_added
|
|
self._on_removed = on_removed
|
|
self._on_cleared = on_cleared
|
|
# Constructing a proxy doesn't require the peer to be present yet —
|
|
# only individual calls fail while beacon isn't up.
|
|
self._proxy = Gio.DBusProxy.new_for_bus_sync(
|
|
Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None,
|
|
FDN_NAME, FDN_PATH, HISTORY_IFACE, None)
|
|
self._proxy.get_connection().signal_subscribe(
|
|
FDN_NAME, HISTORY_IFACE, None, FDN_PATH, None,
|
|
Gio.DBusSignalFlags.NONE, self._on_signal)
|
|
|
|
def _on_signal(self, _conn, _sender, _path, _iface, signal, params) -> None:
|
|
if signal == "HistoryAdded" and self._on_added is not None:
|
|
(entry,) = params.unpack()
|
|
self._on_added(entry)
|
|
elif signal == "HistoryRemoved" and self._on_removed is not None:
|
|
(nid,) = params.unpack()
|
|
self._on_removed(int(nid))
|
|
elif signal == "HistoryCleared" and self._on_cleared is not None:
|
|
self._on_cleared()
|
|
|
|
# -- calls --------------------------------------------------------------
|
|
def list_async(self, callback: Callable[[list[dict]], None]) -> None:
|
|
def done(proxy, result, _data=None) -> None:
|
|
try:
|
|
(entries,) = proxy.call_finish(result).unpack()
|
|
except GLib.GError:
|
|
entries = []
|
|
callback(entries)
|
|
self._proxy.call("List", None, Gio.DBusCallFlags.NONE, -1, None, done, None)
|
|
|
|
def remove_async(self, nid: int) -> None:
|
|
self._proxy.call(
|
|
"Remove", GLib.Variant("(u)", (nid,)), Gio.DBusCallFlags.NONE, -1, None,
|
|
self._ignore_result, None)
|
|
|
|
def clear_async(self) -> None:
|
|
self._proxy.call(
|
|
"Clear", None, Gio.DBusCallFlags.NONE, -1, None, self._ignore_result, None)
|
|
|
|
@staticmethod
|
|
def _ignore_result(proxy, result, _data=None) -> None:
|
|
try:
|
|
proxy.call_finish(result)
|
|
except GLib.GError:
|
|
pass # beacon not running / already gone — nothing to do
|