Add `acli.watch` flicker-free alias-loop

Per-terminal optimized `watch`-like xonsh alias that
runs an arbitrary callable alias in a loop inside the
alt-screen buffer with flicker-free repaint. Supersedes
the inline `acli.ptree` polling .xsh snippet (removed
from `_ptree` docstr in favor of
`acli.watch acli.ptree pytest`).

Deats,
- alt-screen entry/exit (`\033[?1049h/l`) + cursor-hide
  (`\033[?25l/h`) wrapped in try/finally so Ctrl-C always
  returns to a pristine shell.
- per-frame draw uses cursor-home (`\033[H`) + per-line
  EL (`\033[K` before each `\n`) + post-draw erase-down
  (`\033[J`) → stale tail chars from a longer prior
  frame are obvi cleared; no full-screen flash.
- SIGWINCH-aware: terminal resize sets a flag, next
  frame does a full clear (`\033[H\033[2J`) instead of
  the cheap cursor-home path.
- Ctrl-C handling: install `signal.default_int_handler`
  so `KeyboardInterrupt` lands cleanly; prior handler
  restored on exit.
- Output capture: redirect the alias's stdout to
  `StringIO` per frame so we can post-process the EL
  fix. Aliases writing directly to `sys.stdout.buffer`
  / `os.write(1)` bypass capture — EL-fix won't apply
  but loop still works.
- Alias unwrap: xonsh stores callables as either a bare
  callable OR `[fn, *preset_args]`. Both forms handled;
  subprocess-style aliases rejected w/ a friendly err
  msg.
- `argparse` w/ `-n`/`--interval` (default 0.3s); rest
  of argv forwarded as alias args.
- Reg `'acli.watch': watch` in `_TCLI_ALIASES`.

Other,
- Tn `_ptree` `args: list[str]` param.
- Mod-header `Provides:` block updated w/ `acli.watch`
  entry.
- Top-level imports: `os`, `sys`, `signal`, `time`,
  `typing.Callable`.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
trionics.start_or_cancel
Gud Boi 2026-05-14 19:42:00 -04:00
parent f617c8cb73
commit bb239e847f
1 changed files with 165 additions and 7 deletions

View File

@ -25,6 +25,11 @@ Provides:
reaper + optional `/dev/shm/` reaper + optional `/dev/shm/`
+ UDS sock-file sweeps. + UDS sock-file sweeps.
alias for `scripts/tractor-reap`. alias for `scripts/tractor-reap`.
- `acli.watch [-n SEC] <alias-name> run a callable alias in
[alias-args]` an alt-screen loop with
flicker-free repaint
(cursor-home + per-line
EL + post-draw erase-down).
Loading from repo root: Loading from repo root:
xontrib load -p ./xontrib tractor_diag xontrib load -p ./xontrib tractor_diag
@ -43,7 +48,16 @@ helpers) — these aliases are just thin terminal wrappers.
Requires `psutil` for full functionality (`ptree` and the Requires `psutil` for full functionality (`ptree` and the
`hung_dump` tree-walk). Falls back to `pgrep -P` recursion if `hung_dump` tree-walk). Falls back to `pgrep -P` recursion if
missing. missing.
""" """
import os
import sys
import signal
import time
from typing import (
Callable,
)
from pathlib import Path from pathlib import Path
@ -55,10 +69,156 @@ from tractor._testing.trace import (
scan_bindspace, scan_bindspace,
) )
@aliases.unthreadable
def watch(
args: list[str],
) -> int:
'''
A per-term optimized `watch`-like alias for xonsh
that runs an arbitrary callable alias in a loop
inside the alt-screen buffer. Ctrl-C returns to a
pristine shell, SIGWINCH triggers a full redraw,
and the per-frame draw uses cursor-home + per-line
EL + post-draw erase-down so the loop is flicker-
free even when individual lines shrink or grow
between frames.
usage: acli.watch [-n SEC] <alias-name>
[alias-args]...
Examples:
acli.watch acli.ptree pytest
acli.watch -n 1.0 acli.bindspace_scan piker
acli.watch acli.hung_dump pytest
Only callable aliases (Python functions registered
in `aliases`) are supported. Subprocess-style
aliases raise an error — wrap them in a thin
callable if you need watching.
Output capture: the watched alias's stdout is
redirected into a `StringIO` per frame so we can
post-process it (insert `\033[K` before each `\n`).
Aliases that write directly to `sys.stdout.buffer`
or `os.write(1, ...)` bypass capture; for those the
EL-fix won't apply but the loop still functions.
'''
import argparse, io
from contextlib import redirect_stdout
parser = argparse.ArgumentParser(
prog='acli.watch',
description=watch.__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
'-n', '--interval',
type=float,
default=0.3,
help='poll interval in seconds (default: 0.3)',
)
parser.add_argument(
'alias',
help='name of a registered xonsh callable alias',
)
parser.add_argument(
'alias_args',
nargs=argparse.REMAINDER,
help='args forwarded to the watched alias',
)
try:
ns = parser.parse_args(args)
except SystemExit as se:
return int(se.code) if se.code is not None else 0
raw = aliases.get(ns.alias)
if raw is None:
print(
f'[acli.watch] no such alias: {ns.alias!r}'
)
return 1
# xonsh stores callable aliases as a bare callable
# OR wraps them in `[fn, *preset_args]` (depending
# on registration path / version). Unwrap both.
fn: Callable|None = None
preset_args: list = []
if callable(raw):
fn = raw
elif (
isinstance(raw, list)
and raw
and callable(raw[0])
):
fn = raw[0]
preset_args = list(raw[1:])
if fn is None:
kind: str = type(raw).__name__
print(
f'[acli.watch] alias {ns.alias!r} is not a '
f'callable alias (got {kind}); '
f'subprocess-style aliases not supported'
)
return 1
_FD: int = sys.stdout.fileno()
need_full_clear: bool = False
def _on_winch(signum, frame):
nonlocal need_full_clear
need_full_clear = True
prev_winch = signal.signal(
signal.SIGWINCH,
_on_winch,
)
prev_sigint = signal.signal(
signal.SIGINT,
signal.default_int_handler,
)
os.write(_FD, b'\033[?1049h\033[?25l')
try:
while True:
buf = io.StringIO()
with redirect_stdout(buf):
fn(preset_args + ns.alias_args)
if need_full_clear:
os.write(_FD, b'\033[H\033[2J')
need_full_clear = False
else:
os.write(_FD, b'\033[H')
# `\033[K` (EL) before each newline erases
# any stale tail chars left by a longer
# prior-frame version of the same line.
text: str = buf.getvalue()
painted: bytes = (
text.replace('\n', '\033[K\n').encode()
)
os.write(_FD, painted)
os.write(_FD, b'\033[J')
time.sleep(ns.interval)
except KeyboardInterrupt:
pass
finally:
os.write(_FD, b'\033[?25h\033[?1049l')
signal.signal(signal.SIGWINCH, prev_winch)
signal.signal(signal.SIGINT, prev_sigint)
return 0
# --- ptree ---------------------------------------------------- # --- ptree ----------------------------------------------------
def _ptree(args): def _ptree(
args: list[str],
):
''' '''
psutil-backed proc tree; per-proc classification into psutil-backed proc tree; per-proc classification into
severity-ordered buckets so leaked / defunct procs severity-ordered buckets so leaked / defunct procs
@ -69,15 +229,12 @@ def _ptree(args):
See `tractor._testing.trace.dump_proc_tree()` for the See `tractor._testing.trace.dump_proc_tree()` for the
bucket semantics + classification details. bucket semantics + classification details.
As a hot tip, you can use this `xonsh`-script snippet to poll To watch this live with flicker-free repaint
a target actor tree: (alt-screen, per-line EL, SIGWINCH-aware):
.. code-block:: xonsh .. code-block:: xonsh
while 1: acli.watch acli.ptree pytest
acli.ptree pytest
@.imp.time.sleep(.3)
print("\033c", end="")
''' '''
flag_tree: bool = False flag_tree: bool = False
@ -410,6 +567,7 @@ _TCLI_ALIASES: dict = {
'acli.bindspace_scan': _bindspace_scan, 'acli.bindspace_scan': _bindspace_scan,
'acli.dump_all': _dump_all_alias, 'acli.dump_all': _dump_all_alias,
'acli.reap': _tractor_reap, 'acli.reap': _tractor_reap,
'acli.watch': watch,
} }
for _name, _fn in _TCLI_ALIASES.items(): for _name, _fn in _TCLI_ALIASES.items():