A plugin will generally want to:
In both cases the plugin will use deltas which the server uses to signal changes in the Signal K full data model. Delta messages contain the new value associated with a path (not the amount of change from the previous value.)_
See the Signal K Delta Specification for details.
Using the server API, plugins can either:
By specifying a context e.g. 'vessels.self' you can limit the number of delta messages received to those of host vesseel.
To receive all deltas you can specify * as the context.
You can also limit the deltas received by the path you supply.
If you supply a specific path e.g. navigation.position, only updates in the value will be received.
Since paths are hierarchical, paths can contain wildcards e.g._navigation.* which will deliver deltas containing updates to all paths under navigation.
The data received is formatted as per the following example:
{
path: 'navigation.position',
value: { longitude: 24.7366117, latitude: 59.72493 },
context: 'vessel.self',
source: {
label: 'n2k-sample-data',
type: 'NMEA2000',
pgn: 129039,
src: '43'
},
$source: 'n2k-sample-data.43',
timestamp: '2014-08-15T19:00:02.392Z'
}
The server API provides the following methods for retrieving values from the full data model.
getSelfPath(path) returns the value of the supplied path in the vessels.self context.const value = app.getSelfPath('uuid')
app.debug(value) // Should output something like urn:mrn:signalk:uuid:a9d2c3b1-611b-4b00-8628-0b89d014ed60
getPath(path) returns the value of the path (including the context) starting from the root of the full data model.const baseStations = app.getPath('shore.basestations')
A can subscribe to a stream of updates (deltas) by creating the subscription.
Subcriptions are generally manged in the plugin start() and stop() methods to ensure the subscribtions are unsubscribed prior to the plugin stopping to ensure all resources are freed.
The following example illustrates the pattern using the subscriptionmanager API method.
let unsubscribes = []
plugin.start = (options, restartPlugin) => {
app.debug('Plugin started')
let localSubscription = {
context: '*', // Get data for all contexts
subscribe: [
{
path: '*', // Get all paths
period: 5000 // Every 5000ms
}
]
}
app.subscriptionmanager.subscribe(
localSubscription,
unsubscribes,
(subscriptionError) => {
app.error('Error:' + subscriptionError)
},
(delta) => {
delta.updates.forEach((u) => {
app.debug(u)
})
}
)
}
plugin.stop = () => {
unsubscribes.forEach((f) => f())
unsubscribes = []
}
In the start() method create a subscription definition localSubscription which is then passed to app.subscriptionmanager.subscribe() as the first argument, we also pass the unsubscribes array in the second argument.
The third argument is a function that will be called when there's an error.
The final argument is a function that will be called every time an update is received.
In the stop() method each subcription in the unsubscribes array is unsubscribed and the resources released.
announceNewPathsWhen using granular subscriptions (subscribing to specific paths rather than *), you may want to discover what paths are available without receiving continuous updates for all of them. The announceNewPaths option solves this:
let localSubscription = {
context: '*',
announceNewPaths: true, // Announce all matching paths once
subscribe: [
{
path: 'navigation.position', // Only get continuous updates for this path
period: 1000
}
]
}
When announceNewPaths: true is set:
This is useful for:
The announced deltas are regular delta messages - there's no special flag. Your client should track which paths it has seen and can then subscribe to specific ones as needed.
sourcePolicyWhen the server has Source Priority configured, subscriptions receive only the preferred source's data by default. You can override this with the sourcePolicy option:
let localSubscription = {
context: '*',
sourcePolicy: 'all',
subscribe: [
{
path: 'navigation.position',
period: 1000
}
]
}
| Value | Behaviour |
|---|---|
'preferred' |
Only deliver values from the preferred source (default) |
'all' |
Deliver values from all sources regardless of priority configuration |
Use sourcePolicy: 'all' when your plugin needs to see data from every source — for example, a display that compares readings from multiple sensors, or a data logger that records all sources.
Note: sourcePolicy is honoured only when subscribing through app.subscriptionmanager.subscribe() — the recommended API documented above. Plugins that read directly from app.streambundle (getBus(), getSelfStream()) always receive the preferred-only stream and have no way to opt into all sources; new plugins should prefer subscriptionmanager.subscribe().
WebSocket clients can apply the same policy to the entire connection by setting the sourcePolicy query parameter on the streaming endpoint:
ws://localhost:3000/signalk/v1/stream?subscribe=self&sourcePolicy=all
| Query value | Behaviour |
|---|---|
(omitted)/preferred |
Connection delivers preferred-only deltas (default) |
all |
Connection delivers deltas from every source, regardless of priorities |
The query-string default applies to the bootstrap cache replay and to per-message subscriptions that don't carry their own sourcePolicy. A subscribe message can still override it on a per-call basis by including sourcePolicy in the message body.
excludeSources / excludeSelfA plugin may want a priority-resolved view of a path with one or more sources removed from the cascade — for example to see the preferred upstream source while ignoring a known-bad device. excludeSources / excludeSelf provide that: the subscription still receives a single priority-resolved value per path, but the cascade runs without the excluded refs.
Note: Because the cascade runs on the subscription's feed, this delivers a single priority-resolved value — and if the user ranks the plugin itself above the upstream source, that source is held to the fallback timeout on the plugin's own input. That is the right behaviour for a plugin that wants the preferred remaining upstream value, but not for a correction/transform plugin that must see every raw sample at full rate — for that case see Correction and transform plugins below, which uses
sourcePolicy: 'all'.
The same fallback semantics the user configured still apply across the remaining sources:
let localSubscription = {
context: 'vessels.self',
excludeSelf: true,
subscribe: [
{
path: 'environment.wind.speedTrue'
}
]
}
With user ranking myPlugin > sourceB > sourceA:
| Bus state | Delivered to plugin |
|---|---|
sourceB publishing |
sourceB |
sourceB silent past its fallback timeout |
sourceA |
sourceB resumes |
sourceB |
myPlugin (own output) |
(never delivered) |
| Field | Behaviour |
|---|---|
excludeSources: string[] |
Drop these $source refs from the cascade. Explicit form; works in both plugin and WebSocket subscriptions. |
excludeSelf: true |
Plugin-only shorthand. The server resolves it to [plugin.id]. Combine with excludeSources to add explicit refs on top. |
Both fields take effect only when sourcePolicy is 'preferred' (the default). Under sourcePolicy: 'all' they are ignored — 'all' already bypasses the priority cascade and partial filtering would be surprising.
excludeSelf resolves to the plugin's id only — a single ref, not a prefix match. A plugin that publishes under additional labels (e.g. myPlugin.windFromPolars) should use the explicit excludeSources form to list every ref it produces.
WebSocket subscriptions can use excludeSources directly. excludeSelf is meaningless for them (there is no plugin identity to resolve against) and is silently ignored — WebSocket clients should always use the explicit form.
The match is on the $source of the deltas your plugin emits, and excludeSelf resolves to your bare plugin.id. So your emitted deltas must carry $source === plugin.id for the exclusion to apply. The simplest way is to emit with no source object at all — the server then sets $source to your plugin.id:
app.handleMessage(plugin.id, {
context: 'vessels.self',
updates: [
{ values: [{ path: 'environment.wind.speedTrue', value: corrected }] }
]
})
If you do supply a source object, also set a top-level $source: plugin.id on the update — the server keeps an explicit $source and only derives one from the source object when it is absent, so this guarantees the match. (See Sending Deltas below.)
A SignalK plugin can not only read deltas, but can also send them. This is done using the handleMessage() API method and supplying:
Example:
app.handleMessage(
plugin.id,
{
updates: [
{
values: [
{
path: 'environment.outside.temperature',
value: -253
}
]
}
]
},
'v1'
)
Plugins can suggest default metadata for paths they publish using setDefaultMetadata(). This is useful for setting display units, descriptions, or other metadata that improves the user experience in the data browser without overwriting values the user has already configured.
The method uses per-field merge semantics: it only sets metadata fields that don't already exist for the path. If a user has already set displayName for a path, the plugin can still set displayUnits on the same path without overwriting the user's choice.
plugin.start = async (options) => {
// Suggest that energy paths display in Wh instead of the SI unit (J)
await app.setDefaultMetadata('electrical.batteries.house.energy', {
displayUnits: { category: 'energy', targetUnit: 'Wh' }
})
// Suggest display units and a description
await app.setDefaultMetadata('electrical.solar.panelPower', {
displayUnits: { category: 'power', targetUnit: 'W' },
description: 'Total solar panel output'
})
}
The method is idempotent: on subsequent server starts, fields already persisted from a previous call will not be overwritten. It returns true if any new fields were applied, false if all fields already existed.
The displayUnits.category is validated against the path's SI unit. If the category doesn't match, the method returns false and logs a debug message.
There are essentially two strategies a plugin can employ to change a value for a path:
In place strategy is used when the new value is meant to replace the original value completely, for example applying a fixed offset to a heading to compensate for a sensor's mount position. This makes the original value unavailable for other consumers.
With new value strategy both values are available, but using the source priority system the plugin's output can be prioritised over the original value. The user ranks the plugin's output above the raw source in source priority, so downstream consumers get the corrected value and a user interface like a dashboard can display both if needed.
A correction plugin needs its input at full rate, independent of the priority ranking — you correct every reading from the upstream source, not just whichever one a cascade would currently prefer — while the user's ranking decides what consumers see. Subscribe with sourcePolicy: 'all' and skip your own output in the callback:
sourcePolicy: 'all' delivers every source at full rate with no cascade on your input. Skip updates whose $source is your own plugin.id so you don't reprocess your output. The global priority cascade still applies to consumers, so the user's ranking of your output works as expected.Do not use registerDeltaInputHandler for new value strategy. An input handler is for modifying a delta in place as it passes through, not for republishing a value under your own source — a correction plugin emits its corrected value under its own $source and lets source priority choose between that and the raw source, which is a different shape from rewriting the incoming delta. Subscribe-and-emit keeps the raw and corrected values as two distinct sources the user can rank.
excludeSelf (see Excluding Sources) is a different tool and not what you want for a full-rate correction. It runs the priority cascade on your input feed, so it delivers a single ranked-and-throttled upstream value rather than every reading: with the user ranking your plugin rank-0, the real source is held as a lower-ranked fallback and only reaches you after the fallback timeout. excludeSelf fits a plugin that wants the preferred remaining upstream value (e.g. "show source B, fall back to A, never my own C"), not one that corrects each raw sample.
A complete heel-correction example:
let unsubscribes = []
plugin.start = () => {
app.subscriptionmanager.subscribe(
{
context: 'vessels.self',
sourcePolicy: 'all',
subscribe: [{ path: 'navigation.speedThroughWater' }]
},
unsubscribes,
(err) => app.setPluginError(err),
(delta) => {
delta.updates.forEach((u) => {
// $source is a property of the update, not of each value. Skip
// our own output so we don't reprocess the corrected value.
if (u.$source === plugin.id) return
u.values.forEach((pv) => {
if (pv.path !== 'navigation.speedThroughWater') return
const corrected = applyHeelCorrection(pv.value)
// No source object: $source defaults to plugin.id, so the
// guard above recognises and skips this delta on the way back.
app.handleMessage(plugin.id, {
context: 'vessels.self',
updates: [
{
values: [
{ path: 'navigation.speedThroughWater', value: corrected }
]
}
]
})
})
})
}
)
}
plugin.stop = () => {
unsubscribes.forEach((f) => f())
unsubscribes = []
}
A SignalK plugin can not only emit deltas, but can also send data such as NMEA 2000 data.
This is done using the emit() API and specifying the provider as well as the formatted data to send.
Example: Send NMEA using Actisense serial format:
app.emit(
'nmea2000out',
'2017-04-15T14:57:58.468Z,0,262384,0,0,14,01,0e,00,88,b6,02,00,00,00,00,00,a2,08,00'
)
Example: Send NMEA using Canboat JSON format:
app.emit('nmea2000JsonOut', {
pgn: 130306,
'Wind Speed': speed,
'Wind Angle': angle < 0 ? angle + Math.PI * 2 : angle,
Reference: 'Apparent'
})
If you need to send an NMEA2000 message out at startup, e.g get current state from a device you will need to wait until the provider is ready before sending your message.
Example: Send NMEA after the provider is ready:
app.on('nmea2000OutAvailable', () => {
app.emit(
'nmea2000out',
'2017-04-15T14:57:58.468Z,2,6,126720,%s,%s,4,a3,99,01,00'
)
})