esio-spb-builder 0.1.0

Configuration-driven SparkPlug B packet builder and decoder for industrial IoT applications

esio-spb-builder

A configuration-driven SparkplugB packet builder for IoT applications that simplifies the creation and management of SparkplugB protocol messages with proper bdSeq handling and hierarchical device organization.

Features

• 🏗️ Configuration-Driven: Define your edge nodes and devices in JSON/YAML, not code
• 📊 Proper bdSeq Management: Handles the critical birth/death sequence numbers correctly
• 🌳 Hierarchical Structure: Group → Edge Node → Device organization
• 💾 Persistence: Save and load configurations for consistent deployments
• 🔄 Dynamic Updates: Runtime metric updates without recompilation
• ✅ Protocol Compliance: Includes all required SparkplugB metrics (Node Control, bdSeq)
• 🎯 Type Safety: Full type hints and validation
• 🔍 Visualization: Built-in tools to explore and display your configuration

Why This Library?

SparkplugB is powerful but complex. Traditional approaches require:

• Hardcoded enums for metric aliases
• Custom classes for each node/device type
• Manual bdSeq management (easy to get wrong)
• Recompilation for configuration changes

esio-spb-builder solves these problems by:

• Using configuration files instead of code
• Automatically managing aliases and required metrics
• Handling bdSeq correctly for MQTT session management
• Supporting runtime configuration changes

Installation

pip install esio-spb-builder

Development installation:

git clone https://github.com/edgestack/esio-spb-builder
cd esio-spb-builder
pip install -e ".[dev]"

Quick Start

from esio_spb_builder import (
    SpBConfigurationManager,
    EdgeNodeFactory,
    HierarchyDisplay,
)

# Create configuration
config = SpBConfigurationManager()

# Add a group (location/namespace)
config.add_group("oil.field.texas", "Texas oil field site")

# Add an edge node with metrics
config.add_node(
    group_id="oil.field.texas",
    node_id="edge-01",
    description="Primary edge gateway",
    metrics=[
        {"name": "CPU/Load", "data_type": "Float", "value": 0.0},
        {"name": "Memory/Free", "data_type": "UInt64", "value": 0}
    ]
)

# Add a device to the node
config.add_device(
    group_id="oil.field.texas",
    node_id="edge-01",
    device_id="sensor-01",
    description="Temperature sensor",
    metrics=[
        {
            "name": "Temperature",
            "data_type": "Float",
            "value": 20.0,
            "properties": [
                {"name": "EngUnit", "data_type": "String", "value": "°C"}
            ]
        }
    ]
)

# Create factory and build runtime instance
factory = EdgeNodeFactory(config)
edge_node = factory.create_edge_node("oil.field.texas", "edge-01")

# Generate SparkplugB packets
nbirth = edge_node.birth_certificate()
print(f"Publishing to: {nbirth.topic}")
print(f"Payload: {nbirth.serialize()}")

# Update metrics
edge_node.update_metrics({
    "CPU/Load": 45.5,
    "Memory/Free": 8192000000
})

# Send data packet
ndata = edge_node.data_packet()

Configuration Structure

The library uses a hierarchical configuration that mirrors the SparkplugB protocol:

version: "1.0.0"
groups:
  oil.field.texas:
    description: "Texas oil field"
    edge_nodes:
      edge-01:
        description: "Primary edge gateway"
        metrics:
          - name: "CPU/Load"
            alias: 10  # Auto-assigned if not provided
            data_type: "Float"
            value: 0.0
        devices:
          ogi-camera:
            description: "Optical gas imaging camera"
            metrics:
              - name: "Flow/Rate"
                alias: 100
                data_type: "Float"
                value: 0.0
                properties:
                  - name: "EngUnit"
                    data_type: "String"
                    value: "SCFH"

Key Concepts

Required Metrics

The library automatically includes SparkplugB required metrics:

• Node Control/Rebirth (alias 1): Command to request rebirth
• Node Control/Next Server (alias 0): Command for failover
• Node Control/Reboot (alias 2): Command to reboot
• bdSeq: Birth/death sequence for session management

bdSeq Management

The birth/death sequence (bdSeq) is critical for MQTT session management. The library handles this automatically:

# First session
nbirth1 = edge_node.birth_certificate()  # bdSeq included
# ... send data ...
ndeath1 = edge_node.death_certificate()   # bdSeq incremented

# Next session will have incremented bdSeq
nbirth2 = edge_node.birth_certificate()  # New bdSeq

Metric Aliases

Aliases are numeric identifiers for metrics that reduce payload size. The library:

• Auto-assigns aliases if not provided
• Prevents conflicts between metrics
• Preserves aliases when configuration changes
• Tracks deprecated aliases to prevent reuse

Advanced Usage

Loading from File

# Load from JSON
config = SpBConfigurationManager.load_from_file("config.json")

# Load from YAML
config = SpBConfigurationManager.load_from_file("config.yaml")

Visualization Tools

from esio_spb_builder import HierarchyDisplay, MetricExplorer

# Display hierarchy
display = HierarchyDisplay(config)
print(display.display_tree())

# Explore metrics
explorer = MetricExplorer(config)
flow_metrics = explorer.find_metric("flow")
command_metrics = explorer.get_command_metrics()

Command Processing

# Process NCMD command
action = edge_node.process_command(alias=1, value=True)
if action == "rebirth":
    packets = edge_node.rebirth()  # Returns all birth certificates

Change Tracking

# Only send changed metrics in NDATA
edge_node.update_metric("CPU/Load", 50.0)
data_packet = edge_node.data_packet(send_all=False)  # Only CPU/Load

MQTT Integration

import paho.mqtt.client as mqtt
from esio_spb_builder import EdgeNodeFactory

# Create MQTT client
client = mqtt.Client()
client.connect("broker.mqtt.com", 1883)

# Create edge node
factory = EdgeNodeFactory(config)
edge_node = factory.create_edge_node("group", "node")

# Send NBIRTH
nbirth = edge_node.birth_certificate()
client.publish(nbirth.topic, nbirth.serialize())

# Send NDATA periodically
def send_data():
    edge_node.update_metrics(get_current_values())
    ndata = edge_node.data_packet()
    client.publish(ndata.topic, ndata.serialize())

API Reference

Core Classes

• SpBConfigurationManager: Manages the complete configuration
• EdgeNodeFactory: Creates runtime instances from configuration
• DynamicEdgeNode: Runtime edge node that generates packets
• DynamicDevice: Runtime device that generates packets

Packet Types

• NBirthPacket: Node birth certificate
• NDataPacket: Node data
• NDeathPacket: Node death certificate
• NCmdPacket: Node command
• DBirthPacket: Device birth certificate
• DDataPacket: Device data
• DDeathPacket: Device death certificate
• DCmdPacket: Device command

Utilities

• HierarchyDisplay: Visualize configuration structure
• MetricExplorer: Search and analyze metrics
• PacketDecoder: Decode received SparkplugB packets

Testing

Run the test suite:

pytest tests/

Run with coverage:

pytest --cov=esio_spb_builder tests/

Examples

See the examples/ directory for complete examples:

• complete_example.py: Full demonstration of all features
• Configuration samples in JSON and YAML

Architecture

Configuration (JSON/YAML)
         ↓
SpBConfigurationManager
         ↓
EdgeNodeFactory
         ↓
DynamicEdgeNode / DynamicDevice
         ↓
SparkplugB Packets (NBIRTH, NDATA, etc.)
         ↓
MQTT Broker

Comparison with Traditional Approach

Traditional (sparkplug-b-packets style)

# Requires hardcoded enums
class AliasMap(IntEnum):
    CPU_Load = 3
    Memory_Avail = 4

# Requires custom classes
class MyNodeBirth(NBirthPacket):
    def __init__(self):
        super().__init__(metrics=get_metrics())

# Metrics defined in code
def get_metrics():
    return OrderedDict(...)

With esio-spb-builder

# Load configuration
config = SpBConfigurationManager.load_from_file("config.json")

# Create and use
factory = EdgeNodeFactory(config)
node = factory.create_edge_node("group", "node")
birth = node.birth_certificate()

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

MIT License - see LICENSE file for details

Acknowledgments

Built for the EdgeStack IoT platform to simplify SparkplugB implementation and ensure protocol compliance.