116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""Thin wrapper around swaymsg and local process launching.
|
|
|
|
Identical in shape to hosts/thin-client/agent/thinclient_agent/sway_control.py —
|
|
duplicated rather than imported across hosts, same convention as the digest/admin
|
|
canvas SDK split in Phase 13.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import glob
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Contract with configs/sway/config — these strings must match the `set $ws_*` lines.
|
|
WS_SPOTIFY = "1:spotify"
|
|
WS_HOME = "2:home"
|
|
WS_WEB = "3:web"
|
|
|
|
|
|
def runtime_dir() -> str:
|
|
return os.environ.get("XDG_RUNTIME_DIR") or f"/run/user/{os.getuid()}"
|
|
|
|
|
|
class SwayControl:
|
|
def _socket_path(self) -> str | None:
|
|
path = os.environ.get("SWAYSOCK")
|
|
if path and os.path.exists(path):
|
|
return path
|
|
# sway names the socket sway-ipc.<uid>.<pid>.sock, so the path changes every
|
|
# time sway restarts. touchpanel-agent is a system service that outlives the
|
|
# session, so the socket is re-resolved per call instead of cached at startup.
|
|
matches = sorted(glob.glob(os.path.join(runtime_dir(), "sway-ipc.*.sock")))
|
|
return matches[-1] if matches else None
|
|
|
|
def session_env(self) -> dict[str, str]:
|
|
env = dict(os.environ)
|
|
env["XDG_RUNTIME_DIR"] = runtime_dir()
|
|
env.setdefault("DBUS_SESSION_BUS_ADDRESS", f"unix:path={runtime_dir()}/bus")
|
|
env.setdefault("WAYLAND_DISPLAY", "wayland-1")
|
|
sock = self._socket_path()
|
|
if sock:
|
|
env["SWAYSOCK"] = sock
|
|
return env
|
|
|
|
def swaymsg(self, *args: str) -> str | None:
|
|
if self._socket_path() is None:
|
|
log.warning("no sway IPC socket found; dropping command %s", " ".join(args))
|
|
return None
|
|
try:
|
|
result = subprocess.run(
|
|
["swaymsg", *args],
|
|
env=self.session_env(),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
log.warning("swaymsg %s failed: %s", " ".join(args), exc)
|
|
return None
|
|
if result.returncode != 0:
|
|
log.warning("swaymsg %s: %s", " ".join(args), result.stderr.strip())
|
|
return None
|
|
return result.stdout
|
|
|
|
def switch_workspace(self, name: str) -> None:
|
|
log.info("switching to workspace %s", name)
|
|
self.swaymsg("workspace", name)
|
|
|
|
def focus_window(self, criteria: str) -> None:
|
|
self.swaymsg(f"[{criteria}] focus")
|
|
|
|
def is_running(self, pattern: str) -> bool:
|
|
try:
|
|
result = subprocess.run(
|
|
["pgrep", "-u", str(os.getuid()), "-f", pattern],
|
|
capture_output=True,
|
|
timeout=5,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return False
|
|
return result.returncode == 0
|
|
|
|
def launch_app(
|
|
self,
|
|
command: list[str],
|
|
process_pattern: str | None = None,
|
|
workspace: str | None = None,
|
|
focus_criteria: str | None = None,
|
|
) -> None:
|
|
if workspace:
|
|
self.switch_workspace(workspace)
|
|
|
|
if process_pattern and self.is_running(process_pattern):
|
|
log.info("%s already running; focusing instead of launching", command[0])
|
|
if focus_criteria:
|
|
self.focus_window(focus_criteria)
|
|
return
|
|
|
|
log.info("launching %s", " ".join(command))
|
|
try:
|
|
subprocess.Popen(
|
|
command,
|
|
env=self.session_env(),
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
)
|
|
except OSError as exc:
|
|
log.error("could not launch %s: %s", " ".join(command), exc)
|