Implement a Sensor Plugin¶

This guide shows how to implement a sensor plugin using SensorPlugin as the base class.

When to Use SensorPlugin¶

Use SensorPlugin when your plugin:

Reads data from a physical or simulated sensor
Publishes numeric, boolean, or string metrics at regular intervals
May need to reconnect after an unexpected disconnect

Use PluginBase directly when your plugin handles commands but does not produce continuous metric data — see Implement a Custom Plugin.

1. Define the Plugin Class¶

Declare your output metrics in Meta.out_metrics, implement connect (setting self.id) and disconnect:

import asyncio
import time
from tolomeo.metrics import Metric, MetricType, MetricDataType
from tolomeo.plugins import SensorPlugin


class TemperaturePlugin(SensorPlugin):
    class Meta:
        out_metrics = [
            Metric(
                name="temperature",
                unit="°C",
                type=MetricType.TEMPORAL,
                data_type=MetricDataType.NUMBER,
            )
        ]

    async def connect(self) -> bool:
        self.id = "temperature_sensor"   # must be set here
        # open serial port, BLE connection, etc.
        return True

    async def disconnect(self) -> bool:
        # close connection
        return True

2. Push Data¶

SensorPlugin is push-only. Push each reading directly with await self.push_reading(...) from a background task started in after_setup. The dict must contain a timestamp key plus values keyed by the namespaced metric name (f"{self.id}:{metric.name}"):

    async def after_setup(self) -> None:
        await self.task_manager.add_task("read_sensor", self._read_loop)

    async def _read_loop(self) -> None:
        while True:
            temperature = await self._hw.read_celsius()
            await self.push_reading({
                "timestamp": round(time.time()),
                f"{self.id}:temperature": temperature,
            })
            await asyncio.sleep(1)

3. Handle Reconnects (Optional)¶

Register a callback on on_disconnect to be notified of unexpected disconnects:

    async def after_setup(self) -> None:
        self.on_disconnect = self._handle_disconnect
        await self.task_manager.add_task("read_sensor", self._read_loop)

    async def _handle_disconnect(self) -> None:
        from tolomeo.plugins.sensor import SensorState
        self._logger.warning("Sensor disconnected — attempting reconnect")
        self.state = SensorState.DISCONNECTED
        # implement reconnection logic using self.conn_context.reconnect_policy

4. Make the Cadence Configurable (Optional)¶

Hardcoding the reading interval means a deployment cannot change it. Declare a model in Meta.config_model and read self.options instead:

from pydantic import BaseModel, ConfigDict, PositiveFloat


class TemperatureOptions(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)

    interval_s: PositiveFloat = 1.0


class TemperaturePlugin(SensorPlugin):
    # ... (connect, disconnect, after_setup unchanged from sections 1-2; the
    # optional reconnect handling from section 3, if you added it, is unchanged too)

    class Meta:
        out_metrics = [...]              # unchanged from section 1
        config_model = TemperatureOptions  # carried over from section 2, plus config_model

    async def _read_loop(self) -> None:
        while True:
            temperature = await self._hw.read_celsius()
            await self.push_reading({
                "timestamp": round(time.time()),
                f"{self.id}:temperature": temperature,
            })
            await asyncio.sleep(self.options.interval_s)

_read_loop here replaces section 2's version; after_setup still schedules it under the same task name, so the loop keeps running, now sleeping for the configured interval_s instead of a hardcoded one second.

[plugins.temperature_sensor]
interval_s = 5.0

Keep the cadence in the plugin loop rather than inside the sensor object: a sensor that sleeps for its own polling interval cannot be reconfigured without editing it. PositiveFloat also rules out a zero that would spin. See the Configuration Reference.

5. Wire into a Service¶

Use SingleSensorService for the common case of a single plugin — it acquires the plugin and signals the params loop for you:

from tolomeo.services import SingleSensorService
from .plugin import TemperaturePlugin


class TemperatureService(SingleSensorService):
    class Meta:
        plugin_class = TemperaturePlugin

Subclass SensorService and override acquire_plugins when you need more control — several plugins, or discovery:

class MultiSensorService(SensorService):
    class Meta:
        plugin_class = TemperaturePlugin

    async def acquire_plugins(self) -> None:
        if await self.attach_plugin(self.new_plugin()):
            await self.notify_params_changed()

attach_plugin wires the plugin's callbacks (SensorService.wire_plugin binds async_data_handler and async_info_handler), attaches it, and registers it only if the attach succeeded. notify_params_changed then wakes params_sync_loop so the new plugin's params get published — call it after registration, since the loop reads the registry as soon as it wakes.

SensorService declares no plugin_class

SensorPlugin is abstract, so there is no useful default to inherit. Every sensor service names its own concrete plugin; forgetting to raises a RuntimeError naming Meta.plugin_class at construction rather than failing later.

BLE Sensors¶

BLE sensors use BLESensorPlugin, which adds Bleak connection management, GATT characteristic discovery and an FSM-backed state on top of the sensor data path. It is abstract: every BLE device speaks its own wire format, so parse_data has no default and each device needs its own plugin.

from typing import Dict
from tolomeo.plugins.ble import BLESensorPlugin


class ThermoBeaconPlugin(BLESensorPlugin):
    class Meta:
        out_metrics = [Metric("temperature", "Cel")]

    def parse_data(self, data: bytearray) -> Dict:
        raw = int.from_bytes(data[0:2], "little", signed=True)
        return {
            "timestamp": round(time.time()),
            f"{self.id}:temperature": raw / 100.0,
        }

The service side needs nothing but the plugin name — BLESensorService already handles discovery by address, registration and persistence, and its ConnectDevice / DisconnectDevice commands work unchanged:

class ThermoBeaconService(BLESensorService):
    class Meta:
        plugin_class = ThermoBeaconPlugin

A BLE plugin's id is the device address

connect sets self.id from the connected client's address, so a configuration table for a BLE plugin is keyed by the address and must be quoted in TOML:

[plugins."AA:BB:CC:DD:EE:FF"]
notify_interval_s = 2.0

All registered devices share the one plugin_class, so a deployment mixing two sensor models is not expressible yet.

Devices to discover at startup come from .devices.yml by default. Override new_config_manager to keep that list elsewhere — examples/ble_mock returns an in-memory manager so it runs with no config file and no BLE adapter. That example is the fastest way to see the whole flow: python -m examples.ble_mock.demo drives connect → start acquisition → readings → stop → disconnect against a stubbed broker.

Published NATS Output¶

Metrics are published to events.data as SenML records:

[{"bn": "urn:cpt:device:sn:abc123:", "n": "temperature_sensor:temperature", "u": "°C", "v": 23.5, "t": 1700000010.0}]