Skip to content

Build Your First ToloMEO Microservice

This tutorial walks through creating a minimal sensor microservice using Py ToloMEO. You will build a plugin that generates data, a service that connects it to NATS, and run the whole thing locally.

What you will do

Build a service called dummy that publishes a random metric value to events.data every second over NATS.

Source reference: examples/dummy/

Prerequisites

  • Python ≥ 3.10
  • uv installed
  • A running NATS server (nats-server on nats://localhost:4222)
  • Py ToloMEO installed: uv add tolomeo

1. Create the Plugin

A plugin wraps your hardware or data source. Subclass SensorPlugin to get built-in state management and metric registration; it is push-only — there is no internal queue or polling loop, so a plugin pushes each reading directly via await self.push_reading({...}).

Create my_service/plugin.py:

import asyncio
import time
from random import random
from typing import Dict

from tolomeo.commands.plugin import PluginCmd, PluginCmdContext
from tolomeo.metrics import Metric
from tolomeo.plugins import SensorPlugin


class PingCmd(PluginCmd):
    """A simple command that returns the plugin ID."""

    @classmethod
    async def execute(cls, context: PluginCmdContext) -> Dict:
        return {"message": f"pong from {context.plugin.id}"}


class DummyPlugin(SensorPlugin):
    class Meta:
        commands = [PingCmd]
        out_metrics = [Metric("dummy_metric", "")]

    async def connect(self) -> bool:
        self.id = "DummyPlugin"         # must be set inside connect()
        self._logger.info("Connected")
        return True

    async def disconnect(self) -> bool:
        self._logger.info("Disconnected")
        return True

    async def after_setup(self) -> None:
        # Start a background task that generates data
        await self.task_manager.add_task("simulate", self._simulate)

    async def _simulate(self) -> None:
        while True:
            await asyncio.sleep(1)
            await self.push_reading({
                "timestamp": round(time.time()),
                f"{self.id}:dummy_metric": random(),
            })

2. Create the Service

A service manages one or more plugins and handles NATS subscriptions. Subclass SensorService for sensor-style services that publish metrics as plugins push them.

Create my_service/service.py:

from tolomeo.services import SensorService
from .plugin import DummyPlugin


class DummyService(SensorService):
    class Meta:
        plugin_class = DummyPlugin

    async def setup(self) -> None:
        await super().setup()

        plugin = self._plugin_class(logger=self._logger)
        await plugin.attach()          # attach before acquiring the condition lock
        self.plugins[plugin.id] = plugin

        async with self.params_sync.condition:
            self.params_sync.is_ready = True
            self.params_sync.condition.notify_all()

plugin.attach() placement

In this example attach() is called before acquiring the params_sync condition lock, which is safe when the plugin's connect and setup hooks do not need the lock held. The library's built-in SingleSensorService calls attach() inside the condition block instead — use that pattern when you need the lock held during setup to avoid a wakeup race.

3. Wire the Entry Point

Create my_service/main.py:

import asyncio
import logging
from .service import DummyService

logging.basicConfig(level=logging.INFO)


async def main() -> None:
    service = DummyService("dummy")
    try:
        await service.run()
    except asyncio.CancelledError:
        pass


if __name__ == "__main__":
    asyncio.run(main())

4. Run the Service

Start the service:

DEVICE_SERIAL_NUMBER=test001 python -m my_service.main

In a separate terminal, subscribe to NATS to observe the output:

nats sub "events.data"

You should see SenML records arriving every second:

[{"bn": "urn:cpt:device:sn:test001:", "n": "DummyPlugin:dummy_metric", "v": 0.723, "t": 1700000010.0}]

5. Send a Command

nats pub commands.dummy.req \
  '[{"n":"PingCmd","vs":"{\"id\":\"DummyPlugin\"}"}]'

The response appears on events.params:

[{"bn": "urn:cpt:device:sn:test001:", "n": "PingCmd", "vs": "{\"message\": \"pong from DummyPlugin\"}"}]

Next Steps