Skip to content

Basic

Kelvin MQTT Connector

Connects to an MQTT Broker with configurable topic subscriptions, and publishes the received messages payload to the Kelvin Platform as Data streams.

Code

The full code is hosted in the Kelvin App Samples repository on GitHub here.

Repository contents

  • README.md — short instructions for requirements and usage.
  • main.py — the application implementation (async MQTT client + Kelvin SDK integration).
  • app.yaml — Kelvin app metadata, IO and UI schema references, and defaults.
  • requirements.txt — Python dependencies required to run the app.
  • Dockerfile — image recipe for containerizing the app.
  • .dockerignore — items to exclude from Docker build context.
  • mqtt-connector-io-example.csv — example CSV for mapping Asset/DataStream/Topic pairs (used for creating IO configuration).
  • ui_schemas/ — UI schema files referenced by app.yaml (configuration and io default schemas).

Basic Code Explanation

The main.py program connects the Kelvin Connection Application to an MQTT broker and acts as a bridge between incoming MQTT data and the Kelvin Platform.

It starts by initializing the KelvinApp, reading configuration details like the broker’s IP and port through a Pydantic MQTTConfig, and building an internal io_map that links Kelvin data streams to their MQTT topics.

When connected, it listens for messages on those topics, automatically converts each payload to the correct data type, and creates the appropriate Kelvin message type (such as KMessageTypeData) for publishing back into the Kelvin system.

The core logic is encapsulated in MQTTImporterApplication, which manages the app lifecycle (connect, disconnect, configuration updates) and handles the continuous async MQTT read loop.

It subscribes to all relevant topics using aiomqtt, processes each message into Kelvin message objects, and publishes them through the active KelvinApp connection.

Error handling and reconnection logic ensure that the service can automatically recover from broker or network issues, keeping the data flow between MQTT and Kelvin synchronized.