Christian Sass
6cddabc633
Omarchy 4 runs one long-running Quickshell process and its bar accepts
command modules that print {"text", "tooltip", "class"}. That is the
cheapest native frontend Vitals can have: no QML, no websocket, no plugin
manifest, just a command the bar runs on an interval.
bar_widget.py prints that shape from the existing JSON API, with a
workload selector and a tooltip that carries the verdict line. It also
fails soft, because a status bar should never make a session look broken
when the monitor is down.
This is the point of splitting sampling from presentation: a native UI on
another distro is a new frontend, not a rewrite.
Signed-off-by: Christian Sass <csass@binaryranch.com>
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
bar_widget.py — one line of JSON for a status bar.
|
|
|
|
Omarchy's shell, Waybar, and anything else that speaks the same convention can
|
|
run this on an interval and render the result. It is the cheapest possible
|
|
frontend for Vitals: no QML, no websocket, no plugin manifest.
|
|
|
|
{"text": "97%", "tooltip": "6.9 cores in leaked browsers...", "class": "hot"}
|
|
|
|
Usage:
|
|
python3 tools/bar_widget.py # machine total
|
|
python3 tools/bar_widget.py --workload brave # one workload
|
|
python3 tools/bar_widget.py --url http://127.0.0.1:8791
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def fetch(url):
|
|
with urllib.request.urlopen(url + "/api/snapshot", timeout=3) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--url", default="http://127.0.0.1:8791")
|
|
ap.add_argument("--workload", default="", help="substring of a workload name")
|
|
args = ap.parse_args()
|
|
|
|
try:
|
|
snap = fetch(args.url)
|
|
except (urllib.error.URLError, OSError, ValueError):
|
|
# a bar widget must never be the reason a session looks broken
|
|
print(json.dumps({"text": "vitals ?", "tooltip": "Vitals is not responding", "class": "offline"}))
|
|
return 0
|
|
|
|
if args.workload:
|
|
match = [w for w in snap["workloads"] if args.workload.lower() in w["name"].lower()]
|
|
if not match:
|
|
print(json.dumps({"text": "?", "tooltip": f"no workload matching {args.workload}",
|
|
"class": "offline"}))
|
|
return 0
|
|
w = max(match, key=lambda x: x["cpu"])
|
|
text = f'{w["cpu"]:.1f}c'
|
|
tip = [f'{w["name"]}: {w["cpu"]:.2f} cores ({w["pct"]:.0f}%), {w["procs"]} procs']
|
|
if w.get("verdict"):
|
|
tip.append(w["verdict"])
|
|
for t in (w.get("tabs") or [])[:3]:
|
|
tip.append(f' {t["title"][:40]} — {t["task_pct"]:.1f}% page work')
|
|
klass = "hot" if w["pct"] > 40 else "warn" if w["pct"] > 15 else "ok"
|
|
if "leaked" in (w.get("flags") or []):
|
|
klass = "leaked"
|
|
print(json.dumps({"text": text, "tooltip": "\n".join(tip), "class": klass}))
|
|
return 0
|
|
|
|
pct = snap["total_pct"]
|
|
klass = "hot" if pct > 80 else "warn" if pct > 50 else "ok"
|
|
tip = [f'{snap["total_busy"]:.2f} of {snap["cores"]} cores ({pct}%)']
|
|
for w in snap["workloads"][:5]:
|
|
tip.append(f' {w["name"][:28]} {w["cpu"]:.2f}c')
|
|
flagged = [w for w in snap["workloads"] if w.get("flags")]
|
|
for w in flagged:
|
|
tip.append(f' ! {w["name"]}: {", ".join(w["flags"])}')
|
|
print(json.dumps({"text": f"{pct}%", "tooltip": "\n".join(tip), "class": klass}))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|