Mathieu Benoit
320b88b83b
- F2 : nouvelle vue SYSTÈME (état/uptime + charge, CPU %, mémoire, disque, réseau ↓/↑, batterie, température). Les I/O sont déportées en thread (asyncio.to_thread) ; rafraîchissement 2 s (deltas CPU/réseau). Si la température n'est pas lisible (ni /sys/class/thermal ni lm-sensors), on propose d'installer lm-sensors avec la touche « i » (commande selon l'OS — apt/dnf/pacman — affichée puis validée). read_temperature essaie /sys/class/thermal d'abord (sans dépendance). - Vue Kanban GRILLE : clic sur le TITRE d'une case -> elle s'AGRANDIT (row-span, prend toute la hauteur d'une colonne) ; re-clic -> taille normale (classe kbig). Validé headless : vue système peuplée (mém/disque/temp), F2/F3 basculent, clic titre en grille -> kbig on/off. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
904 lines
31 KiB
Python
904 lines
31 KiB
Python
#!/usr/bin/env python3
|
||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||
"""Télémétrie de navigation du CLI TODO.
|
||
|
||
Enregistre les fils d'Ariane visités (« A › B › C » -> compteur) et affiche un
|
||
DIAGRAMME arborescent des fonctionnalités dans un TUI Textual, trié par usage.
|
||
|
||
- record(path) : appelé par Todo._menu_header à chaque affichage de menu ; on
|
||
dédupe les ré-affichages consécutifs pour ne compter que les TRANSITIONS.
|
||
- run_tui() : ouvre l'arbre de navigation (compteurs par menu).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import asyncio
|
||
import glob
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from script.todo.todo_i18n import t
|
||
except Exception: # pragma: no cover - repli si i18n indisponible
|
||
|
||
def t(key: str) -> str:
|
||
return key
|
||
|
||
|
||
# Dernier chemin enregistré : évite de recompter chaque ré-affichage du même
|
||
# menu (une commande qui revient au menu ne compte pas comme une navigation).
|
||
_LAST = [None]
|
||
|
||
|
||
def _path() -> Path:
|
||
base = Path(os.path.expanduser("~/.erplibre"))
|
||
base.mkdir(parents=True, exist_ok=True)
|
||
return base / "todo_telemetry.json"
|
||
|
||
|
||
def load() -> dict:
|
||
try:
|
||
return json.loads(_path().read_text())
|
||
except (OSError, ValueError):
|
||
return {"paths": {}}
|
||
|
||
|
||
def _save(data: dict) -> None:
|
||
try:
|
||
_path().write_text(json.dumps(data, ensure_ascii=False, indent=2))
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def record(path: str) -> None:
|
||
"""Incrémente le compteur du menu `path`. Best-effort : ne lève jamais
|
||
(la télémétrie ne doit jamais casser la navigation)."""
|
||
if not path or path == _LAST[0]:
|
||
return
|
||
_LAST[0] = path
|
||
try:
|
||
data = load()
|
||
paths = data.setdefault("paths", {})
|
||
paths[path] = paths.get(path, 0) + 1
|
||
data["updated"] = int(time.time())
|
||
_save(data)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def reset() -> None:
|
||
_save({"paths": {}})
|
||
|
||
|
||
def _nested(paths: dict) -> dict:
|
||
"""{« A › B › C »: count} -> arbre {nom: {'count', 'children'}}. Chaque
|
||
menu enregistre son propre chemin, donc chaque nœud a son compteur."""
|
||
root: dict = {}
|
||
for path, count in paths.items():
|
||
cur = root
|
||
parts = path.split(" › ")
|
||
for i, name in enumerate(parts):
|
||
node = cur.setdefault(name, {"count": 0, "children": {}})
|
||
if i == len(parts) - 1:
|
||
node["count"] += count
|
||
cur = node["children"]
|
||
return root
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Métadonnées de navigation DÉRIVÉES DU CODE (structure réelle des menus)
|
||
# --------------------------------------------------------------------------- #
|
||
def _str_of(node) -> str | None:
|
||
"""Chaîne d'un nœud AST : littéral, t("…") ou f-string (parties Constant)."""
|
||
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||
return node.value
|
||
if (
|
||
isinstance(node, ast.Call)
|
||
and isinstance(node.func, ast.Name)
|
||
and node.func.id == "t"
|
||
and node.args
|
||
and isinstance(node.args[0], ast.Constant)
|
||
):
|
||
return node.args[0].value
|
||
if isinstance(node, ast.JoinedStr):
|
||
return "".join(
|
||
p.value
|
||
for p in node.values
|
||
if isinstance(p, ast.Constant) and isinstance(p.value, str)
|
||
)
|
||
return None
|
||
|
||
|
||
def _choice_labels(func) -> list:
|
||
"""Libellés NUMÉROTÉS d'un menu « choices = [...] » (dans l'ordre, sections
|
||
exclues car non numérotées par fill_help_info)."""
|
||
for node in ast.walk(func):
|
||
if (
|
||
isinstance(node, ast.Assign)
|
||
and any(
|
||
isinstance(tg, ast.Name) and tg.id == "choices"
|
||
for tg in node.targets
|
||
)
|
||
and isinstance(node.value, ast.List)
|
||
):
|
||
labels = []
|
||
for el in node.value.elts:
|
||
if not isinstance(el, ast.Dict):
|
||
continue
|
||
for k, v in zip(el.keys, el.values):
|
||
if isinstance(k, ast.Constant) and k.value == "prompt_description":
|
||
s = _str_of(v)
|
||
if s:
|
||
labels.append(s)
|
||
return labels
|
||
return []
|
||
|
||
|
||
def _dispatch(func) -> dict:
|
||
"""{numéro: (nom_de_méthode, kwargs)} depuis « status == "N": self.X(...) ».
|
||
Les kwargs LITTÉRAUX (ex. dry_run=True) sont capturés pour pouvoir rejouer
|
||
la commande à l'identique depuis la télémétrie."""
|
||
out = {}
|
||
for node in ast.walk(func):
|
||
if not isinstance(node, ast.If):
|
||
continue
|
||
test = node.test
|
||
if (
|
||
isinstance(test, ast.Compare)
|
||
and isinstance(test.left, ast.Name)
|
||
and test.left.id == "status"
|
||
and len(test.ops) == 1
|
||
and isinstance(test.ops[0], ast.Eq)
|
||
and isinstance(test.comparators[0], ast.Constant)
|
||
):
|
||
try:
|
||
num = int(test.comparators[0].value)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
for stmt in node.body:
|
||
found = None
|
||
for n in ast.walk(stmt):
|
||
if (
|
||
isinstance(n, ast.Call)
|
||
and isinstance(n.func, ast.Attribute)
|
||
and isinstance(n.func.value, ast.Name)
|
||
and n.func.value.id == "self"
|
||
):
|
||
kwargs = {
|
||
kw.arg: kw.value.value
|
||
for kw in n.keywords
|
||
if kw.arg
|
||
and isinstance(kw.value, ast.Constant)
|
||
}
|
||
found = (n.func.attr, kwargs)
|
||
break
|
||
if found:
|
||
out[num] = found
|
||
break
|
||
return out
|
||
|
||
|
||
def _menu_labels(cls) -> dict:
|
||
"""{méthode: label} depuis l'attribut de classe _MENU_LABELS."""
|
||
for node in ast.walk(cls):
|
||
if (
|
||
isinstance(node, ast.Assign)
|
||
and any(
|
||
isinstance(tg, ast.Name) and tg.id == "_MENU_LABELS"
|
||
for tg in node.targets
|
||
)
|
||
and isinstance(node.value, ast.Dict)
|
||
):
|
||
return {
|
||
k.value: v.value
|
||
for k, v in zip(node.value.keys, node.value.values)
|
||
if isinstance(k, ast.Constant) and isinstance(v, ast.Constant)
|
||
}
|
||
return {}
|
||
|
||
|
||
def build_code_tree(todo_path=None) -> dict | None:
|
||
"""Construit l'arbre des menus/commandes EN LISANT le code de todo.py
|
||
(AST). Chaque menu (méthode de _MENU_LABELS) devient un nœud ; ses branches
|
||
« status == N » deviennent des sous-menus (si la cible est un menu) ou des
|
||
commandes (feuilles). None si l'analyse échoue."""
|
||
p = Path(todo_path) if todo_path else (Path(__file__).parent / "todo.py")
|
||
try:
|
||
mod = ast.parse(p.read_text(encoding="utf-8"))
|
||
except (OSError, SyntaxError):
|
||
return None
|
||
cls = next(
|
||
(n for n in ast.walk(mod) if isinstance(n, ast.ClassDef)), None
|
||
)
|
||
if cls is None:
|
||
return None
|
||
methods = {
|
||
n.name: n for n in cls.body if isinstance(n, ast.FunctionDef)
|
||
}
|
||
labels = _menu_labels(cls)
|
||
if "run" not in methods or not labels:
|
||
return None
|
||
seen = set()
|
||
|
||
def build(method):
|
||
node = {
|
||
"label": labels.get(method, method),
|
||
"is_menu": True,
|
||
"children": [],
|
||
}
|
||
if method in seen or method not in methods:
|
||
return node
|
||
seen.add(method)
|
||
func = methods[method]
|
||
disp = _dispatch(func)
|
||
clabels = _choice_labels(func)
|
||
for num in sorted(disp):
|
||
target, kwargs = disp[num]
|
||
if target in labels: # sous-menu
|
||
node["children"].append(build(target))
|
||
else: # commande (feuille) exécutable
|
||
lab = (
|
||
clabels[num - 1]
|
||
if 0 <= num - 1 < len(clabels)
|
||
else target.lstrip("_").replace("_", " ")
|
||
)
|
||
node["children"].append(
|
||
{
|
||
"label": lab,
|
||
"is_menu": False,
|
||
"children": [],
|
||
"method": target,
|
||
"kwargs": kwargs,
|
||
}
|
||
)
|
||
seen.discard(method)
|
||
return node
|
||
|
||
return build("run")
|
||
|
||
|
||
def _command_columns(tree, paths):
|
||
"""Colonnes du Kanban : (label, chemin, count, [commandes]) pour CHAQUE
|
||
menu qui contient des commandes directes (parcours profondeur d'abord)."""
|
||
cols = []
|
||
|
||
def walk(node, path):
|
||
cmds = [c for c in node["children"] if not c["is_menu"]]
|
||
if cmds:
|
||
cols.append((node["label"], path, paths.get(path, 0), cmds))
|
||
for c in node["children"]:
|
||
if c["is_menu"]:
|
||
walk(c, f"{path} › {c['label']}")
|
||
|
||
if tree:
|
||
walk(tree, "TODO")
|
||
return cols
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Télémétrie SYSTÈME (vue F2)
|
||
# --------------------------------------------------------------------------- #
|
||
def _os_id() -> str:
|
||
try:
|
||
for line in open("/etc/os-release", encoding="utf-8"):
|
||
if line.startswith("ID="):
|
||
return line.split("=", 1)[1].strip().strip('"').lower()
|
||
except OSError:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def sensors_install_command():
|
||
"""Commande d'installation de lm-sensors selon l'OS (nos 4 systèmes)."""
|
||
apt = ["sudo", "apt-get", "install", "-y", "lm-sensors"]
|
||
dnf = ["sudo", "dnf", "install", "-y", "lm_sensors"]
|
||
pac = ["sudo", "pacman", "-S", "--needed", "--noconfirm", "lm_sensors"]
|
||
cmd = {
|
||
"ubuntu": apt,
|
||
"debian": apt,
|
||
"linuxmint": apt,
|
||
"fedora": dnf,
|
||
"arch": pac,
|
||
}.get(_os_id())
|
||
if cmd:
|
||
return cmd
|
||
if shutil.which("apt-get"):
|
||
return apt
|
||
if shutil.which("dnf"):
|
||
return dnf
|
||
if shutil.which("pacman"):
|
||
return pac
|
||
return None
|
||
|
||
|
||
def _first_int(path):
|
||
try:
|
||
return int(open(path).read().strip())
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
|
||
def _cpu_sample():
|
||
try:
|
||
with open("/proc/stat") as f:
|
||
vals = [int(x) for x in f.readline().split()[1:]]
|
||
idle = vals[3] + (vals[4] if len(vals) > 4 else 0)
|
||
return idle, sum(vals)
|
||
except (OSError, ValueError, IndexError):
|
||
return None
|
||
|
||
|
||
def _net_sample():
|
||
rx = tx = 0
|
||
try:
|
||
with open("/proc/net/dev") as f:
|
||
for line in f.readlines()[2:]:
|
||
iface, _, rest = line.partition(":")
|
||
if iface.strip() in ("lo", ""):
|
||
continue
|
||
cols = rest.split()
|
||
if len(cols) >= 9:
|
||
rx += int(cols[0])
|
||
tx += int(cols[8])
|
||
except (OSError, ValueError):
|
||
return None
|
||
return rx, tx
|
||
|
||
|
||
def _mem():
|
||
info = {}
|
||
try:
|
||
for line in open("/proc/meminfo"):
|
||
k, _, v = line.partition(":")
|
||
info[k] = int(v.split()[0]) * 1024
|
||
except (OSError, ValueError):
|
||
return None
|
||
total = info.get("MemTotal", 0)
|
||
avail = info.get("MemAvailable", info.get("MemFree", 0))
|
||
return total, total - avail
|
||
|
||
|
||
def _battery():
|
||
for base in glob.glob("/sys/class/power_supply/BAT*"):
|
||
cap = _first_int(os.path.join(base, "capacity"))
|
||
try:
|
||
status = open(os.path.join(base, "status")).read().strip()
|
||
except OSError:
|
||
status = ""
|
||
if cap is not None:
|
||
return cap, status
|
||
return None
|
||
|
||
|
||
def read_temperature():
|
||
"""(source, [°C]) ou None. /sys/class/thermal d'abord (sans dépendance),
|
||
puis lm-sensors si présent."""
|
||
temps = []
|
||
for zt in glob.glob("/sys/class/thermal/thermal_zone*/temp"):
|
||
v = _first_int(zt)
|
||
if v and v > 0:
|
||
temps.append(round(v / 1000.0, 1))
|
||
if temps:
|
||
return "sysfs", temps
|
||
if shutil.which("sensors"):
|
||
try:
|
||
out = subprocess.run(
|
||
["sensors", "-u"], capture_output=True, text=True, timeout=5
|
||
).stdout
|
||
for line in out.splitlines():
|
||
line = line.strip()
|
||
if "_input:" in line:
|
||
try:
|
||
temps.append(round(float(line.split(":")[1]), 1))
|
||
except (ValueError, IndexError):
|
||
pass
|
||
temps = [x for x in temps if x > 0]
|
||
if temps:
|
||
return "sensors", temps
|
||
except (OSError, subprocess.SubprocessError):
|
||
pass
|
||
return None
|
||
|
||
|
||
def system_snapshot(prev, full=True):
|
||
"""Instantané des métriques système. `prev` = (cpu, net, t) pour calculer
|
||
les taux CPU/réseau par delta. `full=False` saute la température (évite le
|
||
subprocess sensors quand la vue système n'est pas affichée). Renvoie
|
||
(métriques, nouveau_prev)."""
|
||
now = time.time()
|
||
cpu, net = _cpu_sample(), _net_sample()
|
||
pcpu, pnet, pt = prev or (None, None, None)
|
||
m = {}
|
||
if cpu and pcpu and cpu[1] > pcpu[1]:
|
||
m["cpu"] = max(
|
||
0, min(100, round(100 * (1 - (cpu[0] - pcpu[0]) / (cpu[1] - pcpu[1]))))
|
||
)
|
||
else:
|
||
m["cpu"] = None
|
||
if net and pnet and pt and now > pt:
|
||
dt = now - pt
|
||
m["net"] = ((net[0] - pnet[0]) / dt, (net[1] - pnet[1]) / dt)
|
||
else:
|
||
m["net"] = None
|
||
m["mem"] = _mem()
|
||
try:
|
||
du = shutil.disk_usage("/")
|
||
m["disk"] = (du.total, du.used, du.free)
|
||
except OSError:
|
||
m["disk"] = None
|
||
m["battery"] = _battery()
|
||
m["temp"] = read_temperature() if full else None
|
||
try:
|
||
m["uptime"] = float(open("/proc/uptime").read().split()[0])
|
||
except (OSError, ValueError):
|
||
m["uptime"] = None
|
||
try:
|
||
m["load"] = os.getloadavg()
|
||
except OSError:
|
||
m["load"] = None
|
||
m["ncpu"] = os.cpu_count() or 1
|
||
return m, (cpu, net, now)
|
||
|
||
|
||
def _fmt_size(nbytes):
|
||
if nbytes is None:
|
||
return "-"
|
||
for unit, div in (("T", 1 << 40), ("G", 1 << 30), ("M", 1 << 20)):
|
||
if nbytes >= div:
|
||
return f"{nbytes / div:.1f}{unit}"
|
||
return f"{nbytes // 1024}K"
|
||
|
||
|
||
def _fmt_rate(bps):
|
||
if bps is None:
|
||
return "?"
|
||
for unit, div in (("Go/s", 1 << 30), ("Mo/s", 1 << 20), ("Ko/s", 1 << 10)):
|
||
if bps >= div:
|
||
return f"{bps / div:.1f} {unit}"
|
||
return f"{int(bps)} o/s"
|
||
|
||
|
||
def _fmt_uptime(secs):
|
||
if not secs:
|
||
return "?"
|
||
secs = int(secs)
|
||
d, r = divmod(secs, 86400)
|
||
h, r = divmod(r, 3600)
|
||
mnt = r // 60
|
||
if d:
|
||
return f"{d}j {h}h"
|
||
if h:
|
||
return f"{h}h{mnt:02d}"
|
||
return f"{mnt}min"
|
||
|
||
|
||
# Modes d'affichage du Kanban (F4 les fait défiler).
|
||
KANBAN_MODES = ("columns", "swimlanes", "grid")
|
||
|
||
|
||
def run_tui(run_app: bool = True, state: dict | None = None):
|
||
"""TUI de télémétrie : vue Arbre (issue du code) et vue Kanban (F3), la
|
||
disposition du Kanban défilant par F4 (colonnes / swimlanes / grille).
|
||
Sélectionner une COMMANDE = l'exécuter. `state` restaure la vue + le
|
||
curseur au retour. Renvoie (action|None, state) ; run_app=False -> l'app."""
|
||
from textual.app import App, ComposeResult
|
||
from textual.containers import (
|
||
Container,
|
||
Grid,
|
||
HorizontalScroll,
|
||
Vertical,
|
||
VerticalScroll,
|
||
)
|
||
from textual.widgets import (
|
||
Footer,
|
||
Header,
|
||
Label,
|
||
ListItem,
|
||
ListView,
|
||
Static,
|
||
Tree,
|
||
)
|
||
|
||
paths = load().get("paths", {})
|
||
code_tree = build_code_tree()
|
||
columns = _command_columns(code_tree, paths) # (label, path, cnt, cmds)
|
||
state = state or {}
|
||
|
||
class CmdItem(ListItem):
|
||
"""Carte : commande à exécuter + son chemin (pour restaurer le curseur)."""
|
||
|
||
def __init__(self, label, method, kwargs, path):
|
||
super().__init__(Label(label))
|
||
self.cmd_method = method
|
||
self.cmd_kwargs = kwargs or {}
|
||
self.cmd_path = path
|
||
|
||
class Telemetry(App):
|
||
CSS = """
|
||
#summary { height: 1; color: $text-muted; }
|
||
Tree { border: solid $accent; }
|
||
#kanban { display: none; height: 1fr; }
|
||
#system { display: none; height: 1fr; padding: 1 2; }
|
||
.kcol { width: 40; border: solid $accent; margin: 0 1 0 0; }
|
||
.ktitle { height: 1; color: $accent; }
|
||
.lane { height: auto; }
|
||
.lanetitle { height: 1; color: $secondary; }
|
||
.kgrid { grid-size: 3; grid-gutter: 1; }
|
||
.kbig { row-span: 3; }
|
||
"""
|
||
BINDINGS = [
|
||
("q", "quit", "Quitter"),
|
||
("f2", "system", "Système"),
|
||
("f3", "toggle_view", "Arbre/Kanban"),
|
||
("f4", "kanban_layout", "Disposition Kanban"),
|
||
("e", "expand_all", "Tout déplier"),
|
||
("i", "install_sensors", "Installer capteurs"),
|
||
("r", "reset", "Réinitialiser"),
|
||
]
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._action = None
|
||
self._mode = state.get("mode", "tree")
|
||
self._kanban_mode = state.get("kanban_mode", "columns")
|
||
self._sys_prev = None
|
||
|
||
# -- helpers de construction de widgets ------------------------------ #
|
||
def _col_widget(self, label, cnt, cmds):
|
||
items = [
|
||
CmdItem(
|
||
f"· {c['label']}",
|
||
c.get("method"),
|
||
c.get("kwargs"),
|
||
c.get("path"),
|
||
)
|
||
for c in cmds
|
||
]
|
||
return VerticalScroll(
|
||
Static(f"{label} ({cnt})", classes="ktitle"),
|
||
ListView(*items),
|
||
classes="kcol",
|
||
)
|
||
|
||
def _kanban_layout_widget(self):
|
||
cols = [
|
||
self._col_widget(label, cnt, cmds)
|
||
for (label, _p, cnt, cmds) in columns
|
||
]
|
||
if not cols:
|
||
return Static(t("No command found."))
|
||
if self._kanban_mode == "grid":
|
||
return Grid(*cols, classes="kgrid")
|
||
if self._kanban_mode == "swimlanes":
|
||
# Une rangée (swimlane) par menu de NIVEAU 1 (Execute, …).
|
||
groups, order = {}, []
|
||
for (label, path, cnt, cmds) in columns:
|
||
parts = path.split(" › ")
|
||
g = parts[1] if len(parts) > 1 else "TODO"
|
||
if g not in groups:
|
||
groups[g] = []
|
||
order.append(g)
|
||
groups[g].append((label, cnt, cmds))
|
||
lanes = []
|
||
for g in order:
|
||
lane_cols = [
|
||
self._col_widget(lb, cn, cm) for (lb, cn, cm) in groups[g]
|
||
]
|
||
lanes.append(
|
||
Vertical(
|
||
Static(f"━━ {g} ━━", classes="lanetitle"),
|
||
HorizontalScroll(*lane_cols),
|
||
classes="lane",
|
||
)
|
||
)
|
||
return VerticalScroll(*lanes)
|
||
return HorizontalScroll(*cols) # columns
|
||
|
||
def compose(self) -> ComposeResult:
|
||
yield Header(show_clock=True)
|
||
yield Static("", id="summary")
|
||
yield Tree("📍 TODO", id="nav")
|
||
yield Container(id="kanban")
|
||
yield Static("", id="system")
|
||
yield Footer()
|
||
|
||
def on_mount(self):
|
||
self.title = t("TODO navigation telemetry")
|
||
self._populate_tree()
|
||
self._update_summary()
|
||
# Télémétrie système rafraîchie toutes les 2 s (deltas CPU/réseau).
|
||
self.set_interval(2.0, self._tick_system)
|
||
# Restaure la vue / la disposition / le curseur au retour.
|
||
self.run_worker(self._restore())
|
||
|
||
def _apply_visibility(self):
|
||
self.query_one("#nav").display = self._mode == "tree"
|
||
self.query_one("#kanban").display = self._mode == "kanban"
|
||
self.query_one("#system").display = self._mode == "system"
|
||
|
||
async def _restore(self):
|
||
if self._mode == "kanban":
|
||
await self._enter_kanban()
|
||
elif self._mode == "tree":
|
||
self._focus_tree_path(state.get("path"))
|
||
self._apply_visibility()
|
||
self._update_summary()
|
||
|
||
# -- vue Kanban ------------------------------------------------------ #
|
||
async def _enter_kanban(self):
|
||
box = self.query_one("#kanban", Container)
|
||
await box.remove_children()
|
||
await box.mount(self._kanban_layout_widget())
|
||
self._apply_visibility()
|
||
self._focus_kanban_path(state.get("path"))
|
||
|
||
def _focus_kanban_path(self, path):
|
||
if not path:
|
||
return
|
||
for lv in self.query(ListView):
|
||
for i, item in enumerate(lv.children):
|
||
if getattr(item, "cmd_path", None) == path:
|
||
lv.index = i
|
||
lv.focus()
|
||
return
|
||
|
||
def _focus_tree_path(self, path):
|
||
if not path:
|
||
return
|
||
tree = self.query_one("#nav", Tree)
|
||
|
||
def walk(node):
|
||
d = getattr(node, "data", None)
|
||
if isinstance(d, dict) and d.get("path") == path:
|
||
return node
|
||
for ch in node.children:
|
||
found = walk(ch)
|
||
if found:
|
||
return found
|
||
return None
|
||
|
||
node = walk(tree.root)
|
||
if node is not None:
|
||
tree.move_cursor(node)
|
||
tree.scroll_to_node(node)
|
||
|
||
def _update_summary(self):
|
||
total = sum(paths.values())
|
||
src = t("tree from code") if code_tree else t("visited paths only")
|
||
extra = (
|
||
f" [{self._kanban_mode}]" if self._mode == "kanban" else ""
|
||
)
|
||
self.query_one("#summary", Static).update(
|
||
f" {total} {t('navigations')} · {len(paths)} {t('menus')} · "
|
||
f"{src} · {t('view')}: {self._mode}{extra} · "
|
||
f"{t('F2 system · F3/F4 views · Enter run')}"
|
||
)
|
||
|
||
# -- vue Système (F2) ----------------------------------------------- #
|
||
async def _tick_system(self):
|
||
# I/O système déportées en thread (comme le dashboard) ; on saute
|
||
# la température (subprocess sensors) tant que la vue n'est pas
|
||
# affichée.
|
||
try:
|
||
m, self._sys_prev = await asyncio.to_thread(
|
||
system_snapshot, self._sys_prev, self._mode == "system"
|
||
)
|
||
except Exception:
|
||
return
|
||
if self._mode == "system":
|
||
self.query_one("#system", Static).update(
|
||
self._render_system(m)
|
||
)
|
||
|
||
def _render_system(self, m):
|
||
lines = []
|
||
load = (
|
||
" · " + t("load") + " "
|
||
+ "/".join(f"{x:.1f}" for x in m["load"])
|
||
if m["load"]
|
||
else ""
|
||
)
|
||
lines.append(
|
||
f" 🖥 {t('State')} : {t('uptime')} "
|
||
f"{_fmt_uptime(m['uptime'])}{load}"
|
||
)
|
||
cpu = m["cpu"]
|
||
lines.append(
|
||
f" ⚙ CPU : {cpu if cpu is not None else '?'} %"
|
||
f" ({m['ncpu']} {t('cores')})"
|
||
)
|
||
if m["mem"]:
|
||
tot, used = m["mem"]
|
||
pct = int(used / tot * 100) if tot else 0
|
||
lines.append(
|
||
f" 🧠 {t('Memory')} : {_fmt_size(used)} / "
|
||
f"{_fmt_size(tot)} ({pct} %)"
|
||
)
|
||
if m["disk"]:
|
||
tot, used, free = m["disk"]
|
||
pct = int(used / tot * 100) if tot else 0
|
||
lines.append(
|
||
f" 💽 {t('Disk')} / : {_fmt_size(used)} / "
|
||
f"{_fmt_size(tot)} ({pct} %) · {t('free')} "
|
||
f"{_fmt_size(free)}"
|
||
)
|
||
if m["net"]:
|
||
rx, tx = m["net"]
|
||
lines.append(
|
||
f" 🌐 {t('Network')} : ↓ {_fmt_rate(rx)} "
|
||
f"↑ {_fmt_rate(tx)}"
|
||
)
|
||
if m["battery"]:
|
||
cap, status = m["battery"]
|
||
lines.append(
|
||
f" 🔋 {t('Battery')} : {cap} % ({status})"
|
||
)
|
||
temp = m["temp"]
|
||
if temp:
|
||
lines.append(
|
||
f" 🌡 {t('Temperature')} : {max(temp[1]):.0f}°C "
|
||
f"({t('max')})"
|
||
)
|
||
else:
|
||
lines.append(
|
||
f" 🌡 {t('Temperature')} : "
|
||
f"{t('lm-sensors absent — press i to install')}"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
# -- actions --------------------------------------------------------- #
|
||
def action_system(self):
|
||
self._mode = "system"
|
||
self._apply_visibility()
|
||
self.run_worker(self._tick_system()) # rafraîchit tout de suite
|
||
self._update_summary()
|
||
|
||
def action_install_sensors(self):
|
||
if self._mode != "system":
|
||
return
|
||
if read_temperature() is not None:
|
||
self.notify(t("Sensors already available."))
|
||
return
|
||
cmd = sensors_install_command()
|
||
if not cmd:
|
||
self.notify(
|
||
t("Unknown package manager for lm-sensors."),
|
||
severity="warning",
|
||
)
|
||
return
|
||
printable = " ".join(cmd)
|
||
with self.suspend():
|
||
print(f"{t('Proposed install command:')}")
|
||
print(f" {printable}")
|
||
print(" sudo sensors-detect --auto")
|
||
ans = input(
|
||
t("Install lm-sensors now? (y/N): ")
|
||
).strip().lower()
|
||
if ans in ("o", "oui", "y", "yes"):
|
||
os.system(printable + " || true")
|
||
os.system("sudo sensors-detect --auto || true")
|
||
self.run_worker(self._tick_system())
|
||
|
||
def on_click(self, event):
|
||
# Grille (mosaïque) : clic sur le TITRE d'une case -> l'agrandir
|
||
# (row-span sur toute la hauteur) ; re-clic -> taille normale.
|
||
if self._mode != "kanban" or self._kanban_mode != "grid":
|
||
return
|
||
w = getattr(event, "widget", None)
|
||
if w is None or "ktitle" not in getattr(w, "classes", ()):
|
||
return
|
||
card = w
|
||
while card is not None and "kcol" not in getattr(
|
||
card, "classes", ()
|
||
):
|
||
card = card.parent
|
||
if card is not None:
|
||
card.toggle_class("kbig")
|
||
|
||
# -- actions --------------------------------------------------------- #
|
||
async def action_toggle_view(self):
|
||
if self._mode == "tree":
|
||
self._mode = "kanban"
|
||
await self._enter_kanban()
|
||
else:
|
||
self._mode = "tree"
|
||
self._apply_visibility()
|
||
self._update_summary()
|
||
|
||
def _populate_tree(self):
|
||
tree = self.query_one("#nav", Tree)
|
||
tree.clear()
|
||
if code_tree is not None:
|
||
tree.root.data = {"path": "TODO"}
|
||
tree.root.set_label(f"📍 TODO ({paths.get('TODO', 0)})")
|
||
self._add_code(tree.root, code_tree["children"], "TODO")
|
||
else:
|
||
nested = _nested(paths)
|
||
todo = nested.get("TODO", {"count": 0, "children": nested})
|
||
tree.root.set_label(f"📍 TODO ({todo['count']})")
|
||
self._add_visited(tree.root, todo["children"])
|
||
tree.root.expand()
|
||
|
||
def _add_code(self, tnode, children, parent_path):
|
||
for c in children:
|
||
path = f"{parent_path} › {c['label']}"
|
||
if c["is_menu"]:
|
||
child = tnode.add(
|
||
f"{c['label']} ({paths.get(path, 0)})",
|
||
data={"path": path},
|
||
expand=True,
|
||
)
|
||
self._add_code(child, c["children"], path)
|
||
else:
|
||
# Feuille EXÉCUTABLE : méthode + chemin portés en data.
|
||
tnode.add_leaf(
|
||
f"· {c['label']}",
|
||
data={
|
||
"method": c.get("method"),
|
||
"kwargs": c.get("kwargs"),
|
||
"path": path,
|
||
},
|
||
)
|
||
|
||
def _add_visited(self, tnode, children):
|
||
for name, node in sorted(
|
||
children.items(), key=lambda kv: -kv[1]["count"]
|
||
):
|
||
child = tnode.add(f"{name} ({node['count']})", expand=True)
|
||
self._add_visited(child, node["children"])
|
||
|
||
# -- sélection / exécution ------------------------------------------- #
|
||
def _run(self, method, kwargs, path):
|
||
if not method:
|
||
return
|
||
self._action = (method, kwargs or {})
|
||
self._exit_state = {
|
||
"mode": self._mode,
|
||
"kanban_mode": self._kanban_mode,
|
||
"path": path,
|
||
}
|
||
self.exit()
|
||
|
||
def on_tree_node_selected(self, event):
|
||
d = getattr(event.node, "data", None)
|
||
if isinstance(d, dict) and d.get("method"):
|
||
self._run(d["method"], d.get("kwargs"), d.get("path"))
|
||
|
||
def on_list_view_selected(self, event):
|
||
if isinstance(event.item, CmdItem):
|
||
self._run(
|
||
event.item.cmd_method,
|
||
event.item.cmd_kwargs,
|
||
event.item.cmd_path,
|
||
)
|
||
|
||
async def action_kanban_layout(self):
|
||
if self._mode != "kanban":
|
||
return
|
||
i = KANBAN_MODES.index(self._kanban_mode)
|
||
self._kanban_mode = KANBAN_MODES[(i + 1) % len(KANBAN_MODES)]
|
||
await self._enter_kanban()
|
||
self._update_summary()
|
||
|
||
def action_expand_all(self):
|
||
if self._mode == "tree":
|
||
self.query_one("#nav", Tree).root.expand_all()
|
||
|
||
def action_reset(self):
|
||
reset()
|
||
paths.clear()
|
||
self._populate_tree()
|
||
self._update_summary()
|
||
self.notify(t("Telemetry reset."))
|
||
|
||
app = Telemetry()
|
||
app._exit_state = state
|
||
if run_app:
|
||
app.run()
|
||
return app._action, getattr(app, "_exit_state", state)
|
||
return app
|