Device Drivers

Teach Sapphire to use a machine or a gadget. Devices is part of core: it owns the Settings > Devices page, the device list, and the three tools Sapphire uses. Your plugin brings a driver: one manifest declaration and one small module. Hardware code and its dependencies stay in your plugin, never in core. User guide: Devices.

Anything can be a device: a thermostat, a television, a robot arm. If your code can reach it, Sapphire can use it, and it gets the same page, the same three tools and the same help as every built-in device.

Reference drivers: core/plugins/ssh (stored logins, user-written commands), the MIDI plugin (two device types, every action runs one of the plugin's own tools; the synth it talks to is a profile, not code), and the three that ship inside core, core/devices/drivers/satellite.py, computer.py and sapphire.py (another Sapphire).

What Sapphire sees

Three tools. Their descriptions carry the fleet: every device's name, place, online or offline, and what it can do, so she usually starts at device_action(device) without a list call. With ten devices or fewer the device argument is an enum of their names.

device_list()                                   every device, online or offline, what it can do
device_status("desktop")                        one device, asked right now
device_action("desktop")                        how it is, EVERY action with the value it takes, one example call
device_action("desktop","ssh")                  the same, one capability
device_action("desktop","ssh","volume","40")    runs one

So an action is two calls: device_action(device), then the run. A wrong guess answers with the screen one level up. Core writes all of this help from what your driver's describe() returns, so every device reads the same to her. One screen is kept under about a thousand tokens: a capability with many actions is cut to its first eight and says how to see the rest.

Sapphire cannot add, change, or remove a device. Only the Devices page can.

Your first driver, start to finish

This is a whole driver: a lamp on your network that takes three commands. Copy the two files, change what _send does, and you have a device.

my-lamp/
  plugin.json           # the manifest, with capabilities.devices
  device_driver.py      # describe, status, run, and optionally validate

plugin.json

{
  "name": "my-lamp",
  "version": "1.0.0",
  "description": "A desk lamp Sapphire can switch and dim",
  "author": "you",
  "capabilities": {
    "devices": [
      {
        "driver": "lamp",
        "label": "Desk lamp",
        "icon": "💡",
        "module": "device_driver.py",
        "capabilities": ["light"],
        "config_schema": [
          {"key": "host", "type": "string", "label": "Address", "placeholder": "192.168.0.20"},
          {"key": "token", "type": "string", "widget": "password", "secret": true, "label": "Token"}
        ]
      }
    ]
  }
}

device_driver.py

import requests

class Problem(Exception):
    """A reason fit to show as it is."""

def _send(config, secrets, path):
    """One request to the lamp. The only part that knows the hardware."""
    try:
        r = requests.get(f"http://{config['host']}{path}",
                         headers={"Authorization": "Bearer " + secrets.get("token")}, timeout=5)
    except requests.exceptions.RequestException:
        raise Problem(f"Could not reach the lamp at {config['host']}. Is it plugged in?")
    if r.status_code != 200:
        raise Problem(f"The lamp answered HTTP {r.status_code}.")
    return r.json()

def validate(config):
    """Rules at save time. Returns (config, error). Optional."""
    config["host"] = str(config.get("host") or "").strip()
    return config, "" if config["host"] else "The lamp's address is needed."

def describe(device, config):
    """What this device can do. Called often: keep it cheap and offline."""
    return {
        "light": {
            "label": "Light",                                # the tab name
            "help": "switch and dim it",                     # under 120 characters
            "actions": {
                "on":  {"help": "switch it on",       "example": ""},
                "off": {"help": "switch it off",      "example": ""},
                "dim": {"help": "brightness, 0-100",  "example": "40"},
            },
        },
    }

def status(device, config, secrets):
    """Is it reachable right now? Never raise for a device that is off."""
    try:
        state = _send(config, secrets, "/state")
    except Problem as e:
        return {"online": False, "detail": str(e)}
    return {"online": True, "detail": config["host"],
            "readings": {"brightness": f"{state['brightness']}%"}}

def run(device, capability, action, value, config, secrets, call_tool):
    """Do one thing. Returns (text, ok). value is always a string, maybe empty."""
    try:
        if action == "on":
            _send(config, secrets, "/on")
            return f"{device['label']} is on.", True
        if action == "off":
            _send(config, secrets, "/off")
            return f"{device['label']} is off.", True
        if action == "dim":
            if not value.strip().isdigit() or not 0 <= int(value) <= 100:
                return f"dim takes a number from 0 to 100, not '{value}'.", False
            _send(config, secrets, f"/dim?to={int(value)}")
            return f"{device['label']} is at {int(value)}%.", True
    except Problem as e:
        return str(e), False
    return f"The lamp has no {capability} / {action}.", False

This example is run by tests/test_docs_device_driver_example.py, straight from this page, through the real engine. If it stops working, that test fails.

Try it

1. Put the folder in user/plugins/.
2. Settings > Plugins: enable it. A plugin that is not signed needs sideloading switched on.
3. Settings > Devices > + Add Device. "Desk lamp" is in the Type list.
4. Open the device. The Light tab has a Try button for each action.
5. Ask Sapphire to dim the lamp. She finds it with device_list.

A change to device_driver.py takes effect when the plugin is reloaded. No restart is needed.

What the manifest fields mean

devices is a list. One plugin may bring several device types, each with its own module. The MIDI plugin brings two: a hardware synth, and a plain MIDI keyboard.

Fields land on the tab of the driver's first capability. Add "capability": "sound" to a field to put it on another tab, or "tab": "Status" to put it with the device's health. How to reach a device belongs on Status when the driver has several capabilities.

Setup. Add "setup": true to the fields a device needs before it can work at all - an address, a key, a user name. Add Device asks for those and nothing else (plus any field that shows only because of one of them, through show_if); every other field keeps its default until the user opens the device window. A driver with no setup fields asks for a name only.

Secrets. A field with "secret": true is never stored in the device row. It goes to the device secrets file, scrambled with the machine-bound key, outside user/ and outside backups. The page shows "Set", never the value. Your driver receives it at call time.

What the functions are given

describe may differ per device. The SSH driver lists the commands the user wrote for that one machine.

example is the value she would pass, not the whole call. Core builds the call. Every example must run exactly as written.

Two optional flags on an action: "owner": true keeps it for the person at the Devices page only (she never sees it in her help, and run() refuses her); "danger": "Formats the card. Every backup on it is erased." makes the page ask for I UNDERSTAND before its Try button runs, with those words; it also keeps the action from her (she cannot answer that question), so danger implies owner. Use both for anything that destroys data. A third, "wait": 300, is the seconds a slow action may take (a firmware update, a long recording): the page's Try button waits that long instead of its usual 30 s, so a slow success is never shown as a timeout. Up to 900.

describe may return fewer capabilities than the manifest lists. A gadget that reports only a light gets only a Light tab.

A device that says what it has

One driver often serves boards that differ: one has a light, the next has none. Ask the device, and pass its answer on in status():

def status(device, config, secrets):
    info = ask_the_board(config)             # {"has": ["light", "button"], ...}
    return {"online": True, "detail": info["name"], "has": info["has"]}

Core keeps the list with the device. From then on the device shows only those capabilities, to her and on the page, also while it is offline. A settings field with "capability": "light" is hidden on a device without a light, and its value is kept.

Do not cache the list yourself. describe is given no secrets and must not call the device.

Settings that live on the device

Some settings belong on the hardware: the looks of a ring, the hours it is lit. Give your driver apply(device, config, secrets) and core calls it after every save. Hand the settings over there:

def apply(device, config, secrets):
    r = requests.put(config["url"] + "/led/looks", json={...}, headers=auth(secrets), timeout=6)
    if r.status_code >= 400:
        raise RuntimeError("the board did not take the looks: " + r.text[:100])

The save holds either way. What you raise is shown beside "Saved": Saved, but the device did not take its settings (satellite: ...). A device that is turned off is not told. Keep apply short; it runs while the user waits.

Rules

• Do not reimplement your plugin. If a tool of yours already does the work, run it with call_tool. It goes through the function manager, so it uses the same state and settings as the tool itself (it runs even when the tool's toolset is off: the driver owns the call).
• Do not import your modules under a second name. A module imported twice is two copies of its state. If you must read a helper, take it from the module the loader already holds (plugins.<your-plugin>.tools.<file>).
• Secrets stay out of text. Never put a secret in a result, a log line, or a URL. Core scrubs results as a second line of defense, not a first.
• Quote what she types. If her value reaches a shell, quote it (shlex.quote). Refuse templates that would put it inside quotes.
• Anything that moves needs a bound. An action that starts something must end without a second call: a duration, a watchdog, or a device-side timeout. A dead controller means OFF.
• A keeper calls status() on its own clock (every 30 s while a device is online, every 90 s while it is offline) and keeps one belief per device: online or offline. One answer makes it online; two misses in a row make it offline; the last state survives a restart. The page and device_list read the belief and never wait on you. device_status and the Test button ask fresh. Keep status() under a few seconds for an offline device.

Testing your driver

A driver is three plain functions, so it is tested like any other code. No Sapphire has to run. Fake the one function that touches the hardware:

from unittest.mock import patch
import device_driver as lamp

DEVICE = {"id": "desk", "label": "Desk lamp", "location": "Office"}
CONFIG = {"host": "192.168.0.20"}

def test_dim():
    with patch.object(lamp, "_send", lambda config, secrets, path: {"ok": True}):
        assert lamp.run(DEVICE, "light", "dim", "40", CONFIG, {}, None) == ("Desk lamp is at 40%.", True)
        assert lamp.run(DEVICE, "light", "dim", "bright", CONFIG, {}, None)[1] is False

Two tests are worth having in every driver:

• Every example in your help really runs. Loop over describe() and run each action with its own example. Sapphire copies those examples.
• Say what the value is. An action may carry values, a short grammar of its value: <color> [pulse|blink] [5s]. Angle brackets for what is needed, square for what may be left out, | for a choice. It sits beside the action's name on her screen, so she does not have to guess, or call again to find out. Put what the words mean in help. A long menu (every color, 128 instruments) stays an action of its own, one call away.
• What is missing is said plainly. An unplugged device must answer with a sentence, never a stack trace.

Names that mean the same everywhere

Use these capability names when they fit. Sapphire then reads every device the same way, and core knows what to do with them.

Any other name works too. It gets a tab and a place in her help like the rest.

Power: a device that can restart or switch off

Name the capability power. Use these action names where they fit, so every device reads the same to her: restart, shutdown, sleep.

• Answer first, act a moment later. The answer has to leave the device.
• Say in the result how the device comes back. After a shutdown, someone has to walk over to it.
• power carries a switch on the Devices page, Sapphire may use this. Core enforces it. Your driver does nothing for it. camera, screen and mic carry the same switch.
• To have the switch start out off, add "locked_by_default": ["power"] to your driver's declaration. Do that for anything the user would miss, like their own computer.

Pictures: a device with a camera

An action may answer with pictures. Return this in place of the text:

return {"text": "A picture from the camera of kitchen.",
        "images": [{"data": base64_text, "media_type": "image/jpeg"}]}, True

She sees the pictures if her model can see. If it cannot, core describes them to her in words. The Try button shows them on the page.

Anything outside those limits is dropped. Name the capability camera and the action look so every camera reads the same to her. Warn the room before you capture: a light, a sound, or both.

Moving pictures on a board's glass

A plugin that paints frames on a satellite's screen (FlashHead puts her talking face there) uses two in-process calls in core.devices.glass, never the driver's private ones. There is no door for it: each board is reached with its own secrets, as every call is.

• face_boards() returns [(device_id, side)]: every enabled satellite whose glass is on and takes a JPEG frame (firmware 0.7.7+, screen.jpeg in its /health). side is the largest square it paints, a multiple of 16.
• face_frame(device_id, jpeg, seconds=3) paints one JPEG of that size straight to the panel. The board's own page comes back seconds after the last frame. It raises when the board did not take it. Send the newest frame only: a small board paints about 8 a second.

Voice: a device with a microphone or a speaker

Core's voice pipeline uses devices through a few conventions. Follow them and your device becomes a satellite with no other code.

A speaker. Declare a capability named speaker with an action say. Its value is the text. Get her voice as bytes from core, then play it:

from core.devices import voice
audio, kind = voice.render(text)        # (bytes, "audio/ogg") or (None, reason)

render applies the privacy gate for the chat that produced the text. When it returns None, return the reason as your result and send nothing.

A small board plays one format. Let it state that format, and have core convert:

plays = {"type": "audio/wav", "rate": 16000, "channels": 1}     # as the board stated it
audio, kind = voice.fit(audio, kind, plays)     # (bytes, "audio/wav") or (None, reason)

audio/wav is always 16 bit PCM. It is the only type today. rate is 8000 to 48000, channels is 1 or 2.

A microphone. Declare a capability named mic, and these config fields:

The device then posts what it heard:

POST /api/devices/<device name>/voice
Authorization: Bearer <voice_key>
Content-Type: audio/wav

<the recording>

A form with the file field audio works too. The full contract of a satellite board is in Satellite Protocol.

Core transcribes it and answers at once with {ok, heard, accepted, chat}. accepted: true means a turn has started. Its answer comes later, through your speaker / say. The device must not wait on this response for her answer.

She reads one line above what was heard, naming the device and its location:

[Voice from device "kitchen" (Kitchen Pi). Location: Kitchen. Your reply is spoken aloud there by itself: just answer, never call say on that device for it.]

A chat that is mid-turn is waited for, up to 20 seconds. One device may have 3 questions waiting. More are refused with busy: true.

A light. The device may hold one stream open to learn what its light should show:

GET /api/devices/<device name>/events
Authorization: Bearer <voice_key>

It is a server-sent event stream. Each data: line is JSON with a state:

A device hears only about its own turns. A stream that opens in the middle of a turn is sent the current state first. Lines that start with : keep the stream alive and carry nothing. The stream ends when the key changes.

Show thinking yourself from the moment accepted arrives, and put a bound on it. If core restarts in the middle of a turn, no idle will come.

To transcribe inside an action of your own, use voice.stt_refusal() first, then voice.transcribe(audio_bytes).

Things that come and go

A keyboard is plugged in. A board is pulled out. A Bluetooth speaker walks out of range. A driver can ask to be kept told, so its heavy part (a synth, an open port, a connection) runs only while it has something to run for.

Add "presence": true to the manifest entry, and up to three functions to the module. All three are optional.

tend is handed the whole truth every time. Never "this one arrived". present is everything that is here and counts as this device. It is called when something changed, when the user saved the device, and every 15 seconds regardless. So tend must be safe to repeat: start what is not running, leave alone what is. In return a missed event is repaired by the next call, and so is a process of yours that fell over.

leaving is true once, when the device is switched off or removed, when your plugin unloads, and when Sapphire stops. present is empty then. Let go of everything.

id must hold across a replug. Use the hardware's own name or serial number, not a number the system hands out afresh each time. name is what the user reads. kind is a short word like USB or Bluetooth.

The pick list

A field of type found lets the user say which of the found things count. Core draws it from discover(), stores the choice, and applies it. Your driver never sees the filter, only present.

(•) Whichever is plugged in          "many": false shows one dropdown
( ) Only these                       "many": true (the default) shows ticks
      [x] Board A    USB
      [ ] Board B    USB

Setting a thing up before it is a device. A thing in the list may carry a note (a line under it, 200 characters) and actions (up to three buttons, [{"do": "pair", "label": "Pair"}]). A button calls your module's setup(thing_id, action, config) -> (text, ok); the words come back under the list and the list is looked at again. While any thing shows a button the list looks again on its own every few seconds, so a state that changes on the other side (a phone that asks to pair) shows up by itself. A phone is paired this way in the Add window, and its driver's validate refuses to add one that is not paired: a setup that does not set the thing up is no setup (core/plugins/kdeconnect).

A board on USB

Two boards are plugged in. Each is its own device, and each device picks its board. The manifest entry:

{
  "driver": "board",
  "label": "USB board",
  "module": "device_driver.py",
  "capabilities": ["light"],
  "presence": true,
  "config_schema": [
    {"key": "which", "type": "found", "many": false, "tab": "Status",
     "label": "Which board", "all_label": "Whichever is plugged in"}
  ]
}

What the module adds to describe, status and run:

import os

BY_ID = "/dev/serial/by-id"       # the system lists boards here by make and serial number
_held = {}                        # device id -> {"path", "port"}

def discover(config):
    """Every board that is plugged in right now."""
    try:
        names = sorted(os.listdir(BY_ID))
    except OSError:
        return []                 # no folder: nothing is plugged in
    return [{"id": name, "name": name.split("_")[1] if "_" in name else name, "kind": "USB"}
            for name in names]

def watch(changed, stopped):
    """Look once a second. If your system can tell you when something changed,
    read that in place of looking."""
    seen = None
    while not stopped.wait(1):
        now = discover({})
        if now != seen:
            seen = now
            changed()

def tend(device, config, secrets, present, leaving):
    """Hold the board's port open while it is here. Let go when it is not."""
    want = "" if leaving or not present else os.path.join(BY_ID, present[0]["id"])
    held = _held.get(device["id"])
    if held and held["path"] != want:
        held["port"].close()
        del _held[device["id"]]
    if want and device["id"] not in _held:
        _held[device["id"]] = {"path": want, "port": open_port(want)}

open_port is yours to write. run then uses _held[device["id"]]["port"], and answers "The board is not plugged in." when there is none.

What core does for you

• One watcher per driver, however many devices it has. changed() costs nothing to call twice: a burst is gathered into one look.
• Each device is tended by itself. One that is slow does not hold the others. The same device is never tended twice at once.
• A watch that ends or fails is started again, a little later each time. The 15 second look covers until then.
• Status gains a connected line, said the same way for every driver.
• Before your plugin unloads, every device of yours is tended with leaving=True. You have 8 seconds.

Rules

• One thing shared by several devices is yours to add up. The MIDI keyboard driver has one synth for all its devices. It keeps what each device wants and plays for all of them together.
• discover must be quick and must not change anything. It runs on every look and when the user opens the pick list.
• Start nothing at import. A thread that starts when your module loads runs in tests, and runs when no device exists. tend is where things start.

Drivers that ship inside core

Three drivers live in core/devices/drivers/ because every install should have them: satellite (a room box with mic, speaker, light and camera), computer (the machine Sapphire runs on) and sapphire (another Sapphire). Their ids are reserved: a plugin cannot register a driver with one of those names, and cannot call itself core. Everything in this guide applies to them unchanged. Their manifest entry is a SPEC dict in the module.

One thing only a core driver may do: name tools of other plugins in its SPEC (uses_tools) and run them through call_tool. The computer sees its screen through the screenshot plugin's own tool that way, so Sapphire has one screenshot program, not two. A plugin's driver runs its own plugin's tools and no others.

Settings fields added for device forms

These work in any plugin's settings schema.

Reference for AI

• Manifest: capabilities.devices: [{driver, label?, icon?, module, capabilities[], config_schema[]?, locked_by_default[]?, presence?}]. Registered by the loader into core/devices/registry.py; unregistered on unload. Bad declarations are skipped with a log line, never a failed load.
• Module functions: describe(device, config) -> {capability: {label, help, actions: {name: {help, example, values?, owner?, danger?, wait?}}}}, status(device, config, secrets) -> {online, detail, readings?, has?, takes?}, optional apply(device, config, secrets) (after a save; raise DeviceError to be heard), run(device, capability, action, value, config, secrets, call_tool) -> (text, ok), optional validate(config) -> (config, error), optional options(device, config, key) -> [{value, label}] for a dynamic select (below).
• A select whose choices are the device's own: {"key", "type": "select", "dynamic": true} in config_schema, and the module's options(device, config, key); device carries has, takes and fingerprint. The satellite driver's wake_model offers the wake words of the family that board runs, from core.wakeword.library. The stored value is always among the choices, so a save round-trips when the driver cannot say. A field with takes ("takes": "wake.model") shows only on a device whose status() answered that name in takes.
• Engine: core/devices/engine.py, loaded on first use. Tools: functions/devices.py. Rows in user/plugin_state/devices.json under key devices: {id, label, location, enabled, created, locked: [capability], parts: [{driver, plugin, config}]}. locked = what she may not use; only capabilities in engine.LOCKABLE can be locked; engine.run(..., owner=True) is the user's own button and passes the lock. Secrets in core/devices/secret_store.py under <driver>.<field>.
• The engine re-checks that the owning plugin is enabled and loaded on every call and drops cached driver modules when the registry generation changes.
• Hosted (managed) installs: the tools refuse, the routes answer 404, the tab is hidden.
• call_tool refuses tools owned by another plugin. It calls execute_function(name, args, allowed_tools={name}, with_success=True).
• Voice: core/devices/voice.py (hear, say, cue, render, transcribe, key_ok, origin_line). hear answers once the words are known and runs the turn on its own thread. Reply lane: cadence.run_turn(chat, text, speak='device:<id>', on_event=...). Doors: POST /api/devices/{id}/voice and GET /api/devices/{id}/events, device key, each 30 a minute per caller address, counted before the key is checked.
• Core drivers: registry.CORE_DRIVERS, registered by the engine on first use with builtin=True.
• Self-report: status() may answer has: [capability]. engine._learn keeps it on the row as parts[].has (written only on a change, names outside the manifest dropped with one warning). engine.capabilities(part, spec) is the manifest list held to it; describe, public, run and voice._voice_part all use it. public hides schema fields whose capability the device lacks. No key = never said = everything.
• Sound format: voice.fit(audio, kind, plays) and voice.wanted(plays). numpy and soundfile only. Going down in rate it low-pass filters first.
• After a save: engine.tell(row) calls each part's apply and returns problems in words; the routes put them in the save's warning. voice.clock() = {now, tz} (POSIX zone from /etc/localtime's footer) rides on the events stream's connected line.
• Routes (core/routes/devices.py; a login session or a Bearer API token, which runs as the owner): GET|POST /api/devices, GET|PUT|DELETE /api/devices/{id}, POST /api/devices/{id}/test, POST /api/devices/{id}/run, GET /api/devices/found/{driver}?device=, POST /api/devices/provision, GET /api/devices/here, /api/devices/flash/*, GET /api/devices/firmware[/{board}/{part}]; with the device's own key: /api/devices/{id}/voice|wake|text|events. Scheduled Device tasks run as the owner too (locks and owner-only actions pass; danger actions cannot be scheduled). Limits: 240 reads and 60 writes a minute. PUT takes locked as a map {capability: bool} that changes only what it names (a list replaces the whole set). found is a reserved device name.
• Presence: core/devices/presence.py, started after the plugin scan (sapphire.py), stopped before plugin services. Manifest "presence": true plus optional module functions discover(config) -> [{id, name, kind}], watch(changed, stopped), tend(device, config, secrets, present, leaving). tend is level triggered: called on a poke (watcher, device saved, plugin loaded), on the 15 s heartbeat, and once with leaving=True, present=[] when the device is disabled or removed, the plugin unloads (presence.release(plugin), called by the loader before it unregisters the drivers) or the app stops. One keeper thread, one watcher thread per driver with an enabled device, tends on a pool of 4, never two at once for one device. Pokes are gathered for 0.3 s.
• Filter: a config_schema field of "type": "found" (one per driver). Stored in the part's config as {all: bool, only: [{id, name}]}. engine.found(driver, config) cleans what discover returns (64 at most; a thing may carry note and actions: [{do, label}]). engine.passing(spec, config, things) applies the filter. "many": false keeps one. engine.setup_found(driver, thing_id, action, config) runs the module's setup() for a thing's button: POST /api/devices/found/{driver}/do {id, do, config?}.