1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
| #!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------------------------------------------------------
# %% Imports
import socket
import json
import os
import signal
import argparse
from dataclasses import dataclass
from time import perf_counter, sleep
from collections import deque
# ---------------------------------------------------------------------------------------------------------------------
# %% Args
# Set built-in defaults (helpful for debugging)
default_N = 3
default_delay_ms = 1000 if perf_counter() < 5 else 0
default_maximize_solos = True
default_maximize_solo_on_close = True
default_collapse_solos_on_open = True
default_apply_on_move = False
default_debug_names = False
default_debug_data = False
# Define script arguments
parser = argparse.ArgumentParser(
description="Script which makes niri behave like an auto-tiler when there are fewer than 'N' windows"
)
parser.add_argument(
"-n",
default=default_N,
type=int,
help=f"Number of windows handled with auto-tiling (default {default_N})",
)
parser.add_argument(
"-delay",
default=default_delay_ms,
type=int,
help=f"Number of milliseconds to delay before listening to niri IPC (default: {default_delay_ms})",
)
parser.add_argument(
"-x",
action="store_false" if default_maximize_solos else "store_true",
help=f"Auto-maximize first window opened on a workspace (default: {default_maximize_solos})",
)
parser.add_argument(
"-xc",
action="store_false" if default_maximize_solo_on_close else "store_true",
help=f"When closing windows, if one window remains, auto-maximize it (default: {default_maximize_solo_on_close})",
)
parser.add_argument(
"-c",
action="store_false" if default_collapse_solos_on_open else "store_true",
help=f"Collapse solo maximized window when opening a second window (default: {default_collapse_solos_on_open})",
)
parser.add_argument(
"-m",
action="store_false" if default_apply_on_move else "store_true",
help=f"Apply tiling logic to windows that are moved into other workspaces (default: {default_apply_on_move})",
)
parser.add_argument(
"-e",
"--maximize_to_edges",
action="store_true",
help="Use maximize-to-edges instead of maximize-column",
)
parser.add_argument(
"-dn",
action="store_false" if default_debug_names else "store_true",
help="Enable event name printing, for debugging",
)
parser.add_argument(
"-dd",
action="store_false" if default_debug_data else "store_true",
help="Enable event data printing, for debugging",
)
parser.add_argument(
"-iw",
type=int,
action="append",
help="Ignore workspace with this id (can be specified multiple times)"
)
# Get script configs
args, _ = parser.parse_known_args()
TILE_TO_N = args.n
STARTUP_DELAY_MS = args.delay
MAXIMIZE_SOLOS = args.x
MAXIMIZE_SOLOS_ON_CLOSE = args.xc
COLLAPSE_SOLOS_ON_OPEN = args.c
APPLY_TO_MOVED_WINDOWS = args.m
USE_MAX_TO_EDGES = args.maximize_to_edges
ENABLE_EVENT_NAME_DEBUG_PRINT = args.dn
ENABLE_EVENT_DATA_DEBUG_PRINT = args.dd
IGNORED_WORKSPACE_IDS = args.iw
# ---------------------------------------------------------------------------------------------------------------------
# %% Data types
@dataclass
class TimeKeeper:
t1: int = 0
t2: int = 0
def get_time_elapsed_ms(self) -> int:
"""Reports the time (in ms) since the last time this function was called"""
self.t1 = self.t2
self.t2 = round(perf_counter() * 1000)
delta_ms = self.t2 - self.t1
return delta_ms
@dataclass
class FocusState:
workspace_id: int = None
window_id: int = None
def copy_inplace(self, other_focus_state):
"""Overwrite current data with data from another object (avoids creating new instances)"""
self.workspace_id = other_focus_state.workspace_id
self.window_id = other_focus_state.window_id
return self
# ---------------------------------------------------------------------------------------------------------------------
# %% Classes
class NiriSocket:
"""Helper used to read & write json messages to a niri socket connection"""
def __init__(self, socket_path: str, buffer_size: int = 4096):
# Sanity check
is_bad_path = socket_path is None or str(socket_path) == ""
assert not is_bad_path, "Cannot connect to niri, no socket path given..."
self._skt = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self._skt.connect(skt_path)
self._bufsize = buffer_size
# Storage for
self._msg_queue = deque([])
self._inprog_str = None
def _read_next(self):
# Read from existing (buffered) messages, if any
if len(self._msg_queue) > 0:
next_msg = self._msg_queue.popleft()
return json.loads(next_msg)
while True:
# Listen for raw (binary) string data from socket
# -> Will return 0 bytes if connection closes
resp_binstr = self._skt.recv(self._bufsize)
if len(resp_binstr) == 0:
print("DEBUG - READNEXT: No data received!")
return {}
# If we have an in-progress result, append the new data to it
resp_str = resp_binstr.decode("utf-8")
if self._inprog_str is not None:
resp_str = "".join((self._inprog_str, resp_str))
self._inprog_str = None
# Stop listening if got at least 1 message
# - Expect response to look like: "message 1\\nmessage 2\\nmessage 3\\n"
# - If incomplete, we'll see something not ending with '\\n': "message 1\\nmessa"
msg_list = resp_str.split("\\n")
last_msg_piece = msg_list.pop()
contains_incomplete_message = len(last_msg_piece) > 0
self._inprog_str = last_msg_piece if contains_incomplete_message else None
if len(msg_list) > 0:
break
# Sanity check, make sure we read something
if len(msg_list) == 0:
raise IOError("Error reading next message (empty message list)!")
# If we have more than 1 message, return only the 'next one
# (future calls to this function will return the queued up messages)
out_msg_str = msg_list[0]
if len(msg_list) > 1:
self._msg_queue.extend(msg_list[1:])
return json.loads(out_msg_str)
def _send_string(self, string: str):
"""Helper used to send simple string messages (e.g. for requests)"""
return self._skt.sendall(f'"{string}"\\n'.encode("utf-8"))
def _send_json(self, json_data: dict):
"""Helper used to send json message (e.g. for actions)"""
json_as_str = json.dumps(json_data, indent=None, separators=(",", ":"))
return self._skt.sendall(("".join([json_as_str, "\\n"])).encode("utf-8"))
def close(self):
self._skt.close()
@staticmethod
def get_niri_socket_path():
return os.environ.get("NIRI_SOCKET")
class NiriRequests(NiriSocket):
"""
Helper used to make requests to niri
See: https://yalter.github.io/niri/niri_ipc/enum.Request.html
"""
def get_version(self):
return self.request("Version")
def request(self, message: str):
self._send_string(message)
# Listen for ok/err response
resp_json = self._read_next()
is_ok_resp = "Ok" in resp_json.keys()
resp_data = resp_json["Ok" if is_ok_resp else "Err"]
return is_ok_resp, resp_data
def read_eventstream(self):
is_ok, evt_resp = self.request("EventStream")
if not is_ok:
print("DEBUG - EventStream response:", evt_resp, sep="\\n")
raise IOError("Error requesting EventStream")
# Read events from stream, forever
while True:
event_json = self._read_next()
event_name = tuple(event_json.keys())[0]
event_data = event_json.get(event_name, None)
yield event_name, event_data
return
class NiriActions(NiriSocket):
"""
Helper used to trigger actions through the niri IPC
See: https://yalter.github.io/niri/niri_ipc/enum.Action.html
"""
def action(self, message: str, **kwargs):
# Build action request
json_data = {"Action": {message: kwargs}}
self._send_json(json_data)
# Listen for ok/err response
resp_json = self._read_next()
is_ok_resp = "Err" not in resp_json.keys()
resp_data = resp_json if is_ok_resp else resp_json["Err"]
return is_ok_resp, resp_data
# ---------------------------------------------------------------------------------------------------------------------
# %% Functions
def catch_sigterm(signum, frame):
"""Turn SIGTERM events into exceptions for graceful shutdown"""
raise InterruptedError
def make_workspace_state_from_WorkspacesChanged(event_data: dict) -> dict[int, dict]:
return {info_dict["id"]: info_dict for info_dict in event_data["workspaces"]}
def make_window_state_from_WindowsChanged(event_data: dict, workspace_state, output_width_lut: dict) -> dict[int, dict]:
state = {}
for info_dict in event_data["windows"]:
win_id = info_dict["id"]
win_aug_data = get_additional_window_data(info_dict, workspace_state, output_width_lut)
info_dict.update(win_aug_data)
state[win_id] = info_dict
return state
def get_windows_by_conditions(window_state: dict[int, dict], **conditions) -> dict[int, dict]:
"""Function used to filter window state data according to key-value conditions"""
meets_conditions = lambda data: all(data[k] == v for k, v in conditions.items())
return {winid: windata for winid, windata in window_state.items() if meets_conditions(windata)}
def get_additional_window_data(
window_data: dict,
workspace_state: dict,
output_width_lut: dict,
max_width_threshold: float = 0.8,
) -> dict:
"""Helper used to generate addition windowing data (particularly 'is_maximized' flag)"""
# Set up augmentation data
win_pos = window_data["layout"]["pos_in_scrolling_layout"]
win_col, win_row = win_pos if win_pos is not None else (None, None)
augment_dict = {
"col_idx": win_col,
"row_idx": win_row,
"is_maximized": False,
}
# Try to figure out if window is maximized
win_wspace_id = window_data["workspace_id"]
win_output = workspace_state.get(win_wspace_id, {}).get("output", None)
output_width = output_width_lut.get(win_output, None)
if output_width is not None:
win_width = window_data["layout"]["window_size"][0]
augment_dict["is_maximized"] = (win_width / output_width) > max_width_threshold
return augment_dict
def toggle_window_maximization(target_window_id: int, focused_window_id: int):
"""Helper used to toggle the maximization state of a window, without messing with current focused window"""
if target_window_id == focused_window_id:
niri_action.action("MaximizeWindowToEdges" if USE_MAX_TO_EDGES else "MaximizeColumn")
else:
niri_action.action("FocusWindow", id=target_window_id)
niri_action.action("MaximizeWindowToEdges" if USE_MAX_TO_EDGES else "MaximizeColumn")
niri_action.action("FocusWindow", id=focused_window_id)
return
def maximize_window(window_state: dict, focus_state: FocusState, target_window_id: int) -> bool:
"""
Helper used to maximize a window if it\'s not already maximized.
This function assumes window state includes \'is_maximized\' flag!
Returns True if the window needed maximization, false otherwise
"""
solo_win_data = window_state[target_window_id]
need_maximization = not solo_win_data["is_maximized"]
if need_maximization:
solo_id = solo_win_data["id"]
toggle_window_maximization(solo_id, focus_state.window_id)
win_state[solo_id]["is_maximized"] = True
return need_maximization
def collapse_window(window_state: dict, focus_state: FocusState, target_window_id: int) -> bool:
"""
Helperused to collapse a maximized window. This function assumes
that the window state includes \'is_maximized\' flag!
Returns: True if window needed collapse, false otherwise
"""
solo_win_data = window_state[target_window_id]
need_collapse = solo_win_data["is_maximized"]
if need_collapse:
solo_id = solo_win_data["id"]
toggle_window_maximization(solo_id, focus_state.window_id)
win_state[solo_id]["is_maximized"] = False
return need_collapse
# ---------------------------------------------------------------------------------------------------------------------
# %% Setup
# Handle startup delay (prevent listening to niri during potentially busy startup)
if STARTUP_DELAY_MS > 0:
sleep(STARTUP_DELAY_MS / 1000)
# Get niri socket from env
skt_path = NiriSocket.get_niri_socket_path()
if skt_path is None or skt_path == "":
print("Couldn\'t find niri socket! (from env: NIRI_SOCKET)")
quit()
# Create separate read/write sockets, since eventstream reader cannot issue actions
niri_reader = NiriRequests(skt_path)
niri_action = NiriActions(skt_path)
# Sanity check. Make sure we have the right version
is_version_ok, version_resp = niri_reader.request("Version")
expected_version, actual_version = "26.04 (8ed0da4)", version_resp.get("Version", "unknown")
if actual_version != expected_version:
print(
"",
"WARNING - Unexpected niri version!",
f"expected: {expected_version}",
f" actual: {actual_version}",
"Errors may occur...",
sep="\\n",
)
# ---------------------------------------------------------------------------------------------------------------------
# %% *** IPC listening loop ***
# Get monitor into
is_outputs_ok, outputs_resp = niri_reader.request("Outputs")
if not is_outputs_ok:
print("Error requesting info about monitors", outputs_resp, sep="\\n")
quit()
output_full_info = {out_key: out_dict["logical"] for out_key, out_dict in outputs_resp["Outputs"].items()}
output_width_lut = {out_key: out_info["width"] for out_key, out_info in output_full_info.items() if out_info is not None}
# Initialize state tracking
prev_focus_state = FocusState()
focus_state = FocusState()
timekeeper = TimeKeeper()
win_state = None
wspace_state = None
# Main listening loop
signal.signal(signal.SIGTERM, catch_sigterm)
try:
init_time = timekeeper.get_time_elapsed_ms()
for evt_name, evt_data in niri_reader.read_eventstream():
# For debugging printouts, add spaces between events that don\'t occur together
time_elapsed_ms = timekeeper.get_time_elapsed_ms()
if ENABLE_EVENT_NAME_DEBUG_PRINT or ENABLE_EVENT_DATA_DEBUG_PRINT:
if time_elapsed_ms > 250:
print("", f"Time elapsed (sec): {(timekeeper.t2 - init_time) // 1000}", sep="\\n")
if ENABLE_EVENT_NAME_DEBUG_PRINT:
print(evt_name)
if ENABLE_EVENT_DATA_DEBUG_PRINT:
print(evt_data)
# Handle all IPC stream events
prev_focus_state.copy_inplace(focus_state)
closed_window_data, newest_window_data = None, None
if evt_name == "WorkspacesChanged":
wspace_state = make_workspace_state_from_WorkspacesChanged(evt_data)
for item in wspace_state.values():
if item["is_focused"]:
focus_state.workspace_id = item["id"]
elif evt_name == "WorkspaceUrgencyChanged":
evt_wspace_id = evt_data["id"]
wspace_state[evt_wspace_id]["is_urgent"] = evt_data["urgent"]
elif evt_name == "WorkspaceActivated":
if evt_data["focused"]:
focus_state.workspace_id = evt_data["id"]
wspace_state[prev_focus_state.workspace_id]["is_focused"] = False
pass
elif evt_name == "WindowsChanged":
win_state = make_window_state_from_WindowsChanged(evt_data, wspace_state, output_width_lut)
for item in win_state.values():
if item["is_focused"]:
focus_state.window_id = item["id"]
elif evt_name == "WindowOpenedOrChanged":
evt_win_id = evt_data["window"]["id"]
evt_win_wspace_id = evt_data["window"]["workspace_id"]
evt_is_new_window = evt_win_id not in win_state.keys()
evt_is_moved_window, prev_win_wspace_id = False, None
if not evt_is_new_window:
prev_win_wspace_id = win_state[evt_win_id]["workspace_id"]
evt_is_moved_window = prev_win_wspace_id != evt_win_wspace_id
if evt_data["window"]["is_focused"]:
focus_state.window_id = evt_win_id
win_aug_data = get_additional_window_data(evt_data["window"], wspace_state, output_width_lut)
win_state[evt_win_id] = {**evt_data["window"], **win_aug_data}
need_check_rearrange = evt_is_new_window or (evt_is_moved_window and APPLY_TO_MOVED_WINDOWS)
newest_window_data = win_state[evt_win_id] if need_check_rearrange else None
elif evt_name == "WindowClosed":
evt_win_id = evt_data["id"]
closed_window_data = win_state.pop(evt_win_id)
elif evt_name == "WindowFocusChanged":
focus_state.window_id = evt_data["id"]
elif evt_name == "WindowFocusTimestampChanged":
evt_win_id = evt_data["id"]
win_state[evt_win_id]["focus_timestamp"] = evt_data["focus_timestamp"]
elif evt_name == "WindowUrgencyChanged":
evt_win_id = evt_data["id"]
win_state[evt_win_id]["is_urgent"] = evt_data["urgent"]
elif evt_name == "WindowLayoutsChanged":
for evt_win_id, evt_new_layout in evt_data["changes"]:
win_state[evt_win_id]["layout"] = evt_new_layout
win_aug_data = get_additional_window_data(win_state[evt_win_id], wspace_state, output_width_lut)
win_state[evt_win_id].update(win_aug_data)
pass
elif evt_name == "OverviewOpenedOrClosed":
evt_is_overview_open = evt_data["is_open"]
elif evt_name == "ConfigLoaded":
pass
elif evt_name == "CastsChanged":
pass
else:
print("Unknown event:", evt_name)
# Handle max-on-close
if closed_window_data is not None:
if MAXIMIZE_SOLOS_ON_CLOSE:
curr_wspace_id = closed_window_data["workspace_id"]
curr_wins = get_windows_by_conditions(win_state, workspace_id=curr_wspace_id, is_floating=False)
if len(curr_wins) == 1:
solo_id = tuple(curr_wins.keys())[0]
maximize_window(win_state, focus_state, solo_id)
pass
# Handle window-creation behaviors
if newest_window_data is not None:
if newest_window_data["is_maximized"] or newest_window_data["is_floating"]:
continue
curr_wspace_id = newest_window_data["workspace_id"]
if IGNORED_WORKSPACE_IDS and curr_wspace_id and (curr_wspace_id in IGNORED_WORKSPACE_IDS):
print(f"Ignored event on workspace {curr_wspace_id}")
continue
curr_tile_wins = get_windows_by_conditions(win_state, workspace_id=curr_wspace_id, is_floating=False)
num_tile_wins = len(curr_tile_wins)
if num_tile_wins == 0 or num_tile_wins > TILE_TO_N:
continue
if MAXIMIZE_SOLOS and num_tile_wins == 1:
solo_id = tuple(curr_tile_wins.keys())[0]
maximize_window(win_state, focus_state, solo_id)
curr_max_wins: dict = get_windows_by_conditions(curr_tile_wins, is_maximized=True)
num_max_wins = len(curr_max_wins)
if COLLAPSE_SOLOS_ON_OPEN and num_max_wins == 1 and num_tile_wins == 2:
solo_max_id = tuple(curr_max_wins.keys())[0]
collapse_window(win_state, focus_state, solo_max_id)
num_max_wins -= 1
is_zero_max_windows = num_max_wins == 0
if is_zero_max_windows and (2 < num_tile_wins <= TILE_TO_N):
is_new_win_onscreen = newest_window_data["col_idx"] == 2
consume_action = "ConsumeOrExpelWindowRight" if is_new_win_onscreen else "ConsumeOrExpelWindowLeft"
niri_action.action(consume_action, id=newest_window_data["id"])
pass
except (KeyboardInterrupt, InterruptedError):
pass
finally:
niri_action.close()
niri_reader.close()
print("", f"({os.path.basename(__file__)}) - Closed niri IPC connection", sep="\\n")
|
