230 lines
8.0 KiB
Python
230 lines
8.0 KiB
Python
"""
|
|
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)
|