Plugin Development

A plugin adds a feature to Sonorium without changing Sonorium: a doorbell chime, an alarm clock, a connector to another home system, a remote control, an importer. It is a folder with a Python file. Sonorium loads it and offers it a set of services: follow what happens (events), act on channels (actions), run at set times (the scheduler), be called by other devices (webhook triggers), add pages and settings to the web UI, play sounds over channels, ramp volume, add audio outputs, take commands from remotes, and use MQTT, Home Assistant and text to speech.

Think of Sonorium as a phone and a plugin as an app. The app doesn't open the phone and rewire it; it asks the phone's services for the camera or the speaker. Those services are documented here and keep working across Sonorium versions: names and fields are only added, never renamed or removed.

This page describes the plugin API of Sonorium 1.7.1 (plugin API 1.1). The repository's guide is docs/PLUGIN_DEVELOPMENT.md, with a second example in docs/examples/trigger_play_channel. Installing and managing plugins as a user is on Plugins; the HTTP routes are on API Reference.

What a plugin can do

A first plugin: Doorbell

A doorbell button (an ESP32, a Zigbee button through Home Assistant, a shell script) calls a URL; Sonorium plays a chime over the chosen channels, ducking what plays. The plugin has settings, a webhook trigger, an event subscription, an event of the plugin and a page with a button.

The files

doorbell/
├── manifest.json
└── plugin.py

doorbell/manifest.json:

{
  "id": "doorbell",
  "name": "Doorbell",
  "version": "1.0.0",
  "description": "Plays a chime over channels when the doorbell trigger is called.",
  "author": "Your name",
  "entry_point": "plugin.py",
  "plugin_class": "Doorbell",
  "api_version": "1.1",
  "requires": ["triggers", "pages", "audio.sounds"]
}

doorbell/plugin.py:

"""Doorbell: a webhook trigger that plays a chime over Sonorium's channels."""
from __future__ import annotations

from datetime import datetime, timezone

from sonorium.plugins.actions import ActionError
from sonorium.plugins.base import BasePlugin


class Doorbell(BasePlugin):
    id = "doorbell"
    name = "Doorbell"
    version = "1.0.0"
    description = "Plays a chime over channels when the doorbell trigger is called."
    author = "Your name"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.rings: list[dict] = []          # newest first, kept in memory for the page

    # Settings: drawn as a form under the plugin in Settings > Plugins
    def get_settings_schema(self) -> dict:
        return {
            "chime_url": {"type": "url", "label": "Chime", "required": True,
                          "help": "An http(s) address of an MP3 or WAV file"},
            "channels": {"type": "string", "label": "Channels", "default": "all",
                         "help": "Channel names or ids, separated by commas, or all"},
            "volume": {"type": "number", "label": "Volume", "min": 0, "max": 200, "default": 100},
        }

    # Lifecycle: subscriptions made here are removed when the plugin is disabled
    async def on_enable(self) -> None:
        self.subscribe("sound.ended", self.on_sound_ended)

    async def on_sound_ended(self, event) -> None:
        if event.data.get("origin") != f"plugin:{self.id}":
            return                            # a sound someone else played
        for ring in self.rings:
            if ring["sound_id"] == event.data.get("sound_id"):
                ring["result"] = event.data.get("reason", "")

    # Webhook trigger: POST or GET /api/triggers/doorbell/ring
    def get_triggers(self) -> list[dict]:
        return [{"id": "ring", "title": "Ring", "methods": ["POST", "GET"],
                 "description": "Play the chime. Optional payload: door (a name shown on the page)."}]

    async def on_trigger(self, trigger: str, payload: dict, context: dict) -> None:
        await self.ring(str(payload.get("door") or "Front door"), context.get("client", ""))

    async def ring(self, door: str, source: str) -> dict:
        url = self.get_setting("chime_url")
        if not url:
            raise ActionError("invalid", "Set the chime in the plugin's settings first")
        wanted = str(self.get_setting("channels") or "all").strip()
        channels = "all" if wanted.lower() == "all" else [c.strip() for c in wanted.split(",") if c.strip()]
        result = await self.audio.announce(channels, sound=url, volume=self.get_setting("volume"))
        sound_id = (result.get("chime") or {}).get("sound_id", "")
        self.rings.insert(0, {"id": sound_id or str(len(self.rings)), "sound_id": sound_id, "door": door,
                              "at": datetime.now(timezone.utc).isoformat(), "from": source,
                              "channels": len(result.get("played") or []), "result": "playing"})
        del self.rings[20:]
        self.emit("rang", {"door": door, "channels": result.get("played") or []})
        return result

    # A page in the left menu (Plugins section)
    def get_pages(self) -> list[dict]:
        return [{"id": "rings", "title": "Doorbell", "icon": "mdi:doorbell"}]

    async def get_page(self, page_id: str) -> dict:
        return {
            "title": "Doorbell",
            "intro": "The latest rings since Sonorium started.",
            "actions": [{"action": "test", "label": "Ring now", "style": "primary"}],
            "refresh": 10,
            "sections": [{
                "id": "rings", "title": "Latest rings",
                "items": [
                    {"type": "table", "id": "rings", "empty": "No rings yet",
                     "columns": [{"key": "at", "label": "When", "format": "datetime"},
                                 {"key": "door", "label": "Door"},
                                 {"key": "channels", "label": "Channels", "format": "number"},
                                 {"key": "result", "label": "Result"}],
                     "rows": self.rings},
                    {"type": "button", "action": "clear", "label": "Clear the list", "style": "secondary",
                     "confirm": "Clear the list of rings?"},
                ],
            }],
        }

    async def page_action(self, page_id: str, action: str, data: dict) -> dict:
        if action == "test":
            try:
                result = await self.ring("Test", "Ring now")
            except ActionError as e:
                return {"success": False, "message": e.message}
            return {"success": True, "message": f"Ringing on {len(result.get('played') or [])} channels"}
        if action == "clear":
            self.rings.clear()
            return {"success": True, "message": "Cleared"}
        return await super().page_action(page_id, action, data)

What each part does:

• get_settings_schema() makes three fields under the plugin in Settings › Plugins: the chime's address, the channels, the volume.
• on_enable() subscribes to sound.ended, so the page can show how each ring ended. The subscription is removed when the plugin is switched off.
• get_triggers() declares one webhook trigger, ring, served at /api/triggers/doorbell/ring. on_trigger() runs after Sonorium has checked the caller's token and answered 202.
• ring() sends an announcement with self.audio.announce(): a chime, no text, on the channels from the settings. Announcements start a stopped channel's speakers for their length and override zones and quiet hours.
• self.emit("rang", …) publishes plugin.doorbell.rang, which other plugins and programs can follow.
• get_pages(), get_page() and page_action() add Doorbell to the left menu: a table of rings, a Ring now button and Clear the list.

Install it

1. Zip the folder, so the zip holds doorbell/manifest.json and doorbell/plugin.py.
   • Linux and macOS: zip -r doorbell.zip doorbell
   • Windows (PowerShell): Compress-Archive -Path doorbell -DestinationPath doorbell.zip
2. Open Settings in the left menu.
3. Click Plugins.
4. Click Upload plugin (top right).
5. Pick doorbell.zip. Doorbell appears under Installed, switched off.
6. Click the On switch on the Doorbell row.
7. Click the arrow at the left of the row to show its settings.
8. Type the chime's address in Chime. Any MP3 or WAV file reachable over http(s) works; Sonorium's own test chime is http://192.0.2.20:8008/api/test-tone.mp3.
9. Type the channel names in Channels (for example Kitchen, Living Room), or leave all.

From a terminal instead. Installing and configuring plugins needs an Admin, and keys can't have that permission, so these calls work while Require login is off (or with a signed-in Admin's cookie and CSRF token, see API Reference):

B=http://192.0.2.20:8008/api
curl -F "file=@doorbell.zip" $B/plugins/upload
curl -X PUT $B/plugins/doorbell/enable
curl -X PUT $B/plugins/doorbell/settings -H "Content-Type: application/json" \
     -d '{"settings": {"chime_url": "http://192.0.2.20:8008/api/test-tone.mp3", "channels": "Kitchen", "volume": 80}}'

Ring it

A trigger needs a key or a token. Either:

• a key with Run triggers (Settings › Keys › New key, see API Reference), which opens every trigger of every plugin; or
• a token for this trigger alone: POST /api/plugins/doorbell/triggers/token with {"trigger": "ring"} (Admin). It answers {"token": "…", "trigger": "ring"} once.

TOKEN=EXAMPLEtoken0000000000000000000000000000000
curl -X POST http://192.0.2.20:8008/api/triggers/doorbell/ring \
     -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" -d '{"door": "Back door"}'

{"accepted": true, "event_id": 44}

A device that can only open a URL sends the same as a GET:

http://192.0.2.20:8008/api/triggers/doorbell/ring?door=Garage&token=EXAMPLEtoken0000000000000000000000000000000

See it run

• The chime plays on the channels. A stopped channel's speakers start for it, then stop again.
• Doorbell is in the left menu, in the Plugins section. Its table lists each ring; Result changes from playing to finished when the chime has played. Ring now rings it from the page.
• The event stream shows the call, the plugin's event and the end of the chime:

event: trigger.received
data: {"id": 44, "type": "trigger.received", "origin": "sonorium", "data": {"plugin_id": "doorbell", "trigger": "ring", "payload": {"door": "Back door"}, "method": "POST"}, ...}

event: plugin.doorbell.rang
data: {"id": 45, "type": "plugin.doorbell.rang", "origin": "plugin:doorbell", "data": {"door": "Back door", "channels": ["2c62267b"]}, ...}

event: sound.ended
data: {"id": 61, "type": "sound.ended", "origin": "sonorium", "data": {"sound_id": "s_27ad7fdc1282", "channel_id": "2c62267b", "name": "Kitchen", "label": "Chime", "origin": "plugin:doorbell", "reason": "finished", "capped": false, "played": 3.0}, ...}

Note the two origins on sound.ended: the event's origin is sonorium (the mixer published it), while data.origin says who played the sound. That is why on_sound_ended() checks event.data["origin"].

Folder layout and manifest

my_plugin/
├── manifest.json      what the plugin is
├── plugin.py          the plugin class (or the file named in entry_point)
└── ...                anything else it needs: modules, images, bin/

Installed plugins live in the plugins folder of Sonorium's config folder (/config/sonorium/plugins/ on the Home Assistant app and Docker), kept across updates. A folder whose name starts with . is skipped, and so is one without plugin.py.

manifest.json:

Without manifest.json, Sonorium makes one from the folder name (version 1.0.0). Write one anyway: the version decides updates.

Import only from sonorium.plugins.* (base, actions, events, scheduler, theme_api) and, for an audio output, sonorium.core.audio_outputs (OutputType, Sink). Everything else in Sonorium is internal and changes without notice.

The plugin class

from sonorium.plugins.base import BasePlugin


class MyPlugin(BasePlugin):
    id = "my_plugin"
    name = "My Plugin"
    version = "1.0.0"
    description = "What it does, in one line."
    author = "Your name"

Sonorium creates one instance: MyPlugin(plugin_dir, settings, audio_path). Call super().__init__(*args, **kwargs) first when you define __init__.

Attributes and properties Sonorium sets:

Methods a plugin may define (all optional):

Helpers it inherits:

Everything runs on Sonorium's event loop: write async code, and run slow blocking work with await asyncio.to_thread(...) so playback controls stay responsive. Log with the standard logging module (logger sonorium); the lines appear in Settings › Logs.

Lifecycle

Time limits: an event handler 30 s, a scheduled job 5 minutes, a trigger handler 60 s, get_page() 10 s, page_action() and on_connection_changed() 30 s, test_connection() 20 s, text to speech 60 s. A handler that raises or runs over is logged and published as an error event; a page call answers 502 The plugin failed: …. Nothing a plugin does holds up playback or other plugins.

Settings

Settings are values the user sets once. Sonorium draws them under the plugin in Settings › Plugins (the arrow at the left of the row) and saves each one as it changes.

def get_settings_schema(self):
    return {
        "channel": {"type": "string", "label": "Channel", "default": ""},
        "volume": {"type": "number", "label": "Volume", "min": 0, "max": 100, "default": 40},
        "voice": {"type": "select", "label": "Voice", "options": [{"value": "en", "label": "English"}, "de"]},
    }

Every field may have label, help, default and required. A field of an unknown type is left out (and logged). PUT /api/plugins/{id}/settings with {"settings": {…}} checks each value against its field: an unknown name or a wrong value answers 400 and nothing is saved. Then on_settings_changed(settings) runs, with self.settings already updated.

get_ui_schema() can add buttons under the form; each calls handle_action(action, data) (POST /api/plugins/{id}/action with {"action", "data"}). A result with "refresh_themes": true makes Sonorium rescan the themes. New plugins usually use a page instead.

Events

Sonorium announces what happens as events: a type, a time, an origin and data. A plugin subscribes to the types it needs, like following a few accounts on a news feed.

import logging

async def on_enable(self):
    self.subscribe("channel.started", self.on_started)          # one type
    self.subscribe(["speaker.*", "error"], self.on_trouble)     # a prefix, or several

async def on_started(self, event):
    # event.type, event.data, event.time (ISO 8601, UTC), event.origin, event.id
    logging.getLogger("sonorium").info("%s started", event.data["name"])

• A pattern is a type (channel.started), a prefix (channel.*) or *.
• subscribe() returns a subscription; its close() removes it before the plugin is switched off.
• An async handler runs on the event loop; a plain function runs in a worker thread.
• Each subscription has a queue of 256 events: a slow handler falls behind and then loses the oldest. A handler that raises or runs over 30 s is logged and reported as an error event.
• event.origin is sonorium (the web UI, the API, Home Assistant, timers) or plugin:<id> for changes a plugin made through its actions. Check it to ignore your own changes. Events published by the mixer (sound.*, ramp.*) carry the requester in data.origin.

self.emit(name, data) publishes plugin.<id>.<name>; name is lowercase letters, digits, _, - and ., up to 64 characters. Other plugins can subscribe to it, and programs see it on the event stream.

The list of event types and their data is on API Reference › Event catalogue.

Actions

self.actions is what a plugin may do with channels. Each method is async, returns plain data and does what the same change in the web UI does (events, Home Assistant entities, fades). A failure raises ActionError (from sonorium.plugins.actions import ActionError) with .code (not_found, invalid, conflict, unavailable) and .message. Changes made here have the origin plugin:<id>, and they act as an Admin (zones don't refuse them).

A channel here:

{"id": "2c62267b", "name": "Kitchen", "theme_id": "df297db5-a98d-4372-8029-2c49d1e280e4", "theme_name": "Tavern",
 "preset_id": "slow_night_at_the_inn", "sequence_id": null, "playlist_id": null, "volume": 40, "is_playing": true,
 "speakers_muted": false, "announcements": true, "speakers": ["media_player.kitchen"], "speaker_group_id": null,
 "entity_slug": "kitchen"}

entity_slug is the part of the channel's Home Assistant entity ids after sonorium_.

channel = await self.actions.create_channel(theme_id="df297db5-a98d-4372-8029-2c49d1e280e4",
                                            speakers=["media_player.bedroom"], name="Wake up", volume=20)
await self.actions.start_channel(channel["id"])

Scheduler

A plugin registers jobs; Sonorium keeps them in a file, so they survive restarts, and calls the plugin when one is due. Like an alarm clock that remembers its alarms through a power cut.

async def on_enable(self):
    self.schedule("wake", "weekly", {"time": "07:00", "days": ["mon", "tue", "wed", "thu", "fri"]},
                  name="Weekday wake-up", data={"channel": "Bedroom"}, missed="skip")

async def on_schedule(self, job):
    channel = next(c for c in await self.actions.list_channels() if c["name"] == job["data"]["channel"])
    await self.actions.start_channel(channel["id"])

self.schedule(job_id, kind, spec, *, name="", data=None, missed="skip", callback=None, timezone=None) adds a job, or replaces the job with the same id, and returns it. A bad job raises ScheduleError (from sonorium.plugins.scheduler import ScheduleError).

Times without an offset are local to the time zone. On a daylight-saving change, a time in the skipped hour runs at the same clock time after the change, and a time that happens twice runs once.

The job (self.scheduled_jobs(), GET /api/plugins/{id}/schedules):

{"id": "evening", "plugin_id": "doorbell", "name": "Evening chime", "kind": "weekly",
 "spec": {"time": "18:30", "days": ["mon", "fri"]}, "data": {"door": "Kitchen"}, "missed": "skip",
 "enabled": true, "timezone": null, "created": "2026-10-12T16:21:09.246207+00:00",
 "next_run": "2026-10-12T23:30:00+00:00", "last_run": null, "last_status": null, "last_error": null}

last_status: ok, error, missed or skipped. on_schedule(job) gets these fields plus scheduled_for and missed. self.unschedule(job_id) removes a job (True if it existed). Jobs of a plugin that is off skip to their next time. Each run publishes schedule.fired; a job that raises or runs over 5 minutes is recorded in last_error and published as error.

Over HTTP (Admin): GET /api/schedules, PUT /api/schedules/settings {"timezone"}, GET|POST /api/plugins/{id}/schedules ({"id", "kind", "spec", "name", "data", "missed", "timezone", "enabled"}), DELETE /api/plugins/{id}/schedules/{job}, PUT /api/plugins/{id}/schedules/{job}/enabled {"enabled": bool}.

Webhook triggers

A trigger is a URL that a device or another system calls to make the plugin act: a motion sensor, an ESP32 button, an IR receiver, a Home Assistant automation.

def get_triggers(self):
    return [{"id": "play", "title": "Play", "methods": ["POST", "GET"],
             "description": "Play a channel. Payload: channel, volume."}]

async def on_trigger(self, trigger, payload, context):
    # context: {"method": "POST", "client": "192.0.2.31", "event_id": 44}
    ...

Sonorium serves each trigger at:

POST /api/triggers/{plugin_id}/{trigger}        JSON object or form fields = payload
GET  /api/triggers/{plugin_id}/{trigger}?k=v    query values = payload (as text)

Sonorium answers 202 {"accepted": true, "event_id": n} at once and runs on_trigger() afterwards, so a slow handler never keeps the device waiting. token is removed from the payload.

Tokens. Every call needs one of:

• a key with Run triggers (opens every trigger of every plugin);
• the plugin's token (opens all its triggers): POST /api/plugins/{id}/triggers/token with {};
• one trigger's token (opens only that one; give each device a separate token): the same with {"trigger": "play"}.

A token is 32 random bytes, shown once; making a new one replaces the old one. DELETE /api/plugins/{id}/triggers/token[?trigger=play] revokes it; GET /api/plugins/{id}/triggers lists the triggers, their URLs and which tokens are set. Send it as Authorization: Bearer …, X-Sonorium-Token: … or ?token=…. Triggers need a token or a key even while sign-in is off.

Each accepted call publishes trigger.received.

From an ESP32 (Arduino), when a button is pressed:

#include <WiFi.h>
#include <HTTPClient.h>

const char* URL = "http://192.0.2.20:8008/api/triggers/doorbell/ring";
const char* TOKEN = "EXAMPLEtoken0000000000000000000000000000000";

void ring() {
  HTTPClient http;
  http.begin(URL);
  http.addHeader("Authorization", String("Bearer ") + TOKEN);
  http.addHeader("Content-Type", "application/json");
  int code = http.POST("{\"door\": \"Front door\"}");   // 202 = accepted
  http.end();
}

From a Home Assistant automation, with a rest_command whose URL is the trigger's and whose Authorization header is Bearer and the token.

Pages

A plugin can add entries to the left menu, each with a page. The plugin describes the page as data (sections of text, fields, tables, buttons) and Sonorium draws it in the app's style, so a plugin page looks and works like the rest of the app. A plugin can't send HTML or scripts.

def get_pages(self):
    return [{"id": "alarms", "title": "Alarms", "icon": "mdi:alarm"}]

async def get_page(self, page_id):
    return {
        "title": "Alarms",
        "intro": "Plays a channel at set times.",
        "actions": [{"action": "add", "label": "Add alarm", "style": "primary"}],
        "refresh": 30,
        "sections": [{
            "id": "add", "title": "New alarm",
            "items": [
                {"type": "field", "name": "time", "field": {"type": "time", "label": "Time"}, "value": "07:00"},
                {"type": "field", "name": "days", "field": {"type": "multiselect", "label": "Days",
                                                            "options": ["mon", "tue", "wed", "thu", "fri"]}},
                {"type": "button", "action": "add", "label": "Add", "style": "primary"},
            ]}, {
            "id": "list", "title": "Alarms",
            "items": [
                {"type": "table", "id": "alarms", "empty": "No alarms yet",
                 "columns": [{"key": "time", "label": "Time"}, {"key": "days", "label": "Days"},
                             {"key": "state", "label": "State", "format": "status"}],
                 "rows": [{"id": "a1", "time": "07:00", "days": "Mon-Fri", "state": {"state": "ok", "text": "On"}}],
                 "row_actions": [{"id": "delete", "label": "Delete", "style": "danger",
                                  "confirm": "Delete this alarm?"}]},
            ]}],
    }

async def page_action(self, page_id, action, data):
    # data: {"values": {field name: value}, "row": the row's id for a row action, else None}
    if action == "add":
        ...
        return {"success": True, "message": "Alarm added"}
    if action == "delete":
        ...
        return {"success": True, "message": "Deleted"}
    return await super().page_action(page_id, action, data)

The menu entry (get_pages()): id (letters, digits, _, -; up to 64), title, icon. It appears in the Plugins section of the left menu while the plugin is on.

The page (get_page()), built fresh each time it is opened:

Items:

Field types: the settings types plus textarea, time (HH:MM), date (YYYY-MM-DD), datetime (ISO 8601) and multiselect (a list of option values).

A status column takes a state, or {"state": "ok", "text": "On"} for a chip with words. On a phone, the columns other than the first, status and ⋯ fold into a line under the first.

Icons are a name (alarm, mdi:alarm-clock) or a small inline SVG (at most 8 KB; no scripts, event attributes, links or foreignObject).

Buttons. A button or row action sends the values of every field on the page. Sonorium checks them against the fields (400 for a wrong value), and refuses an action the page doesn't show (404), before calling page_action(). The result {"success", "message", "refresh", "data"?} shows message as a notice; the page reloads unless "refresh": false.

Limits: 50 sections, 200 items per section, 20 columns, 1000 rows, 4000 characters of text. An item Sonorium doesn't know is left out.

Over HTTP (pg.use): GET /api/extensions lists every page; GET /api/plugins/{id}/pages/{page} returns one; POST /api/plugins/{id}/pages/{page}/actions/{action} with {"values", "row"} clicks a button. The Doorbell's page:

{"id": "rings", "title": "Doorbell", "icon": {"type": "name", "value": "mdi:doorbell"}, "plugin_id": "doorbell",
 "path": "/api/plugins/doorbell/pages/rings", "intro": "The latest rings since Sonorium started.", "refresh": 10,
 "actions": [{"type": "button", "action": "test", "label": "Ring now", "style": "primary", "confirm": null}],
 "sections": [{"id": "rings", "title": "Latest rings", "description": "", "items": [
   {"type": "table", "id": "rings", "columns": [{"key": "at", "label": "When", "format": "datetime"}, "..."],
    "rows": [{"id": "s_27ad7fdc1282", "door": "Back door", "at": "2026-10-12T16:18:46.621436+00:00",
              "channels": 1, "result": "finished"}], "empty": "No rings yet", "row_actions": []},
   {"type": "button", "action": "clear", "label": "Clear the list", "style": "secondary",
    "confirm": "Clear the list of rings?"}]}]}

Connections

A plugin that talks to another system (a hub, a chat bot, a cloud service) adds a row to Settings › Connection, next to Home Assistant and MQTT: its icon, name (marked Plugin), status, what it points at, Test, Edit (a dialog with the fields) and Remove. Saving needs no restart.

def get_connections(self):
    return [{"id": "hub", "title": "Hubitat", "icon": "hub", "test": True,
             "description": "The Maker API of your Hubitat hub.",
             "fields": {
                 "host": {"type": "string", "label": "Hub address", "required": True},
                 "app_id": {"type": "number", "label": "Maker API app id", "min": 1},
                 "token": {"type": "secret", "label": "Access token"},
             }}]

async def test_connection(self, connection_id, values):
    ok = await self.try_hub(values["host"], values.get("token"))
    return {"success": ok, "message": "Connected" if ok else "No answer from the hub"}

def connection_status(self, connection_id):
    return {"state": "connected", "message": "12 devices"}

async def on_connection_changed(self, connection_id, values):
    await self.reconnect(values)

connection_status() states: connected, disconnected, error, not_configured, unknown. The plugin reads the saved values, secrets included, with self.connection_values("hub").

Secret fields are write-only: the API never returns them; GET shows token_set: true instead. Saving with an empty secret keeps the saved one; "clear": ["token"] removes it. Secrets are kept apart from other values in a file only Sonorium's user can read.

Over HTTP (Admin): GET /api/plugins/{id}/connections/{conn} (fields, values without secrets, status), PUT … with {"values": {…}, "clear": [names]} (checked against the fields first), POST …/test with {"values": {…}} (the saved values with the form's unsaved ones on top).

Sounds and announcements

self.audio (capability audio.sounds) plays sounds over channels. A sound is mixed into each channel's stream, so every speaker, browser and cast device of the channel hears it. While it plays, the channel ducks (gets quieter) and comes back afterwards, the way a radio presenter talks over the music.

result = await self.audio.play(["kitchen-id", "office-id"], "https://example.com/ghost.mp3",
                               volume=90, duck={"db": 18, "attack": 0.3, "release": 2.0},
                               policy="now", label="Ghost")

await self.audio.play(channels, source=None, *, samples=None, sample_rate=None, volume=100, duck=True, policy="queue", priority=0, max_seconds=None, start_stopped=False, override=False, label="", announcement=False)

It returns {"sound_id", "length", "capped", "channels": {id: {"state", "message"}}}; a channel's state is playing, queued, starting or skipped (with the reason in message). A channel whose speakers are muted plays the sound to browsers only.

Announcements take one path, whoever sends them (Home Assistant, MQTT, POST /api/announce, a plugin): a chime and/or text spoken, ducking, admin-level, starting stopped channels.

result = await self.audio.announce("Kids Room", "Dinner is ready", sound="https://example.com/chime.mp3")

channels: ids, names (not case-sensitive), a list, or "all". sound: a URL, a path or bytes, played first. duck: True, False, a number of dB, or {"db", "attack", "release"}. It returns {"played", "skipped", "channels", "chime", "speech", "notice"}: played the channel ids it plays on, skipped those with Announcements off, chime and speech the play results, notice why the text wasn't spoken when only the chime played. Text with no text to speech and no sound raises ActionError (unavailable); an unknown channel name not_found.

Each channel has an Announcements switch (on by default; PUT /api/sessions/{id}/announcements, and switch.sonorium_<channel>_announcements in Home Assistant). With it off, no announcement plays there.

Gates. A plugin that owns channels (zones) can refuse sounds that aren't admin-level. One gate per plugin:

def gate(channel, request):
    # channel: the channel (channel.id, channel.name); request: sound_id, label, origin, priority
    return "Quiet hours" if self.quiet(channel.id) else None     # None allows it

self.audio.add_gate(gate)

A sound with override=True skips every gate.

Timing. The sound is exact to the stream. Network speakers (Cast, Sonos, AirPlay, DLNA) buffer a few seconds, so they play it that much later: fine for an announcement, noticeable for an effect that has to match a motion sensor. Audio outputs and browsers are close to instant.

Text to speech

A provider turns text into audio for announcements (capability audio.tts). The built-in Announcements plugin provides Home Assistant's (homeassistant), ready while Home Assistant is linked. The first provider registered is the default; an announcement may name one.

async def on_enable(self):
    self.audio.register_tts("piper", "Piper (local)", self.speak, ready=lambda: True)

async def speak(self, message, language=None, **options) -> bytes:
    return wav_or_mp3_bytes

Providers are removed when the plugin is switched off. GET /api/plugin-framework lists them as tts_providers.

Volume ramps

A ramp makes a channel louder or quieter gradually (capability audio.ramps): an alarm that rises over 20 minutes, a sleep timer that fades out over an hour. The gain is worked out for each sample, so it is smooth. It applies to the channel's mix: announcements play at their own level, and the speakers' volume stays as it is (set it with set_volume(), then ramp within it).

await self.ramps.start("2c62267b", to=1.0, seconds=1200, from_level=0.0, curve="db", start=True)  # wake-up
await self.ramps.start("2c62267b", to=0.0, seconds=3600, then="stop")                             # sleep timer
await self.ramps.cancel("2c62267b", reset=False)                                                  # hold where it is
await self.ramps.status("2c62267b")      # {"channel_id", "level", "ramp": {...} or None}

await self.ramps.start(channel, to, seconds, *, from_level=None, curve="db", then=None, start=False)

It returns {"ramp_id", "channel_id", "from", "to", "seconds", "curve", "then"}. cancel(channel, reset=False) holds the level reached (reset=True: back to 1) and returns {"ramp_id", "level", "reset"} or None. A finished ramp holds its level until the channel stops or another ramp starts. Events: ramp.started, ramp.finished, ramp.cancelled.

Audio outputs

A channel normally goes to speakers that pull its stream over the network. An output type sends it somewhere else (capability audio.outputs): the sound card of the machine Sonorium runs on, a Bluetooth speaker, a file. The plugin registers the type; Sonorium asks it for its devices, and for a sink when a channel is sent to one.

from sonorium.core.audio_outputs import OutputType, Sink
from sonorium.plugins.base import BasePlugin


class HostAudio(OutputType):
    id = "host_audio"
    title = "This computer"

    def available(self):
        try:
            import sounddevice
            sounddevice.query_devices()
            return True, ""
        except Exception as e:
            return False, f"No sound card here ({e})"

    async def discover(self):
        import sounddevice
        return [{"id": str(i), "name": d["name"]} for i, d in enumerate(sounddevice.query_devices())
                if d["max_output_channels"] > 0]

    def open(self, device_id, sample_rate, options):
        return HostSink(int(device_id), sample_rate)


class HostSink(Sink):
    latency = 0.1

    def __init__(self, device, rate):
        import sounddevice
        self.stream = sounddevice.OutputStream(device=device, samplerate=rate, channels=1, dtype="int16")
        self.stream.start()

    def write(self, block):            # int16, shape (1, n), mono, at sample_rate
        self.stream.write(block.reshape(-1, 1))     # may block: it runs in a separate thread

    def close(self):
        self.stream.stop()
        self.stream.close()


class HostAudioPlugin(BasePlugin):
    id = "host_audio"
    name = "Host audio"
    version = "1.0.0"

    async def on_enable(self):
        self.audio_outputs.register_type(HostAudio())

self.audio_outputs also has unregister_types(), await types(), await devices(type_id), await attach(channel, type_id, device_id, options=None), await detach(channel, attachment_id) and await attachments(channel=None). Types are removed when the plugin is switched off.

sounddevice isn't part of every install: the plugin brings it with "requirements": ["sounddevice"]. Where host audio can work: Windows, Linux and macOS installs; Docker on Linux with the sound device (/dev/snd) or the host's sound server socket passed in; not the Home Assistant app.

A user sends a channel to a device with POST /api/sessions/{id}/audio-outputs {"type", "device", "options"}. A channel's outputs are remembered and start again when it plays. Each reads the channel's stream like a speaker: it keeps the channel from going idle and follows the speakers-muted switch.

Command maps

An input plugin turns what it receives (an IR code, a button on an RF remote, an MQTT message) into a code and hands it over (capability commands). The user decides what each code does in a command map: a list of codes, each with an action, a channel and its arguments.

def get_command_maps(self):
    return [{"id": "remote", "title": "TV remote",
             "codes": [{"code": "KEY_VOLUMEUP", "label": "Volume +"}]}]     # known codes, optional

async def on_ir(self, code):
    await self.commands.received("remote", code)

The plugin shows the map on its page with {"type": "command_map", "map": "remote"}. There, Learn waits for the next code the plugin receives (press the button, up to 60 s), and Add adds one by hand. An entry:

{"code": "KEY_VOLUMEUP", "label": "Volume +", "action": "channel.volume_up",
 "channel": "2c62267b", "args": {"step": 5}, "enabled": true}

The actions run through the plugin's actions, so they behave like the same change in the UI. Every code received publishes command.received. Over HTTP (Admin): GET /api/command-actions, GET /api/plugins/{id}/command-maps, PUT …/command-maps/{map} {"entries": [...]}, POST …/learn, POST …/test {"code"}.

MQTT

The broker Sonorium is connected to (capability mqtt): the Mosquitto app on Home Assistant, or the one set in Settings › Connection.

async def on_enable(self):
    self.mqtt.subscribe("zigbee2mqtt/front_door", self.on_door)     # "+" and "#" wildcards

async def on_door(self, topic, payload):
    if payload.get("contact") is False:
        await self.ring("Front door", "MQTT")

await self.mqtt.publish("doorbell/state", {"last_ring": "Front door"}, retain=True)   # a dict is sent as JSON

Messages reach Sonorium's own entities first, then plugins. Subscriptions are removed when the plugin is switched off.

Home Assistant

Calls to Home Assistant's REST API through the connection Sonorium already has (capability homeassistant): the Supervisor on the Home Assistant app, the saved URL and token on Docker and Windows. The token never reaches the plugin, and only Home Assistant's /api/... paths are allowed.

if self.homeassistant.available:
    status, states = await self.homeassistant.request("GET", "/api/states")
    audio = await self.homeassistant.tts("Dinner is ready", engine="tts.piper")

Failures raise ActionError (unavailable).

Themes from a plugin

A plugin that creates a theme or adds to one (an importer, a generator) goes through ThemeEdit, so it never writes metadata.json or presets.json itself. Nothing is written before commit(); a failure leaves the themes as they were.

from pathlib import Path

edit = self.new_theme("Rainy Tavern", description="Rain on the windows of a busy inn", icon="🌧️")
fire = edit.add_file(downloaded_bytes, "Fireplace.mp3", settings={"playback_mode": "continuous"})
edit.add_file(Path("/tmp/door.wav"), group="Doors", settings={"playback_mode": "sparse", "presence": 0.4})
edit.set_group("Doors", {"volume": 0.8})
edit.add_preset("Quiet night", {fire: {"volume": 0.5}})
edit.add_extra_file("ATTRIBUTION.md", "Sounds by ...")
result = edit.commit()     # {"theme_id", "folder", "name", "is_new", "added"}

Track settings: playback_mode (continuous Background, sparse Intermittent, presence Ebb & Flow, unassigned), volume, presence, volume_low, rise (0-1), muted, seamless_loop, exclusive, pause_between (booleans), fade_in, fade_out, pause_min, pause_max (seconds). The theme format is in docs/THEME_FORMAT.md. A plugin action that changed themes returns "refresh_themes": true so Sonorium rescans them.

Python packages and programs

Python packages (capability packages.pip): "requirements": ["pyserial==3.5"] in the manifest, up to 30, each a package name with optional extras and version conditions (no URLs, paths or pip options).

• Off until an Admin switches on Settings › Advanced › Plugin packages (then restarts). A package comes from the internet and runs with Sonorium's rights, like the plugin. A plugin that needs packages while the switch is off is refused with that sentence.
• Installed with pip, wheels only (nothing is built and no setup script runs), into plugin_packages/<plugin id>/ in the config folder, once, and again when the requirements or Python change. Removed with the plugin. GET /api/plugins/{id}/packages shows their state.
• That folder is added to the end of Python's search path: Sonorium's packages always win.

Programs (capability packages.binaries), such as ffmpeg:

"binaries": {"ffmpeg": {
    "linux-x86_64":  {"path": "bin/linux-x86_64/ffmpeg",  "sha256": "…"},
    "linux-aarch64": {"path": "bin/linux-aarch64/ffmpeg", "sha256": "…"},
    "windows-amd64": {"path": "bin/windows-amd64/ffmpeg.exe", "sha256": "…"}}}

Each file is checked when the plugin loads: it must be inside the plugin's folder (no links) and match its SHA-256; a mismatch refuses the plugin. self.binary("ffmpeg") is the path for this platform, or None. Platforms: linux-x86_64, linux-aarch64, windows-amd64, darwin-arm64, … On the Home Assistant app, build for musl (or statically).

API version and capabilities

The plugin API has a version, MAJOR.MINOR; Sonorium 1.7.1 has 1.1. Adding a service or a field raises the minor version: a plugin written for 1.0 runs on 1.1 and every later 1.x, because nothing it uses is renamed or removed. The major version goes up only for a change old plugins would notice.

Sonorium reads api_version and requires before it loads any of the plugin's code. A plugin that needs a newer API, another major version, or a capability this install doesn't offer is refused with a sentence that says why, for example:

Remote needs plugin API 1.3; this Sonorium has 1.1. Update Sonorium to use it.

A refused plugin is listed in Settings › Plugins with that sentence (compatible: false and problem in GET /api/plugins), the web UI shows an alert, and none of its code runs.

GET /api/plugin-framework gives the version, every capability and those this install offers now (offered). Each 1.1 service is a facade on the plugin (self.audio, self.ramps, self.audio_outputs, self.commands, self.mqtt, self.homeassistant); every method that can fail raises ActionError, as self.actions does.

Packaging, installing and updating

A zip. Zip the plugin folder so the zip holds my_plugin/plugin.py and my_plugin/manifest.json (a zip with the files at its top works too; the id then comes from the manifest, or the zip's name). Sonorium uses the first plugin.py it finds.

Upload (an Admin):

1. Open Settings in the left menu.
2. Click Plugins.
3. Click Upload plugin.
4. Pick the zip.
5. Switch the plugin on with its On switch.

⋯ › Plugin requirements on that page lists what a zip has to contain. Or POST /api/plugins/upload (multipart, field file). The upload is refused (400) without a version in the manifest or class, with a version that isn't MAJOR.MINOR.PATCH, with a path outside the folder, or without plugin.py. A plugin with the same id is replaced; its settings, jobs, tokens and connection values are kept. The answer has the plugin as GET /api/plugins/{id} shows it, or "status": "warning" when the files were installed but the plugin didn't load (the log says why).

Uninstall: ⋯ › Uninstall… on its row, or DELETE /api/plugins/{id}.

The Catalog. Settings › Plugins › Catalog lists the plugins in plugins/catalog.json of the Sonorium repository, each with its zip next to it:

{"version": 1, "updated": "2026-10-11",
 "plugins": [{"id": "theme_merge", "name": "Theme Merge",
              "description": "Merge two themes together - combines tracks, groups and presets",
              "version": "2.0.0", "author": "Sonorium", "category": "utilities", "zip_file": "theme_merge_v2.0.0.zip"}]}

Install downloads and installs the zip; Update appears when the Catalog's version is newer than the installed one, so raise version for every release. The zip is fetched from the plugins folder next to catalog.json. Built-in plugins that were uninstalled are listed there too, to install again.

Storage and security

• Plugins run inside Sonorium with its rights: they can read its files and the network. Install plugins you trust, as you would an add-on.
• Settings are in state.json; jobs, plain connection values, the scheduler's time zone, command maps and audio output choices in plugin_framework.json; token hashes and secret connection values in plugin_secrets.json (readable only by the user Sonorium runs as). All in Sonorium's config folder, kept across updates. Keep anything else the plugin needs to remember in a file in the config folder, or in memory as the Doorbell does.
• Tokens (triggers, the event stream) are 32 random bytes. Sonorium keeps only their SHA-256 hash and compares in constant time; a wrong token and a missing plugin get the same answer. token= is masked in the log.
• The management routes (installing, settings, tokens, jobs, connections) need an Admin. The trigger and event stream routes are the ones meant for other devices, and they need a token or a key.
• A plugin never logs secrets or tokens; connection_values() is for the plugin's use only.

Testing

The quickest test is the Doorbell's: install the zip on a test Sonorium, switch it on, call its trigger and page with curl, and follow GET /api/events/stream?types=* while you do. Errors in a handler appear as error events and in Settings › Logs.

The repository's tests load plugins without Home Assistant or speakers; copy the pattern of tests/test_plugin_framework_api.py:

1. Build a session manager with fake speakers (build() in tests/test_sequences_playlists.py).
2. Make a PluginHost (with a fresh EventBus) and a PluginManager(state_store, plugins_dir=..., host=host), with the plugin folder copied into plugins_dir.
3. Mount create_plugin_framework_router(manager, host) in a FastAPI app and use TestClient as a context manager; run async calls with client.portal.call(...) (manager.initialize, host.start, manager.enable_plugin).
4. Call the triggers, pages and actions over HTTP and check the events in bus.history() and the channels' state.

Run the suite with python -m pytest -q tests.