dasandClaude Opus 5.5
a2bffc7fe9
build / build (push) Successful in 38m26s
- das-shell 0.1.0: the Das two-line bash prompt by default (replaces the PS1 of Void's skel .bashrc once), dprompt (Textual prompt builder with presets and live preview) and dbanner (SSH login banner editor: canvas tools, fields with values, widgets, das-banner service for cached ones, Das OS logo and tryzub pictures). Python tests in make test. - das-server 0.3.0 depends on das-shell; the server installer mentions dprompt and dbanner; the desktop image includes it. - run-qemu.sh forwards the guest's SSH to localhost:2222. - Hardware fixes: Ventoy menu icon, boot menu titles "Das OS <version>". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
416 lines
14 KiB
Python
416 lines
14 KiB
Python
"""Values for banner fields ({hostname}, {ip}, ...) and widgets: blocks of lines
|
||
from built-in functions, scripts, commands or files.
|
||
|
||
"live" widgets run at login with a short timeout; "cached" ones are run by the
|
||
das-banner service, and the login only reads their last output from CACHE_DIR.
|
||
"""
|
||
|
||
import concurrent.futures
|
||
import datetime
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import time
|
||
|
||
from . import CACHE_DIR, DATA_DIR, colors
|
||
from .i18n import t
|
||
|
||
LIVE_TIMEOUT = 1.5
|
||
CACHED_TIMEOUT = 60
|
||
|
||
|
||
# --- field values -----------------------------------------------------------------
|
||
|
||
def _read(path):
|
||
try:
|
||
with open(path, encoding="utf-8", errors="replace") as f:
|
||
return f.read()
|
||
except OSError:
|
||
return ""
|
||
|
||
|
||
def _ips():
|
||
try:
|
||
out = subprocess.run(["ip", "-4", "-o", "addr", "show", "scope", "global"],
|
||
capture_output=True, text=True, timeout=1).stdout
|
||
ips = [line.split()[3].split("/")[0] for line in out.splitlines() if len(line.split()) > 3]
|
||
if ips:
|
||
return ips
|
||
except (OSError, subprocess.SubprocessError):
|
||
pass
|
||
try:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
|
||
s.connect(("192.0.2.1", 9)) # no packet is sent
|
||
return [s.getsockname()[0]]
|
||
except OSError:
|
||
return []
|
||
|
||
|
||
def _os_name():
|
||
for line in _read("/etc/os-release").splitlines():
|
||
if line.startswith("PRETTY_NAME="):
|
||
return line.split("=", 1)[1].strip().strip('"')
|
||
return "Linux"
|
||
|
||
|
||
def _duration(seconds):
|
||
days, rest = divmod(int(seconds), 86400)
|
||
hours, rest = divmod(rest, 3600)
|
||
minutes = rest // 60
|
||
parts = []
|
||
if days:
|
||
parts.append(t(f"{days} дн", f"{days} d"))
|
||
if hours or days:
|
||
parts.append(t(f"{hours} год", f"{hours} h"))
|
||
if not days:
|
||
parts.append(t(f"{minutes} хв", f"{minutes} min"))
|
||
return " ".join(parts)
|
||
|
||
|
||
def _uptime():
|
||
try:
|
||
return _duration(float(_read("/proc/uptime").split()[0]))
|
||
except (IndexError, ValueError):
|
||
return "?"
|
||
|
||
|
||
def _meminfo():
|
||
info = {}
|
||
for line in _read("/proc/meminfo").splitlines():
|
||
key, _, value = line.partition(":")
|
||
try:
|
||
info[key] = int(value.split()[0]) * 1024
|
||
except (IndexError, ValueError):
|
||
pass
|
||
return info
|
||
|
||
|
||
def human(n):
|
||
for unit in ("B", "K", "M", "G", "T", "P"):
|
||
if abs(n) < 1024 or unit == "P":
|
||
return f"{n:.0f}{unit}" if unit in ("B", "K") or n >= 100 else f"{n:.1f}{unit}"
|
||
n /= 1024
|
||
return str(n)
|
||
|
||
|
||
def _mem():
|
||
m = _meminfo()
|
||
total = m.get("MemTotal", 0)
|
||
used = total - m.get("MemAvailable", 0)
|
||
return f"{human(used)}/{human(total)}"
|
||
|
||
|
||
def _cpu():
|
||
for line in _read("/proc/cpuinfo").splitlines():
|
||
if line.startswith("model name"):
|
||
return " ".join(line.split(":", 1)[1].split())
|
||
return os.uname().machine
|
||
|
||
|
||
def _users():
|
||
try:
|
||
out = subprocess.run(["who"], capture_output=True, text=True, timeout=1).stdout
|
||
return str(len({line.split()[0] for line in out.splitlines() if line.strip()}))
|
||
except (OSError, subprocess.SubprocessError):
|
||
return "?"
|
||
|
||
|
||
PROVIDERS = {
|
||
"hostname": lambda: socket.gethostname().split(".")[0],
|
||
"fqdn": socket.getfqdn,
|
||
"ip": lambda: (_ips() or ["?"])[0],
|
||
"ips": lambda: " ".join(_ips()),
|
||
"os": _os_name,
|
||
"kernel": lambda: os.uname().release,
|
||
"arch": lambda: os.uname().machine,
|
||
"uptime": _uptime,
|
||
"load": lambda: " ".join(_read("/proc/loadavg").split()[:3]),
|
||
"mem": _mem,
|
||
"cpu": _cpu,
|
||
"cores": lambda: str(os.cpu_count() or 1),
|
||
"date": lambda: datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"users": _users,
|
||
"user": lambda: os.environ.get("USER") or os.environ.get("LOGNAME") or "?",
|
||
}
|
||
|
||
VAR_HELP = {
|
||
"hostname": ("ім'я комп'ютера", "host name"),
|
||
"fqdn": ("повне ім'я в мережі", "fully qualified name"),
|
||
"ip": ("перша IPv4-адреса", "first IPv4 address"),
|
||
"ips": ("усі IPv4-адреси", "all IPv4 addresses"),
|
||
"os": ("назва системи", "system name"),
|
||
"kernel": ("версія ядра", "kernel version"),
|
||
"arch": ("архітектура", "architecture"),
|
||
"uptime": ("час роботи", "uptime"),
|
||
"load": ("навантаження", "load average"),
|
||
"mem": ("пам'ять зайнято/всього", "memory used/total"),
|
||
"cpu": ("процесор", "processor"),
|
||
"cores": ("кількість ядер", "number of cores"),
|
||
"date": ("дата й час", "date and time"),
|
||
"users": ("користувачів у системі", "users logged in"),
|
||
"user": ("хто входить", "who logs in"),
|
||
}
|
||
|
||
|
||
class Values(dict):
|
||
"""Field values, computed only when a field asks for them."""
|
||
|
||
def __init__(self, user_vars=None):
|
||
super().__init__()
|
||
self.user_vars = dict(user_vars or {})
|
||
|
||
def get(self, key, default=None):
|
||
if key in self.user_vars:
|
||
return self.user_vars[key]
|
||
if key not in self and key in PROVIDERS:
|
||
try:
|
||
self[key] = PROVIDERS[key]()
|
||
except Exception: # a broken value must not break the login
|
||
self[key] = "?"
|
||
return super().get(key, default)
|
||
|
||
|
||
# --- built-in widgets -------------------------------------------------------------
|
||
|
||
def _bar(percent, width=10):
|
||
full = round(percent / 100 * width)
|
||
color = "green" if percent < 70 else ("yellow" if percent < 90 else "red")
|
||
return (colors.sgr(color, None, True) + "█" * full + colors.sgr("bright_black") + "░" * (width - full)
|
||
+ colors.RESET)
|
||
|
||
|
||
PSEUDO_FS = {"proc", "sysfs", "devtmpfs", "devpts", "tmpfs", "cgroup", "cgroup2", "securityfs",
|
||
"pstore", "efivarfs", "bpf", "debugfs", "tracefs", "configfs", "fusectl", "mqueue",
|
||
"hugetlbfs", "autofs", "binfmt_misc", "overlay", "squashfs", "nsfs", "ramfs", "rpc_pipefs",
|
||
"fuse.portal", "fuse.gvfsd-fuse", "iso9660", "zram"}
|
||
|
||
|
||
def _mounts():
|
||
seen, out = set(), []
|
||
for line in _read("/proc/mounts").splitlines():
|
||
parts = line.split()
|
||
if len(parts) < 3 or parts[2] in PSEUDO_FS or not parts[0].startswith("/"):
|
||
continue
|
||
mount = parts[1].replace("\\040", " ")
|
||
if parts[0] in seen or mount.startswith(("/snap", "/var/lib/docker", "/.snapshots", "/run")):
|
||
continue
|
||
seen.add(parts[0])
|
||
out.append(mount)
|
||
return out
|
||
|
||
|
||
def w_disks(args, values):
|
||
lines = [colors.sgr("white", None, True) + t("Диски:", "Disks:") + colors.RESET]
|
||
rows = []
|
||
for mount in args.get("mounts") or _mounts():
|
||
try:
|
||
st = os.statvfs(mount)
|
||
except OSError:
|
||
continue
|
||
total = st.f_blocks * st.f_frsize
|
||
if not total:
|
||
continue
|
||
used = total - st.f_bfree * st.f_frsize
|
||
rows.append((mount, total, used))
|
||
width = max([len(r[0]) for r in rows] + [1])
|
||
for mount, total, used in rows:
|
||
pct = round(used * 100 / total)
|
||
lines.append(f" {mount.ljust(width)} {human(total):>6} {_bar(pct)} {pct:3d}%")
|
||
return lines
|
||
|
||
|
||
def w_memory(args, values):
|
||
m = _meminfo()
|
||
lines = []
|
||
for label, total, free in ((t("Пам'ять", "Memory"), m.get("MemTotal", 0), m.get("MemAvailable", 0)),
|
||
("Swap", m.get("SwapTotal", 0), m.get("SwapFree", 0))):
|
||
if not total:
|
||
continue
|
||
pct = round((total - free) * 100 / total)
|
||
lines.append(f"{colors.sgr('white', None, True)}{label + ':':<9}{colors.RESET} "
|
||
f"{human(total - free):>6}/{human(total):<6} {_bar(pct)} {pct:3d}%")
|
||
return lines
|
||
|
||
|
||
def w_uptime(args, values):
|
||
return [f"{colors.sgr('white', None, True)}{t('Працює:', 'Uptime:')}{colors.RESET} "
|
||
f"{colors.sgr('red', None, True)}{values.get('uptime')}{colors.RESET}"]
|
||
|
||
|
||
def w_load(args, values):
|
||
return [f"{colors.sgr('white', None, True)}{t('Навантаження:', 'Load:')}{colors.RESET} "
|
||
f"{values.get('load')} ({t('ядер', 'cores')}: {values.get('cores')})"]
|
||
|
||
|
||
def w_network(args, values):
|
||
lines = [colors.sgr("white", None, True) + t("Мережа:", "Network:") + colors.RESET]
|
||
try:
|
||
out = subprocess.run(["ip", "-o", "addr", "show", "scope", "global"],
|
||
capture_output=True, text=True, timeout=1).stdout
|
||
except (OSError, subprocess.SubprocessError):
|
||
return []
|
||
for line in out.splitlines():
|
||
p = line.split()
|
||
if len(p) > 3:
|
||
lines.append(f" {p[1]:<10} {colors.sgr('cyan', None, True)}{p[3]}{colors.RESET}")
|
||
return lines if len(lines) > 1 else []
|
||
|
||
|
||
BUILTINS = {
|
||
"uptime": (w_uptime, ("Час роботи", "Uptime")),
|
||
"load": (w_load, ("Навантаження", "Load average")),
|
||
"memory": (w_memory, ("Пам'ять і swap", "Memory and swap")),
|
||
"disks": (w_disks, ("Диски", "Disks")),
|
||
"network": (w_network, ("Мережа", "Network")),
|
||
}
|
||
|
||
|
||
def widget_dir():
|
||
return os.path.join(DATA_DIR, "banner", "widgets")
|
||
|
||
|
||
def script_meta(path):
|
||
"""# dbanner-<key>: value lines of a widget script."""
|
||
meta = {}
|
||
for line in _read(path).splitlines()[:20]:
|
||
if line.startswith("# dbanner-"):
|
||
key, _, value = line[10:].partition(":")
|
||
meta[key.strip()] = value.strip()
|
||
return meta
|
||
|
||
|
||
def available():
|
||
"""(name, title, mode) of every built-in widget and widget script."""
|
||
out = [(name, t(*names), "live") for name, (_, names) in BUILTINS.items()]
|
||
try:
|
||
scripts = sorted(f for f in os.listdir(widget_dir()) if f.endswith(".sh") and f != "lib.sh")
|
||
except OSError:
|
||
scripts = []
|
||
for f in scripts:
|
||
meta = script_meta(os.path.join(widget_dir(), f))
|
||
out.append((f[:-3], t(meta.get("title-uk", f[:-3]), meta.get("title-en", f[:-3])),
|
||
meta.get("mode", "live")))
|
||
return out
|
||
|
||
|
||
def describe(w):
|
||
"""A short line about a widget for lists."""
|
||
kind = w.get("type", "builtin")
|
||
if kind == "builtin":
|
||
names = {n: title_ for n, title_, _ in available()}
|
||
return names.get(w.get("name"), w.get("name", "?"))
|
||
if kind == "command":
|
||
return "$ " + w.get("cmd", "")
|
||
return t("файл ", "file ") + w.get("path", "")
|
||
|
||
|
||
def mode_of(w):
|
||
if w.get("mode"):
|
||
return w["mode"]
|
||
if w.get("type", "builtin") == "builtin" and w.get("name") not in BUILTINS:
|
||
return script_meta(os.path.join(widget_dir(), w.get("name", "") + ".sh")).get("mode", "live")
|
||
return "live"
|
||
|
||
|
||
def interval_of(w):
|
||
if w.get("interval"):
|
||
return int(w["interval"])
|
||
if w.get("type", "builtin") == "builtin":
|
||
meta = script_meta(os.path.join(widget_dir(), w.get("name", "") + ".sh"))
|
||
if meta.get("interval", "").isdigit():
|
||
return int(meta["interval"])
|
||
return 600
|
||
|
||
|
||
def cache_path(w):
|
||
key = hashlib.sha1(json.dumps(w, sort_keys=True).encode()).hexdigest()[:16]
|
||
return os.path.join(CACHE_DIR, key + ".txt")
|
||
|
||
|
||
def _env(design_vars):
|
||
env = dict(os.environ)
|
||
for key, value in (design_vars or {}).items():
|
||
env["DBANNER_" + key.upper()] = str(value)
|
||
env["DBANNER_LANG"] = "uk" if t("uk", "en") == "uk" else "en"
|
||
return env
|
||
|
||
|
||
def run(w, values, design_vars=None, timeout=None):
|
||
"""Runs a widget now and returns its lines (empty when it fails)."""
|
||
kind = w.get("type", "builtin")
|
||
try:
|
||
if kind == "file":
|
||
text = _read(os.path.expanduser(w.get("path", "")))
|
||
lines = text.rstrip("\n").split("\n") if text.strip() else []
|
||
elif kind == "builtin" and w.get("name") in BUILTINS:
|
||
lines = BUILTINS[w["name"]][0](w.get("args") or {}, values)
|
||
else:
|
||
if kind == "builtin":
|
||
cmd = [os.path.join(widget_dir(), w.get("name", "") + ".sh")]
|
||
else:
|
||
cmd = ["sh", "-c", w.get("cmd", "")]
|
||
limit = timeout or float(w.get("timeout") or (CACHED_TIMEOUT if mode_of(w) == "cached" else LIVE_TIMEOUT))
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, errors="replace",
|
||
timeout=limit, env=_env(design_vars), stdin=subprocess.DEVNULL)
|
||
out = proc.stdout.rstrip("\n")
|
||
lines = out.split("\n") if out.strip() else []
|
||
except Exception: # a broken widget must not break the login
|
||
return []
|
||
if lines and w.get("title"):
|
||
lines = [colors.sgr("white", None, True) + w["title"] + colors.RESET] + lines
|
||
return lines
|
||
|
||
|
||
def read_cache(w):
|
||
text = _read(cache_path(w))
|
||
return text.rstrip("\n").split("\n") if text.strip() else []
|
||
|
||
|
||
def blocks(design, values, use_cache=True):
|
||
"""Lines of every widget of a design, in order; live ones run in parallel."""
|
||
widgets = design.get("widgets", [])
|
||
result = [[] for _ in widgets]
|
||
jobs = {}
|
||
# No "with": leaving it would wait for a widget that hangs
|
||
pool = concurrent.futures.ThreadPoolExecutor(max_workers=8)
|
||
for i, w in enumerate(widgets):
|
||
if w.get("enabled") is False:
|
||
continue
|
||
if use_cache and mode_of(w) == "cached":
|
||
result[i] = read_cache(w)
|
||
else:
|
||
jobs[pool.submit(run, w, values, design.get("vars"))] = i
|
||
try:
|
||
for fut in concurrent.futures.as_completed(jobs, timeout=LIVE_TIMEOUT + 1 if use_cache else None):
|
||
result[jobs[fut]] = fut.result()
|
||
except concurrent.futures.TimeoutError:
|
||
pass
|
||
pool.shutdown(wait=False, cancel_futures=True)
|
||
return result
|
||
|
||
|
||
def refresh(design, force=False):
|
||
"""Runs the cached widgets whose output is older than their interval."""
|
||
values = None
|
||
os.makedirs(CACHE_DIR, mode=0o755, exist_ok=True)
|
||
for w in design.get("widgets", []):
|
||
if w.get("enabled") is False or mode_of(w) != "cached":
|
||
continue
|
||
path = cache_path(w)
|
||
try:
|
||
age = time.time() - os.stat(path).st_mtime
|
||
except OSError:
|
||
age = None
|
||
if not force and age is not None and age < interval_of(w):
|
||
continue
|
||
if values is None:
|
||
values = Values(design.get("vars"))
|
||
lines = run(w, values, design.get("vars"))
|
||
tmp = path + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines) + ("\n" if lines else ""))
|
||
os.chmod(tmp, 0o644)
|
||
os.replace(tmp, path)
|