__init__.py
Python script, ASCII text executable
1import pkgutil 2import importlib 3import inspect 4import types 5import sys 6from unittest import case 7 8import pywayland.protocol 9import pywayland.protocol_core 10 11import wl4pgi 12 13from gi.repository import GObject, GLib, Gio 14 15protocol_classes: dict[str, list[type]] = {} 16wrappers_by_name: dict[str, type] = {} 17 18# Find all protocols and the classes within 19for finder, name, is_package in pkgutil.walk_packages(pywayland.protocol.__path__, "pywayland.protocol."): 20try: 21module = importlib.import_module(name) 22except Exception: 23continue 24 25classes = [cls for _, cls in inspect.getmembers(module, inspect.isclass) 26if cls.__module__ == module.__name__ and issubclass(cls, pywayland.protocol_core.Interface)] 27 28# Each class has its own module like pywayland.protocol.protocol_name.class_name 29protocol_name = name.rsplit(".", 2)[1] 30 31if protocol_name not in protocol_classes: 32protocol_classes[protocol_name] = [] 33 34if classes: 35protocol_classes[protocol_name].extend(classes) 36 37 38class GWaylandProxy(GObject.GObject): 39"""Base class for PyGObject proxies for Wayland.""" 40__gtype_name__ = "GWaylandProxy" 41 42def __new__(cls, proxy): 43if hasattr(proxy, "gproxy"): 44return proxy.gproxy 45 46self = super().__new__(cls) 47proxy.gproxy = self 48 49return self 50 51def __init__(self, proxy): 52if hasattr(self, "internal"): 53return 54 55GObject.GObject.__init__(self) 56self.internal = proxy 57 58for event in self.events: 59self.internal.dispatcher[event] = (lambda e: 60lambda *args: self.emit(e, *(args[1:])) 61)(event) 62 63 64def make_gtype_for_pywayland(interface: type): 65"""Generate a PyGObject proxy class for a PyWayland interface.""" 66methods = {} 67request_names = [] 68 69# Define requests 70for request in interface.requests: 71# print("->", request.name) 72# print(request.arguments) 73 74def make_request_method(req): 75def request_method(self, *args): 76raw_args = [(arg.internal if isinstance(arg, GWaylandProxy) else arg) for 77arg in args] 78raw_result = getattr(self.internal, req.name)(*raw_args) 79 80if isinstance(raw_result, pywayland.protocol_core.Interface): 81return wrappers_by_name[type(raw_result).__name__](raw_result) 82if isinstance(raw_result, pywayland.protocol_core.Proxy): 83return wrappers_by_name[type(raw_result).__name__.removesuffix("Proxy")](raw_result) 84 85return raw_result 86 87return request_method 88 89methods[request.name] = make_request_method(request) 90request_names.append(request.name) 91 92proxy_type = interface.proxy_class 93 94# Define events 95signals = {} 96for event in interface.events: 97# print("<-", event.name) 98# print(event.arguments) 99signals[event.name] = ( 100GObject.SignalFlags.RUN_FIRST, 101None, 102tuple([GObject.TYPE_PYOBJECT] * len(event.arguments)), 103) 104 105methods["__gsignals__"] = signals 106methods["requests"] = request_names 107methods["events"] = list(signals.keys()) 108methods["proxy_type"] = proxy_type 109methods["interface_type"] = interface 110 111# Define constructor 112def __init__(self, proxy): 113if not isinstance(proxy, proxy_type): 114raise TypeError(f"Proxy instantiated with incorrect type {type(proxy)}") 115GWaylandProxy.__init__(self, proxy) 116 117methods["__init__"] = __init__ 118 119return type(interface.__name__, (GWaylandProxy,), methods) 120 121 122class VirtualSubModule(types.ModuleType): 123"""Holder for a particular Wayland protocol.""" 124def __init__(self, protocol_name): 125super().__init__(f"{__name__}.{protocol_name}") 126self.protocol_name = protocol_name 127self._generate() 128 129def _generate(self): 130self._class_list = protocol_classes[self.protocol_name] 131 132for klass in self._class_list: 133setattr(self, klass.__name__, make_gtype_for_pywayland(klass)) 134wrappers_by_name[klass.__name__] = getattr(self, klass.__name__) 135 136 137# Inject the protocols 138for protocol, classes in protocol_classes.items(): 139submodule = VirtualSubModule(protocol) 140sys.modules[submodule.__name__] = submodule 141setattr(wl4pgi, protocol, submodule)