mohanki/digitaltwin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
mohanki
2026-08-24 15:35:32 +05:30
commit fcade251a6
51 changed files with 11565 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/ingest/mqttSource.js
+100
View File
@@ -0,0 +1,100 @@
/**
* MQTT telemetry source - DOCUMENTED STUB.
*
* This is not implemented, and it says so honestly rather than pretending. What
* it does provide is the exact shape of the work: the tag map, the frame
* assembly, and where analytics plugs in. Wiring this to a real broker is a
* day's work, not a rewrite, because everything downstream consumes frames.
*
* To implement:
* 1. npm i mqtt
* 2. Fill TAG_MAP with the customer's actual topic names.
* 3. Implement start() as marked below.
* 4. Set TELEMETRY_SOURCE=mqtt and MQTT_URL in .env
*
* Sparkplug B note: most industrial brokers publish Sparkplug B protobuf rather
* than plain JSON on flat topics. If so, add `sparkplug-payload` to decode
* NBIRTH/NDATA messages and map metric aliases instead of topic strings.
*/
import { TelemetrySource } from './source.js';
import { AnalyticsEngine } from '../analytics/alarms.js';
/**
* Maps a broker topic to a station signal.
*
* The demo model expects the signals declared in server/sim/stations.js. Any
* topic not mapped here is ignored; any signal not supplied by the broker simply
* has no data, and the UI shows it as such rather than inventing a value.
*/
export const TAG_MAP = {
// 'plant/line1/conveyor01/belt_speed': { station: 'CONV-01', signal: 'beltSpeed' },
// 'plant/line1/conveyor01/motor_current': { station: 'CONV-01', signal: 'motorAmps' },
// 'plant/line1/cnc02/vibration_rms': { station: 'CNC-02', signal: 'vibration' },
// 'plant/line1/cnc02/spindle_load': { station: 'CNC-02', signal: 'spindleLoad' },
// 'plant/line1/oven03/zone2_pv': { station: 'OVN-03', signal: 'zone2Temp' },
// 'plant/line1/oven03/zone2_sp': { station: 'OVN-03', signal: 'setpoint' },
// 'plant/line1/ins04/reject_rate': { station: 'INS-04', signal: 'rejectRate' },
// 'plant/line1/pkg05/units_per_min': { station: 'PKG-05', signal: 'unitsPerMin' },
};
export class MqttSource extends TelemetrySource {
constructor({ url, username, password, topicPrefix } = {}) {
super('mqtt');
this.url = url;
this.username = username;
this.password = password;
this.topicPrefix = topicPrefix;
this.analytics = new AnalyticsEngine();
}
/** A real broker feed is read-only: you observe the plant, you do not drive it. */
get capabilities() {
return { timeControl: false, faultInjection: false, setpointControl: false };
}
async start() {
throw new Error(
'MQTT source is not configured. This is a documented stub.\n' +
'To enable it: npm i mqtt, populate TAG_MAP in server/ingest/mqttSource.js ' +
'with your topic names, implement start(), then set TELEMETRY_SOURCE=mqtt ' +
'and MQTT_URL in .env.\n' +
'Run with TELEMETRY_SOURCE=simulated for the demo.',
);
/* Implementation outline:
*
* const mqtt = await import('mqtt');
* this.client = mqtt.connect(this.url, { username: this.username, password: this.password });
* this.client.on('connect', () => this.client.subscribe(Object.keys(TAG_MAP)));
*
* // Accumulate the latest value per tag. Industrial tags publish on change,
* // at wildly different rates, so you assemble a frame on a timer rather
* // than trying to emit one per message.
* this.client.on('message', (topic, payload) => {
* const tag = TAG_MAP[topic];
* if (!tag) return;
* this.values[`${tag.station}.${tag.signal}`] = Number(payload.toString());
* });
*
* this.timer = setInterval(() => this.assembleAndEmit(), 500);
*
* assembleAndEmit() builds the same frame shape SimulatedSource emits:
* stations with their signals, KPI rollups (see server/sim/kpi.js - the
* KpiTracker works on any counter source, not just the simulator), then
* this.analytics.update(snapshot) and this.emit(frame).
*
* Two things that bite in the real world:
* - Staleness. Track a per-tag last-seen timestamp and mark a station
* offline when its tags go quiet, exactly as the F4 dropout fault does.
* Never let a stale value render as if it were live.
* - Units. Vibration in in/s, temperature in F, and pressure in psi are all
* common. Convert at the boundary here, not downstream.
*/
}
async stop() {
if (this.timer) clearInterval(this.timer);
if (this.client) this.client.end();
}
}
server/ingest/opcuaSource.js
+119
View File
@@ -0,0 +1,119 @@
/**
* OPC-UA telemetry source - DOCUMENTED STUB.
*
* OPC-UA is usually the right answer when the customer already has a PLC or SCADA
* layer, because it gives you a browsable address space, real subscriptions, and
* server-side timestamps rather than a flat topic namespace.
*
* To implement:
* 1. npm i node-opcua
* 2. Fill NODE_MAP with the customer's actual NodeIds (browse the server first).
* 3. Implement start() as marked below.
* 4. Set TELEMETRY_SOURCE=opcua and OPCUA_ENDPOINT in .env
*
* For testing without a plant, Prosys OPC-UA Simulation Server or the
* node-opcua sample server both work locally.
*/
import { TelemetrySource } from './source.js';
import { AnalyticsEngine } from '../analytics/alarms.js';
/**
* Maps an OPC-UA NodeId to a station signal.
*
* NodeIds are namespace-qualified and installation-specific - never guess them.
* Browse the server's address space and read them off.
*/
export const NODE_MAP = {
// 'ns=2;s=Line1.CONV01.BeltSpeed': { station: 'CONV-01', signal: 'beltSpeed' },
// 'ns=2;s=Line1.CNC02.VibrationRMS': { station: 'CNC-02', signal: 'vibration' },
// 'ns=2;s=Line1.CNC02.SpindleLoad': { station: 'CNC-02', signal: 'spindleLoad' },
// 'ns=2;s=Line1.OVN03.Zone2PV': { station: 'OVN-03', signal: 'zone2Temp' },
// 'ns=2;s=Line1.OVN03.Zone2SP': { station: 'OVN-03', signal: 'setpoint' },
// 'ns=2;s=Line1.INS04.RejectRate': { station: 'INS-04', signal: 'rejectRate' },
// 'ns=2;s=Line1.PKG05.UnitsPerMin': { station: 'PKG-05', signal: 'unitsPerMin' },
};
export class OpcUaSource extends TelemetrySource {
constructor({ endpoint, securityMode, username, password } = {}) {
super('opcua');
this.endpoint = endpoint;
this.securityMode = securityMode;
this.username = username;
this.password = password;
this.analytics = new AnalyticsEngine();
}
/**
* Read-only by default, deliberately.
*
* OPC-UA can write back to a PLC, and a twin that can change a real setpoint is
* a genuinely useful thing - but it is also a safety-critical action that needs
* interlocks, an audit trail, and the customer's explicit sign-off. Do not turn
* setpointControl on here because the demo UI has a slider.
*/
get capabilities() {
return { timeControl: false, faultInjection: false, setpointControl: false };
}
async start() {
throw new Error(
'OPC-UA source is not configured. This is a documented stub.\n' +
'To enable it: npm i node-opcua, populate NODE_MAP in ' +
'server/ingest/opcuaSource.js with NodeIds browsed from your server, ' +
'implement start(), then set TELEMETRY_SOURCE=opcua and OPCUA_ENDPOINT in .env.\n' +
'Run with TELEMETRY_SOURCE=simulated for the demo.',
);
/* Implementation outline:
*
* const { OPCUAClient, MessageSecurityMode, SecurityPolicy, AttributeIds,
* ClientSubscription, TimestampsToReturn } = await import('node-opcua');
*
* this.client = OPCUAClient.create({ endpointMustExist: false });
* await this.client.connect(this.endpoint);
* this.session = await this.client.createSession(
* this.username ? { userName: this.username, password: this.password } : undefined);
*
* this.subscription = await this.session.createSubscription2({
* requestedPublishingInterval: 500,
* publishingEnabled: true,
* });
*
* for (const [nodeId, tag] of Object.entries(NODE_MAP)) {
* const item = await this.subscription.monitor(
* { nodeId, attributeId: AttributeIds.Value },
* { samplingInterval: 500, queueSize: 10, discardOldest: true },
* TimestampsToReturn.Both);
* item.on('changed', (dataValue) => {
* // Honour the status code. A Bad or Uncertain value must NOT be
* // rendered as live data - that is how a twin starts lying.
* if (!dataValue.statusCode.isGood()) {
* this.markStale(tag);
* return;
* }
* this.values[`${tag.station}.${tag.signal}`] = {
* value: dataValue.value.value,
* // Prefer the SOURCE timestamp: it is when the PLC sampled the
* // sensor, not when the message happened to reach us.
* t: dataValue.sourceTimestamp ?? dataValue.serverTimestamp,
* };
* });
* }
*
* this.timer = setInterval(() => this.assembleAndEmit(), 500);
*
* assembleAndEmit() builds the same frame shape SimulatedSource emits, then
* calls this.analytics.update(snapshot) and this.emit(frame). The analytics
* layer needs no changes at all: TrendTracker and BaselineBank work on
* timestamped values regardless of where they came from.
*/
}
async stop() {
if (this.timer) clearInterval(this.timer);
if (this.subscription) await this.subscription.terminate();
if (this.session) await this.session.close();
if (this.client) await this.client.disconnect();
}
}
server/ingest/simulatedSource.js
+119
View File
@@ -0,0 +1,119 @@
/**
* The simulated telemetry source: owns the model, the clock, and the analytics.
*
* Real time advances at TICK_MS. Simulated time advances at TICK_MS * speed, so
* the operator can run the plant at 60x and watch a twenty-minute degradation
* play out in twenty seconds. Nothing downstream knows or cares.
*/
import { ProductionLine } from '../sim/line.js';
import { AnalyticsEngine } from '../analytics/alarms.js';
import { TelemetrySource } from './source.js';
/** Real milliseconds between broadcast frames. */
export const TICK_MS = 500;
/** Selectable clock multipliers. 0 is paused. */
export const SPEEDS = [0, 1, 5, 20, 60];
/**
* Simulated seconds to run before serving the first frame.
*
* The demo must open on a plant that has been running, not one that just booted.
* Without this the rolling OEE window contains a few seconds of loss-free data and
* reads a perfect 100%, which is precisely the "obviously fabricated" impression
* the model works hard to avoid; the learned anomaly baselines are also not ready,
* so nothing can be detected for the first several minutes.
*/
export const PREWARM_SEC = 1800;
export class SimulatedSource extends TelemetrySource {
constructor({ seed, prewarmSec } = {}) {
super('simulated');
this.line = new ProductionLine(seed);
this.analytics = new AnalyticsEngine();
this.speed = 1;
this.timer = null;
this.prewarmSec = Number.isFinite(prewarmSec) ? prewarmSec : PREWARM_SEC;
}
get capabilities() {
return { timeControl: true, faultInjection: true, setpointControl: true };
}
async start() {
if (this.timer) return;
if (this.prewarmSec > 0) {
const t0 = Date.now();
const step = (TICK_MS / 1000);
const iterations = Math.floor(this.prewarmSec / step);
// Run through the normal tick path so the replay ring and the analytics
// baselines end up in exactly the state they would reach organically.
for (let i = 0; i < iterations; i++) this.tick(step);
console.log(
`[source] pre-warmed ${(this.prewarmSec / 60).toFixed(0)} simulated minutes in ${Date.now() - t0} ms`,
);
} else {
// At minimum emit one frame so a connecting client has something to render.
this.tick(0);
}
this.timer = setInterval(() => this.tick((TICK_MS / 1000) * this.speed), TICK_MS);
}
async stop() {
if (this.timer) clearInterval(this.timer);
this.timer = null;
}
tick(simSeconds) {
if (simSeconds > 0) this.line.step(simSeconds);
const snap = this.line.snapshot();
const analytics = this.analytics.update(snap);
this.emit({
...snap,
analytics,
events: this.line.events.slice(-40).reverse(),
sim: {
speed: this.speed,
paused: this.speed === 0,
speeds: SPEEDS,
tickMs: TICK_MS,
source: this.name,
},
});
}
setSpeed(speed) {
const s = Number(speed);
if (!SPEEDS.includes(s)) throw new Error(`Unsupported speed ${speed}. Allowed: ${SPEEDS.join(', ')}`);
this.speed = s;
this.line.logEvent('action', 'LINE-1', s === 0 ? 'Simulation paused.' : `Simulation speed set to ${s}x.`);
return s;
}
setSetpoint(v) { return this.line.setSetpoint(v); }
setLineSpeed(v) { return this.line.setLineSpeed(v); }
injectFault(id) { return this.line.injectFault(id); }
clearFault(id) { return this.line.clearFault(id); }
toolChange() { return this.line.toolChange(); }
reset() {
this.line.reset();
// The learned baselines belong to the old run; keeping them would flag the
// fresh line as anomalous.
this.analytics.reset();
this.replay = [];
// Re-warm, or Reset would leave the dashboard showing a perfect 100% OEE and
// an anomaly detector with nothing learned - worse than before the reset.
if (this.prewarmSec > 0) {
const step = TICK_MS / 1000;
const iterations = Math.floor(this.prewarmSec / step);
for (let i = 0; i < iterations; i++) this.tick(step);
} else {
this.tick(0);
}
}
}
server/ingest/source.js
+90
View File
@@ -0,0 +1,90 @@
/**
* The telemetry ingest seam.
*
* Everything downstream of this interface - analytics, alarms, the API, the UI,
* the copilot - only ever sees frames. It has no idea whether those frames came
* from a simulator, an MQTT broker, or an OPC-UA server.
*
* That is the whole point of putting a seam here rather than running the
* simulator in the browser: "can it take our data?" is answered by implementing
* one class, not by rewriting the application.
*
* A frame is the ProductionLine snapshot plus the analytics block:
* { t, wallT, lineId, stations[], buffers[], bufferCapacity, kpi, faults[],
* controls, totals, analytics: { alarms[], predictions[], trends[] },
* sim: { speed, paused, tickMs, source } }
*/
/** Frames of history retained for replay to newly connected clients. */
export const REPLAY_FRAMES = 240;
export class TelemetrySource {
constructor(name) {
this.name = name;
this.listeners = new Set();
this.latest = null;
/**
* A short ring of recent frames, trimmed to what the charts need.
*
* Without this, opening the dashboard gives you empty trend charts that take
* minutes of wall-clock to fill - so the first thing a customer sees is a
* dashboard with no history on it. Replaying this on connect means the charts
* are populated the instant the page loads.
*/
this.replay = [];
}
/**
* What this source supports. The UI hides controls a source cannot honour, so
* a read-only historian replay does not show fault-injection buttons that
* would silently do nothing.
*/
get capabilities() {
return { timeControl: false, faultInjection: false, setpointControl: false };
}
onFrame(cb) {
this.listeners.add(cb);
return () => this.listeners.delete(cb);
}
emit(frame) {
this.latest = frame;
this.replay.push({
t: frame.t,
kpi: {
oee: frame.kpi.oee,
availability: frame.kpi.availability,
performance: frame.kpi.performance,
quality: frame.kpi.quality,
},
stations: frame.stations.map((s) => ({ id: s.id, online: s.online, signals: s.signals })),
});
if (this.replay.length > REPLAY_FRAMES) this.replay.shift();
for (const cb of this.listeners) {
try {
cb(frame);
} catch (err) {
console.error(`[${this.name}] frame listener failed:`, err);
}
}
}
async start() {
throw new Error(`${this.name}: start() not implemented`);
}
async stop() {}
// --- optional control surface; sources that cannot do these should throw ---
setSpeed() { throw new Error(`${this.name} does not support time control`); }
setSetpoint() { throw new Error(`${this.name} does not support setpoint control`); }
setLineSpeed() { throw new Error(`${this.name} does not support line speed control`); }
injectFault() { throw new Error(`${this.name} does not support fault injection`); }
clearFault() { throw new Error(`${this.name} does not support fault injection`); }
toolChange() { throw new Error(`${this.name} does not support maintenance actions`); }
reset() { throw new Error(`${this.name} does not support reset`); }
}