roundabout,
created on Monday, 1 September 2025, 13:00:36 (1756731636),
received on Monday, 1 September 2025, 13:00:40 (1756731640)
Author identity: Vlad <vlad.muntoiu@gmail.com>
2f32ee07acb5d479e865357c400c5cdafbff5049
wl4pgi/__init__.py
@@ -0,0 +1,141 @@
import pkgutil
import importlib
import inspect
import types
import sys
from unittest import case
import pywayland.protocol
import pywayland.protocol_core
import wl4pgi
from gi.repository import GObject, GLib, Gio
protocol_classes: dict[str, list[type]] = {}
wrappers_by_name: dict[str, type] = {}
# Find all protocols and the classes within
for finder, name, is_package in pkgutil.walk_packages(pywayland.protocol.__path__, "pywayland.protocol."):
try:
module = importlib.import_module(name)
except Exception:
continue
classes = [cls for _, cls in inspect.getmembers(module, inspect.isclass)
if cls.__module__ == module.__name__ and issubclass(cls, pywayland.protocol_core.Interface)]
# Each class has its own module like pywayland.protocol.protocol_name.class_name
protocol_name = name.rsplit(".", 2)[1]
if protocol_name not in protocol_classes:
protocol_classes[protocol_name] = []
if classes:
protocol_classes[protocol_name].extend(classes)
class GWaylandProxy(GObject.GObject):
"""Base class for PyGObject proxies for Wayland."""
__gtype_name__ = "GWaylandProxy"
def __new__(cls, proxy):
if hasattr(proxy, "gproxy"):
return proxy.gproxy
self = super().__new__(cls)
proxy.gproxy = self
return self
def __init__(self, proxy):
if hasattr(self, "internal"):
return
GObject.GObject.__init__(self)
self.internal = proxy
for event in self.events:
self.internal.dispatcher[event] = (lambda e:
lambda *args: self.emit(e, *(args[1:]))
)(event)
def make_gtype_for_pywayland(interface: type):
"""Generate a PyGObject proxy class for a PyWayland interface."""
methods = {}
request_names = []
# Define requests
for request in interface.requests:
# print("->", request.name)
# print(request.arguments)
def make_request_method(req):
def request_method(self, *args):
raw_args = [(arg.internal if isinstance(arg, GWaylandProxy) else arg) for
arg in args]
raw_result = getattr(self.internal, req.name)(*raw_args)
if isinstance(raw_result, pywayland.protocol_core.Interface):
return wrappers_by_name[type(raw_result).__name__](raw_result)
if isinstance(raw_result, pywayland.protocol_core.Proxy):
return wrappers_by_name[type(raw_result).__name__.removesuffix("Proxy")](raw_result)
return raw_result
return request_method
methods[request.name] = make_request_method(request)
request_names.append(request.name)
proxy_type = interface.proxy_class
# Define events
signals = {}
for event in interface.events:
# print("<-", event.name)
# print(event.arguments)
signals[event.name] = (
GObject.SignalFlags.RUN_FIRST,
None,
tuple([GObject.TYPE_PYOBJECT] * len(event.arguments)),
)
methods["__gsignals__"] = signals
methods["requests"] = request_names
methods["events"] = list(signals.keys())
methods["proxy_type"] = proxy_type
methods["interface_type"] = interface
# Define constructor
def __init__(self, proxy):
if not isinstance(proxy, proxy_type):
raise TypeError(f"Proxy instantiated with incorrect type {type(proxy)}")
GWaylandProxy.__init__(self, proxy)
methods["__init__"] = __init__
return type(interface.__name__, (GWaylandProxy,), methods)
class VirtualSubModule(types.ModuleType):
"""Holder for a particular Wayland protocol."""
def __init__(self, protocol_name):
super().__init__(f"{__name__}.{protocol_name}")
self.protocol_name = protocol_name
self._generate()
def _generate(self):
self._class_list = protocol_classes[self.protocol_name]
for klass in self._class_list:
setattr(self, klass.__name__, make_gtype_for_pywayland(klass))
wrappers_by_name[klass.__name__] = getattr(self, klass.__name__)
# Inject the protocols
for protocol, classes in protocol_classes.items():
submodule = VirtualSubModule(protocol)
sys.modules[submodule.__name__] = submodule
setattr(wl4pgi, protocol, submodule)
wl4pgi_example.py
@@ -0,0 +1,52 @@
from pywayland.client import Display, EventQueue
import wl4pgi
import gi
from gi.repository import GLib, Gio
# The display is NOT converted to GObject since they have a lot of special behaviour. Also
# you still have to roundtrip, but only after creating GObject adapters.
display = Display()
display.connect()
event_queue = EventQueue(display)
def on_geometry(output, x, y, pw, ph, subpixel, make, model, transform):
print(output, x, y, pw, ph, subpixel, make, model, transform)
def on_geometry_1(output, x, y, pw, ph, subpixel, make, model, transform):
print("Since it uses GObject signals, multiple handlers work without problems")
print(f"Also, custom attributes work: {output.test}. There's always zero or one GObjects "
f"per Wayland object.")
def on_global(registry, name, interface, version):
if interface == "wl_output":
# GObjects adapters only represent a proxy. Thus, to get the interface class for use
# in this signal, you must go into pywayland.
output = registry.bind(name, wl4pgi.wayland.WlOutput.interface_type, version)
output.test = 0
# Multiple handlers, transparently and with a single display and a single registry.
output.connect("geometry", on_geometry)
output.connect("geometry", on_geometry_1)
display.roundtrip()
# Starting from the registry, operation is within PyGObject.
registry = wl4pgi.wayland.WlRegistry(display.get_registry())
registry.connect("global", on_global)
display.roundtrip()
# The next parts would be removed for GTK integration.
def on_display_event(self, source, condition):
if condition & GLib.IO_IN:
self.display.dispatch(queue=self.event_queue)
return True
fd = display.get_fd()
GLib.io_add_watch(fd, GLib.IO_IN, on_display_event)
main_loop = GLib.MainLoop()
main_loop.run()