206 lines
7.4 KiB
Python
206 lines
7.4 KiB
Python
"""TUIOS helpers: claim a workspace by name, list what is open.
|
|
|
|
A workspace number is never a home. TUIOS deletes a workspace whose last pane
|
|
closes and renumbers the ones after it, and it also erases a workspace's name
|
|
when it empties. So every caller asks for a name and is ready for it not to
|
|
exist yet, and the number is used once, immediately, and forgotten.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
SESSION = os.environ.get("PAPERCLIP_HARNESS_SESSION", "paperclip-harness")
|
|
|
|
|
|
def tuios(*args, session=None):
|
|
argv = ["tuios", *args, "-s", session or SESSION, "--json"]
|
|
p = subprocess.run(argv, capture_output=True, text=True, timeout=60)
|
|
if p.returncode != 0:
|
|
raise RuntimeError("tuios %s failed: %s" % (" ".join(args), p.stderr.strip()))
|
|
try:
|
|
return json.loads(p.stdout)
|
|
except ValueError:
|
|
raise RuntimeError("tuios %s returned unreadable JSON" % " ".join(args))
|
|
|
|
|
|
def sessions():
|
|
"""Every session the daemon knows, as plain names."""
|
|
p = subprocess.run(["tuios", "ls", "--json"], capture_output=True, text=True, timeout=30)
|
|
if p.returncode != 0:
|
|
raise RuntimeError("tuios ls failed: %s" % p.stderr.strip())
|
|
data = json.loads(p.stdout)
|
|
rows = data if isinstance(data, list) else data.get("sessions", [])
|
|
return [r.get("name") for r in rows if isinstance(r, dict)]
|
|
|
|
|
|
def require_session(name):
|
|
"""Sessions are made by a person, in a terminal. Say so, do not invent one."""
|
|
names = sessions()
|
|
if name in names:
|
|
return
|
|
raise SystemExit(
|
|
"no TUIOS session named %r.\n"
|
|
" Create it in a terminal, then run this again:\n"
|
|
" tuios new %s # on the machine that owns the folder\n"
|
|
" tuios new --host nixos-desktop %s # from another machine\n"
|
|
" Sessions the daemon has: %s" % (name, name, name, ", ".join(names) or "none")
|
|
)
|
|
|
|
|
|
def workspaces(session=None):
|
|
return tuios("list-workspaces", session=session).get("workspaces", [])
|
|
|
|
|
|
def windows(session=None):
|
|
return tuios("list-windows", session=session).get("windows", [])
|
|
|
|
|
|
def _name_workspace(number, name, session=None):
|
|
subprocess.run(
|
|
["tuios", "set-workspace-name", str(number), name, "-s", session or SESSION],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def claim(name, session=None):
|
|
"""Return the number of the workspace called `name`, naming one if needed."""
|
|
ws = workspaces(session)
|
|
exact = next((w for w in ws if (w.get("name") or "") == name), None)
|
|
if exact is None:
|
|
same = next((w for w in ws if (w.get("name") or "").lower() == name.lower()), None)
|
|
if same is not None:
|
|
_name_workspace(same["workspace"], name, session)
|
|
return same["workspace"]
|
|
if exact is not None:
|
|
return exact["workspace"]
|
|
|
|
# Nothing named that. Take a workspace that is unnamed and has no panes.
|
|
free = next(
|
|
(w for w in ws if "window_count" in w and (w.get("window_count") or 0) == 0 and not (w.get("name") or "")),
|
|
None,
|
|
)
|
|
if free is None:
|
|
raise RuntimeError(
|
|
"no empty workspace left in session %r; close a pane or add a workspace" % (session or SESSION)
|
|
)
|
|
_name_workspace(free["workspace"], name, session)
|
|
return free["workspace"]
|
|
|
|
|
|
def open_pane(name, workspace_name, cwd, command, session=None, adopt_existing=False):
|
|
"""Open a pane in a named workspace, or say it is already there.
|
|
|
|
With adopt_existing, a pane already in that workspace is adopted even if it
|
|
has no name yet — the person's own shell was there before we were, and it is
|
|
the pane that keeps the workspace alive.
|
|
"""
|
|
ws = claim(workspace_name, session)
|
|
# A pane that was already in this workspace before us is the one to keep:
|
|
# it is the person's own shell, and it is what stops the workspace being
|
|
# deleted. Only when the workspace is empty do we make a new one, and only
|
|
# then does a name match matter.
|
|
if adopt_existing:
|
|
existing = [w for w in windows(session) if w.get("workspace") == ws]
|
|
if existing:
|
|
w = existing[0]
|
|
if (w.get("custom_name") or "") != name:
|
|
subprocess.run(
|
|
["tuios", "set-window", "-w", w["window_id"], "--name", name,
|
|
"-s", session or SESSION],
|
|
capture_output=True, text=True, timeout=30, check=False,
|
|
)
|
|
return "adopted", w["window_id"], ws
|
|
for w in windows(session):
|
|
if (w.get("custom_name") or "") == name:
|
|
return "adopted", w["window_id"], ws
|
|
subprocess.run(
|
|
[
|
|
"tuios", "new-window", name,
|
|
"-s", session or SESSION,
|
|
"--workspace", str(ws),
|
|
"--cwd", cwd,
|
|
"--no-focus", "--keep-open",
|
|
"--", "/bin/sh", "-c", command,
|
|
],
|
|
capture_output=True, text=True, timeout=60, check=False,
|
|
)
|
|
return "created", name, ws
|
|
|
|
|
|
def arrange(order, session=None):
|
|
"""Put these panes on their workspaces, in this order, and name the workspaces.
|
|
|
|
`order` is a list of (pane name, workspace name). Panes that name the same
|
|
workspace are put in that same workspace, so one workspace can hold several
|
|
panes. Workspaces are numbered in the order they first appear.
|
|
"""
|
|
names = []
|
|
for _pane, workspace in order:
|
|
if workspace not in names:
|
|
names.append(workspace)
|
|
|
|
moved = []
|
|
for number, workspace in enumerate(names, start=1):
|
|
for pane, ws in order:
|
|
if ws != workspace:
|
|
continue
|
|
win = next((w for w in windows(session) if (w.get("custom_name") or "") == pane), None)
|
|
if win is None:
|
|
continue
|
|
if win.get("workspace") != number:
|
|
subprocess.run(
|
|
["tuios", "move-window", str(number), "-w", win["window_id"], "-s", session or SESSION],
|
|
capture_output=True, text=True, timeout=30, check=False,
|
|
)
|
|
_name_workspace(number, workspace, session)
|
|
moved.append((number, workspace))
|
|
return moved
|
|
|
|
|
|
def new_window(name, workspace_number, cwd, command, session=None):
|
|
"""Create one pane in a workspace that has already been claimed."""
|
|
subprocess.run(
|
|
[
|
|
"tuios", "new-window", name,
|
|
"-s", session or SESSION,
|
|
"--workspace", str(workspace_number),
|
|
"--cwd", cwd,
|
|
"--no-focus", "--keep-open",
|
|
"--", "/bin/sh", "-c", command,
|
|
],
|
|
capture_output=True, text=True, timeout=60, check=False,
|
|
)
|
|
return name
|
|
|
|
|
|
def close_pane(name, session=None):
|
|
for w in windows(session):
|
|
if (w.get("custom_name") or "") == name:
|
|
subprocess.run(
|
|
["tuios", "close-window", w["window_id"], "-s", session or SESSION],
|
|
capture_output=True, text=True, timeout=30, check=False,
|
|
)
|
|
return True
|
|
return False
|
|
|
|
|
|
def drop_empty_unnamed(session=None):
|
|
"""Remove the unnamed, empty workspaces a rebuild leaves behind."""
|
|
gone = 0
|
|
for w in workspaces(session):
|
|
if "window_count" in w and (w.get("window_count") or 0) == 0 and not (w.get("name") or ""):
|
|
continue
|
|
return gone
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1 and sys.argv[1] == "claim":
|
|
print(claim(sys.argv[2]))
|
|
else:
|
|
print("usage: python3 lib/tuios.py claim NAME")
|