Connectivity

Overview

The Industrial components are set through plain properties: an alarm list, a process value, a pen record. They do not include a transport, so they can be used with any data source. Smart.Industrial.Connect provides a documented way to route a tag from a data source into a component property, with data quality and staleness handling included.

Connect is not a protocol implementation. A browser has no raw TCP socket, so it cannot communicate with OPC UA binary, Modbus TCP or EtherNet/IP directly. A web HMI reaches a PLC through a gateway or broker that terminates the industrial protocol and publishes the data over WebSocket, server-sent events or HTTP. The adapters in this module use the transports available in a browser. The OPC UA and Sparkplug helpers map the payload format that a gateway emits; they do not implement the protocols.

Connect provides three things that a simple polling loop does not:

• Quality is carried with the value. A reading consists of a value, a quality and a time. Without the quality, a stale sensor looks like a live one.
• Missing data is detected. A tag that stops arriving is not shown as live. With staleAfter set, the binding changes the quality to stale automatically.
• Local time is marked as such. When an adapter supplies no timestamp, the sample is stamped with the local time and marked with timestampIsLocal.

Sessions

A session is a connection to one data source. Sessions are created explicitly rather than globally, because an application can have more than one source, for example a plant broker and an instrument stream.

const Connect = Smart.Industrial.Connect;

const plant = Connect.open('websocket', {
    url: 'wss://gateway.example/tags',
    //The messages your gateway expects. These three are examples.
    subscribeMessage: function (tag) { return { op: 'subscribe', tag: tag }; },
    writeMessage: function (tag, value) { return { op: 'write', tag: tag, value: value }; },
    historyUrl: '/api/history'     //asked with tags, from and to in the query string
});
const rig = Connect.open('poll', { url: '/api/tags', interval: 500 });   //GET /api/tags?tags=DUT.V,DUT.I

rig.subscribe(['DUT.V', 'DUT.I'], function (sample) { /* ... */ });
plant.subscribe(['FIC-101.PV', 'FIC-101.SP'], function (sample) {
    //{ tag, value, quality, timestamp, timestampIsLocal? }
});

plant.write('FIC-101.SP', 82).then(function () { /* the write was sent */ });
plant.history('FIC-101.PV', Date.now() - 3600000, Date.now()).then(function (samples) { /* in time order */ });
plant.latest('FIC-101.PV');   //the most recent sample, or undefined
plant.status;                  //connecting | open | reconnecting | error | closed
plant.close();

The write() and history() methods are optional on an adapter. A session whose adapter has no write rejects write() as read-only, and one without a history source rejects history(). The websocket adapter can write only when writeMessage is configured and has history only when historyUrl (or a history function) is; the poll adapter asks only for the tags that are subscribed. A write that fails, including one made while the websocket is closed or reconnecting, rejects the promise; write() does not throw. A resolved write means that the adapter sent it; with the websocket adapter that is when the message was put on the socket, not when the gateway confirmed it. The confirmation is the new value arriving. The faceplate is built for this: it raises a request and does not change its own value, so the screen continues to show the value reported by the plant, also when a write is rejected or lost.

The adapters

websocket. openMessage (a message, or a function that returns one) is sent first on every connection, before the subscriptions; a gateway's authentication belongs there rather than in the URL. subscribeMessage and unsubscribeMessage return the message sent for a tag; without them the adapter sends nothing and receives what the gateway sends. Without writeMessage the session is read-only, and a write while the socket is not open throws an error. A message that is not a string is sent as JSON. parse turns a received message into one sample or an array of samples; the default parses JSON.

poll. Each interval, the adapter requests url?tags= followed by the subscribed tags, one request at a time, and emits what parse returns (by default the response body: a sample or an array of samples). request(url, tags) replaces the HTTP request entirely. History is requested from historyUrl with tags, from and to in the query string, and parseHistory converts the response.

A page that already has an MQTT client can attach it in a few lines:

//client is the page's MQTT client; decode() and encode() are its Sparkplug B (protobuf) codec.
const origin = {};   //tag -> the topic and the metric it came from

const broker = Connect.open('manual', {
    attach: function (emit) {
        client.on('message', function (topic, payload) {
            //Sparkplug B: one payload carries many metrics. The mapper returns one sample per
            //metric, and its tag is the topic, a slash and the metric name.
            Connect.sparkplug(topic, decode(payload)).forEach(function (sample) {
                origin[sample.tag] = { topic: topic, metric: sample.tag.slice(topic.length + 1) };
                emit(sample);
            });
        });
    },
    write: function (tag, value) {
        //A write is a command on the same device: spBv1.0/group/DDATA/edge/device becomes .../DCMD/...
        const from = origin[tag];

        client.publish(from.topic.replace(/\/([ND])(DATA|BIRTH)\//, '/$1CMD/'),
            encode({ metrics: [{ name: from.metric, value: value }] }));
        return true;
    }
});

Connect.opcua(nodeId, dataValue) does the same for the DataValue format emitted by an OPC UA gateway. It prefers the source timestamp to the server timestamp and maps the severity bits of the status code to good, uncertain or bad; a DataValue that carries no status code is mapped to good, because OPC UA leaves the status code out when it is Good. Connect.sparkplug() maps a null metric to bad and any other metric to good, unless a quality(metric) function is passed in its options. A custom transport is registered with Connect.adapter(name, factory). The factory receives the configuration and { emit(sample), state(status, error) } and returns { subscribe, unsubscribe, write, history, close }, each of them optional.

bind(): tags to properties

Connect.bind(faceplate, {
    processValue: { tag: 'FIC-101.PV', staleAfter: 3000 },
    setpoint: 'FIC-101.SP',
    output: 'FIC-101.OP'
}, plant);

Connect.bind(document.getElementById('flowTile'), {
    value: { tag: 'FIC-101.PV', deadband: 0.2 }
}, plant);

The map is property → tag or property → { tag, transform, deadband, staleAfter, quality, unit, displayUnit, scale }. Without a session argument, bind() uses the session opened last. bind() returns an unbind function that should be called when the screen is removed.

• staleAfter: the number of milliseconds without a sample after which the component's quality becomes stale, on the binding that sets the quality (see quality). A component without a quality property, such as the status tile, cannot show it. Set it from the scan rate; for example, three seconds on a 4 Hz tag corresponds to twelve missed samples.
• deadband: a change smaller than this value is not assigned. This avoids a render for every small change of a noisy sensor.
• quality: whether this binding sets the component's quality property. The default is true for the first binding on a component that has a quality property and false for the others, so that two tags do not overwrite each other's quality.
• transform: a function applied to the value before it is assigned.
• unit, displayUnit, scale: unit conversion, described in the Engineering Units guide.

stream() and trend(): tags to records

A strip chart or a trend takes records instead of properties: one object per instant with a field per pen. stream() collects the samples that arrive within an interval and pushes one record per interval in which something arrived, stamped with the newest source time among them. A field that received nothing in the interval keeps its last value, so every pen that has had a sample has a value in each row; a sample without a value is pushed as NaN, a gap in the pen. The last value is kept only for the staleness interval: the field's own staleAfter, the staleAfter option, or the element's staleAfter. After that the field is left out of the records, so a tag that has gone silent is not re-sent as if it were live and the trend marks its pen stale. When a trend pen names a qualityField, each sample's quality is written into that field of the record, live and from history.

Connect.stream(stripChart, {
    flow: { tag: 'FIC-101.PV', session: plant },
    voltage: { tag: 'DUT.V', session: rig }
}, { interval: 250 });

//The same, plus history: when the trend raises historyRequest for a part of its window
//it holds no records for, trend() reads that interval from the session's history() and
//pushes it into the trend.
Connect.trend(historianTrend, {
    temperature: 'TIC-101.PV',
    pressure: 'PIC-102.PV'
}, { interval: 1000, history: true });

A field can name its own session, which allows a trend to show a pen from the plant broker next to a pen from an instrument. Two separate stream() calls on one chart do not combine their fields: each call pushes records with its own fields only.

Sample format

{
    tag: 'FIC-101.PV',
    value: 78.4,
    quality: 'good',            //good | uncertain | bad | stale
    timestamp: 1758210000000,   //milliseconds since the epoch
    timestampIsLocal: true      //only present when the adapter sent no timestamp
}

The quality defaults to good. Most transports offered by a gateway carry no quality information, and marking every reading from them as uncertain would put a warning on every screen. The websocket, SSE and poll adapters pass on the quality field of the message as it is; the opcua and sparkplug mappers derive it from the payload. An adapter whose source reports quality is expected to map it.

A missing value is not a zero. Number(null) is 0, so a binding that only checks isFinite would turn a dropped sample into a reading of zero. A sample whose value is null or undefined does not change the property: bind() keeps the last value and sets the quality to the sample's own quality, or to bad when the sample says good, and stream() pushes a gap. Other values are assigned as they arrive, with two conversions: a value assigned to a boolean property becomes true when it is a non-zero number, '1', 'true' or 'on' and false otherwise, and a number assigned to a text property becomes text.

Demo

The connectivity screen runs two sessions side by side, stops one of them to show the stale state, and shows the sample stream in an inspector. Its code is the screen's index.js, next to its page.