""" The extension API. Drop a ``.py`` file into ``backend/plugins/`` that builds a ``Plugin`` and the dashboard grows a new panel for it - no frontend work, no server edits. Each control you declare is rendered by the browser from the manifest this module produces, and clicking it calls your handler. Minimal example:: from backend.plugin_api import Plugin plugin = Plugin(id="hello", name="Hello", icon="\U0001F44B") @plugin.action("wave", label="Wave hello") async def wave(ctx): await ctx.bridge.play_preset(motion=1002, area=2) return "Waved" Handlers may be async or plain functions. Whatever they return is shown as the result toast; raise an exception and the message is surfaced as an error. """ from __future__ import annotations import inspect import time from dataclasses import dataclass, field from typing import Any, Callable @dataclass class Context: """Everything a handler is given. Passed as the first argument.""" bridge: Any hub: Any config: dict storage: dict plugin_id: str @property def state(self): """Live robot state - the same object the dashboard renders.""" return self.bridge.state async def log(self, message: str, level: str = "info") -> None: await self.hub.emit(level, f"plugin:{self.plugin_id}", message) async def push(self, key: str, value: Any) -> None: """Publish a value into the shared state under ``custom``.""" bucket = self.bridge.state.custom.setdefault(self.plugin_id, {}) bucket[key] = value def record(self, key: str, value: float) -> None: """Append a sample to a chartable time series.""" self.hub.record(f"{self.plugin_id}.{key}", value) @dataclass class Control: key: str kind: str label: str handler: Callable | None = None options: dict = field(default_factory=dict) def manifest(self) -> dict: return {"key": self.key, "kind": self.kind, "label": self.label, **self.options} class Plugin: """A named group of controls that appears as its own card in the Extensions tab.""" def __init__(self, id: str, name: str, description: str = "", icon: str = "◆", order: int = 100, tab: str = "extensions"): self.id = id self.name = name self.description = description self.icon = icon self.order = order self.tab = tab self.controls: list[Control] = [] self.readouts: list[dict] = [] self._on_start: Callable | None = None self._on_tick: Callable | None = None self._tick_interval = 1.0 self.storage: dict = {} # -- control declarations ---------------------------------------------- def action(self, key: str, label: str = "", style: str = "default", confirm: str = "", icon: str = "", help: str = ""): """A button. ``style`` is one of default | primary | warn | danger.""" def wrap(fn): self.controls.append(Control(key, "action", label or key, fn, { "style": style, "confirm": confirm, "icon": icon, "help": help, })) return fn return wrap def slider(self, key: str, label: str = "", min: float = 0.0, max: float = 1.0, step: float = 0.01, default: float = 0.0, unit: str = "", live: bool = False, help: str = ""): """A continuous value. ``live=True`` fires while dragging.""" def wrap(fn): self.controls.append(Control(key, "slider", label or key, fn, { "min": min, "max": max, "step": step, "default": default, "unit": unit, "live": live, "help": help, })) return fn return wrap def toggle(self, key: str, label: str = "", default: bool = False, help: str = ""): def wrap(fn): self.controls.append(Control(key, "toggle", label or key, fn, { "default": default, "help": help, })) return fn return wrap def select(self, key: str, label: str = "", options: list | None = None, default: Any = None, help: str = ""): """``options`` is a list of {"value": ..., "label": ...} or bare strings.""" def wrap(fn): normalised = [] for option in (options or []): if isinstance(option, dict): normalised.append(option) else: normalised.append({"value": option, "label": str(option)}) self.controls.append(Control(key, "select", label or key, fn, { "options": normalised, "default": default, "help": help, })) return fn return wrap def text(self, key: str, label: str = "", default: str = "", placeholder: str = "", multiline: bool = False, submit_label: str = "Send", help: str = ""): def wrap(fn): self.controls.append(Control(key, "text", label or key, fn, { "default": default, "placeholder": placeholder, "multiline": multiline, "submit_label": submit_label, "help": help, })) return fn return wrap def number(self, key: str, label: str = "", min: float | None = None, max: float | None = None, step: float = 1.0, default: float = 0.0, unit: str = "", help: str = ""): def wrap(fn): self.controls.append(Control(key, "number", label or key, fn, { "min": min, "max": max, "step": step, "default": default, "unit": unit, "help": help, })) return fn return wrap def color(self, key: str, label: str = "", default: str = "#3987e5", help: str = ""): def wrap(fn): self.controls.append(Control(key, "color", label or key, fn, { "default": default, "help": help, })) return fn return wrap # -- readouts ---------------------------------------------------------- def readout(self, key: str, label: str, unit: str = "", chart: bool = False, format: str = "number", precision: int = 2) -> None: """ Declare a value this plugin publishes via ``ctx.push(key, value)``. ``chart=True`` also draws a sparkline from ``ctx.record(key, value)``. """ self.readouts.append({ "key": key, "label": label, "unit": unit, "chart": chart, "format": format, "precision": precision, }) # -- lifecycle --------------------------------------------------------- def on_start(self, fn): """Called once when the plugin loads.""" self._on_start = fn return fn def on_tick(self, interval: float = 1.0): """Called repeatedly on a background timer.""" def wrap(fn): self._on_tick = fn self._tick_interval = max(0.1, float(interval)) return fn return wrap # -- introspection ----------------------------------------------------- def manifest(self) -> dict: return { "id": self.id, "name": self.name, "description": self.description, "icon": self.icon, "order": self.order, "tab": self.tab, "controls": [c.manifest() for c in self.controls], "readouts": self.readouts, "has_tick": self._on_tick is not None, "tick_interval": self._tick_interval, } def find(self, key: str) -> Control | None: return next((c for c in self.controls if c.key == key), None) async def invoke(handler: Callable, ctx: Context, value: Any = None) -> Any: """Call a handler, passing ``value`` only when it takes one.""" signature = inspect.signature(handler) args = [ctx] if len(signature.parameters) > 1: args.append(value) result = handler(*args) if inspect.isawaitable(result): result = await result return result def now_ms() -> int: return int(time.time() * 1000)