Skip to content

Add OTA Update Support

This tutorial extends a running Py ToloMEO service with OTA update capability. By the end you will have a service that accepts OTA commands, manages a download task, and transitions through the OTA state machine.

What you will do

Build a service with an OTAPlugin that responds to NotifyOTAUpdate, StartOTADownload, and InstallOTAUpdate commands, manages a download task, and transitions through the OTA state machine.

Source reference: examples/ota_mock/

Prerequisites

1. Subclass OTAPlugin

OTAPlugin handles the download lifecycle automatically. You only need to implement install_update (performs the actual firmware flash) and monitor_update (tracks progress).

Create my_service/ota_plugin.py:

import asyncio
import logging
from tolomeo.plugins.ota import OTAPlugin


class MyOTAPlugin(OTAPlugin):

    async def connect(self) -> bool:
        self.id = "swupdate"
        self._logger.info("OTA plugin connected")
        return True

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

    async def install_update(self) -> None:
        """Trigger the actual firmware install and start monitoring."""
        self._logger.info("Starting installation from %s", self.fsm.pending_update.download_path)
        self.fsm.install()

        # Start a task to monitor progress
        await self.task_manager.add_task("monitor_update", self.monitor_update, auto_restart=False)

    async def monitor_update(self) -> None:
        """Simulate or perform the installation and report success/failure."""
        self._logger.info("Monitoring update in state: %s", self.state)

        # Replace the sleep + fsm.update_success() with real install logic
        await asyncio.sleep(5)
        self.fsm.update_success()
        self._logger.info("Installation succeeded")

2. Wire the Plugin into a Service

Create my_service/ota_service.py:

from tolomeo.services.ota import OTAService
from .ota_plugin import MyOTAPlugin


class MyOTAService(OTAService):
    class Meta:
        plugin_class = MyOTAPlugin

3. Wire the Entry Point

Create my_service/ota_main.py:

import asyncio
import logging
from .ota_service import MyOTAService

logging.basicConfig(level=logging.INFO)


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


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

4. Run the Service

DEVICE_SERIAL_NUMBER=test001 python -m my_service.ota_main

5. Walk Through the Update Flow

Open a NATS subscriber to watch status events:

nats sub events.infos

Step 1 — Notify an update

Send NotifyOTAUpdate with the firmware URL. Status transitions to update_available.

Step 2 — Start the download

Send StartOTADownload. Status transitions: downloading → update_ready.

Step 3 — Install

Send InstallOTAUpdate. Status transitions: updating → update_successful.

For the exact NATS command syntax and full payload schemas, see Trigger OTA Updates via NATS.

6. Handle a Download Failure

If the download fails (bad URL, network error), the FSM moves to download_failed. Send AbortOTAUpdate to return to update_available and retry. See Recover from Failures for the full command.

Next Steps