Compare commits

..

4 Commits

Author SHA1 Message Date
znetsixe
ef34c82f13 P11.6 wiki regen + Phase 10 private-test rewrites where applicable
For all 11 nodes with auto-gen markers: wiki/Home.md sections 5 (topic
contract) and 9 (data model) regenerated via npm run wiki:all. New
Unit column shows '<measure> (default <unit>)' for declared topics,
'—' otherwise. Effect column now uses descriptor.description (P11.2
field) overriding the generic per-prefix fallback.

For rotatingMachine + reactor: Phase 10 test rewrites — 3 + 8 files
moved off private nodeClass internals (_attachInputHandler, _commands,
_pendingExtras, _registerChild, _tick, etc.) to the public
BaseNodeAdapter surface (node.handlers.input, node.source.*).
+6 / +7 net new tests.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 19:44:15 +02:00
znetsixe
778b2e0c79 P11.5 + B2.1/B2.2: per-command units + description (where applicable)
Adds  to scalar setters whose payloads are
plain numbers OR {value, unit}. Skipped where payload is compound or
mode-dependent (control-%, {F, C: [...]}, etc.) — documented inline.
Every command gains a description field for wikiGen consumption.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 17:41:17 +02:00
znetsixe
c44d5959ad P9.3: wiki/Home.md following 14-section visual-first template + wiki:* scripts
Auto-generated topic-contract + data-model sections via shared wikiGen
script. Hand-written Mermaid diagrams for position-in-platform, code
map, child registration, lifecycle, configuration, state chart (where
applicable).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 15:17:39 +02:00
znetsixe
e02cd1a7a7 P6: convert valveGroupControl to BaseDomain + BaseNodeAdapter + concern split
Refactor of valveGroupControl to use the platform infrastructure (BaseDomain, BaseNodeAdapter,
ChildRouter, commandRegistry, statusBadge). Extracts concerns into
focused modules per .claude/refactor/MODULE_SPLIT.md generic template.
Tests stay green; CONTRACT.md generated; legacy aliases preserved.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-10 22:09:24 +02:00
11 changed files with 997 additions and 915 deletions

67
CONTRACT.md Normal file
View File

@@ -0,0 +1,67 @@
# valveGroupControl — Contract
Hand-maintained for Phase 6; the `## Inputs` table is generated from
`src/commands/index.js` (see Phase 9 generator). Keep ≤ 80 lines.
## Inputs (msg.topic on Port 0)
| Canonical | Aliases (deprecated) | Payload | Effect |
|---|---|---|---|
| `set.mode` | `setMode` | `string` — one of `auto`, `virtualControl`, `fysicalControl`, `maintenance` | Switches the control strategy via `source.setMode(payload)`. |
| `set.position` | `setpoint` | `any` | Reserved for future per-valve positional override; currently a debug-logged no-op pending Phase 7. |
| `child.register` | `registerChild` | `string` — the child node's Node-RED id | Resolves the child via `RED.nodes.getNode` and registers it through `childRegistrationUtils.registerChild(childObj.source, msg.positionVsParent)`. |
| `cmd.execSequence` | `execSequence` | `{ source, action, parameter }` | Forwards to `source.handleInput(source, action, parameter)`. |
| `data.totalFlow` | `totalFlowChange` | numeric, `{ value, position?, variant?, unit? }`, or `{ source, action, ... }` | Updates total measured/predicted flow at the configured position; drives `calcValveFlows` to re-distribute across valves. |
| `cmd.emergencyStop` | `emergencyStop`, `emergencystop` | optional `{ source }` | Runs the `emergencystop` sequence via `handleInput`. |
| `set.reconcileInterval` | `setReconcileInterval` | numeric — seconds (> 0) | Re-tunes the periodic flow-reconciliation interval. Min clamp 100 ms. |
Aliases log a one-time deprecation warning the first time they fire.
## Outputs (msg.topic on Port 0/1/2)
- **Port 0 (process):** `msg.topic = config.general.name`. Payload built by
`outputUtils.formatMsg(..., 'process')` from `getOutput()` — delta-compressed
(only changed fields are emitted). Output keys follow
`<position>_<variant>_<type>` plus `mode` and `maxDeltaP`.
- **Port 1 (InfluxDB telemetry):** same shape as Port 0, formatted with the
`'influxdb'` formatter.
- **Port 2 (registration):** at startup the node sends one
`{ topic: 'child.register', payload: <node.id>, positionVsParent }`
to the upstream parent.
## Events emitted by `source.emitter` / `source.measurements.emitter`
- `output-changed` (`source.emitter`) — public output state shifted; the
adapter listens and pushes Ports 0/1.
- `fluidContractChange` (`source.emitter`) — group-level fluid contract
(status / serviceType / sourceCount) changed. Parents (e.g. an upstream
valve registering this VGC as its parent) subscribe to react.
- `reconcileIntervalChange` (`source.emitter`) — emitted by
`setReconcileIntervalSeconds`; the adapter restarts the tick loop.
- `flow.predicted.atequipment` (`source.measurements.emitter`) — total
predicted group flow (sum of per-valve assigned flows).
- `pressure.predicted.deltaMax` (`source.measurements.emitter`) — max
delta-P across registered valves.
The exact set is data-driven by which sources/valves register and what
they publish; downstream consumers subscribe by event name.
## Children registered by this node
valveGroupControl accepts two child classes through the
`childRegistrationUtils` handshake:
- `valve` — an individual valve. Stored in `source.valves[id]`. VGC binds
to the child's `positionChange` (via `child.state.emitter`) and
`deltaPChange` (via `child.emitter`) events to re-distribute flow and
re-compute group max delta-P.
- `machine` / `rotatingmachine` / `machinegroup` / `machinegroupcontrol` /
`pumpingstation` / `valvegroupcontrol` — an upstream **source**. Stored
in `source.sources[id]`. VGC subscribes to the source's
`flow.predicted.*` / `flow.measured.*` events to drive `updateFlow`,
and reads the child's `getFluidContract()` (if present) plus
`fluidContractChange` events to aggregate the group's upstream service
type (`getFluidContract()` exposes the resolved view).
Position labels accepted from children are `upstream`, `downstream`,
`atEquipment` (and case variants — normalised internally).

View File

@@ -4,7 +4,10 @@
"description": "Valve group control module",
"main": "valveGroupControl.js",
"scripts": {
"test": "node --test test/basic/*.test.js test/integration/*.test.js test/edge/*.test.js"
"test": "node --test test/basic/*.test.js test/integration/*.test.js test/edge/*.test.js",
"wiki:contract": "node ../generalFunctions/scripts/wikiGen.js contract ./src/commands/index.js --write ./wiki/Home.md",
"wiki:datamodel": "node ../generalFunctions/scripts/wikiGen.js datamodel ./src/specificClass.js --write ./wiki/Home.md",
"wiki:all": "npm run wiki:contract && npm run wiki:datamodel"
},
"repository": {
"type": "git",

65
src/commands/handlers.js Normal file
View File

@@ -0,0 +1,65 @@
'use strict';
// Handler functions for valveGroupControl commands. Pure functions:
// source = domain (specificClass) instance
// msg = Node-RED input message
// ctx = { node, RED, send, logger } — provided by BaseNodeAdapter
function _logger(source, ctx) {
return ctx?.logger || source?.logger || null;
}
exports.setMode = (source, msg) => {
source.setMode(msg.payload);
};
exports.setPosition = (source, msg) => {
// Reserved for future per-valve positional override; currently a no-op
// pending Phase 7 topic standardisation of valve setpoint payloads.
_logger(source, null)?.debug?.(`set.position received (no-op): ${JSON.stringify(msg.payload ?? null)}`);
};
exports.registerChild = (source, msg, ctx) => {
const log = _logger(source, ctx);
const childId = msg.payload;
const childObj = ctx?.RED?.nodes?.getNode?.(childId);
if (!childObj || !childObj.source) {
log?.warn?.(`registerChild: child '${childId}' not found or has no .source`);
return;
}
source.childRegistrationUtils.registerChild(childObj.source, msg.positionVsParent);
};
exports.execSequence = async (source, msg) => {
const { source: seqSource, action: seqAction, parameter } = msg.payload || {};
await source.handleInput(seqSource, seqAction, parameter);
};
exports.totalFlowChange = async (source, msg) => {
const payload = msg.payload || {};
if (payload && typeof payload === 'object' && Object.prototype.hasOwnProperty.call(payload, 'source')) {
const src = payload.source || 'parent';
const action = payload.action || 'totalFlowChange';
await source.handleInput(src, action, payload);
return;
}
await source.handleInput('parent', 'totalFlowChange', payload);
};
exports.emergencyStop = async (source, msg) => {
const payload = msg.payload || {};
const src = payload.source || 'parent';
await source.handleInput(src, 'emergencystop');
};
exports.setReconcileInterval = (source, msg) => {
const log = _logger(source, null);
const nextSec = Number(msg.payload);
if (!Number.isFinite(nextSec) || nextSec <= 0) {
log?.warn?.(`Invalid reconcile interval payload '${msg.payload}'. Expected seconds > 0.`);
return;
}
if (typeof source.setReconcileIntervalSeconds === 'function') {
source.setReconcileIntervalSeconds(nextSec);
}
};

63
src/commands/index.js Normal file
View File

@@ -0,0 +1,63 @@
'use strict';
// valveGroupControl command registry. Consumed by BaseNodeAdapter via
// `static commands = require('./commands')`. Canonical names follow
// CONTRACTS.md §1; legacy names live in `aliases` and emit a one-time
// deprecation warning at runtime.
const handlers = require('./handlers');
module.exports = [
{
topic: 'set.mode',
aliases: ['setMode'],
payloadSchema: { type: 'string' },
description: 'Switch the valve group between auto / manual control modes.',
handler: handlers.setMode,
},
{
topic: 'set.position',
aliases: ['setpoint'],
payloadSchema: { type: 'any' },
// Control-percent setpoint — no `units` (no `percent` measure in convert).
description: 'Set the group-level valve position (currently a no-op pending Phase 7).',
handler: handlers.setPosition,
},
{
topic: 'child.register',
aliases: ['registerChild'],
payloadSchema: { type: 'string' },
description: 'Register a child valve with this group.',
handler: handlers.registerChild,
},
{
topic: 'cmd.execSequence',
aliases: ['execSequence'],
payloadSchema: { type: 'object' },
description: 'Run a group-wide sequence (startup / shutdown / emergencystop).',
handler: handlers.execSequence,
},
{
topic: 'data.totalFlow',
aliases: ['totalFlowChange'],
payloadSchema: { type: 'any' },
// Compound payload `{source, action, ...}` in some shapes — no scalar
// normalisation. The handler routes by payload.source.
description: 'Notify the group that the total flow setpoint has changed.',
handler: handlers.totalFlowChange,
},
{
topic: 'cmd.emergencyStop',
aliases: ['emergencyStop', 'emergencystop'],
payloadSchema: { type: 'any' },
description: 'Trigger an emergency stop across all valves in the group.',
handler: handlers.emergencyStop,
},
{
topic: 'set.reconcileInterval',
aliases: ['setReconcileInterval'],
payloadSchema: { type: 'any' },
description: 'Update the reconciliation interval (seconds).',
handler: handlers.setReconcileInterval,
},
];

View File

@@ -0,0 +1,126 @@
'use strict';
// Per-valve flow distribution. Splits the group's total flow across
// available valves proportional to Kv, then asks each valve back what
// flow it actually accepted and re-balances the residual. Also surfaces
// max delta-P across the group for downstream readers.
const DEFAULT_RECONCILIATION = Object.freeze({ maxPasses: 2, residualTolerance: 0.001 });
function isValveAvailable(valve) {
const currentState = valve?.state?.getCurrentState?.();
const mode = valve?.currentMode;
const kv = Number(valve?.kv);
return (
currentState !== 'off'
&& currentState !== 'maintenance'
&& mode !== 'maintenance'
&& Number.isFinite(kv)
&& kv > 0
);
}
function listAvailableValves(valves) {
return Object.entries(valves)
.filter(([, valve]) => isValveAvailable(valve))
.map(([id, valve]) => ({ id, valve }));
}
function _readAcceptedFlow(valve, flowUnit) {
const accepted = Number(
valve?.measurements
?.type('flow')
?.variant('predicted')
?.position('downstream')
?.getCurrentValue(flowUnit)
);
return Number.isFinite(accepted) ? accepted : null;
}
function solveFlowDistribution(totalFlow, availableEntries, reconciliation, flowUnit) {
const totalKv = availableEntries.reduce((sum, { valve }) => sum + Number(valve.kv), 0);
if (!Number.isFinite(totalKv) || totalKv <= 0) {
return { flowsById: {}, residual: Number(totalFlow) || 0, passes: 0 };
}
const targetById = Object.fromEntries(availableEntries.map(({ id }) => [id, 0]));
let residual = Number(totalFlow);
let passes = 0;
const maxPasses = Math.max(1, Number(reconciliation?.maxPasses) || DEFAULT_RECONCILIATION.maxPasses);
const tolerance = Math.max(0, Number(reconciliation?.residualTolerance) || DEFAULT_RECONCILIATION.residualTolerance);
while (passes < maxPasses && Number.isFinite(residual) && Math.abs(residual) > tolerance) {
availableEntries.forEach(({ id, valve }) => {
const share = (Number(valve.kv) / totalKv) * residual;
targetById[id] = Number(targetById[id]) + share;
valve.updateFlow('predicted', targetById[id], 'downstream', flowUnit);
});
let acceptedTotal = 0;
availableEntries.forEach(({ id, valve }) => {
const accepted = _readAcceptedFlow(valve, flowUnit);
if (Number.isFinite(accepted)) {
targetById[id] = accepted;
acceptedTotal += accepted;
return;
}
acceptedTotal += Number(targetById[id]) || 0;
});
residual = Number(totalFlow) - acceptedTotal;
passes += 1;
}
return { flowsById: targetById, residual: Number.isFinite(residual) ? residual : 0, passes };
}
function distributeFlow(vgc) {
const flowUnit = vgc.unitPolicy.output('flow');
const totalFlowMeasured = vgc._read('flow', 'measured', 'atEquipment', flowUnit);
const totalFlowPredicted = vgc._read('flow', 'predicted', 'atEquipment', flowUnit);
const totalFlow = Number.isFinite(totalFlowMeasured) ? totalFlowMeasured : totalFlowPredicted;
if (!Number.isFinite(totalFlow)) return;
const availableEntries = listAvailableValves(vgc.valves);
const availableIds = new Set(availableEntries.map((entry) => entry.id));
const totalKv = availableEntries.reduce((sum, { valve }) => sum + Number(valve.kv), 0);
if (!availableEntries.length || !Number.isFinite(totalKv) || totalKv <= 0) {
vgc.logger.warn('No available valves with valid Kv, setting assigned flow to 0.');
for (const valve of Object.values(vgc.valves)) {
valve.updateFlow('predicted', 0, 'downstream', flowUnit);
}
vgc._write('flow', 'predicted', 'atEquipment', 0, flowUnit);
vgc.lastFlowSolve = { passes: 0, residual: Number(totalFlow) || 0, targetTotal: Number(totalFlow) || 0, assignedTotal: 0 };
return;
}
const solve = solveFlowDistribution(totalFlow, availableEntries, vgc.flowReconciliation, flowUnit);
let assignedTotal = 0;
for (const [id, valve] of Object.entries(vgc.valves)) {
const flow = availableIds.has(id) ? (solve.flowsById[id] || 0) : 0;
valve.updateFlow('predicted', flow, 'downstream', flowUnit);
assignedTotal += flow;
}
vgc._write('flow', 'predicted', 'atEquipment', assignedTotal, flowUnit);
vgc.lastFlowSolve = { passes: solve.passes, residual: solve.residual, targetTotal: totalFlow, assignedTotal };
calcMaxDeltaP(vgc);
}
function calcMaxDeltaP(vgc) {
const pUnit = vgc.unitPolicy.output('pressure');
let maxDeltaP = 0;
for (const [id, valve] of Object.entries(vgc.valves)) {
const deltaP = Number(
valve.measurements.type('pressure').variant('predicted').position('delta').getCurrentValue(pUnit)
);
if (!Number.isFinite(deltaP)) continue;
vgc.logger.debug(`Delta P for valve ${id}: ${deltaP}`);
if (deltaP > maxDeltaP) maxDeltaP = deltaP;
}
vgc.maxDeltaP = maxDeltaP;
vgc._write('pressure', 'predicted', 'deltaMax', maxDeltaP, pUnit);
}
module.exports = { distributeFlow, calcMaxDeltaP, listAvailableValves, isValveAvailable, DEFAULT_RECONCILIATION };

47
src/io/output.js Normal file
View File

@@ -0,0 +1,47 @@
'use strict';
// getOutput + getStatusBadge composition for valveGroupControl. Keeps
// the orchestrator under its file-size budget.
const { statusBadge } = require('generalFunctions');
function _outputUnitForType(unitPolicy, type) {
switch (String(type || '').toLowerCase()) {
case 'flow': return unitPolicy.output('flow');
case 'pressure': return unitPolicy.output('pressure');
default: return null;
}
}
function getOutput(vgc) {
const out = {};
const measurements = vgc.measurements;
Object.entries(measurements.measurements || {}).forEach(([type, variants]) => {
Object.entries(variants || {}).forEach(([variant, positions]) => {
const unit = _outputUnitForType(vgc.unitPolicy, type);
Object.keys(positions || {}).forEach((position) => {
const value = measurements.type(type).variant(variant).position(position).getCurrentValue(unit || undefined);
if (value != null) out[`${position}_${variant}_${type}`] = value;
});
});
});
out.mode = vgc.currentMode;
out.maxDeltaP = vgc.maxDeltaP;
return out;
}
function getStatusBadge(vgc) {
const flowUnit = vgc.unitPolicy.output('flow');
const measured = vgc.measurements.type('flow').variant('measured').position('atEquipment').getCurrentValue(flowUnit);
const predicted = vgc.measurements.type('flow').variant('predicted').position('atEquipment').getCurrentValue(flowUnit);
const raw = Number.isFinite(measured) ? measured : predicted;
const totalFlow = Number.isFinite(raw) ? Math.round(raw) : 0;
const available = vgc.getAvailableValves();
const status = available.length > 0 ? `${available.length} valve(s) connected` : 'No valves';
return statusBadge.text(
`${vgc.currentMode} | flow=${totalFlow} ${flowUnit} | ${status}`,
{ fill: available.length > 0 ? 'green' : 'red', shape: 'dot' }
);
}
module.exports = { getOutput, getStatusBadge };

View File

@@ -1,258 +1,33 @@
const { outputUtils, configManager, convert } = require("generalFunctions");
const Specific = require("./specificClass");
'use strict';
class nodeClass {
/**
* Create a MeasurementNode.
* @param {object} uiConfig - Node-RED node configuration.
* @param {object} RED - Node-RED runtime API.
* @param {object} nodeInstance - The Node-RED node instance.
* @param {string} nameOfNode - The name of the node, used for
*/
constructor(uiConfig, RED, nodeInstance, nameOfNode) {
// Preserve RED reference for HTTP endpoints if needed
this.node = nodeInstance; // This is the Node-RED node instance, we can use this to send messages and update status
this.RED = RED; // This is the Node-RED runtime API, we can use this to create endpoints if needed
this.name = nameOfNode; // This is the name of the node, it should match the file name and the node type in Node-RED
this.source = null; // Will hold the specific class instance
const { BaseNodeAdapter } = require('generalFunctions');
const ValveGroupControl = require('./specificClass');
const commands = require('./commands');
// Load default & UI config
this._loadConfig(uiConfig, this.node);
this._reconcileIntervalMs = this._resolveReconcileIntervalMs(uiConfig);
// Tick-driven: a periodic reconcile pass re-balances per-valve flow if
// a child's accepted value drifts between event-driven recalcs.
class nodeClass extends BaseNodeAdapter {
static DomainClass = ValveGroupControl;
static commands = commands;
static tickInterval = 1000;
static statusInterval = 1000;
// Instantiate core Measurement class
this._setupSpecificClass();
buildDomainConfig() { return {}; }
// Wire up event and lifecycle handlers
this._bindEvents();
this._registerChild();
this._startTickLoop();
this._attachInputHandler();
this._attachCloseHandler();
extraSetup() {
this.source?.emitter?.on?.('reconcileIntervalChange', (ms) => this._restartTick(ms));
}
/**
* Load and merge default config with user-defined settings.
* @param {object} uiConfig - Raw config from Node-RED UI.
*/
_loadConfig(uiConfig, node) {
const cfgMgr = new configManager();
this.defaultConfig = cfgMgr.getConfig(this.name);
// Resolve flow unit with validation before building config
const flowUnit = this._resolveUnitOrFallback(uiConfig.unit, 'volumeFlowRate', 'm3/h', 'flow');
const resolvedUiConfig = { ...uiConfig, unit: flowUnit };
// Build config: base sections (no domain-specific config for group controller)
this.config = cfgMgr.buildConfig(this.name, resolvedUiConfig, node.id);
// Utility for formatting outputs
this._output = new outputUtils();
}
_resolveUnitOrFallback(candidate, expectedMeasure, fallbackUnit, label) {
const raw = typeof candidate === "string" ? candidate.trim() : "";
const fallback = String(fallbackUnit || "").trim();
if (!raw) {
return fallback;
}
try {
const desc = convert().describe(raw);
if (expectedMeasure && desc.measure !== expectedMeasure) {
throw new Error(`expected '${expectedMeasure}' but got '${desc.measure}'`);
}
return raw;
} catch (error) {
this.node?.warn?.(`Invalid ${label} unit '${raw}' (${error.message}). Falling back to '${fallback}'.`);
return fallback;
}
}
_resolveReconcileIntervalMs(uiConfig) {
const raw = Number(
uiConfig?.reconcileIntervalSeconds
?? uiConfig?.reconcileIntervalSec
?? uiConfig?.reconcileEverySeconds
?? 1
);
const sec = Number.isFinite(raw) && raw > 0 ? raw : 1;
return Math.max(100, Math.round(sec * 1000));
}
_updateNodeStatus() {
const vg = this.source;
const mode = vg.currentMode;
const flowUnit = vg?.unitPolicy?.output?.flow || this.config.general.unit || "m3/h";
const measuredFlow = vg.measurements.type("flow").variant("measured").position("atEquipment").getCurrentValue(flowUnit);
const predictedFlow = vg.measurements.type("flow").variant("predicted").position("atEquipment").getCurrentValue(flowUnit);
const totalFlowRaw = Number.isFinite(measuredFlow) ? measuredFlow : predictedFlow;
const totalFlow = Number.isFinite(totalFlowRaw) ? Math.round(totalFlowRaw) : 0;
const availableValves = Array.isArray(vg.getAvailableValves?.()) ? vg.getAvailableValves() : [];
// const totalCapacity = Math.round(vg.dynamicTotals.flow.max * 1) / 1; ADD LATER?
// Determine overall status based on available valves
const status =
availableValves.length > 0
? `${availableValves.length} valve(s) connected`
: "No valves";
// Generate status text in a single line
const text = `${mode} | flow=${totalFlow} ${flowUnit} | ${status}`;
return {
fill: availableValves.length > 0 ? "green" : "red",
shape: "dot",
text,
};
}
/**
* Instantiate the core logic and store as source.
*/
_setupSpecificClass() {
this.source = new Specific(this.config);
this.node.source = this.source; // Store the source in the node instance for easy access
}
/**
* Bind events to Node-RED status updates. Using internal emitter. --> REMOVE LATER WE NEED ONLY COMPLETE CHILDS AND THEN CHECK FOR UPDATES
*/
_bindEvents() {
}
/**
* Register this node as a child upstream and downstream.
* Delayed to avoid Node-RED startup race conditions.
*/
_registerChild() {
setTimeout(() => {
this.node.send([
null,
null,
{
topic: "registerChild",
payload: this.node.id,
positionVsParent:
this.config?.functionality?.positionVsParent || "atEquipment",
},
]);
}, 100);
}
/**
* Start the periodic tick loop to drive the Measurement class.
*/
_startTickLoop() {
setTimeout(() => {
this._tickInterval = setInterval(() => this._tick(), this._reconcileIntervalMs);
// Update node status on nodered screen every second ( this is not the best way to do this, but it works for now)
this._statusInterval = setInterval(() => {
const status = this._updateNodeStatus();
this.node.status(status);
}, 1000);
}, 1000);
}
/**
* Execute a single tick: update measurement, format and send outputs.
*/
_tick() {
if (typeof this.source?.calcValveFlows === 'function') {
this.source.calcValveFlows();
}
const raw = this.source.getOutput();
const processMsg = this._output.formatMsg(raw, this.config, "process");
const influxMsg = this._output.formatMsg(raw, this.config, "influxdb");
// Send only updated outputs on ports 0 & 1
this.node.send([processMsg, influxMsg]);
}
/**
* Attach the node's input handler, routing control messages to the class.
*/
_attachInputHandler() {
this.node.on(
"input",
async (msg, send, done) => {
const vg = this.source;
const RED = this.RED;
try {
switch (msg.topic) {
case "registerChild": {
const childId = msg.payload;
const childObj = RED.nodes.getNode(childId);
if (!childObj || !childObj.source) {
vg.logger.warn(`registerChild skipped: missing child/source for id=${childId}`);
break;
}
vg.childRegistrationUtils.registerChild(childObj.source, msg.positionVsParent);
break;
}
case 'setMode':
vg.setMode(msg.payload);
break;
case 'setReconcileInterval': {
const nextSec = Number(msg.payload);
if (!Number.isFinite(nextSec) || nextSec <= 0) {
vg.logger.warn(`Invalid reconcile interval payload '${msg.payload}'. Expected seconds > 0.`);
break;
}
this._reconcileIntervalMs = Math.max(100, Math.round(nextSec * 1000));
clearInterval(this._tickInterval);
this._tickInterval = setInterval(() => this._tick(), this._reconcileIntervalMs);
vg.logger.info(`Flow reconciliation interval updated to ${nextSec}s (${this._reconcileIntervalMs}ms).`);
break;
}
case 'execSequence': {
const { source: seqSource, action: seqAction, parameter } = msg.payload;
vg.handleInput(seqSource, seqAction, parameter);
break;
}
case 'totalFlowChange': {
const payload = msg.payload || {};
if (payload && typeof payload === "object" && Object.prototype.hasOwnProperty.call(payload, "source")) {
const tfcSource = payload.source || "parent";
const tfcAction = payload.action || "totalFlowChange";
vg.handleInput(tfcSource, tfcAction, payload);
} else {
vg.handleInput("parent", "totalFlowChange", payload);
}
break;
}
case 'emergencystop':
case 'emergencyStop': {
const payload = msg.payload || {};
const esSource = payload.source || "parent";
vg.handleInput(esSource, "emergencystop");
break;
}
default:
vg.logger.warn(`Unknown topic: ${msg.topic}`);
break;
}
} catch (error) {
vg.logger.error(`Input handler failure: ${error.message}`);
}
if (typeof done === 'function') done();
}
);
}
/**
* Clean up timers and intervals when Node-RED stops the node.
*/
_attachCloseHandler() {
this.node.on("close", (done) => {
clearInterval(this._tickInterval);
clearInterval(this._statusInterval);
this.source?.destroy?.();
if (typeof done === 'function') done();
});
_restartTick(ms) {
const next = Math.max(100, Math.round(Number(ms) || 0));
if (!next) return;
if (this._tickInterval) clearInterval(this._tickInterval);
this._tickInterval = setInterval(() => {
try { this.source.tick?.(); }
catch (err) { this.source?.logger?.error?.(`tick threw: ${err.message}`); }
this._emitOutputs();
}, next);
}
}
module.exports = nodeClass; // Export the class for Node-RED to use
module.exports = nodeClass;

View File

@@ -0,0 +1,186 @@
'use strict';
// Upstream-source registration + fluid-contract reconciliation.
// Sources are non-valve upstream children (rotatingMachine, MGC, PS, …)
// that publish flow events and optionally a service-type contract.
// VGC aggregates their contracts into one group-level view that valves
// can read for compatibility checks.
const SERVICE_TYPES = new Set(['gas', 'liquid']);
const SOURCE_SOFTWARE_TYPES = new Set([
'machine',
'rotatingmachine',
'machinegroup',
'machinegroupcontrol',
'pumpingstation',
'valvegroupcontrol',
]);
const SOURCE_FLOW_EVENTS = [
'flow.predicted.downstream',
'flow.predicted.atEquipment',
'flow.predicted.atequipment',
'flow.measured.downstream',
'flow.measured.atEquipment',
'flow.measured.atequipment',
];
const DEFAULT_SOURCE_SERVICE_TYPE = Object.freeze({
machine: 'liquid',
rotatingmachine: 'liquid',
machinegroup: 'liquid',
machinegroupcontrol: 'liquid',
pumpingstation: 'liquid',
});
function normalizeServiceType(value) {
const raw = String(value || '').trim().toLowerCase();
return SERVICE_TYPES.has(raw) ? raw : null;
}
function isSourceSoftwareType(softwareType) {
return SOURCE_SOFTWARE_TYPES.has(String(softwareType || '').trim().toLowerCase());
}
function isSourceLike(child, softwareType) {
if (isSourceSoftwareType(softwareType)) return true;
return typeof child?.getFluidContract === 'function';
}
function extractFluidContract(child, softwareType, logger) {
let contract = null;
if (typeof child?.getFluidContract === 'function') {
try { contract = child.getFluidContract(); }
catch (error) { logger?.warn?.(`Failed to read child fluid contract: ${error.message}`); }
}
const status = String(contract?.status || '').trim().toLowerCase();
if (status === 'conflict') return { status: 'conflict', serviceType: null };
const fromContract = normalizeServiceType(contract?.serviceType);
if (fromContract) return { status: 'resolved', serviceType: fromContract };
const direct = normalizeServiceType(
child?.serviceType || child?.expectedServiceType || child?.config?.asset?.serviceType
);
if (direct) return { status: 'resolved', serviceType: direct };
const fallback = DEFAULT_SOURCE_SERVICE_TYPE[String(softwareType || '').trim().toLowerCase()] || null;
if (fallback) return { status: 'inferred', serviceType: fallback };
return { status: 'unknown', serviceType: null };
}
function _diff(prev, next) {
return (
prev.status !== next.status
|| prev.serviceType !== next.serviceType
|| prev.sourceCount !== next.sourceCount
|| (prev.message || '') !== (next.message || '')
);
}
function refreshFluidContract(vgc) {
const contracts = Object.values(vgc.sources).map((s) => s?.fluidContract || null).filter(Boolean);
const serviceTypes = Array.from(new Set(
contracts.map((c) => normalizeServiceType(c.serviceType)).filter(Boolean)
));
const hasConflict = contracts.some((c) => String(c.status || '').toLowerCase() === 'conflict');
const sourceCount = Object.keys(vgc.sources).length;
let next;
if (hasConflict || serviceTypes.length > 1) {
next = {
status: 'conflict', serviceType: null, upstreamServiceTypes: serviceTypes, sourceCount,
message: `Conflicting upstream fluids detected: ${serviceTypes.join(', ') || 'unknown'}.`,
};
} else if (serviceTypes.length === 1) {
next = {
status: 'resolved', serviceType: serviceTypes[0], upstreamServiceTypes: serviceTypes, sourceCount,
message: `Upstream fluid resolved as ${serviceTypes[0]}.`,
};
} else {
next = {
status: 'unknown', serviceType: null, upstreamServiceTypes: [], sourceCount,
message: 'No upstream fluid sources registered.',
};
}
const prev = vgc.fluidContract || {};
vgc.fluidContract = next;
if (_diff(prev, next)) vgc.emitter.emit('fluidContractChange', vgc.getFluidContract());
}
function registerSource(vgc, child, positionVsParent, softwareType) {
const id = child?.config?.general?.id || child?.config?.general?.name || `source-${Object.keys(vgc.sources).length + 1}`;
if (vgc._sourceListeners.has(id)) unbindSource(vgc, id);
child.positionVsParent = positionVsParent;
vgc.sources[id] = child;
bindSource(vgc, id, child);
vgc.sources[id].fluidContract = extractFluidContract(child, softwareType, vgc.logger);
refreshFluidContract(vgc);
vgc.logger.info(`Source '${id}' (${softwareType || 'unknown'}) registered at ${positionVsParent}.`);
return true;
}
function bindSource(vgc, sourceId, source) {
const listeners = { flow: [], onFluidContractChange: null };
if (source?.measurements?.emitter?.on) {
SOURCE_FLOW_EVENTS.forEach((eventName) => {
const handler = (eventData = {}) => {
const value = Number(eventData.value);
if (!Number.isFinite(value)) return;
const variant = String(eventName).split('.')[1] === 'measured' ? 'measured' : 'predicted';
const unit = eventData.unit || vgc.unitPolicy.output('flow');
vgc.updateFlow(variant, value, 'atEquipment', unit);
};
source.measurements.emitter.on(eventName, handler);
listeners.flow.push({ emitter: source.measurements.emitter, eventName, handler });
});
}
if (source?.emitter?.on) {
listeners.onFluidContractChange = () => {
if (!vgc.sources[sourceId]) return;
vgc.sources[sourceId].fluidContract = extractFluidContract(source, source?.config?.functionality?.softwareType, vgc.logger);
refreshFluidContract(vgc);
};
source.emitter.on('fluidContractChange', listeners.onFluidContractChange);
}
vgc._sourceListeners.set(sourceId, { source, listeners });
}
function unbindSource(vgc, sourceId) {
const entry = vgc._sourceListeners.get(sourceId);
if (!entry) return;
const { source, listeners } = entry;
listeners.flow.forEach(({ emitter, eventName, handler }) => {
if (typeof emitter?.off === 'function') emitter.off(eventName, handler);
else if (typeof emitter?.removeListener === 'function') emitter.removeListener(eventName, handler);
});
if (listeners.onFluidContractChange) {
if (typeof source?.emitter?.off === 'function') source.emitter.off('fluidContractChange', listeners.onFluidContractChange);
else if (typeof source?.emitter?.removeListener === 'function') source.emitter.removeListener('fluidContractChange', listeners.onFluidContractChange);
}
vgc._sourceListeners.delete(sourceId);
}
function getFluidContract(vgc) {
const s = vgc.fluidContract || {};
return {
status: s.status || 'unknown',
serviceType: s.serviceType || null,
upstreamServiceTypes: Array.isArray(s.upstreamServiceTypes) ? [...s.upstreamServiceTypes] : [],
sourceCount: Number(s.sourceCount) || 0,
message: s.message || '',
source: 'valvegroupcontrol',
};
}
module.exports = {
isSourceLike,
isSourceSoftwareType,
registerSource,
unbindSource,
refreshFluidContract,
getFluidContract,
SOURCE_SOFTWARE_TYPES,
};

View File

@@ -1,161 +1,65 @@
/**
* @file valveGroupControl.js
*/
'use strict';
const EventEmitter = require('events');
const { logger, configUtils, configManager, state, MeasurementContainer, childRegistrationUtils, convert } = require('generalFunctions');
// ValveGroupControl — S88 Unit orchestrator coordinating valve children.
// Concern modules under src/{groupOps,sources,io,commands} carry the
// real work; this file stitches them together: registration, valve event
// routing, source fluid-contract aggregation, mode/sequence dispatch.
const CANONICAL_UNITS = Object.freeze({
pressure: 'Pa',
flow: 'm3/s',
});
const { BaseDomain, UnitPolicy, state } = require('generalFunctions');
const flowDist = require('./groupOps/flowDistribution');
const sources = require('./sources/fluidContract');
const io = require('./io/output');
const DEFAULT_IO_UNITS = Object.freeze({
pressure: 'mbar',
flow: 'm3/h',
});
// Source softwareTypes after BaseDomain canonicalisation
// (rotatingmachine→machine, machinegroupcontrol→machinegroup).
const SOURCE_SOFTWARE_TYPES = ['machine', 'machinegroup', 'pumpingstation', 'valvegroupcontrol'];
const KNOWN_POSITIONS = new Set(['upstream', 'downstream', 'atEquipment']);
const SERVICE_TYPES = new Set(['gas', 'liquid']);
const SOURCE_SOFTWARE_TYPES = new Set([
'machine',
'rotatingmachine',
'machinegroup',
'machinegroupcontrol',
'pumpingstation',
'valvegroupcontrol',
]);
const SOURCE_FLOW_EVENTS = [
'flow.predicted.downstream',
'flow.predicted.atEquipment',
'flow.predicted.atequipment',
'flow.measured.downstream',
'flow.measured.atEquipment',
'flow.measured.atequipment',
];
const DEFAULT_SOURCE_SERVICE_TYPE = Object.freeze({
machine: 'liquid',
rotatingmachine: 'liquid',
machinegroup: 'liquid',
machinegroupcontrol: 'liquid',
pumpingstation: 'liquid',
});
const DEFAULT_FLOW_RECONCILIATION = Object.freeze({
maxPasses: 2,
residualTolerance: 0.001,
});
class ValveGroupControl extends BaseDomain {
static name = 'valveGroupControl';
class ValveGroupControl {
constructor(valveGroupControlConfig = {}) {
this.emitter = new EventEmitter();
this.configManager = new configManager();
this.defaultConfig = this.configManager.getConfig('valveGroupControl');
this.configUtils = new configUtils(this.defaultConfig);
this.config = this.configUtils.initConfig(valveGroupControlConfig);
this.unitPolicy = this._buildUnitPolicy(this.config);
static unitPolicy = UnitPolicy.declare({
canonical: { flow: 'm3/s', pressure: 'Pa' },
output: { flow: 'm3/h', pressure: 'mbar' },
requireUnitForTypes: ['pressure', 'flow'],
});
configure() {
this.config = this.configUtils.updateConfig(this.config, {
general: { unit: this.unitPolicy.output.flow },
general: { unit: this.unitPolicy.output('flow') },
});
this.logger = new logger(this.config.general.logging.enabled, this.config.general.logging.logLevel, this.config.general.name);
this.measurements = new MeasurementContainer({
autoConvert: true,
defaultUnits: {
pressure: this.unitPolicy.output.pressure,
flow: this.unitPolicy.output.flow,
},
preferredUnits: {
pressure: this.unitPolicy.output.pressure,
flow: this.unitPolicy.output.flow,
},
canonicalUnits: this.unitPolicy.canonical,
storeCanonical: true,
strictUnitValidation: true,
throwOnInvalidUnit: true,
requireUnitForTypes: ['pressure', 'flow'],
}, this.logger);
this.child = {};
this.valves = {};
this._valveListeners = new Map();
this.sources = {};
this._sourceListeners = new Map();
this.fluidContract = {
status: 'unknown',
serviceType: null,
upstreamServiceTypes: [],
sourceCount: 0,
message: 'No upstream fluid sources registered.',
status: 'unknown', serviceType: null, upstreamServiceTypes: [],
sourceCount: 0, message: 'No upstream fluid sources registered.',
};
this.flowReconciliation = { ...DEFAULT_FLOW_RECONCILIATION };
this.lastFlowSolve = {
passes: 0,
residual: 0,
targetTotal: 0,
assignedTotal: 0,
};
this.flowReconciliation = { ...flowDist.DEFAULT_RECONCILIATION };
this.lastFlowSolve = { passes: 0, residual: 0, targetTotal: 0, assignedTotal: 0 };
this.maxDeltaP = 0;
this.currentMode = this.config.mode.current;
this.childRegistrationUtils = new childRegistrationUtils(this);
this.state = new state({}, this.logger);
this.state.stateManager.currentState = 'operational';
}
registerOnChildEvents() {}
_resolveRegistrationContext(child, positionVsParentOrSoftwareType) {
const fromArg = String(positionVsParentOrSoftwareType || '').trim();
if (KNOWN_POSITIONS.has(fromArg)) {
return {
positionVsParent: fromArg,
softwareType: child?.config?.functionality?.softwareType || null,
};
this.router.onRegister('valve', (child) => this._registerValve(child));
for (const swType of SOURCE_SOFTWARE_TYPES) {
this.router.onRegister(swType, (child, canonicalKey) => this._registerSource(child, canonicalKey));
}
return {
positionVsParent: child?.positionVsParent || 'atEquipment',
softwareType: fromArg || child?.config?.functionality?.softwareType || null,
};
}
_isValveLike(child) {
return Boolean(
child
&& typeof child.updateFlow === 'function'
&& child.state
&& typeof child.state.getCurrentState === 'function'
&& child.state && typeof child.state.getCurrentState === 'function'
&& child.measurements
);
}
_isSourceLike(child, softwareType) {
const type = String(softwareType || child?.config?.functionality?.softwareType || '').trim().toLowerCase();
if (SOURCE_SOFTWARE_TYPES.has(type)) {
return true;
}
return typeof child?.getFluidContract === 'function';
}
registerChild(child, positionVsParentOrSoftwareType) {
const ctx = this._resolveRegistrationContext(child, positionVsParentOrSoftwareType);
const softwareType = String(ctx.softwareType || child?.config?.functionality?.softwareType || '').trim().toLowerCase();
if (softwareType === 'valve' || (!softwareType && this._isValveLike(child))) {
return this._registerValve(child, ctx.positionVsParent);
}
if (this._isSourceLike(child, softwareType)) {
return this._registerSource(child, ctx.positionVsParent, softwareType);
}
this.logger.warn(`registerChild skipped: unsupported child type '${softwareType || 'unknown'}'`);
return false;
}
_registerValve(child, positionVsParent) {
_registerValve(child) {
if (!this._isValveLike(child)) {
this.logger.warn('registerChild skipped: child is not valve-like');
return false;
@@ -165,314 +69,154 @@ class ValveGroupControl {
this.logger.debug(`registerChild skipped: valve ${id} already registered`);
return true;
}
const positionVsParent = child.positionVsParent
|| child.config?.functionality?.positionVsParent
|| 'atEquipment';
child.positionVsParent = positionVsParent;
this.valves[id] = child;
this._bindValveEvents(id, child);
this.calcValveFlows();
this.calcMaxDeltaP();
this._refreshFluidContract();
sources.refreshFluidContract(this);
this.logger.info(`Valve '${id}' registered at ${positionVsParent}.`);
return true;
}
_registerSource(child, positionVsParent, softwareType) {
const id = child?.config?.general?.id || child?.config?.general?.name || `source-${Object.keys(this.sources).length + 1}`;
if (this._sourceListeners.has(id)) {
this._unbindSourceEvents(id);
}
child.positionVsParent = positionVsParent;
this.sources[id] = child;
this._bindSourceEvents(id, child);
const contract = this._extractFluidContractFromChild(child, softwareType);
this.sources[id].fluidContract = contract;
this._refreshFluidContract();
this.logger.info(`Source '${id}' (${softwareType || 'unknown'}) registered at ${positionVsParent}.`);
return true;
_registerSource(child, softwareType) {
const positionVsParent = child.positionVsParent
|| child.config?.functionality?.positionVsParent
|| 'atEquipment';
return sources.registerSource(this, child, positionVsParent, softwareType);
}
_bindValveEvents(valveId, valve) {
const handlers = {
onPositionChange: () => {
this.logger.debug(`Valve ${valveId} position changed, recalculating flows.`);
this.calcValveFlows();
},
onDeltaPChange: () => {
this.logger.debug(`Valve ${valveId} deltaP changed, recalculating max deltaP.`);
this.calcMaxDeltaP();
},
onPositionChange: () => { this.logger.debug(`Valve ${valveId} position changed, recalculating flows.`); this.calcValveFlows(); },
onDeltaPChange: () => { this.logger.debug(`Valve ${valveId} deltaP changed, recalculating max deltaP.`); this.calcMaxDeltaP(); },
};
if (valve.state?.emitter?.on) {
valve.state.emitter.on('positionChange', handlers.onPositionChange);
}
if (valve.emitter?.on) {
valve.emitter.on('deltaPChange', handlers.onDeltaPChange);
}
if (valve.state?.emitter?.on) valve.state.emitter.on('positionChange', handlers.onPositionChange);
if (valve.emitter?.on) valve.emitter.on('deltaPChange', handlers.onDeltaPChange);
this._valveListeners.set(valveId, { valve, handlers });
}
_unbindValveEvents(valveId) {
const listener = this._valveListeners.get(valveId);
if (!listener) {
return;
}
const { valve, handlers } = listener;
if (handlers.onPositionChange && valve.state?.emitter?.off) {
valve.state.emitter.off('positionChange', handlers.onPositionChange);
}
if (handlers.onDeltaPChange && valve.emitter?.off) {
valve.emitter.off('deltaPChange', handlers.onDeltaPChange);
}
const entry = this._valveListeners.get(valveId);
if (!entry) return;
const { valve, handlers } = entry;
if (handlers.onPositionChange && valve.state?.emitter?.off) valve.state.emitter.off('positionChange', handlers.onPositionChange);
if (handlers.onDeltaPChange && valve.emitter?.off) valve.emitter.off('deltaPChange', handlers.onDeltaPChange);
this._valveListeners.delete(valveId);
}
_bindSourceEvents(sourceId, source) {
const listeners = {
flow: [],
onFluidContractChange: null,
};
if (source?.measurements?.emitter?.on) {
SOURCE_FLOW_EVENTS.forEach((eventName) => {
const handler = (eventData = {}) => {
this._handleSourceFlowEvent(eventName, eventData);
};
source.measurements.emitter.on(eventName, handler);
listeners.flow.push({
emitter: source.measurements.emitter,
eventName,
handler,
});
});
}
if (source?.emitter?.on) {
listeners.onFluidContractChange = () => {
const contract = this._extractFluidContractFromChild(
source,
source?.config?.functionality?.softwareType
);
if (!this.sources[sourceId]) {
return;
}
this.sources[sourceId].fluidContract = contract;
this._refreshFluidContract();
};
source.emitter.on('fluidContractChange', listeners.onFluidContractChange);
}
this._sourceListeners.set(sourceId, { source, listeners });
}
_unbindSourceEvents(sourceId) {
const listener = this._sourceListeners.get(sourceId);
if (!listener) {
return;
}
const { source, listeners } = listener;
listeners.flow.forEach(({ emitter, eventName, handler }) => {
if (typeof emitter?.off === 'function') {
emitter.off(eventName, handler);
} else if (typeof emitter?.removeListener === 'function') {
emitter.removeListener(eventName, handler);
}
});
if (listeners.onFluidContractChange) {
if (typeof source?.emitter?.off === 'function') {
source.emitter.off('fluidContractChange', listeners.onFluidContractChange);
} else if (typeof source?.emitter?.removeListener === 'function') {
source.emitter.removeListener('fluidContractChange', listeners.onFluidContractChange);
}
}
this._sourceListeners.delete(sourceId);
}
_handleSourceFlowEvent(eventName, eventData = {}) {
const value = Number(eventData.value);
if (!Number.isFinite(value)) {
return;
}
const eventParts = String(eventName || '').split('.');
const variant = eventParts[1] === 'measured' ? 'measured' : 'predicted';
const unit = eventData.unit || this.unitPolicy.output.flow;
this.updateFlow(variant, value, 'atEquipment', unit);
}
_normalizeOptionalServiceType(value) {
const raw = String(value || '').trim().toLowerCase();
if (SERVICE_TYPES.has(raw)) {
return raw;
}
return null;
}
_deriveDefaultServiceTypeForSoftwareType(softwareType) {
const key = String(softwareType || '').trim().toLowerCase();
return DEFAULT_SOURCE_SERVICE_TYPE[key] || null;
}
_extractFluidContractFromChild(child, softwareType) {
let contract = null;
if (typeof child?.getFluidContract === 'function') {
try {
contract = child.getFluidContract();
} catch (error) {
this.logger.warn(`Failed to read child fluid contract: ${error.message}`);
}
}
const contractStatus = String(contract?.status || '').trim().toLowerCase();
if (contractStatus === 'conflict') {
return { status: 'conflict', serviceType: null };
}
const serviceTypeFromContract = this._normalizeOptionalServiceType(contract?.serviceType);
if (serviceTypeFromContract) {
return { status: 'resolved', serviceType: serviceTypeFromContract };
}
const directType = this._normalizeOptionalServiceType(
child?.serviceType
|| child?.expectedServiceType
|| child?.config?.asset?.serviceType
);
if (directType) {
return { status: 'resolved', serviceType: directType };
}
const fallbackType = this._deriveDefaultServiceTypeForSoftwareType(softwareType);
if (fallbackType) {
return { status: 'inferred', serviceType: fallbackType };
}
return { status: 'unknown', serviceType: null };
}
_refreshFluidContract() {
const contracts = Object.values(this.sources)
.map((source) => source?.fluidContract || null)
.filter(Boolean);
const serviceTypes = Array.from(new Set(
contracts
.map((contract) => this._normalizeOptionalServiceType(contract.serviceType))
.filter(Boolean)
));
const hasConflict = contracts.some((contract) => String(contract.status || '').toLowerCase() === 'conflict');
let next = null;
if (hasConflict || serviceTypes.length > 1) {
next = {
status: 'conflict',
serviceType: null,
upstreamServiceTypes: serviceTypes,
sourceCount: Object.keys(this.sources).length,
message: `Conflicting upstream fluids detected: ${serviceTypes.join(', ') || 'unknown'}.`,
};
} else if (serviceTypes.length === 1) {
next = {
status: 'resolved',
serviceType: serviceTypes[0],
upstreamServiceTypes: serviceTypes,
sourceCount: Object.keys(this.sources).length,
message: `Upstream fluid resolved as ${serviceTypes[0]}.`,
};
} else {
next = {
status: 'unknown',
serviceType: null,
upstreamServiceTypes: [],
sourceCount: Object.keys(this.sources).length,
message: 'No upstream fluid sources registered.',
};
}
const prev = this.fluidContract || {};
const changed = (
prev.status !== next.status
|| prev.serviceType !== next.serviceType
|| prev.sourceCount !== next.sourceCount
|| (prev.message || '') !== (next.message || '')
);
this.fluidContract = next;
if (changed) {
this.emitter.emit('fluidContractChange', this.getFluidContract());
}
}
getFluidContract() {
const state = this.fluidContract || {};
return {
status: state.status || 'unknown',
serviceType: state.serviceType || null,
upstreamServiceTypes: Array.isArray(state.upstreamServiceTypes) ? [...state.upstreamServiceTypes] : [],
sourceCount: Number(state.sourceCount) || 0,
message: state.message || '',
source: 'valvegroupcontrol',
};
}
registerOnChildEvents() {}
destroy() {
for (const valveId of this._valveListeners.keys()) {
this._unbindValveEvents(valveId);
}
for (const sourceId of this._sourceListeners.keys()) {
this._unbindSourceEvents(sourceId);
for (const id of this._valveListeners.keys()) this._unbindValveEvents(id);
for (const id of this._sourceListeners.keys()) sources.unbindSource(this, id);
}
// ── measurement read/write helpers used by concern modules ─────────
_outputUnitForType(type) {
switch (String(type || '').toLowerCase()) {
case 'flow': return this.unitPolicy.output('flow');
case 'pressure': return this.unitPolicy.output('pressure');
default: return null;
}
}
_isValveAvailable(valve) {
const currentState = valve.state.getCurrentState();
const mode = valve.currentMode;
const kv = Number(valve.kv);
return (
currentState !== 'off'
&& currentState !== 'maintenance'
&& mode !== 'maintenance'
&& Number.isFinite(kv)
&& kv > 0
);
_read(type, variant, position, unit = null) {
const u = unit || this._outputUnitForType(type);
return this.measurements.type(type).variant(variant).position(position).getCurrentValue(u || undefined);
}
getAvailableValves() {
return Object.entries(this.valves)
.filter(([, valve]) => this._isValveAvailable(valve))
.map(([id, valve]) => ({ id, valve }));
_write(type, variant, position, value, unit = null, timestamp = Date.now()) {
const v = Number(value);
if (!Number.isFinite(v)) return;
this.measurements.type(type).variant(variant).position(position).value(v, timestamp, unit || undefined);
}
// ── public surface used by adapter, tests, commands, valves ────────
getAvailableValves() { return flowDist.listAvailableValves(this.valves); }
calcValveFlows() { flowDist.distributeFlow(this); this.notifyOutputChanged(); }
calcMaxDeltaP() { flowDist.calcMaxDeltaP(this); }
getFluidContract() { return sources.getFluidContract(this); }
isValidSourceForMode(source, mode) {
const allowedSourcesSet = this.config.mode.allowedSources[mode] || [];
return allowedSourcesSet.has(source);
}
setMode(newMode) {
const availableModes = Array.isArray(this.defaultConfig?.mode?.current?.rules?.values)
? this.defaultConfig.mode.current.rules.values.map((m) => m.value)
: Object.keys(this.config?.mode?.allowedSources || {});
if (!availableModes.includes(newMode)) {
this.logger.warn(`Invalid mode '${newMode}'. Allowed modes are: ${availableModes.join(', ')}`);
return;
}
this.currentMode = newMode;
this.logger.info(`Mode successfully changed to '${newMode}'.`);
this.notifyOutputChanged();
}
async executeSequence(sequenceName) {
const sequence = this.config.sequences[sequenceName];
if (!sequence || sequence.size === 0) {
this.logger.warn(`Sequence '${sequenceName}' not defined.`);
return;
}
this.logger.info(` --------- Executing sequence: ${sequenceName} -------------`);
for (const stateName of sequence) {
try { await this.state.transitionToState(stateName); }
catch (error) { this.logger.error(`Error during sequence '${sequenceName}': ${error}`); break; }
}
}
updateFlow(variant, value, position, unit = this.unitPolicy.output('flow')) {
if (value === null || value === undefined) {
this.logger.warn(`Received null or undefined value for flow update. Variant: ${variant}, Position: ${position}`);
return;
}
if (variant !== 'measured' && variant !== 'predicted') {
this.logger.warn(`Unrecognized variant '${variant}' for flow update.`);
return;
}
this.logger.debug(`Updating ${variant} flow for position ${position} with value ${value}`);
this._write('flow', variant, position, value, unit);
this.calcValveFlows();
}
updateMeasurement(variant, subType, value, position, unit) {
this.logger.debug(`---------------------- updating ${subType} ------------------ `);
if (subType === 'flow') {
this.updateFlow(variant, value, position, unit || this.unitPolicy.output('flow'));
return;
}
this.logger.error(`Type '${subType}' not recognized for measured update.`);
}
async handleInput(source, action, parameter) {
if (!this.isValidSourceForMode(source, this.currentMode)) {
const warningTxt = `Source '${source}' is not valid for mode '${this.currentMode}'.`;
this.logger.warn(warningTxt);
return { status: false, feedback: warningTxt };
}
this.logger.info(`Handling input from source '${source}' with action '${action}' in mode '${this.currentMode}'.`);
try {
const flowUnit = this.unitPolicy.output('flow');
switch (action) {
case 'execSequence':
await this.executeSequence(parameter);
break;
case 'totalFlowChange': {
case 'totalFlowChange':
if (parameter && typeof parameter === 'object' && Object.prototype.hasOwnProperty.call(parameter, 'value')) {
await this.updateFlow(parameter.variant || 'measured', parameter.value, parameter.position || 'atEquipment', parameter.unit || this.unitPolicy.output.flow);
await this.updateFlow(parameter.variant || 'measured', parameter.value, parameter.position || 'atEquipment', parameter.unit || flowUnit);
} else if (parameter && typeof parameter === 'object' && Object.prototype.hasOwnProperty.call(parameter, 'q')) {
await this.updateFlow('measured', Number(parameter.q), 'atEquipment', parameter.unit || this.unitPolicy.output.flow);
await this.updateFlow('measured', Number(parameter.q), 'atEquipment', parameter.unit || flowUnit);
} else {
await this.updateFlow('measured', Number(parameter), 'atEquipment', this.unitPolicy.output.flow);
await this.updateFlow('measured', Number(parameter), 'atEquipment', flowUnit);
}
break;
}
case 'emergencyStop':
case 'emergencystop':
this.logger.warn(`Emergency stop activated by '${source}'.`);
@@ -493,288 +237,19 @@ class ValveGroupControl {
}
}
setMode(newMode) {
const availableModes = Array.isArray(this.defaultConfig?.mode?.current?.rules?.values)
? this.defaultConfig.mode.current.rules.values.map((vgc) => vgc.value)
: Object.keys(this.config?.mode?.allowedSources || {});
if (!availableModes.includes(newMode)) {
this.logger.warn(`Invalid mode '${newMode}'. Allowed modes are: ${availableModes.join(', ')}`);
return;
}
this.currentMode = newMode;
this.logger.info(`Mode successfully changed to '${newMode}'.`);
setReconcileIntervalSeconds(sec) {
const ms = Math.max(100, Math.round(Number(sec) * 1000));
this.emitter.emit('reconcileIntervalChange', ms);
this.logger.info(`Flow reconciliation interval updated to ${sec}s (${ms}ms).`);
}
_buildUnitPolicy(config = {}) {
const flowUnit = this._resolveUnitOrFallback(
config?.general?.unit,
'volumeFlowRate',
DEFAULT_IO_UNITS.flow
);
// Periodic reconciliation — adapter fires this each tickInterval. Keeps
// per-valve assigned flow in sync if a child's accepted value drifts
// between event-driven recalcs.
tick() { this.calcValveFlows(); }
return {
canonical: { ...CANONICAL_UNITS },
output: {
flow: flowUnit,
pressure: DEFAULT_IO_UNITS.pressure,
},
};
}
_resolveUnitOrFallback(candidate, expectedMeasure, fallbackUnit) {
const fallback = String(fallbackUnit || '').trim();
const raw = typeof candidate === 'string' ? candidate.trim() : '';
if (!raw) {
return fallback;
}
try {
const desc = convert().describe(raw);
if (expectedMeasure && desc.measure !== expectedMeasure) {
throw new Error(`expected '${expectedMeasure}', got '${desc.measure}'`);
}
return raw;
} catch (error) {
this.logger?.warn?.(`Invalid unit '${raw}' (${error.message}); falling back to '${fallback}'.`);
return fallback;
}
}
_outputUnitForType(type) {
switch (String(type || '').toLowerCase()) {
case 'flow':
return this.unitPolicy.output.flow;
case 'pressure':
return this.unitPolicy.output.pressure;
default:
return null;
}
}
_readMeasurement(type, variant, position, unit = null) {
const requestedUnit = unit || this._outputUnitForType(type);
return this.measurements
.type(type)
.variant(variant)
.position(position)
.getCurrentValue(requestedUnit || undefined);
}
_writeMeasurement(type, variant, position, value, unit = null, timestamp = Date.now()) {
const valueNum = Number(value);
if (!Number.isFinite(valueNum)) {
return;
}
this.measurements
.type(type)
.variant(variant)
.position(position)
.value(valueNum, timestamp, unit || undefined);
}
async executeSequence(sequenceName) {
const sequence = this.config.sequences[sequenceName];
if (!sequence || sequence.size === 0) {
this.logger.warn(`Sequence '${sequenceName}' not defined.`);
return;
}
this.logger.info(` --------- Executing sequence: ${sequenceName} -------------`);
for (const stateName of sequence) {
try {
await this.state.transitionToState(stateName);
} catch (error) {
this.logger.error(`Error during sequence '${sequenceName}': ${error}`);
break;
}
}
}
updateFlow(variant, value, position, unit = this.unitPolicy.output.flow) {
if (value === null || value === undefined) {
this.logger.warn(`Received null or undefined value for flow update. Variant: ${variant}, Position: ${position}`);
return;
}
switch (variant) {
case 'measured':
this.logger.debug(`Updating measured flow for position ${position} with value ${value}`);
this._writeMeasurement('flow', 'measured', position, value, unit);
this.calcValveFlows();
break;
case 'predicted':
this.logger.debug(`Updating predicted flow for position ${position} with value ${value}`);
this._writeMeasurement('flow', 'predicted', position, value, unit);
this.calcValveFlows();
break;
default:
this.logger.warn(`Unrecognized variant '${variant}' for flow update.`);
break;
}
}
updateMeasurement(variant, subType, value, position, unit) {
this.logger.debug(`---------------------- updating ${subType} ------------------ `);
switch (subType) {
case 'flow':
this.updateFlow(variant, value, position, unit || this.unitPolicy.output.flow);
break;
default:
this.logger.error(`Type '${subType}' not recognized for measured update.`);
break;
}
}
calcValveFlows() {
const totalFlowMeasured = this._readMeasurement('flow', 'measured', 'atEquipment', this.unitPolicy.output.flow);
const totalFlowPredicted = this._readMeasurement('flow', 'predicted', 'atEquipment', this.unitPolicy.output.flow);
const totalFlow = Number.isFinite(totalFlowMeasured) ? totalFlowMeasured : totalFlowPredicted;
if (!Number.isFinite(totalFlow)) {
return;
}
const availableEntries = this.getAvailableValves();
const availableIds = new Set(availableEntries.map((entry) => entry.id));
const totalKv = availableEntries.reduce((sum, { valve }) => sum + Number(valve.kv), 0);
if (!availableEntries.length || !Number.isFinite(totalKv) || totalKv <= 0) {
this.logger.warn('No available valves with valid Kv, setting assigned flow to 0.');
for (const valve of Object.values(this.valves)) {
valve.updateFlow('predicted', 0, 'downstream', this.unitPolicy.output.flow);
}
this._writeMeasurement('flow', 'predicted', 'atEquipment', 0, this.unitPolicy.output.flow);
this.lastFlowSolve = {
passes: 0,
residual: Number(totalFlow) || 0,
targetTotal: Number(totalFlow) || 0,
assignedTotal: 0,
};
return;
}
const solve = this._solveFlowDistribution(totalFlow, availableEntries);
let assignedTotal = 0;
for (const [id, valve] of Object.entries(this.valves)) {
const flow = availableIds.has(id) ? (solve.flowsById[id] || 0) : 0;
valve.updateFlow('predicted', flow, 'downstream', this.unitPolicy.output.flow);
assignedTotal += flow;
}
this._writeMeasurement('flow', 'predicted', 'atEquipment', assignedTotal, this.unitPolicy.output.flow);
this.lastFlowSolve = {
passes: solve.passes,
residual: solve.residual,
targetTotal: totalFlow,
assignedTotal,
};
this.calcMaxDeltaP();
}
_readValveAcceptedFlow(valve) {
const accepted = Number(
valve?.measurements
?.type('flow')
?.variant('predicted')
?.position('downstream')
?.getCurrentValue(this.unitPolicy.output.flow)
);
return Number.isFinite(accepted) ? accepted : null;
}
_solveFlowDistribution(totalFlow, availableEntries) {
const totalKv = availableEntries.reduce((sum, { valve }) => sum + Number(valve.kv), 0);
if (!Number.isFinite(totalKv) || totalKv <= 0) {
return { flowsById: {}, residual: Number(totalFlow) || 0, passes: 0 };
}
const targetById = {};
availableEntries.forEach(({ id }) => {
targetById[id] = 0;
});
let residual = Number(totalFlow);
let passes = 0;
const maxPasses = Math.max(1, Number(this.flowReconciliation?.maxPasses) || DEFAULT_FLOW_RECONCILIATION.maxPasses);
const tolerance = Math.max(0, Number(this.flowReconciliation?.residualTolerance) || DEFAULT_FLOW_RECONCILIATION.residualTolerance);
while (passes < maxPasses && Number.isFinite(residual) && Math.abs(residual) > tolerance) {
availableEntries.forEach(({ id, valve }) => {
const kv = Number(valve.kv);
const share = (kv / totalKv) * residual;
const nextTarget = Number(targetById[id]) + share;
targetById[id] = nextTarget;
valve.updateFlow('predicted', nextTarget, 'downstream', this.unitPolicy.output.flow);
});
let acceptedTotal = 0;
availableEntries.forEach(({ id, valve }) => {
const accepted = this._readValveAcceptedFlow(valve);
if (Number.isFinite(accepted)) {
targetById[id] = accepted;
acceptedTotal += accepted;
return;
}
acceptedTotal += Number(targetById[id]) || 0;
});
residual = Number(totalFlow) - acceptedTotal;
passes += 1;
}
return {
flowsById: targetById,
residual: Number.isFinite(residual) ? residual : 0,
passes,
};
}
calcMaxDeltaP() {
let maxDeltaP = 0;
for (const [id, valve] of Object.entries(this.valves)) {
const deltaP = Number(
valve.measurements
.type('pressure')
.variant('predicted')
.position('delta')
.getCurrentValue(this.unitPolicy.output.pressure)
);
if (!Number.isFinite(deltaP)) {
continue;
}
this.logger.debug(`Delta P for valve ${id}: ${deltaP}`);
if (deltaP > maxDeltaP) {
maxDeltaP = deltaP;
}
}
this.maxDeltaP = maxDeltaP;
this._writeMeasurement('pressure', 'predicted', 'deltaMax', maxDeltaP, this.unitPolicy.output.pressure);
}
getOutput() {
const output = {};
Object.entries(this.measurements.measurements || {}).forEach(([type, variants]) => {
Object.entries(variants || {}).forEach(([variant, positions]) => {
Object.keys(positions || {}).forEach((position) => {
const value = this._readMeasurement(type, variant, position, this._outputUnitForType(type));
if (value != null) {
output[`${position}_${variant}_${type}`] = value;
}
});
});
});
output.mode = this.currentMode;
output.maxDeltaP = this.maxDeltaP;
return output;
}
getOutput() { return io.getOutput(this); }
getStatusBadge() { return io.getStatusBadge(this); }
}
module.exports = ValveGroupControl;

View File

@@ -84,10 +84,43 @@ test('valveGroupControl distributes total flow according to supplier-curve Kv an
valve2.destroy();
});
test('valveGroupControl rejects non-valve-like child payload', () => {
test('valveGroupControl skips a non-valve-like payload registered as a valve', () => {
const group = buildGroup();
const result = group.registerChild({ config: { functionality: { softwareType: 'valve' } } }, 'atEquipment');
assert.equal(result, false);
// Router dispatches by softwareType; the _registerValve handler rejects
// non-valve-like children (missing updateFlow/state/measurements) by
// returning false from its branch — the registry side-effect (valves[])
// stays empty even though BaseDomain's registerChild returns true.
group.registerChild({ config: { functionality: { softwareType: 'valve' } } }, 'valve');
assert.equal(Object.keys(group.valves).length, 0);
group.destroy();
});
test('valveGroupControl router dispatches valve registration by softwareType, honouring config positionVsParent', async () => {
const valve = (function buildValveAtUpstream() {
const Valve = require('../../../valve/src/specificClass');
return new Valve(
{
general: { name: 'valve-upstream', logging: { enabled: false, logLevel: 'error' } },
asset: { supplier: 'binder', category: 'valve', type: 'control', model: 'ECDV', unit: 'm3/h' },
functionality: { positionVsParent: 'upstream', softwareType: 'valve' },
},
{
general: { logging: { enabled: false, logLevel: 'error' } },
movement: { speed: 1 },
time: { starting: 0, warmingup: 0, stopping: 0, coolingdown: 0 },
}
);
})();
primeValve(valve, 50);
const group = buildGroup();
// childRegistrationUtils consumes positionVsParent (2nd arg) and forwards
// softwareType='valve' to the parent — the router fans out from there.
assert.equal(await group.childRegistrationUtils.registerChild(valve, 'upstream'), true);
assert.equal(Object.keys(group.valves).length, 1);
const registered = Object.values(group.valves)[0];
assert.equal(registered.positionVsParent, 'upstream');
group.destroy();
valve.destroy();
});

242
wiki/Home.md Normal file
View File

@@ -0,0 +1,242 @@
# valveGroupControl
> **Reflects code as of `e02cd1a` · regenerated `2026-05-11` via `npm run wiki:all`**
> If this banner is stale, the page may be out of date. Treat as informative, not authoritative.
## 1. What this node is
**valveGroupControl** (VGC) is an S88 Unit that distributes a group-level flow target across registered `valve` children proportional to their current Kv, aggregates the per-valve accepted flow back into a group total, and surfaces the maximum delta-P across the group. It also reconciles each child's fluid-contract advertisement into a single group-level service-type view.
## 2. Position in the platform
```mermaid
flowchart LR
src[machine / MGC / PS<br/>upstream source]:::unit -.flow.predicted.-> vgc[valveGroupControl<br/>Unit]:::unit
vgc -->|set.position| v1[valve A]:::equip
vgc -->|set.position| v2[valve B]:::equip
v1 -->|evt.deltaPChange| vgc
v2 -->|evt.deltaPChange| vgc
vgc -->|child.register| parent[upstream parent]:::unit
classDef unit fill:#50a8d9,color:#000
classDef equip fill:#86bbdd,color:#000
```
S88 colours: Unit `#50a8d9`, Equipment `#86bbdd`. Source of truth: `.claude/rules/node-red-flow-layout.md`.
## 3. Capability matrix
| Capability | Status | Notes |
|---|---|---|
| Proportional flow distribution by Kv | ✅ | `groupOps/flowDistribution.js`. |
| Two-pass residual reconciliation | ✅ | `maxPasses: 2`, `residualTolerance: 0.001`. |
| Periodic tick re-balance | ✅ | Runs each adapter tick; `set.reconcileInterval` re-tunes. |
| Group `maxDeltaP` aggregation | ✅ | Recomputed on any child `deltaPChange`. |
| Upstream fluid-contract aggregation | ✅ | `sources/fluidContract.js`. |
| Sequence dispatch to all valves | ✅ | `cmd.execSequence` → per-valve `handleInput`. |
| Per-valve positional override | ❌ | `set.position` reserved; no-op in current build. |
| Multi-source aggregation | ✅ | Multiple machines / MGCs may register as sources. |
## 4. Code map
```mermaid
flowchart TB
subgraph nodeRED["nodeClass.js — adapter (BaseNodeAdapter)"]
nc["buildDomainConfig()<br/>static DomainClass, commands"]
end
subgraph domain["specificClass.js — orchestrator (BaseDomain)"]
sc["ValveGroupControl.configure()<br/>registerChild dispatch<br/>tick → calcValveFlows"]
end
subgraph concerns["src/ concern modules"]
gops["groupOps/<br/>flowDistribution"]
srcs["sources/<br/>fluidContract"]
cmds["commands/<br/>topic registry + handlers"]
io["io/<br/>output + status badge"]
end
nc --> sc
sc --> gops
sc --> srcs
sc --> io
nc --> cmds
```
| Module | Owns | Read first if you're changing… |
|---|---|---|
| `groupOps/` | Flow distribution, residual solver, max-deltaP aggregation | How the group divides flow. |
| `sources/` | Upstream-source registration, fluid-contract reconciliation | Service-type aggregation, source-event subscriptions. |
| `commands/` | Input-topic registry + per-topic handlers | New input topics, payload validation. |
| `io/` | Port-0 output shape + status badge | What lands on the wire. |
## 5. Topic contract
> **Auto-generated** from `src/commands/index.js`. Do NOT hand-edit between the markers. Re-run `npm run wiki:contract`.
<!-- BEGIN AUTOGEN: topic-contract -->
| Canonical topic | Aliases | Payload | Unit | Effect |
|---|---|---|---|---|
| `set.mode` | `setMode` | `string` | — | Switch the valve group between auto / manual control modes. |
| `set.position` | `setpoint` | `any` | — | Set the group-level valve position (currently a no-op pending Phase 7). |
| `child.register` | `registerChild` | `string` | — | Register a child valve with this group. |
| `cmd.execSequence` | `execSequence` | `object` | — | Run a group-wide sequence (startup / shutdown / emergencystop). |
| `data.totalFlow` | `totalFlowChange` | `any` | — | Notify the group that the total flow setpoint has changed. |
| `cmd.emergencyStop` | `emergencyStop`, `emergencystop` | `any` | — | Trigger an emergency stop across all valves in the group. |
| `set.reconcileInterval` | `setReconcileInterval` | `any` | — | Update the reconciliation interval (seconds). |
<!-- END AUTOGEN: topic-contract -->
## 6. Child registration
```mermaid
flowchart LR
subgraph kids["accepted children (softwareType)"]
v["valve"]:::equip
src["machine / rotatingmachine /<br/>machinegroup / pumpingstation /<br/>valvegroupcontrol"]:::unit
end
v -->|positionChange| handler1[onPositionChange<br/>calcValveFlows]
v -->|deltaPChange| handler2[onDeltaPChange<br/>calcMaxDeltaP]
src -->|flow.predicted.*| handler3[updateFlow<br/>at atEquipment]
src -->|fluidContractChange| handler4[sources.refresh<br/>aggregate contract]
classDef equip fill:#86bbdd,color:#000
classDef unit fill:#50a8d9,color:#000
```
| softwareType | onRegister side-effect | Subscribed events |
|---|---|---|
| `valve` | Stored in `this.valves[id]`; binds `positionChange` + `deltaPChange`; triggers `calcValveFlows` + `calcMaxDeltaP`. | `positionChange`, `deltaPChange`. |
| `machine` / `rotatingmachine` | Stored as upstream source; reads `getFluidContract()` (or default `liquid`). | `flow.predicted.*`, `flow.measured.*`, `fluidContractChange`. |
| `machinegroup` / `machinegroupcontrol` | Same as machine. | Same as machine. |
| `pumpingstation` | Same as machine. | Same as machine. |
| `valvegroupcontrol` | Allowed for cascaded VGCs; not exercised in production. | Same. |
## 7. Lifecycle — what one tick / event does
```mermaid
sequenceDiagram
participant src as upstream source
participant vgc as VGC
participant v1 as valve A
participant v2 as valve B
participant out as Port-0
src->>vgc: flow.predicted.downstream (m3/s)
vgc->>vgc: _write flow.predicted.atEquipment
vgc->>vgc: calcValveFlows()
Note over vgc: solveFlowDistribution<br/>(by Kv share)
vgc->>v1: updateFlow('predicted', shareA, 'downstream')
vgc->>v2: updateFlow('predicted', shareB, 'downstream')
v1-->>vgc: accepted flow + deltaPChange
v2-->>vgc: accepted flow + deltaPChange
vgc->>vgc: residual pass (up to maxPasses)
vgc->>vgc: calcMaxDeltaP
vgc->>out: msg{topic, payload (delta-compressed)}
```
## 11. Distribution loop — flow-share solver
VGC has no FSM. The loop below replaces the state chart for this section.
```mermaid
sequenceDiagram
participant tick as adapter tick / event
participant vgc as VGC
participant solver as solveFlowDistribution
participant valves as valve children
tick->>vgc: calcValveFlows()
vgc->>vgc: read flow.measured / predicted .atEquipment
vgc->>solver: target=totalFlow, entries=availableValves
loop ≤ maxPasses while |residual| > tolerance
solver->>valves: updateFlow share by (Kv / totalKv)
valves-->>solver: accepted flow back
solver->>solver: residual = target - sum(accepted)
end
solver-->>vgc: { flowsById, residual, passes }
vgc->>vgc: write flow.predicted.atEquipment = assignedTotal
vgc->>vgc: notifyOutputChanged()
```
## 8. Data model — `getOutput()`
What lands on Port 0. Composed in `io/output`, then delta-compressed by `outputUtils.formatMsg`.
<!-- BEGIN AUTOGEN: data-model -->
| Key | Type | Unit | Sample |
|---|---|---|---|
| `maxDeltaP` | number | — | `0` |
| `mode` | string | — | `"auto"` |
<!-- END AUTOGEN: data-model -->
Measurement-derived keys follow the `<position>_<variant>_<type>` shape (e.g. `atEquipment_predicted_flow`) and are emitted only when the container holds a finite value.
## 9. Configuration — editor form ↔ config keys
```mermaid
flowchart TB
subgraph editor["Node-RED editor form"]
f1[Mode]
f2[Sequences]
f3[Allowed sources per mode]
f4[Reconciliation interval]
end
subgraph config["Domain config slice"]
c1[mode.current]
c2[sequences.<name>]
c3[mode.allowedSources.<mode>]
c4[flowReconciliation (runtime)]
end
f1 --> c1
f2 --> c2
f3 --> c3
f4 --> c4
```
| Form field | Config key | Default | Range | Where used |
|---|---|---|---|---|
| Mode | `mode.current` | per schema | enum | `setMode` |
| Sequences | `sequences.<name>` | per schema | Set of state names | `executeSequence` |
| Allowed sources | `mode.allowedSources.<mode>` | per schema | Set of source ids | `isValidSourceForMode` |
| Reconcile maxPasses | `flowReconciliation.maxPasses` | `2` | ≥ 1 | `solveFlowDistribution` |
| Reconcile tolerance | `flowReconciliation.residualTolerance` | `0.001` | ≥ 0 | `solveFlowDistribution` |
| Reconcile interval | (runtime, via `set.reconcileInterval`) | adapter default | ≥ 100 ms | `setReconcileIntervalSeconds` |
## 10. State chart
Skipped — VGC has no FSM of its own. State semantics belong to the child valves; VGC is a coordinator. See section 11 for the flow-distribution loop.
## 11. Examples
| Tier | File | What it shows | Mandatory? |
|---|---|---|---|
| Basic | `examples/01-Basic.flow.json` | Inject `data.totalFlow` + 2 valves + dashboard | ✅ |
| Integration | `examples/02-Integration.flow.json` | VGC + valves + upstream rotatingMachine | ✅ |
| Dashboard | `examples/03-Dashboard.flow.json` | Live FlowFuse charts (per-valve flow, max ΔP) | ⭕ |
Screenshots under `wiki/_partial-screenshots/valveGroupControl/` when produced. Docker compose snippet under `examples/README.md`.
## 12. Debug recipes
| Symptom | First thing to check | Where to look |
|---|---|---|
| All valves show `assigned flow = 0` | `getAvailableValves()` empty; check Kv > 0 and state ≠ `off` / `maintenance`. | `groupOps/flowDistribution.isValveAvailable` |
| Residual never converges | `flowReconciliation.maxPasses` too low for the valve curve shape. | `lastFlowSolve.residual` |
| Group `maxDeltaP` stale | Child `deltaPChange` not subscribed; valve emitter not exposed. | `_bindValveEvents` |
| Service-type stays `unknown` | No source registered yet or all sources advertise unknown. | `sources.refreshFluidContract` |
| `data.totalFlow` ignored | Mode rejects the source; check `mode.allowedSources`. | `isValidSourceForMode` |
> Never ship `enableLog: 'debug'` in a demo — fills the container log within seconds and obscures real errors. Use only for live debugging.
## 13. When you would NOT use this node
- Use VGC for **2+ parallel valves** sharing a header. For a single valve, wire `valve` directly under the upstream parent.
- Don't use VGC to coordinate series valves — the flow-distribution model assumes parallel branches with a shared upstream pressure.
- Skip VGC when the upstream source already publishes per-branch flow setpoints; route those straight to each valve.
## 14. Known limitations / current issues
| # | Issue | Tracked in |
|---|---|---|
| 1 | `set.position` is reserved but currently a no-op — per-valve override is pending Phase 7. | CONTRACT.md `## Inputs` |
| 2 | Residual solver assumes Kv share is a good first estimate; pathological curves can need more passes than the default `2`. | `groupOps/flowDistribution.js` |
| 3 | Cascaded `valvegroupcontrol` registration accepted but not test-covered. | CONTRACT.md `## Children` |