89 lines
3.4 KiB
Nix
89 lines
3.4 KiB
Nix
{ pkgs }:
|
|
pkgs.writeScriptBin "niri-shortcut-swap-monitor-workspaces" ''
|
|
#!${pkgs.python3}/bin/python3
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
|
|
|
|
def disposable(workspace):
|
|
return workspace["active_window_id"] is None and workspace["name"] is None
|
|
|
|
|
|
def swap(connection):
|
|
with connection.makefile("rw", encoding="utf-8") as stream:
|
|
def request(message):
|
|
stream.write(json.dumps(message) + "\n")
|
|
stream.flush()
|
|
reply = json.loads(stream.readline())
|
|
if "Err" in reply:
|
|
raise RuntimeError(reply["Err"])
|
|
return reply["Ok"]
|
|
|
|
def action(name, **arguments):
|
|
request({"Action": {name: arguments}})
|
|
|
|
workspaces = request("Workspaces")["Workspaces"]
|
|
source = next((ws for ws in workspaces if ws["is_focused"]), None)
|
|
active = {
|
|
ws["output"]: ws for ws in workspaces
|
|
if ws["output"] is not None and ws["is_active"]
|
|
}
|
|
if source is None or source["output"] not in active or len(active) < 2:
|
|
return
|
|
outputs = sorted(active)
|
|
target = active[outputs[(outputs.index(source["output"]) + 1) % len(outputs)]]
|
|
if source["active_window_id"] is None and target["active_window_id"] is None:
|
|
return
|
|
|
|
# IDs survive moves; numeric workspace positions do not.
|
|
moves = ((source, target), (target, source))
|
|
for workspace, destination in moves:
|
|
action(
|
|
"MoveWorkspaceToMonitor",
|
|
reference={"Id": workspace["id"]},
|
|
output=destination["output"],
|
|
)
|
|
|
|
# Activate the original monitor last to keep keyboard focus there.
|
|
for workspace, destination in moves:
|
|
current = request("Workspaces")["Workspaces"]
|
|
arriving = next((
|
|
ws for ws in current
|
|
if ws["id"] == workspace["id"] and ws["output"] == destination["output"]
|
|
), None)
|
|
if arriving is None and disposable(workspace):
|
|
# Niri may discard an unnamed empty workspace during the swap.
|
|
arriving = max((
|
|
ws for ws in current
|
|
if ws["output"] == destination["output"] and disposable(ws)
|
|
), key=lambda ws: ws["idx"], default=None)
|
|
if arriving is None:
|
|
raise RuntimeError("workspace or monitor disappeared during swap")
|
|
reference = {"Id": arriving["id"]}
|
|
# Focus empty workspaces before reordering so Niri retains them.
|
|
action("FocusWorkspace", reference=reference)
|
|
action("MoveWorkspaceToIndex", reference=reference, index=destination["idx"])
|
|
|
|
|
|
def main():
|
|
lock_path = os.path.join(os.environ["XDG_RUNTIME_DIR"], "niri-swap-workspaces.lock")
|
|
with open(lock_path, "a") as lock:
|
|
try:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
return
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
|
|
connection.settimeout(5)
|
|
connection.connect(os.environ["NIRI_SOCKET"])
|
|
swap(connection)
|
|
|
|
|
|
try:
|
|
main()
|
|
except (OSError, ValueError, KeyError, RuntimeError) as error:
|
|
sys.exit(f"niri workspace swap: {error}")
|
|
''
|