64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""Subprocess helpers built on Gio so nothing blocks the GTK main loop.
|
|
|
|
Trimmed copy of astal-menu/lib/proc.py's contract: a command prints JSON (or plain
|
|
text) to stdout, diagnostics to stderr, non-zero exit on failure. These helpers run
|
|
such commands asynchronously and hand the parsed result back on the main thread.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shlex
|
|
from typing import Callable, Sequence
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gio", "2.0")
|
|
from gi.repository import Gio, GLib # noqa: E402
|
|
|
|
|
|
def _as_argv(cmd: Sequence[str] | str) -> list[str]:
|
|
return shlex.split(cmd) if isinstance(cmd, str) else list(cmd)
|
|
|
|
|
|
def run_text(cmd: Sequence[str] | str, cb: Callable[[bool, str, str], None]) -> None:
|
|
"""Run cmd, call cb(ok, stdout, stderr) on the main thread when done."""
|
|
try:
|
|
proc = Gio.Subprocess.new(
|
|
_as_argv(cmd),
|
|
Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_PIPE,
|
|
)
|
|
except GLib.Error as err:
|
|
GLib.idle_add(cb, False, "", str(err))
|
|
return
|
|
|
|
def _done(p: Gio.Subprocess, res: Gio.AsyncResult) -> None:
|
|
try:
|
|
_, out, errout = p.communicate_utf8_finish(res)
|
|
except GLib.Error as err:
|
|
cb(False, "", str(err))
|
|
return
|
|
cb(p.get_successful(), out or "", errout or "")
|
|
|
|
proc.communicate_utf8_async(None, None, _done)
|
|
|
|
|
|
def run_json(cmd: Sequence[str] | str, cb: Callable[[bool, object], None]) -> None:
|
|
"""Run cmd expecting JSON on stdout; call cb(ok, data)."""
|
|
|
|
def _text(ok: bool, out: str, err: str) -> None:
|
|
if not ok or not out.strip():
|
|
cb(False, err.strip() or "no output")
|
|
return
|
|
try:
|
|
cb(True, json.loads(out))
|
|
except json.JSONDecodeError as exc:
|
|
cb(False, f"bad json: {exc}")
|
|
|
|
run_text(cmd, _text)
|
|
|
|
|
|
def fire(cmd: Sequence[str] | str) -> None:
|
|
"""Run cmd and discard the result — for launch/dispatch calls no one awaits."""
|
|
run_text(cmd, lambda *_a: None)
|