# SPDX-License-Identifier: Apache-2.0 """The contract and the registry of the tool plugins. A *tool* is one menu item and one CLI subcommand. To make a tool, subclass :class:`Tool`, decorate it with :func:`register`, and implement :meth:`Tool.run`. The same registry controls the interactive menu and the Typer CLI. :meth:`Tool.execute` puts each run of a tool in a logged ``runs`` row. Thus a new tool gets storage and logging with no more work. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any from ..core import exitcodes # The wrapper for a secret parameter, and its reader. They live with the other rules that # keep secrets out of the logs (refer to meshterm.core.secrets). The tools import them # from here, next to Tool and ToolResult. from ..core.secrets import Secret as Secret from ..core.secrets import loggable from ..core.secrets import reveal as reveal if TYPE_CHECKING: # do not import typer or the context at module load, for a fast start import typer from ..context import AppContext from ..ui.report import Report @dataclass(slots=True) class ToolResult: """The result of a run of a tool. Attributes: summary: A summary that can be serialized to JSON. It is stored in the record of the run. message: An optional closing message for a person. It is the closing line of the *menu*. A scripted run does not print it (refer to :func:`~meshterm.cli._execute_and_render`), because on the command line it repeats the output above it and the exit status below it. artifacts: The paths of the files that the tool made (for example, generated visualizations). exit_code: The status with which a scripted run must exit. Refer to :mod:`meshterm.core.exitcodes`. A tool that worked keeps ``OK``. A tool that ran correctly and found nothing to report sets ``NO_RESULT``. Thus a caller can tell an empty mesh from a full one, and does not have to count lines. A *failure* is raised, not returned, so this value never holds one. report: **The answer**, as data (refer to :mod:`meshterm.ui.report`). A scripted run puts it here, and the CLI boundary gives it to a renderer. The boundary does the same with ``exit_code``: the tool says what occurred, and other code changes that into bytes or into a process status. Before, the tool printed the answer instead. Then the answer was gone before anything could ask for it in another format. That is why ``--json`` got to two commands and stopped. ``None`` for the menu path, and for a feature that has no scripted face (the map, the dashboard). These continue to work with no change. It is not part of ``summary``. ``summary`` is the record in the *run log* of what this run did, and each run writes it to the ``runs`` table, also a menu run. If ``report`` is part of it, a five-hour ``monitor`` capture goes into SQLite. That is the cost of one field fewer. """ summary: dict[str, Any] = field(default_factory=dict) message: str | None = None artifacts: list[str] = field(default_factory=list) exit_code: int = exitcodes.OK report: Report | None = None class Tool(ABC): """The base class for all MeshTerm features. Class Attributes: name: The CLI identifier (kebab-case). It is unique among the tools. title: The name that the interactive menu item and the result screen of the tool show. It starts with a capital letter. It is the same as the name of the screen that the item opens, so that the menu never shows one name and opens another. If it is empty, ``name`` is used. help: The one-line description in the menu and in ``--help`` (with no period at the end). icon: The emoji that the interactive menu draws before the title (only there: the result screen keeps the plain title). Use only glyphs of one code point with emoji presentation, so that the width of two cells is the same on all terminals (refer to :mod:`meshterm.ui.tui.emoji_width`). category: The group label that organizes the interactive menu. order: The sort key in a category (a lower value comes first). menu_visible: Whether the tool is in the interactive menu. Set ``False`` for a tool that is only for the CLI (for example, diagnostics at startup, which have no place in a connected session). Such a tool registers a CLI subcommand as usual. popup: Whether the menu stays pushed while this tool runs. Then the prompts and the result of the tool float over the menu as modal dialogs (the pattern of the quit dialog), and do not replace the screen. This is for quick tools of the size of a dialog. A tool that pushes its own full screen keeps the default. """ name: str = "" title: str = "" help: str = "" icon: str = "" category: str = "General" order: int = 100 menu_visible: bool = True popup: bool = False @abstractmethod async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: """Do the work of the tool. Args: ctx: The shared application context (console, device, repository, and other items). params: The validated parameters for this run. Returns: A :class:`ToolResult` that describes the result. """ async def prompt_params(self, ctx: AppContext) -> dict[str, Any]: """Ask the user for the parameters of a menu run. The default implementation has no parameters. A tool overrides this method to ask the user through :attr:`AppContext.ui`: a dialog or a select list in the menu. On the CLI, it asks nothing, because the arguments came from the command line. Args: ctx: The shared application context. Returns: A dict of parameters that :meth:`run` accepts. """ return {} def register_cli(self, app: typer.Typer) -> None: """Register this tool as a Typer subcommand. The default registers a command with no arguments. A tool with parameters overrides this method to declare typed options, and then calls :meth:`execute`. Args: app: The Typer application that gets the command. """ from ..cli import run_tool_command @app.command(name=self.name, help=self.help) def _command() -> None: run_tool_command(self, {}) async def execute(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: """Run the tool, with a log of the run and a capture of its errors. This method opens a ``runs`` row before the run. After the run, it closes the row with the final status and the summary, also when the run failed. Args: ctx: The shared application context. params: The parameters for this run. Returns: The :class:`ToolResult` from :meth:`run`. Raises: Exception: Each error from :meth:`run`, raised again after it is stored. """ import typer from ..core.connection import DeviceCommandError, is_connection_lost from ..core.device_config import DeviceConfigError from ..core.preferences import PreferenceError from ..core.selection import DeviceSelectionError run_id = ctx.repo.start_run(self.name, params, ctx.profile_name) # The copy with each secret hidden: a password, a channel secret, or a PIN in the # parameters must never reach the log (refer to meshterm.core.secrets). ctx.log.debug("run %s start: tool=%s params=%s", run_id, self.name, loggable(params)) try: result = await self.run(ctx, {**params, "_run_id": run_id}) except ( DeviceSelectionError, DeviceConfigError, DeviceCommandError, PreferenceError, # When a tool refuses one of its own arguments (a `--to` that it cannot # resolve), it raises the same error as the parser. It is a usage error, not a # fault. typer.BadParameter, typer.Abort, typer.Exit, ) as exc: # An expected condition for the user (no device or an ambiguous device, a bad # config or preference value, a bad argument, or a temporary failure of a # command). Store it, but do not write a traceback. The callers print the # message cleanly. ctx.repo.finish_run(run_id, "error", {"error": str(exc)}) ctx.log.debug("run %s aborted: %s", run_id, exc) raise except Exception as exc: # noqa: BLE001 - we store it, then raise it again ctx.repo.finish_run(run_id, "error", {"error": str(exc)}) if is_connection_lost(exc): # The device was disconnected or switched off during a command. Each # layer above knows how to say that in one line. A stack of serial # internals tells how the driver found the problem, and that is a # different question. ctx.log.warning("run %s lost the device connection: %s", run_id, exc) else: ctx.log.exception("run %s failed: %s", run_id, exc) raise ctx.repo.finish_run(run_id, "ok", result.summary) ctx.log.debug("run %s ok", run_id) return result _REGISTRY: dict[str, Tool] = {} #: The explicit order of the tool categories in the menu. The menu asks "What would you #: like to do?", so its sections answer that question. Each section names an *activity*, #: not a subject, and holds all the items for that activity. Message is first (the #: features for each day, and the list from which you select a recipient). Then come Watch #: (what the mesh does, what it did, and what to get notifications about) and Explore #: (where the nodes are, how they get to each other, and the boards that score the walks). #: Then come the two owners of a setting: This node (the radio in your hand) and Other #: nodes (the radio of another person, over the mesh). #: #: This app ends the list as the third and last *owner*, after the radio in your hand and #: the radio of another person over the mesh: it is the program in front of you. It holds #: the one item that changes how MeshTerm behaves (Preferences), and the four pages that #: say what MeshTerm is. These are questions that no activity above can hold, and that #: nothing on the mesh can answer. This app is last because the user who has nothing more #: to do is the user who looks for it. Categories that are not in this list come last, in #: alphabetical order. Thus a new category is always in the menu. _CATEGORY_ORDER = [ "Message", "Watch", "Explore", "This node", "Other nodes", "This app", ] def _category_rank(category: str) -> int: """Return the sort rank of a category (its position in ``_CATEGORY_ORDER``, or last).""" try: return _CATEGORY_ORDER.index(category) except ValueError: return len(_CATEGORY_ORDER) def register(cls: type[Tool]) -> type[Tool]: """A class decorator that makes an instance of a tool and adds it to the registry. Args: cls: The :class:`Tool` subclass to register. Returns: The class with no change (thus the decorator is transparent). Raises: ValueError: If the tool has no ``name``, or if the name is already registered. """ instance = cls() if not instance.name: raise ValueError(f"{cls.__name__} must define a non-empty 'name'.") if instance.name in _REGISTRY: raise ValueError(f"Duplicate tool name: {instance.name!r}") _REGISTRY[instance.name] = instance return cls def all_tools() -> list[Tool]: """Return all the registered tools, sorted by category rank, then order, then name. :data:`_CATEGORY_ORDER` sets the order of the categories (Mesh first). Thus the main tools, Chat and Channels, come first in the menu. In a category, ``order`` and then ``name`` decide a tie. Returns: The registered tool instances, in a stable menu order. """ return sorted( _REGISTRY.values(), key=lambda t: (_category_rank(t.category), t.category, t.order, t.name), ) def get_tool(name: str) -> Tool | None: """Find a registered tool by its name. Args: name: The ``name`` of the tool. Returns: The tool instance, or ``None`` if no tool has that name. """ return _REGISTRY.get(name)