Sparkplug B

Read and command Sparkplug B tags, and publish Synnax channels as a Sparkplug B edge node.

Sparkplug B is a specification on top of MQTT that SCADA systems and industrial gateways use, such as Ignition with the Cirrus Link modules. It sets the topics, a binary payload, and the rules for the state of each edge node. Synnax works with it in two ways:

Host application -> the MQTT read task reads tags, and the MQTT write task sends commands for them. The same tasks can hold plain topics and tags together.
Edge node -> the Sparkplug edge node task publishes Synnax channels as tags, so a SCADA system sees them.

Before you start, connect an MQTT broker.

Identify a tag

An edge node publishes tags for itself and for its devices. Four values identify a tag:

Group -> the group ID of the edge node.
Edge node -> the edge node ID.
Device -> the device ID. Leave it empty for a tag of the edge node.
Tag -> the name of the tag, which can hold /, such as Motor/RPM.

The IDs cannot hold /, +, or #.

Host application ID

The Sparkplug B settings of the broker control the host application:

Host ID -> with an ID, Synnax publishes the retained message spBv1.0/STATE/<host_id>. It says that Synnax is online while the Core has a connection to the broker. Empty makes Synnax a passive host that publishes nothing.
Groups -> the group IDs that the Add data dialog searches. Empty searches all groups.

Set a host ID only when your edge nodes name Synnax as their primary host and hold their data until that host is online. Each host ID must be in use by one application only.

Births, deaths, and history

An edge node declares its tags in a birth message. When a read task starts, it asks each edge node for a new birth, so that it starts from the current value of each tag. It asks again when it finds a gap in the sequence numbers of an edge node, or a tag that the last birth did not declare. It sends at most one request every 5 seconds to each edge node.

Each value of a tag carries the time at which the edge node took it, and the task writes that time to the index channel of the tag. Tags change at different times, so give each tag an index channel of its own.

An edge node can store values while it is offline and send them after its next birth, marked as historical. The task writes each one into the outage that it belongs to, or into the time before the task started. These values reach stored data but not live plots. The task writes them after the first live value of the tag, which usually comes in the birth, or when the task stops. A task with data saving off, and a tag on a virtual channel, store nothing, so they drop historical values with no warning.

Read tags

In the devices toolbar, right-click the broker and select Create read task.
Click Add data, then select Sparkplug B. The dialog lists the edge nodes that publish now.
Select an edge node to see its tags. The Console asks the edge node for a rebirth to get them.
Select the tags to record. For a tag that is not listed, click Add a tag by name.
Click Add N channels.
Click (Start).

The Console creates a channel and an index channel for each tag.

Python does not create channels. Give each tag a channel with its own index.

import synnax as sy

broker = client.devices.retrieve(name="Plant broker")
time = client.channels.create(
    name="pump1_speed_time", is_index=True, data_type=sy.DataType.TIMESTAMP
)
speed = client.channels.create(
    name="pump1_speed", index=time.key, data_type=sy.DataType.FLOAT32
)

task = sy.mqtt.ReadTask(
    name="Line 1 tags",
    device=broker.key,
    entries=[
        sy.mqtt.SparkplugReadEntry(
            group="Plant",
            edge_node="Line1",
            device="Pump1",
            tag="speed",
            channel=speed.key,
            data_type="float32",
        )
    ],
)
client.tasks.configure(task)
task.start()

Command tags

Each value written to the command channel becomes one NCMD message, or one DCMD message for a tag of a device. Sparkplug B sends each command at quality of service 0, with no reply. To confirm a command, read the tag back with a read task.

In the devices toolbar, right-click the broker and select Create write task.
Click Add targets, then select Sparkplug B.
Select an edge node, then select the tags to command.
Click Add N targets.
Select a target and check its Sparkplug B type. It must match the type that the edge node declares.
Click (Start).

The Console creates a command channel for each tag.

import synnax as sy

broker = client.devices.retrieve(name="Plant broker")
cmd_time = client.channels.create(
    name="pump1_setpoint_cmd_time", is_index=True, data_type=sy.DataType.TIMESTAMP
)
cmd = client.channels.create(
    name="pump1_setpoint_cmd", index=cmd_time.key, data_type=sy.DataType.FLOAT64
)

task = sy.mqtt.WriteTask(
    name="Pump 1 commands",
    device=broker.key,
    targets=[
        sy.mqtt.SparkplugWriteTarget(
            group="Plant",
            edge_node="Line1",
            device="Pump1",
            tag="setpoint",
            channel=cmd.key,
            tag_type="double",
        )
    ],
)
client.tasks.configure(task)
task.start()

Run an edge node

The edge node task makes the Core one Sparkplug B edge node. Each tag publishes one Synnax channel, and a SCADA system such as Ignition sees the channels as tags, with their births, deaths, and sequence numbers.

In the devices toolbar, right-click the broker and select Create Sparkplug edge node. You can also run the “Create Sparkplug edge node” command.
Enter the Group and the Edge node ID.
Click Add tag, then select the channel to publish. The tag takes the name of the channel. Set the name and the Sparkplug B type of the tag.
To let a host write to a tag, right-click it and select Add command channel.
Set Authority, the control authority of the writes that commands make.
Click (Start).

import synnax as sy

broker = client.devices.retrieve(name="Plant broker")
temperature = client.channels.retrieve("line1_temperature")
setpoint = client.channels.retrieve("line1_setpoint")
setpoint_cmd = client.channels.retrieve("line1_setpoint_cmd")

task = sy.mqtt.EdgeTask(
    name="Line 1 edge node",
    device=broker.key,
    group="Plant",
    edge_node="Line1",
    authority=200,
    tags=[
        sy.mqtt.EdgeTag(name="temperature", channel=temperature.key),
        sy.mqtt.EdgeTag(
            name="setpoint",
            channel=setpoint.key,
            command_channel=setpoint_cmd.key,
        ),
    ],
)
client.tasks.configure(task)
task.start()

The task publishes each sample written to a channel as an NDATA message, with the timestamp of the sample. Its birth declares every tag with its data type. A tag whose channel has had no sample since the task started is null in the birth.

The task is its own MQTT client, so the broker sends its death message when the connection drops. It publishes its birth again when a host asks for a rebirth, and after it connects again.

A command for a tag writes the value to the command channel of that tag, with the authority of the task. A tag with no command channel rejects commands. A writer with a higher authority on the channel keeps control, so the command is dropped with a warning.

Tag reference

A tag in a read task has these settings:

Setting	Default	Notes
Group `group`		See identify a tag.
Edge node `edge_node`		See identify a tag.
Device `device`	Empty	Empty for a tag of the edge node.
Tag `tag`		The name of the tag.
Channel `channel`		The channel that records the value.
Data type `data_type`	`float64`	The data type of the channel.

In code, a tag to read is SparkplugReadEntry, of type "sparkplug". A tag to command is SparkplugWriteTarget, with the same identity fields, a command channel, and Sparkplug B type (tag_type), which is double by default.

The edge node task, of type mqtt_sparkplug_edge, has these settings:

Setting	Default	Notes
Group `group`		The group ID of the edge node.
Edge node `edge_node`		The edge node ID.
Authority `authority`	0	0 to 255. The authority of the writes of commands.
Auto start `auto_start`	Off	Starts the task each time the Core starts.

Each of its tags (EdgeTag) has a name, a channel, a tag_type (double by default), and a command_channel (0 rejects commands). Every birth declares the tags bdSeq and Node Control/Rebirth, so no tag of the task can use these names.

Data types

The read task converts each value to the data type of the channel. It reads the integer, float, double, boolean, string, text, uuid, and date_time types of Sparkplug B. It does not read data_set, bytes, file, template, or array tags.

A Sparkplug B type is one of int8, int16, int32, int64, uint8, uint16, uint32, uint64, float, double, boolean, string, or date_time. The task rejects a channel whose data type can never fit, such as a string channel for a double tag, and drops a value that is out of range for the type with a warning.

Troubleshooting

... is offline
Cause
An edge node or a device published a death message, or did not answer the rebirth request. The warning clears when it publishes a birth.
Fix
Check that the edge node runs and has a connection to the broker.
... is not in the birth of its ...
Cause
A tag of the task is not in the last birth of its edge node or device. The name can be wrong, or the edge node no longer publishes the tag.
Fix
Check the Group, Edge node, Device, and Tag against the tags in the Add data dialog. Names are case sensitive.
tags: tag ... appears more than once tags: tag ... and tag ... write to the same command channel ...
Cause
Two tags of an edge node task have the same name, or share one command channel.
Fix
Give each tag its own name and its own command channel.

For problems with the connection, see Connect an MQTT broker.

Write task

Next in National Instruments

Configure a device