Plugin Development

Plugin Architecture

casectl plugins are Python packages that implement a simple Protocol interface. No inheritance required – if your class has the right methods, it’s a plugin (Python calls this structural subtyping, via typing.Protocol).

Plugins are discovered automatically via Python entry points, loaded by the PluginHost, and managed through the daemon lifecycle.

The Protocol Interface

Your plugin must define four class attributes and implement four methods:

from casectl.plugins.base import CasePlugin, PluginContext, PluginStatus

class MyPlugin:
    """No inheritance needed -- just implement the Protocol."""

    name = "my-plugin"
    version = "0.1.0"
    description = "What my plugin does"
    min_daemon_version = "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        """Called once during plugin initialisation.
        Use ctx to register routes, subscribe to events, access hardware."""
        self._ctx = ctx
        self._hardware = ctx.get_hardware()

    async def start(self) -> None:
        """Called when the daemon starts.
        Launch background tasks, begin your plugin's main work."""
        pass

    async def stop(self) -> None:
        """Called when the daemon shuts down.
        Cancel tasks, clean up resources."""
        pass

    def get_status(self) -> dict[str, Any]:
        """Return current plugin status for health checks."""
        return {"status": PluginStatus.HEALTHY}

PluginContext

The PluginContext provides your plugin with access to the daemon’s subsystems:

Method	Return	Purpose
`register_routes(router)`	`None`	Register a FastAPI router (auto-mounted at `/api/plugins/<name>/`)
`register_config(schema)`	`None`	Register a Pydantic config schema for your plugin
`register_commands(group)`	`None`	Register a Click command group for the CLI
`await get_config()`	`dict`	Get your plugin’s config section (`plugins.<name>`)
`get_hardware()`	`HardwareRegistry`	Access expansion board, OLED, system info
`on_event(event, handler)`	`None`	Subscribe to an event
`emit_event(event, data)`	`None`	Publish an event (sync, fire-and-forget)
`logger`	`Logger`	Per-plugin logger instance

Adding API Routes

Plugins can register their own API endpoints. The daemon auto-mounts your router at /api/plugins/<plugin-name>/ – don’t add a prefix yourself:

from fastapi import APIRouter

class MyPlugin:
    async def setup(self, ctx: PluginContext) -> None:
        self._ctx = ctx

        router = APIRouter(tags=["my-plugin"])

        @router.get("/status")
        async def status():
            return self.get_status()

        @router.post("/action")
        async def do_action():
            return {"result": "done"}

        ctx.register_routes(router)

The routes above become /api/plugins/my-plugin/status and /api/plugins/my-plugin/action.

Hardware Access

Access hardware through the HardwareRegistry returned by ctx.get_hardware():

hw = ctx.get_hardware()

# Properties (may be None if hardware is absent):
hw.expansion    # ExpansionBoard (I2C 0x21) -- fan duty, LED, case temp
hw.oled         # OledDevice (I2C 0x3C) -- SSD1306 display
hw.system_info  # SystemInfo -- CPU temp, usage, memory, disk, IP

All I2C calls must go through asyncio.to_thread() to avoid blocking the event loop:

temp = await asyncio.to_thread(hw.system_info.get_cpu_temperature)

Event Bus

Subscribing to events

async def setup(self, ctx: PluginContext) -> None:
    ctx.on_event("metrics_updated", self._on_metrics)

async def _on_metrics(self, data: dict) -> None:
    cpu_temp = data.get("cpu_temp", 0.0)
    # react to new metrics...

Publishing events

ctx.emit_event("my-plugin.alert", {
    "message": "Temperature exceeded threshold",
    "value": 85.0,
})

Events are broadcast to all subscribers and also pushed to WebSocket clients.

Available events

Event	Source	Payload	Frequency
`metrics_updated`	system-monitor	CPU, memory, disk, temps, fan duty	Every 2s
`daemon.started`	daemon	`None`	Once at startup
`daemon.stopping`	daemon	`None`	Once at shutdown

Entry Point Registration

[project.entry-points."casectl.plugins"]
my-plugin = "my_package.plugin:MyPlugin"

The key (my-plugin) becomes the plugin’s route prefix. The name class attribute is the canonical runtime identifier.

Example: Temperature Logger Plugin

A complete example that logs temperature readings:

import asyncio
import logging
from pathlib import Path
from datetime import datetime
from typing import Any

from casectl.plugins.base import PluginContext, PluginStatus

class TempLoggerPlugin:
    name = "temp-logger"
    version = "0.1.0"
    description = "Logs CPU temperature to a file"
    min_daemon_version = "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        self._ctx = ctx
        self._hw = ctx.get_hardware()
        self._log_path = Path.home() / ".config" / "casectl" / "temp.log"
        self._task: asyncio.Task | None = None

    async def start(self) -> None:
        self._task = asyncio.create_task(self._log_loop())

    async def stop(self) -> None:
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    def get_status(self) -> dict[str, Any]:
        return {
            "status": PluginStatus.HEALTHY if self._task else PluginStatus.STOPPED,
            "log_file": str(self._log_path),
        }

    async def _log_loop(self) -> None:
        while True:
            try:
                si = self._hw.system_info
                if si is not None:
                    temp = await asyncio.to_thread(si.get_cpu_temperature)
                    timestamp = datetime.now().isoformat()
                    with open(self._log_path, "a") as f:
                        f.write(f"{timestamp},{temp}\n")
            except Exception as exc:
                logging.getLogger(__name__).warning("Failed to log temp: %s", exc)
            await asyncio.sleep(60)

[project.entry-points."casectl.plugins"]
temp-logger = "my_package.temp_logger:TempLoggerPlugin"

pip install -e .
# Restart the daemon -- the plugin appears automatically

Testing Without Hardware

casectl’s test fixtures mock the I2C bus. Your plugin tests can do the same:

import pytest
from unittest.mock import MagicMock, AsyncMock
from casectl.plugins.base import PluginContext
from casectl.hardware import HardwareRegistry

@pytest.fixture
def mock_context():
    hw = HardwareRegistry(expansion=None, oled=None, system_info=MagicMock())
    ctx = PluginContext(
        plugin_name="temp-logger",
        hardware_registry=hw,
        config_manager=AsyncMock(),
        event_bus=MagicMock(),
    )
    return ctx

@pytest.mark.asyncio
async def test_plugin_setup(mock_context):
    plugin = TempLoggerPlugin()
    await plugin.setup(mock_context)
    assert plugin._hw is not None

def test_plugin_status_before_start(mock_context):
    plugin = TempLoggerPlugin()
    # Before setup, task is None
    plugin._task = None
    status = plugin.get_status()
    assert "stopped" in str(status["status"]).lower()

Run with: pytest

Built-in Plugins

casectl ships with 8 built-in plugins. These serve as both core features and reference implementations for plugin developers:

Plugin	Description
`fan-control`	Temperature-curve PWM control with follow-temp/follow-rpi/manual/custom/off modes
`led-control`	Rainbow, breathing, follow-temp, manual colour (16 named + hex), off
`oled-display`	4 cycling screens (clock, metrics, temperature, fan duty) with power control
`system-monitor`	Periodic metrics collection and event bus emission
`prometheus`	`/metrics` endpoint in Prometheus text format
`mqtt`	Bidirectional MQTT with Home Assistant auto-discovery, state publishing, and command handling
`automation`	Event-driven rules engine with conditions, actions, priority classes, and cooldown
`alerting`	Alert dispatch via webhook, ntfy.sh, and SMTP channels

The automation plugin can trigger actions on any other plugin. For example, a rule can monitor metrics_updated events and, when CPU temperature exceeds a threshold, set the fan to full speed via the fan target and send a notification via the alert target.

Community Plugins

casectl ships with a complete example plugin at examples/plugins/casectl-example-plugin/ in the repository. Use it as a starting point – it includes working code, a full README with API documentation, and test examples.