refactor: share the sway helpers

classroom-dvd-session needs to ask sway whether VLC has taken the
screen, which is the same tree walking classroom-window already does.
One copy, in a module both import; classroom-window loses a third of its
length.

The state machine is worth protecting through a move like that, so its
regression test comes with it: the exact swaymsg commands for launch,
focus, minimise, restore, Super+M, Super+Shift+M, the scratchpad rescue
and sway not running at all.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
ekklesia
2026-09-04 23:17:18 -05:00
co-authored by Claude Opus 5
parent 86dff744f7
commit 719b856040
3 changed files with 281 additions and 111 deletions
+12 -111
View File
@@ -10,18 +10,23 @@ scratchpad approach did.
"""
import argparse
import json
import os
import re
import subprocess
import sys
HIDDEN_WORKSPACE = "__ekklesia_hidden"
sys.path.insert(0, "/usr/local/lib/ekklesia-classroom")
sys.path.insert(
0,
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lib"),
)
# sway parks scratchpad windows on a workspace of its own. Nothing here
# puts them there, but treating it as hidden means a window left in the
# scratchpad by an older classroom build can still be restored.
HIDDEN_WORKSPACES = {HIDDEN_WORKSPACE, "__i3_scratch"}
from classroom_sway import ( # noqa: E402
HIDDEN_WORKSPACE,
command,
matches,
visible_workspace,
windows,
)
STATE_DIR = os.path.join(
os.environ.get("XDG_RUNTIME_DIR") or "/tmp",
@@ -30,110 +35,6 @@ STATE_DIR = os.path.join(
STATE_FILE = os.path.join(STATE_DIR, "minimized")
def swaymsg(*arguments):
"""Run swaymsg, returning stdout or None when sway is unavailable."""
try:
result = subprocess.run(
["swaymsg", *arguments],
capture_output=True,
text=True,
)
except (OSError, subprocess.SubprocessError):
return None
if result.returncode != 0:
return None
return result.stdout
def command(argument):
"""Run a sway command, ignoring criteria that match nothing."""
swaymsg("--", argument)
def query(kind):
output = swaymsg("-t", kind, "-r")
if not output:
return None
try:
return json.loads(output)
except ValueError:
return None
def collect(node, workspace, found):
"""Flatten the sway tree into the windows the classroom cares about."""
if node.get("type") == "workspace":
workspace = node.get("name")
is_window = node.get("pid") is not None and not node.get("nodes")
if is_window:
properties = node.get("window_properties") or {}
found.append(
{
"id": node.get("id"),
"app_id": node.get("app_id") or "",
"class": properties.get("class") or "",
"instance": properties.get("instance") or "",
"focused": bool(node.get("focused")),
"workspace": workspace,
"hidden": workspace in HIDDEN_WORKSPACES,
}
)
for child in node.get("nodes", []) + node.get("floating_nodes", []):
collect(child, workspace, found)
return found
def windows():
tree = query("get_tree")
if tree is None:
return []
return collect(tree, None, [])
def matches(window, patterns):
"""Match a window against app_id / X11 class / X11 instance."""
identities = [
window["app_id"],
window["class"],
window["instance"],
]
for pattern in patterns:
expression = re.compile(pattern, re.IGNORECASE)
for identity in identities:
if identity and expression.fullmatch(identity):
return True
return False
def visible_workspace():
"""The workspace a restored window should come back to."""
workspaces = query("get_workspaces") or []
for workspace in workspaces:
if workspace.get("focused") and workspace.get("name") not in HIDDEN_WORKSPACES:
return workspace.get("name")
for workspace in workspaces:
if workspace.get("visible") and workspace.get("name") not in HIDDEN_WORKSPACES:
return workspace.get("name")
return "1"
def read_state():
try:
with open(STATE_FILE, "r", encoding="utf-8") as handle: