diff --git a/script/todo/todo.py b/script/todo/todo.py index 214118d..6e4275d 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -117,6 +117,61 @@ class TODO: set_lang("en") print(t("Language changed to: English")) + def run(self): + with open(self.config_file.get_logo_ascii_file_path()) as my_file: + print(my_file.read()) + self._ask_language() + print(t("Opening TODO ...")) + print(f"đŸ€– {t('=> Enter your choice by number and press Enter!')}") + help_info = f"""{self._menu_header()} +[1] {t("Execute")} +[2] {t("Install")} +[3] {t("Question")} +[4] {t("Fork - Open TODO in a new tab")} +[5] {t("Navigation telemetry (TUI)")} +[6] {t("Configuration")} +[0] {t("Quit")} +""" + while True: + try: + status = click.prompt(help_info) + except NameError: + print("Do") + print(f"source ./{VENV_ERPLIBRE}/bin/activate && make") + sys.exit(1) + except ImportError: + print("Do") + print(f"source ./{VENV_ERPLIBRE}/bin/activate && make") + sys.exit(1) + except click.exceptions.Abort: + sys.exit(0) + print() + if status == "0": + break + elif status == "1": + self.prompt_execute() + elif status == "2": + self.prompt_install() + elif status == "3": + self.execute_prompt_ia() + elif status == "4": + # cmd = ( + # f"gnome-terminal --tab -- bash -c 'source" + # f" ./{VENV_ERPLIBRE}/bin/activate;make todo'" + # ) + cmd = "make todo" + self.execute.exec_command_live(cmd, source_erplibre=True) + elif status == "5": + self._todo_telemetry_tui() + elif status == "6": + self.prompt_configuration() + # elif status == "3" or status == "install": + # print("install") + else: + print(t("Command not found !")) + + print(status) + def prompt_execute(self): help_info = f"""{self._menu_header()} @@ -291,6 +346,107 @@ class TODO: "prompt_configuration": "Configuration", } + def _menu_header(self): + """En-tĂȘte de menu : fil d'Ariane (dĂ©rivĂ© de la pile d'appels) suivi de + la ligne « Commande : ». Le fil situe le menu courant et se copie pour + dĂ©crire sans ambiguĂŻtĂ© oĂč l'on se trouve.""" + crumbs = [] + for frame_info in reversed(inspect.stack()): + if frame_info.frame.f_locals.get("self") is not self: + continue + label = self._MENU_LABELS.get(frame_info.function) + if label and (not crumbs or crumbs[-1] != label): + crumbs.append(label) + header = "" + if crumbs: + header = "📍 " + " â€ș ".join(crumbs) + "\n" + # TĂ©lĂ©mĂ©trie de navigation (best-effort, ne casse jamais le menu). + try: + from script.todo import todo_telemetry + + todo_telemetry.record(" â€ș ".join(crumbs)) + except Exception: + pass + return header + t("Command:") + + def _todo_telemetry_tui(self): + """Ouvre le TUI de tĂ©lĂ©mĂ©trie (arbre/Kanban). Une commande choisie est + exĂ©cutĂ©e au retour (hors du TUI) ; on propose ensuite de REVENIR (l'Ă©tat + et la position du curseur sont restaurĂ©s) ou de quitter.""" + from script.todo.todo_telemetry import run_tui + + from script.todo import textual_setup + + if not textual_setup.ensure(): + return + state = None + while True: + try: + result = run_tui(state=state) + except ImportError: + return + if not result: + return + action, state = result + if not action: + return # quittĂ© sans choisir de commande + method, kwargs = action + # Fil d'Ariane : la commande Ă©tant lancĂ©e DEPUIS la tĂ©lĂ©mĂ©trie (et + # non via la navigation), aucun menu n'a affichĂ© le chemin. On le + # montre ici (dernier segment traduit + icĂŽne) et on l'enregistre. + path = state.get("path") if isinstance(state, dict) else None + if path: + segs = path.split(" â€ș ") + segs[-1] = t(segs[-1]) + print(f"\n📍 {' â€ș '.join(segs)}") + try: + from script.todo import todo_telemetry + + todo_telemetry.record(path) + except Exception: + pass + fn = getattr(self, method, None) + if not callable(fn): + print(f"{t('Command not found !')} ({method})") + else: + try: + fn(**(kwargs or {})) + except Exception as exc: + print(f"{t('Command failed: ')}{exc}") + # Revenir (curseur restaurĂ©) ou quitter ? + ans = input(f"\n{t('Back to telemetry (r) or quit (Enter)? ')}") + if ans.strip().lower() not in ("r", "revenir", "o", "oui", "y"): + return + + # PrĂ©fĂ©rences Ă©ditables depuis le menu Configuration : clĂ©, libellĂ©, et + # valeurs proposĂ©es (valeur stockĂ©e -> libellĂ© affichĂ©). Une seule table : + # l'Ă©cran, la lecture et l'Ă©criture en dĂ©coulent. + _PREF_CHOICES = { + "qemu_deploy_ui": ( + "QEMU deployment interface", + ( + ("ask", "Ask every time"), + ("tui", "TUI form"), + ("cli", "Classic questions (line by line)"), + ), + ), + "qemu_deploy_progress": ( + "Display while deploying", + ( + ("cli", "CLI output (easy to copy)"), + ("tui", "TUI, collapsible blocks per VM"), + ), + ), + "migration_ui": ( + "Odoo migration interface", + ( + ("ask", "Ask every time"), + ("tui", "TUI form"), + ("cli", "Classic questions (line by line)"), + ), + ), + } + def fill_help_info(self, choices): # Une entrĂ©e {"section": "..."} affiche un titre de section SANS # consommer de numĂ©ro : la numĂ©rotation reste continue sur les vraies diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 6ddc9f1..55d558f 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -1649,6 +1649,42 @@ TRANSLATIONS = { "fr": "Rien Ă  copier.", "en": "Nothing to copy.", }, + "TODO navigation telemetry": { + "fr": "TODO — tĂ©lĂ©mĂ©trie de navigation", + "en": "TODO navigation telemetry", + }, + "navigations": { + "fr": "navigations", + "en": "navigations", + }, + "menus": { + "fr": "menus", + "en": "menus", + }, + "Telemetry reset.": { + "fr": "TĂ©lĂ©mĂ©trie rĂ©initialisĂ©e.", + "en": "Telemetry reset.", + }, + "tree from code": { + "fr": "arbre issu du code", + "en": "tree from code", + }, + "visited paths only": { + "fr": "chemins visitĂ©s seulement", + "en": "visited paths only", + }, + "Enter = run, F3 = view": { + "fr": "EntrĂ©e = exĂ©cuter, F3 = vue", + "en": "Enter = run, F3 = view", + }, + "Enter = run, F3/F4 = views": { + "fr": "EntrĂ©e = exĂ©cuter, F3/F4 = vues", + "en": "Enter = run, F3/F4 = views", + }, + "view": { + "fr": "vue", + "en": "view", + }, "avg": { "fr": "moy", "en": "avg", @@ -1657,6 +1693,63 @@ TRANSLATIONS = { "fr": "DerniĂšre install :", "en": "Last install:", }, + "No command found.": { + "fr": "Aucune commande trouvĂ©e.", + "en": "No command found.", + }, + "Back to telemetry (r) or quit (Enter)? ": { + "fr": "Revenir Ă  la tĂ©lĂ©mĂ©trie (r) ou quitter (EntrĂ©e) ? ", + "en": "Back to telemetry (r) or quit (Enter)? ", + }, + "F2 system · F3/F4 views · Enter run": { + "fr": "F2 systĂšme · F3/F4 vues · EntrĂ©e exĂ©cute", + "en": "F2 system · F3/F4 views · Enter run", + }, + "State": {"fr": "État", "en": "State"}, + "uptime": {"fr": "actif depuis", "en": "uptime"}, + "load": {"fr": "charge", "en": "load"}, + "cores": {"fr": "cƓurs", "en": "cores"}, + "Memory": {"fr": "MĂ©moire", "en": "Memory"}, + "Disk": {"fr": "Disque", "en": "Disk"}, + "Network": {"fr": "RĂ©seau", "en": "Network"}, + "Battery": {"fr": "Batterie", "en": "Battery"}, + "Temperature": {"fr": "TempĂ©rature", "en": "Temperature"}, + "lm-sensors absent — press i to install": { + "fr": "lm-sensors absent — appuyez sur i pour installer", + "en": "lm-sensors absent — press i to install", + }, + "Sensors already available.": { + "fr": "Capteurs dĂ©jĂ  disponibles.", + "en": "Sensors already available.", + }, + "Sensors now available.": { + "fr": "Capteurs dĂ©sormais disponibles.", + "en": "Sensors now available.", + }, + "Still no temperature (reboot/modprobe may be needed).": { + "fr": "Toujours pas de tempĂ©rature (redĂ©marrage/modprobe requis ?).", + "en": "Still no temperature (reboot/modprobe may be needed).", + }, + "Unknown package manager for lm-sensors.": { + "fr": "Gestionnaire de paquets inconnu pour lm-sensors.", + "en": "Unknown package manager for lm-sensors.", + }, + "Proposed install command:": { + "fr": "Commande d'installation proposĂ©e :", + "en": "Proposed install command:", + }, + "Install lm-sensors now? (y/N): ": { + "fr": "Installer lm-sensors maintenant ? (o/N) : ", + "en": "Install lm-sensors now? (y/N): ", + }, + "Command failed: ": { + "fr": "Échec de la commande : ", + "en": "Command failed: ", + }, + "Your distribution may package it as python3-textual.": { + "fr": "Votre distribution le fournit peut-ĂȘtre en python3-textual.", + "en": "Your distribution may package it as python3-textual.", + }, "Resize a VM disk": { "fr": "📐 Redimensionner le disque d'une VM", "en": "📐 Resize a VM disk", @@ -1987,6 +2080,18 @@ TRANSLATIONS = { "fr": "Attente de l'arrĂȘt de la VM
", "en": "Waiting for the VM to shut down...", }, + "timeout": { + "fr": "dĂ©lai max", + "en": "timeout", + }, + "shutting down": { + "fr": "arrĂȘt en cours", + "en": "shutting down", + }, + "remaining": { + "fr": "restantes", + "en": "remaining", + }, "VM is off.": { "fr": "VM Ă©teinte.", "en": "VM is off.", diff --git a/script/todo/todo_telemetry.py b/script/todo/todo_telemetry.py new file mode 100644 index 0000000..1dcb9d8 --- /dev/null +++ b/script/todo/todo_telemetry.py @@ -0,0 +1,1245 @@ +#!/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_entries(func) -> list: + """EntrĂ©es NUMÉROTÉES d'un menu « choices = [...] », dans l'ordre : chaque + commande est {« label », « section »}, la section Ă©tant le dernier marqueur + {"section": 
} rencontrĂ© (fill_help_info ne numĂ©rote pas les sections).""" + 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) + ): + entries, section = [], None + for el in node.value.elts: + if not isinstance(el, ast.Dict): + continue + d = {} + for k, v in zip(el.keys, el.values): + if isinstance(k, ast.Constant): + d[k.value] = _str_of(v) + if d.get("section"): + section = d["section"] + continue + lab = d.get("prompt_description") or d.get( + "prompt_description_key" + ) + if lab: + entries.append({"label": lab, "section": section}) + return entries + 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 _config_list(config_key, todo_dir): + """EntrĂ©es d'une liste de config de todo.json (ex. « code_from_makefile »), + cherchĂ©e Ă  n'importe quel niveau. [] si absente.""" + try: + data = json.loads( + (Path(todo_dir) / "todo.json").read_text(encoding="utf-8") + ) + except (OSError, ValueError): + return [] + found = [] + + def walk(o): + if isinstance(o, dict): + for k, v in o.items(): + if k == config_key and isinstance(v, list): + found.extend(v) + walk(v) + elif isinstance(o, list): + for x in o: + walk(x) + + walk(data) + return found + + +def _len_choices_offset(comp): + """K pour « str(len(choices)) » -> 0 et « str(len(choices) - N) » -> N ; + None si le nƓud n'est pas de cette forme.""" + if not ( + isinstance(comp, ast.Call) + and isinstance(comp.func, ast.Name) + and comp.func.id == "str" + and comp.args + ): + return None + arg = comp.args[0] + + def is_len_choices(x): + return ( + isinstance(x, ast.Call) + and isinstance(x.func, ast.Name) + and x.func.id == "len" + and x.args + and isinstance(x.args[0], ast.Name) + and x.args[0].id == "choices" + ) + + if is_len_choices(arg): + return 0 + if ( + isinstance(arg, ast.BinOp) + and isinstance(arg.op, ast.Sub) + and is_len_choices(arg.left) + and isinstance(arg.right, ast.Constant) + ): + return int(arg.right.value) + return None + + +def _dispatch_len(func) -> dict: + """{K: mĂ©thode} pour « status == str(len(choices) - K): self.M() ».""" + out = {} + for node in ast.walk(func): + if not isinstance(node, ast.If): + continue + test = node.test + if not ( + 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) + ): + continue + k = _len_choices_offset(test.comparators[0]) + if k is None: + continue + method = None + for stmt in node.body: + 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" + ): + method = n.func.attr + break + if method: + break + if method: + out[k] = method + return out + + +def _choices_children(func, todo_dir): + """Commandes d'un menu bĂąti par « choices » (config_file.get_config + + choices.append/extend) avec dispatch « str(len(choices)-N) ». Renvoie une + liste de (label, mĂ©thode, kwargs) dans l'ordre affichĂ©, ou None si le motif + ne s'applique pas. Les entrĂ©es de CONFIG rejouent via + execute_from_configuration ; les entrĂ©es APPENDÉES via leur mĂ©thode.""" + def _dict_label(dnode): + d = {} + for k, v in zip(dnode.keys, dnode.values): + if isinstance(k, ast.Constant): + d[k.value] = _str_of(v) + return d.get("prompt_description") or d.get("prompt_description_key") + + # On collecte affectations et append AVEC leur n° de ligne pour rejouer + # dans l'ORDRE (les entrĂ©es « menu_entry = {
}; choices.append(menu_entry) » + # rĂ©utilisent la mĂȘme variable -> il faut suivre la derniĂšre valeur). + config_key = None + events = [] # (lineno, kind, payload) + for node in ast.walk(func): + if ( + isinstance(node, ast.Assign) + and len(node.targets) == 1 + and isinstance(node.targets[0], ast.Name) + ): + var = node.targets[0].id + val = node.value + if isinstance(val, ast.Dict): + events.append((node.lineno, "assign", (var, val))) + elif ( + var == "choices" + and isinstance(val, ast.Call) + and isinstance(val.func, ast.Attribute) + and val.func.attr == "get_config" + and val.args + and isinstance(val.args[0], ast.Constant) + ): + config_key = val.args[0].value + elif ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "append" + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "choices" + and node.args + ): + events.append((node.lineno, "append", node.args[0])) + events.sort(key=lambda e: e[0]) + var_dict, appended = {}, [] + for _ln, kind, payload in events: + if kind == "assign": + var_dict[payload[0]] = payload[1] + else: # append : arg = Dict littĂ©ral ou Name (variable) + arg = payload + dnode = ( + arg + if isinstance(arg, ast.Dict) + else var_dict.get(arg.id) + if isinstance(arg, ast.Name) + else None + ) + if dnode is not None: + lab = _dict_label(dnode) + if lab: + appended.append(lab) + if config_key is None and not appended: + return None + children = [] + for entry in _config_list(config_key, todo_dir) if config_key else []: + lab = ( + entry.get("prompt_description_key") + or entry.get("prompt_description") + or "?" + ) + children.append((lab, "execute_from_configuration", {"instance": entry})) + len_disp = _dispatch_len(func) + n_config = len(children) # figĂ© : `children` grossit dans la boucle + n_total = n_config + len(appended) + for j, lab in enumerate(appended): + pos = n_config + j # 0-based dans la liste globale (stable) + method = len_disp.get(n_total - 1 - pos) # None si non mappĂ© + children.append((lab, method, {})) + return children + + +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") + todo_dir = p.parent # todo.json est Ă  cĂŽtĂ© de 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) + centries = _choice_entries(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 + entry = ( + centries[num - 1] + if 0 <= num - 1 < len(centries) + else None + ) + lab = ( + entry["label"] + if entry + else target.lstrip("_").replace("_", " ") + ) + node["children"].append( + { + "label": lab, + "is_menu": False, + "children": [], + "method": target, + "kwargs": kwargs, + "section": entry["section"] if entry else None, + } + ) + # Menus bĂątis par « choices » (get_config + append) sans dispatch + # littĂ©ral « status == "N" » : Code, Update, Git, Database
 + if not disp: + for lab, tmethod, tkwargs in ( + _choices_children(func, todo_dir) or [] + ): + if tmethod in labels: # une entrĂ©e qui ouvre un sous-menu + node["children"].append(build(tmethod)) + else: + node["children"].append( + { + "label": lab, + "is_menu": False, + "children": [], + "method": tmethod, + "kwargs": tkwargs, + "section": None, + } + ) + 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") + +# Les vues dĂ©filent avec F3 ; chaque vue affiche le NOM de la suivante. +VIEWS = ("tree", "kanban", "list") +VIEW_LABELS = { + "tree": "Arbre", + "kanban": "Kanban", + "list": "Liste", + "system": "SystĂšme", +} + + +def _disp(label: str) -> str: + """LibellĂ© d'affichage d'une commande : passe par t() pour rĂ©cupĂ©rer la + traduction ET l'icĂŽne (les valeurs i18n portent l'icĂŽne). Renvoie le + libellĂ© inchangĂ© s'il n'est pas une clĂ© de traduction.""" + return t(label) if label else label + + +_SENTINEL = object() + + +def _group_by_section(cmds): + """ItĂšre les commandes en groupes (section, [cmds]) dans l'ordre, la + section pouvant ĂȘtre None (aucune).""" + groups, cur, bucket = [], _SENTINEL, [] + for c in cmds: + sec = c.get("section") + if sec != cur: + if bucket: + groups.append((cur, bucket)) + cur, bucket = sec, [] + bucket.append(c) + if bucket: + groups.append((cur, bucket)) + return groups + + +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; } + #list { 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; } + .ksec { height: 1; color: $secondary; text-style: italic; } + .listmenu { height: 1; color: $accent; text-style: bold; } + .listsec { height: 1; color: $secondary; text-style: italic; } + .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", "Vue"), + ("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): + # Regroupe par section pour aider au choix ; icĂŽnes via _disp(). + children = [Static(f"{label} ({cnt})", classes="ktitle")] + for section, group in _group_by_section(cmds): + if section: + children.append( + Static(f"── {_disp(section)} ──", classes="ksec") + ) + children.append( + ListView( + *[ + CmdItem( + f"· {_disp(c['label'])}", + c.get("method"), + c.get("kwargs"), + c.get("path"), + ) + for c in group + ] + ) + ) + return VerticalScroll(*children, classes="kcol") + + def _list_view_widget(self): + # Vue Liste : tous les menus empilĂ©s verticalement, chaque menu + # avec ses sections et ses commandes (icĂŽnes), pour aider le choix. + blocks = [] + for (label, path, cnt, cmds) in columns: + blocks.append( + Static(f"▾ {path} ({cnt})", classes="listmenu") + ) + for section, group in _group_by_section(cmds): + if section: + blocks.append( + Static( + f" ── {_disp(section)} ──", classes="listsec" + ) + ) + blocks.append( + ListView( + *[ + CmdItem( + f" · {_disp(c['label'])}", + c.get("method"), + c.get("kwargs"), + c.get("path"), + ) + for c in group + ] + ) + ) + if not blocks: + return Static(t("No command found.")) + return VerticalScroll(*blocks) + + 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 Container(id="list") + 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("#list").display = self._mode == "list" + self.query_one("#system").display = self._mode == "system" + + async def _restore(self): + if self._mode == "kanban": + await self._enter_kanban() + elif self._mode == "list": + await self._enter_list() + 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")) + + # -- vue Liste ------------------------------------------------------- # + async def _enter_list(self): + box = self.query_one("#list", Container) + await box.remove_children() + await box.mount(self._list_view_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 _next_view(self): + cur = self._mode if self._mode in VIEWS else "tree" + return VIEWS[(VIEWS.index(cur) + 1) % len(VIEWS)] + + def _update_f3_hint(self): + # Le footer F3 annonce la PROCHAINE vue (ex. « → Liste »). + nxt = VIEW_LABELS.get(self._next_view(), self._next_view()) + try: + self.bind("f3", "toggle_view", description=f"→ {nxt}") + self.refresh_bindings() + except Exception: + pass + + 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 "" + ) + cur_lbl = VIEW_LABELS.get(self._mode, self._mode) + nxt_lbl = VIEW_LABELS.get(self._next_view(), self._next_view()) + self._update_f3_hint() + self.query_one("#summary", Static).update( + f" {total} {t('navigations')} · {len(paths)} {t('menus')} · " + f"{src} · {t('view')}: {cur_lbl}{extra} · " + f"F3 → {nxt_lbl} · " + 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") + # RafraĂźchit IMMÉDIATEMENT la vue systĂšme : on rĂ©-Ă©chantillonne + # (tempĂ©rature incluse) et on rĂ©Ă©crit la case pour que le message + # « installer » disparaisse si les capteurs sont dĂ©sormais lisibles. + self._sys_prev = None + try: + m, self._sys_prev = system_snapshot(self._sys_prev, full=True) + self.query_one("#system", Static).update( + self._render_system(m) + ) + except Exception: + pass + self.refresh() + if read_temperature() is not None: + self.notify(t("Sensors now available.")) + else: + self.notify( + t("Still no temperature (reboot/modprobe may be needed)."), + severity="warning", + ) + + 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): + # Cycle Arbre -> Kanban -> Liste -> Arbre (depuis SystĂšme on + # revient dans le cycle sur la vue prĂ©cĂ©dente). + cur = self._mode if self._mode in VIEWS else "tree" + self._mode = VIEWS[(VIEWS.index(cur) + 1) % len(VIEWS)] + if self._mode == "kanban": + await self._enter_kanban() + elif self._mode == "list": + await self._enter_list() + 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