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.

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 SingleSensorService for sensor-style services that publish metrics as plugins push them.

Create my_service/service.py:

from tolomeo.services import SingleSensorService
from .plugin import DummyPlugin


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

That is the whole service. SingleSensorService acquires the plugin, wires its data and info callbacks, and wakes the params loop once it is registered.

from tolomeo.services import SensorService

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

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

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

Declaring a Command Payload¶

A command receives the decoded message body as context.payload, a plain dict. To have the payload checked before the command runs, and its shape announced to clients, declare a model. Subclass CommandPayload and set it as the command's Meta.payload_model:

from tolomeo.commands import CommandPayload, PluginCmd, PluginCmdContext


class SetIntervalPayload(CommandPayload):
    seconds: int = 1


class SetIntervalCmd(PluginCmd):
    class Meta:
        payload_model = SetIntervalPayload

    @classmethod
    async def execute(cls, context: PluginCmdContext) -> dict:
        params = cls.params_of(context)
        return {"seconds": params.seconds}

Register it in the plugin's Meta.commands as usual. Before execute runs, the payload is validated and the result is stored in context.params; cls.params_of(context) returns it. context.payload stays the raw dict.

• CommandPayload ignores unknown fields, so a client that sends extra keys keeps working. Fields without a default are required.
• A plugin command's model never declares id: the framework removes it to find the plugin and adds it to the announced schema.
• A command without a Meta.payload_model is not validated, so existing commands keep working.
• A subclass inherits its parent's model unless its own Meta sets another.
• Use CommandStr for string fields that should also accept JSON numbers and booleans: they are stringified, so 123 becomes "123".

A payload that fails validation is answered with an error record that names each offending field, for example Invalid payload for 'SetIntervalCmd': seconds: Input should be a valid integer, unable to parse string as an integer. To reply in the command's own format instead, override on_invalid_payload:

from tolomeo.exceptions import InvalidPayloadError


class SetIntervalCmd(PluginCmd):
    class Meta:
        payload_model = SetIntervalPayload

    @classmethod
    async def on_invalid_payload(cls, context: PluginCmdContext, error: InvalidPayloadError) -> dict:
        return {"result": "error", "error": str(error)}

    ...

The service also announces the schema of every command, so a frontend can build a form for it; see SenML Message Format.

Next Steps¶

• Add OTA support to your service
• Implement a sensor plugin — deeper dive into metric types and connection management