wilderjds
/
keyd
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

keyd-application-mapper: Refactor wayland code

Browse Source 
Clean up Wayland() and allow binding to multiple protocol objects simultaneously
master
Raheman Vaiya 1 year ago
parent 74217ba4a7
commit f20bd7a441
1 changed files with 49 additions and 44 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 93
      scripts/keyd-application-mapper

93
scripts/keyd-application-mapper
Unescape View File

@ -176,7 +176,7 @@ class KDE():
# so this only works for wlroots based compositors :(.
class Wayland():
def __init__(self, interface_name):
def __init__(self, *interfaces):
path = os.getenv("WAYLAND_DISPLAY")
if path == None:
raise Exception("WAYLAND_DISPLAY not set (is wayland running?)")
@ -186,7 +186,8 @@ class Wayland():
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect(path)
self._bind_interface(interface_name)
self._bind_interfaces(interfaces)
def send_msg(self, object_id, opcode, payload):
opcode |= (len(payload)+8) << 16
@ -208,24 +209,31 @@ class Wayland():
return b[4:4+struct.unpack('I', b[:4])[0]-1].decode('utf8')
# Will bind the requested interface to object id 4
def _bind_interface(self, name):
def _bind_interfaces(self, interfaces):
# bind the registry object to id 2
self.send_msg(1, 1, b'\x02\x00\x00\x00')
# solicit a sync message to bookend the registry events
self.send_msg(1, 0, b'\x03\x00\x00\x00')
interface_object_number = 4
while True:
(obj, event, payload) = self.recv_msg()
if obj == 2 and event == 0: # registry.global event
wl_name = struct.unpack('I', payload[0:4])[0]
wl_interface = self.read_string(payload[4:])
if wl_interface == name:
self.send_msg(2, 0, payload+b'\x04\x00\x00\x00')
return
for interface in interfaces:
if wl_interface == interface:
bind_payload = payload + struct.pack('I', interface_object_number)
self.send_msg(2, 0, bind_payload)
setattr(self, interface, interface_object_number)
interface_object_number += 1
if obj == 3: # sync message
raise Exception(f"Could not find interface {name}")
for interface in interfaces:
if not hasattr(self, interface):
raise Exception(f"Could not find interface {interface}")
return
class Wlroots():
@ -246,45 +254,42 @@ class Wlroots():
continue
(obj, event, payload) = msg
if obj == 4 and event == 0:
if obj == self.wl.zwlr_foreign_toplevel_manager_v1 and event == 0:
# zwlr_foreign_toplevel_manager_v1::toplevel event
windows[struct.unpack('I', payload)[0]] = {'title': '', 'appid': ''}
continue
if obj not in windows:
continue
if event == 0:
# zwlr_foreign_toplevel_handle_v1::title event
windows[obj]['title'] = self.wl.read_string(payload)
elif event == 1:
# zwlr_foreign_toplevel_handle_v1::app_id event
windows[obj]['appid'] = self.wl.read_string(payload)
elif event == 4:
# zwlr_foreign_toplevel_handle_v1::state event
if active_window == windows[obj]:
active_window = None
window_is_active = False
array_size = struct.unpack('I', payload[0:4])[0]
for i in range(0, array_size, 4):
start_offset = i + 4
end_offset = start_offset + 4
state = struct.unpack('I', payload[start_offset:end_offset])[0]
# zwlr_foreign_toplevel_handle_v1::state enum -> activated
if state == 2:
window_is_active = True
if window_is_active:
active_window = windows[obj]
elif event == 5 and active_window == windows[obj]:
# zwlr_foreign_toplevel_handle_v1::done event
self.on_window_change(active_window['appid'], active_window['title'])
elif event == 6:
# zwlr_foreign_toplevel_handle_v1::closed event
closed_window = windows.pop(obj)
if closed_window == active_window:
active_window = None
elif obj in windows:
if event == 0:
# zwlr_foreign_toplevel_handle_v1::title event
windows[obj]['title'] = self.wl.read_string(payload)
elif event == 1:
# zwlr_foreign_toplevel_handle_v1::app_id event
windows[obj]['appid'] = self.wl.read_string(payload)
elif event == 4:
# zwlr_foreign_toplevel_handle_v1::state event
if active_window == windows[obj]:
active_window = None
window_is_active = False
array_size = struct.unpack('I', payload[0:4])[0]
for i in range(0, array_size, 4):
start_offset = i + 4
end_offset = start_offset + 4
state = struct.unpack('I', payload[start_offset:end_offset])[0]
# zwlr_foreign_toplevel_handle_v1::state enum -> activated
if state == 2:
window_is_active = True
if window_is_active:
active_window = windows[obj]
elif event == 5 and active_window == windows[obj]:
# zwlr_foreign_toplevel_handle_v1::done event
self.on_window_change(active_window['appid'], active_window['title'])
elif event == 6:
# zwlr_foreign_toplevel_handle_v1::closed event
closed_window = windows.pop(obj)
if closed_window == active_window:
active_window = None
class Cosmic():
def __init__(self, on_window_change):

Write Preview
Loading…
Cancel
Save