By using this site, you agree to have cookies stored on your device, strictly for functional purposes, such as storing your session and preferences.

Dismiss

 __init__.py

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