Skip to content

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. Wire into a Service

Use SingleSensorService for the common case of a single plugin, or subclass SensorService directly for full control:

from tolomeo.services import SensorService
from .plugin import TemperaturePlugin


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

    async def setup(self) -> None:
        await super().setup()
        plugin = self._plugin_class(logger=self._logger)
        async with self.params_sync.condition:
            await plugin.attach()      # attach inside the condition lock — matches SingleSensorService
            self.plugins[plugin.id] = plugin
            self.params_sync.is_ready = True
            self.params_sync.condition.notify_all()

plugin.attach() placement

Calling attach() inside the condition block (as shown above) ensures no consumer can observe is_ready = True before the plugin is fully set up. If connect or setup hooks are long-running and do not need the lock, you may call attach() before acquiring it — see the Build Your First Service tutorial for that alternative pattern.

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}]

See also