leaf-device 2.2.2

Sparkplug B MQTT device SDK for the LEAF IoT platform

leaf-device

Sparkplug B MQTT device SDK for the LEAF IoT platform.

leaf-device handles Sparkplug B protobuf payload construction, birth/data/command topic management, and optional self-managed MQTT connection for Python-based IoT devices and simulators.

Installation

pip install leaf-device

Requires Python 3.10+ and a running MQTT broker.

Quick Start

Standalone device (simulator style)

Define your device metrics in a JSON file:

{
  "device_id": "my-device-01",
  "namespace": "default",
  "metrics": [
    {"name": "latitude",    "data_type": "double",  "units": "°",   "value": 35.9},
    {"name": "temperature", "data_type": "double",  "units": "°C"},
    {"name": "relay",       "data_type": "boolean", "is_writable": true, "value": false}
  ]
}

Then run a device loop:

import time
from leaf_device import LeafDevice

class MyDevice(LeafDevice):
    def on_message(self, client, userdata, message):
        """Handle incoming DCMD commands."""
        for metric in self.parse_metrics(message.payload):
            self.set_metric_value(metric.name, metric.boolean_value)
        self.publish_message()  # acknowledge

device = MyDevice("device.json")
device.connect("mqtt-broker-host", 1883)
device.client.subscribe(device.cmd_topic)

while True:
    device.set_metric_value("temperature", 22.5)
    device.publish_message()
    time.sleep(5)

GPS simulation

from leaf_device import LeafDevice, GpsSimulator

device = LeafDevice("device.json")
gps = GpsSimulator(lat=35.9, lon=-84.3, speed_kmh=50, radius_km=1, update_rate_hz=1)

device.connect("localhost", 1883)
while True:
    gps.move()
    lat, lon = gps.get_position()
    device.set_metric_value("latitude", lat)
    device.set_metric_value("longitude", lon)
    device.publish_message()
    import time; time.sleep(1)

Framework / mediator pattern (hardware devices)

For devices that manage their own MQTT connection separately from sensor logic, use the mediator framework:

from leaf_device import LeafDevice
from leaf_device.framework import DriverInterface, DriverEvent, DriverEventType
from leaf_device.framework import Mediator, MqttClientDriver

class MySensorDriver(DriverInterface):
    def __init__(self, mediator: Mediator, config: str):
        self._mediator = mediator
        self._leaf = LeafDevice(config)

    def notify(self) -> None:
        payload = self._leaf.get_message()
        self._mediator.post_event(DriverEvent(DriverEventType.Status, payload))

    def update(self, event: DriverEvent) -> None:
        if event.event_type == DriverEventType.Command:
            for metric in self._leaf.parse_metrics(event.message.payload):
                self._leaf.set_metric_value(metric.name, metric.boolean_value)

mediator = Mediator()
sensor = MySensorDriver(mediator, "device.json")
mqtt = MqttClientDriver("broker-host", mediator, "my-device-01")

mediator.register_driver(sensor)
mediator.register_driver(mqtt)
mediator.start()
mqtt.start()

Device JSON schema

Field	Type	Required	Description
device_id	string	yes	Unique device identifier
namespace	string	no	LEAF namespace (default: "default")
metrics	array	yes	List of metric definitions

Each metric:

Field	Type	Required	Description
name	string	yes	Metric name
data_type	string	yes	One of: double, float, boolean, int, int32, string
units	string	no	SI unit string (e.g. "°C", "m/s")
value	any	no	Initial value
is_writable	bool	no	Accepts DCMD commands (default false)
min	number	no	Minimum value hint
max	number	no	Maximum value hint

API reference

LeafDevice

Method	Description
__init__(config, device_version)	Load device definition from JSON path or dict
connect(mqtt_host, mqtt_port)	Connect to broker and send DBIRTH
set_metric_value(name, value)	Update a metric in the outbound payload
get_message()	Serialize payload to Sparkplug B protobuf bytes
parse_metrics(payload_bytes)	Deserialize a DCMD payload
publish_message()	Publish to DDATA topic (standalone mode)
on_message(client, userdata, message)	Override to handle DCMD commands

GpsSimulator

Method	Description
__init__(lat, lon, speed_kmh, radius_km, update_rate_hz)	Initialise circular GPS path
move()	Advance position by one time-step
get_position()	Return (latitude, longitude) tuple

Framework (leaf_device.framework)

Symbol	Description
DriverInterface	Abstract base class for all mediator drivers
Mediator	Thread-safe event bus (subclasses threading.Thread)
MqttClientDriver	MQTT publish/subscribe driver
DriverEvent	Immutable event object with .event_type and .message
DriverEventType	Enum: NoEvent, Status, Command

MQTT topics

leaf/{namespace}/DBIRTH/{device_id}   — Device birth (sent on connect)
leaf/{namespace}/DDATA/{device_id}    — Periodic telemetry
leaf/{namespace}/DCMD/{device_id}     — Commands from server to device

Development

# Install with dev dependencies
uv pip install -e ".[dev]"

# Run unit tests
pytest tests/unit/ -v

# Run integration tests (requires a broker on localhost:1883)
pytest tests/integration/ -v

# Lint and format
flake8 src/ tests/
black src/ tests/
isort src/ tests/
mypy src/

License

MIT — see LICENSE.

The bundled sparkplug_b_pb2.py is generated from sparkplug_b.proto (Eclipse Public License 2.0, Eclipse Sparkplug Working Group).