Something went wrong. Try again.
A Material 3 inspired desktop shell for Niri and Hyprland created with Ignis.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768from __future__ import annotations
import asyncioimport logging
from gi.repository import Gio, GLib
log = logging.getLogger("exo.dbus")
PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"OBJECT_MANAGER_INTERFACE = "org.freedesktop.DBus.ObjectManager"
_running: set[asyncio.Future] = set()
def call( bus: Gio.DBusConnection, destination: str, path: str, interface: str, method: str, params: GLib.Variant | None = None, timeout_ms: int = 30_000,) -> asyncio.Future: future = asyncio.get_running_loop().create_future()
def on_reply(connection, result): if future.cancelled(): return try: future.set_result(connection.call_finish(result)) except GLib.Error as error: future.set_exception(error)
bus.call( destination, path, interface, method, params, None, Gio.DBusCallFlags.NO_AUTO_START, timeout_ms, None, on_reply, ) return future
def remote_error_name(error: GLib.Error) -> str: name = Gio.DBusError.get_remote_error(error) or "" return name.rsplit(".", 1)[-1]
def spawn(coroutine, what: str) -> None: task = asyncio.ensure_future(coroutine) _running.add(task)
def finished(done): _running.discard(done) if done.cancelled(): return error = done.exception() if error is not None: log.error("%s failed: %s", what, error, exc_info=error)
task.add_done_callback(finished)