P4 wave 2: convert MGC to BaseDomain + extract control/ + io/

specificClass.js: 1808 → 336 lines.
  MachineGroup extends BaseDomain. Configure() wires GroupOperatingPoint,
  TotalsCalculator, GroupEfficiency, DemandDispatcher (built but unused —
  see OPEN_QUESTIONS); ChildRouter handles registration + measurement
  events; tick is event-driven (no setInterval, recomputes on pressure
  events).

  src/control/strategies.js (210 lines, new) — extracted equalFlowControl
  + prioPercentageControl from the orchestrator to fit the line budget.
  src/io/output.js (69 lines, new) — extracted getOutput + getStatusBadge
  composition.

  Public surface preserved: machines / setMode / setScaling / handleInput
  / isMachineActive / handlePressureChange / dynamicTotals / absoluteTotals
  / absDistFromPeak / relDistFromPeak. _delayedCall + _dispatchInFlight
  inline gate kept (tests await handleInput; LatestWinsGate.fire is
  void) — see OPEN_QUESTIONS for the deferred decision.

nodeClass.js: 280 → 20 lines.
  Extends BaseNodeAdapter. tickInterval=null (event-driven), commands
  registry from src/commands/. buildDomainConfig returns {} (MGC has
  no node-specific domain slice).

53 basic + 23 integration + 1 edge tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
znetsixe
2026-05-10 21:32:11 +02:00
parent 619b1311d2
commit bb2f3bea82
4 changed files with 551 additions and 2005 deletions

210
src/control/strategies.js Normal file
View File

@@ -0,0 +1,210 @@
'use strict';
// Priority-based control strategies for machineGroupControl.
//
// equalFlowControl: distribute demand equally across priority-ordered active
// machines, falling back to start/stop the next priority when the current
// active set can't deliver.
//
// prioPercentageControl: percentage-style ctrl distribution (only valid with
// normalized scaling).
//
// Both extracted verbatim from specificClass during the P4 refactor; the
// orchestrator wires them in via the strategies map below. They depend on
// the same group-curve helpers the optimizer uses, so allocation and power
// evaluation stay on the equalised group operating point.
const { POSITIONS } = require('generalFunctions');
const { groupFlow, groupCalcPower } = require('../groupOps/groupCurves');
function sortMachinesByPriority(machines, priorityList) {
if (priorityList && Array.isArray(priorityList)) {
return priorityList
.filter(id => machines[id])
.map(id => ({ id, machine: machines[id] }));
}
return Object.entries(machines)
.map(([id, machine]) => ({ id, machine }))
.sort((a, b) => a.id - b.id);
}
function filterOutUnavailableMachines(list) {
return list.filter(({ machine }) => {
const state = machine.state.getCurrentState();
const validActionForMode = machine.isValidActionForMode('execsequence', 'auto');
return !(state === 'off' || state === 'coolingdown' || state === 'stopping'
|| state === 'emergencystop' || !validActionForMode);
});
}
function capFlowDemand(Qd, dynamicTotals, logger) {
if (Qd < dynamicTotals.flow.min && Qd > 0) {
logger?.warn?.(`Flow demand ${Qd} below min ${dynamicTotals.flow.min}; capping.`);
return dynamicTotals.flow.min;
}
if (Qd > dynamicTotals.flow.max) {
logger?.warn?.(`Flow demand ${Qd} above max ${dynamicTotals.flow.max}; capping.`);
return dynamicTotals.flow.max;
}
return Qd;
}
async function equalFlowControl(ctx, Qd, _powerCap = Infinity, priorityList = null) {
const { mgc } = ctx;
try {
mgc.equalizePressure();
const dynamicTotals = mgc.calcDynamicTotals();
Qd = capFlowDemand(Qd, dynamicTotals, mgc.logger);
let machinesInPriorityOrder = sortMachinesByPriority(mgc.machines, priorityList);
machinesInPriorityOrder = filterOutUnavailableMachines(machinesInPriorityOrder);
const flowDistribution = [];
let totalFlow = 0;
let totalPower = 0;
const totalCog = 0;
const activeTotals = mgc.totals.activeTotals();
switch (true) {
case (Qd < activeTotals.flow.min && activeTotals.flow.min !== 0): {
let availableFlow = activeTotals.flow.min;
for (let i = machinesInPriorityOrder.length - 1; i >= 0 && availableFlow > Qd; i--) {
const m = machinesInPriorityOrder[i];
if (mgc.isMachineActive(m.id)) {
flowDistribution.push({ machineId: m.id, flow: 0 });
availableFlow -= groupFlow(m.machine).currentFxyYMin;
}
}
const remaining = machinesInPriorityOrder.filter(({ id }) =>
mgc.isMachineActive(id) && !flowDistribution.some(it => it.machineId === id));
const distributedFlow = Qd / remaining.length;
for (const m of remaining) {
flowDistribution.push({ machineId: m.id, flow: distributedFlow });
totalFlow += distributedFlow;
totalPower += groupCalcPower(m.machine, distributedFlow);
}
break;
}
case (Qd > activeTotals.flow.max): {
let i = 1;
while (totalFlow < Qd && i <= machinesInPriorityOrder.length) {
Qd = Qd / i;
if (groupFlow(machinesInPriorityOrder[i - 1].machine).currentFxyYMax >= Qd) {
for (let i2 = 0; i2 < i; i2++) {
if (!mgc.isMachineActive(machinesInPriorityOrder[i2].id)) {
flowDistribution.push({ machineId: machinesInPriorityOrder[i2].id, flow: Qd });
totalFlow += Qd;
totalPower += groupCalcPower(machinesInPriorityOrder[i2].machine, Qd);
}
}
}
i++;
}
break;
}
default: {
const countActive = machinesInPriorityOrder.filter(({ id }) => mgc.isMachineActive(id)).length;
Qd /= countActive;
for (let i = 0; i < countActive; i++) {
flowDistribution.push({ machineId: machinesInPriorityOrder[i].id, flow: Qd });
totalFlow += Qd;
totalPower += groupCalcPower(machinesInPriorityOrder[i].machine, Qd);
}
break;
}
}
const fUnit = mgc._unitView.canonical.power;
const flUnit = mgc._unitView.canonical.flow;
mgc.operatingPoint.writeOwn('power', 'predicted', POSITIONS.AT_EQUIPMENT, totalPower, fUnit);
mgc.operatingPoint.writeOwn('flow', 'predicted', POSITIONS.AT_EQUIPMENT, totalFlow, flUnit);
mgc.measurements.type('efficiency').variant('predicted').position(POSITIONS.AT_EQUIPMENT).value(totalFlow / totalPower);
mgc.measurements.type('Ncog').variant('predicted').position(POSITIONS.AT_EQUIPMENT).value(totalCog);
await Promise.all(flowDistribution.map(async ({ machineId, flow }) => {
const machine = mgc.machines[machineId];
const currentState = machine.state.getCurrentState();
if (flow > 0) {
await machine.handleInput('parent', 'flowmovement', mgc._canonicalToOutputFlow(flow));
if (currentState === 'idle') {
await machine.handleInput('parent', 'execsequence', 'startup');
}
} else if (currentState === 'operational' || currentState === 'accelerating' || currentState === 'decelerating') {
await machine.handleInput('parent', 'execsequence', 'shutdown');
}
}));
} catch (err) {
mgc.logger?.error?.(err);
}
}
async function prioPercentageControl(ctx, input, priorityList = null) {
const { mgc } = ctx;
try {
if (input < 0) { await mgc.turnOffAllMachines(); return; }
if (input > 100) input = 100;
const numOfMachines = Object.keys(mgc.machines).length;
const procentTotal = numOfMachines * input;
const machinesNeeded = Math.ceil(procentTotal / 100);
const activeTotals = mgc.totals.activeTotals();
const machinesActive = activeTotals.countActiveMachines;
const machinesInPriorityOrder = sortMachinesByPriority(mgc.machines, priorityList);
const ctrlDistribution = [];
if (machinesNeeded > machinesActive) {
machinesInPriorityOrder.forEach(({ id }, index) => {
if (index < machinesNeeded) ctrlDistribution.push({ machineId: id, ctrl: 0 });
});
}
if (machinesNeeded < machinesActive) {
machinesInPriorityOrder.forEach(({ id }, index) => {
if (mgc.isMachineActive(id)) {
ctrlDistribution.push({ machineId: id, ctrl: index < machinesNeeded ? 100 : -1 });
}
});
}
if (machinesNeeded === machinesActive) {
const ctrlPerMachine = procentTotal / machinesActive;
machinesInPriorityOrder.forEach(({ id }) => {
if (mgc.isMachineActive(id)) {
ctrlDistribution.push({ machineId: id, ctrl: Math.max(0, Math.min(ctrlPerMachine, 100)) });
}
});
}
await Promise.all(ctrlDistribution.map(async ({ machineId, ctrl }) => {
const machine = mgc.machines[machineId];
const currentState = machine.state.getCurrentState();
if (ctrl < 0 && (currentState === 'operational' || currentState === 'accelerating' || currentState === 'decelerating')) {
await machine.handleInput('parent', 'execsequence', 'shutdown');
} else if (currentState === 'idle' && ctrl >= 0) {
await machine.handleInput('parent', 'execsequence', 'startup');
} else if (currentState === 'operational' && ctrl > 0) {
await machine.handleInput('parent', 'execmovement', ctrl);
}
}));
const totalPower = [];
const totalFlow = [];
Object.values(mgc.machines).forEach(machine => {
const p = mgc.operatingPoint.readChild(machine, 'power', 'predicted', POSITIONS.AT_EQUIPMENT, mgc._unitView.canonical.power);
const f = mgc.operatingPoint.readChild(machine, 'flow', 'predicted', POSITIONS.DOWNSTREAM, mgc._unitView.canonical.flow);
if (p !== null) totalPower.push(p);
if (f !== null) totalFlow.push(f);
});
const sumP = totalPower.reduce((a, b) => a + b, 0);
const sumF = totalFlow.reduce((a, b) => a + b, 0);
mgc.operatingPoint.writeOwn('power', 'predicted', POSITIONS.AT_EQUIPMENT, sumP, mgc._unitView.canonical.power);
mgc.operatingPoint.writeOwn('flow', 'predicted', POSITIONS.AT_EQUIPMENT, sumF, mgc._unitView.canonical.flow);
if (sumP > 0) {
mgc.measurements.type('efficiency').variant('predicted').position(POSITIONS.AT_EQUIPMENT).value(sumF / sumP);
}
} catch (err) {
mgc.logger?.error?.(err);
}
}
module.exports = { equalFlowControl, prioPercentageControl, capFlowDemand, sortMachinesByPriority, filterOutUnavailableMachines };

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

@@ -0,0 +1,69 @@
'use strict';
// Output + status-badge composition for machineGroupControl. Kept off the
// orchestrator so specificClass stays under the file-size budget. Both
// functions take the live MGC instance and reach for the same public surface
// the rest of the package already uses (measurements, dynamicTotals, mode).
const { statusBadge, POSITIONS } = require('generalFunctions');
function _outputUnitForType(unitView, type) {
switch (String(type || '').toLowerCase()) {
case 'flow': return unitView.output.flow;
case 'power': return unitView.output.power;
case 'pressure': return unitView.output.pressure;
case 'temperature': return unitView.output.temperature;
default: return null;
}
}
function getOutput(mgc) {
const out = {};
const { measurements, _unitView, mode, scaling, absDistFromPeak, relDistFromPeak } = mgc;
measurements.getTypes().forEach(type => {
measurements.getVariants(type).forEach(variant => {
const unit = _outputUnitForType(_unitView, type);
const read = (pos) => measurements.type(type).variant(variant).position(pos).getCurrentValue(unit || undefined);
const dn = read(POSITIONS.DOWNSTREAM);
const at = read(POSITIONS.AT_EQUIPMENT);
const up = read(POSITIONS.UPSTREAM);
if (dn != null) out[`downstream_${variant}_${type}`] = dn;
if (up != null) out[`upstream_${variant}_${type}`] = up;
if (at != null) out[`atEquipment_${variant}_${type}`] = at;
if (dn != null && up != null) {
const diff = measurements.type(type).variant(variant)
.difference({ from: POSITIONS.DOWNSTREAM, to: POSITIONS.UPSTREAM, unit });
if (diff?.value != null) out[`differential_${variant}_${type}`] = diff.value;
}
});
});
out.mode = mode;
out.scaling = scaling;
out.absDistFromPeak = absDistFromPeak;
out.relDistFromPeak = relDistFromPeak;
return out;
}
function getStatusBadge(mgc) {
const totalFlow = mgc.measurements.type('flow').variant('predicted').position(POSITIONS.AT_EQUIPMENT)
.getCurrentValue(mgc._unitView.output.flow) ?? 0;
const totalPower = mgc.measurements.type('power').variant('predicted').position(POSITIONS.AT_EQUIPMENT)
.getCurrentValue(mgc._unitView.output.power) ?? 0;
const totalCapacity = mgc.dynamicTotals?.flow?.max ?? 0;
const available = Object.values(mgc.machines).filter(m => {
const s = m?.state?.getCurrentState?.();
const md = m?.currentMode;
return s && s !== 'off' && s !== 'maintenance' && md !== 'maintenance';
});
const status = available.length > 0 ? `${available.length} machine(s)` : 'No machines';
let scalingSymbol;
switch ((mgc.scaling || '').toLowerCase()) {
case 'absolute': scalingSymbol = 'Ⓐ'; break;
case 'normalized': scalingSymbol = 'Ⓝ'; break;
default: scalingSymbol = mgc.mode || ''; break;
}
const text = ` ${mgc.mode || 'Unknown'} | ${scalingSymbol}: 💨=${Math.round(totalFlow)}/${Math.round(totalCapacity)} | ⚡=${Math.round(totalPower)} | ${status}`;
return statusBadge.text(text, { fill: available.length > 0 ? 'green' : 'red', shape: 'dot' });
}
module.exports = { getOutput, getStatusBadge };

View File

@@ -1,280 +1,20 @@
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 MachineGroup = require('./specificClass');
const commands = require('./commands');
// Load default & UI config
this._loadConfig(uiConfig, this.node);
// Event-driven: the domain emits 'output-changed' from handlePressureChange
// (pump events) and from handleInput/turnOff. No tick loop needed.
class nodeClass extends BaseNodeAdapter {
static DomainClass = MachineGroup;
static commands = commands;
static tickInterval = null;
static statusInterval = 1000;
// Instantiate core Measurement class
this._setupSpecificClass();
// Wire up event and lifecycle handlers
this._bindEvents();
this._registerChild();
this._startTickLoop();
this._attachInputHandler();
this._attachCloseHandler();
}
/**
* 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);
const flowUnit = this._resolveUnitOrFallback(uiConfig.unit, 'volumeFlowRate', 'm3/h', 'flow');
// Build config: base sections (no domain-specific config for group controller)
this.config = cfgMgr.buildConfig(this.name, uiConfig, 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;
}
}
_updateNodeStatus() {
//console.log('Updating node status...');
const mg = this.source;
const mode = mg.mode;
const scaling = mg.scaling;
// Add safety checks for measurements
const totalFlow = mg.measurements
?.type("flow")
?.variant("predicted")
?.position("atequipment")
?.getCurrentValue(mg?.unitPolicy?.output?.flow || 'm3/h') || 0;
const totalPower = mg.measurements
?.type("power")
?.variant("predicted")
?.position("atEquipment")
?.getCurrentValue(mg?.unitPolicy?.output?.power || 'kW') || 0;
// Calculate total capacity based on available machines with safety checks
const availableMachines = Object.values(mg.machines || {}).filter((machine) => {
// Safety check: ensure machine and machine.state exist
if (!machine || !machine.state || typeof machine.state.getCurrentState !== 'function') {
mg.logger?.warn(`Machine missing or invalid: ${machine?.config?.general?.id || 'unknown'}`);
return false;
}
const state = machine.state.getCurrentState();
const mode = machine.currentMode;
return !(
state === "off" ||
state === "maintenance" ||
mode === "maintenance"
);
});
const totalCapacity = Math.round((mg.dynamicTotals?.flow?.max || 0) * 1) / 1;
// Determine overall status based on available machines
const status = availableMachines.length > 0
? `${availableMachines.length} machine(s) connected`
: "No machines";
let scalingSymbol = "";
switch ((scaling || "").toLowerCase()) {
case "absolute":
scalingSymbol = "Ⓐ";
break;
case "normalized":
scalingSymbol = "Ⓝ";
break;
default:
scalingSymbol = mode || "";
break;
}
const text = ` ${mode || 'Unknown'} | ${scalingSymbol}: 💨=${Math.round(totalFlow)}/${totalCapacity} | ⚡=${Math.round(totalPower)} | ${status}`;
return {
fill: availableMachines.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() {
this.source.emitter.on("mAbs", (val) => {
this.node.status({
fill: "green",
shape: "dot",
text: `${val} ${this.config.general.unit}`,
});
});
}
/**
* 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(), 1000);
// 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() {
const raw = this.source.getOutput();
const processMsg = this._output.formatMsg(raw, this.source.config, "process");
const influxMsg = this._output.formatMsg(raw, this.source.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 mg = 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) {
mg.logger.warn(`registerChild skipped: missing child/source for id=${childId}`);
break;
}
mg.logger.debug(`Registering child: ${childId}, found: ${!!childObj}, source: ${!!childObj?.source}`);
mg.childRegistrationUtils.registerChild(childObj.source, msg.positionVsParent);
mg.logger.debug(`Total machines after registration: ${Object.keys(mg.machines || {}).length}`);
break;
}
case "setMode": {
const mode = msg.payload;
mg.setMode(mode);
break;
}
case "setScaling": {
const scaling = msg.payload;
mg.setScaling(scaling);
break;
}
case "Qd": {
const Qd = parseFloat(msg.payload);
const sourceQd = "parent";
if (isNaN(Qd)) {
mg.logger.error(`Invalid demand value: ${msg.payload}`);
break;
}
try {
await mg.handleInput(sourceQd, Qd);
msg.topic = mg.config.general.name;
msg.payload = "done";
send(msg);
} catch (error) {
mg.logger.error(`Failed to process Qd: ${error.message}`);
}
break;
}
default:
mg.logger.warn(`Unknown topic: ${msg.topic}`);
break;
}
} catch (error) {
mg.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.node.status({}); // clear node status badge
if (typeof done === 'function') done();
});
buildDomainConfig() {
return {};
}
}
module.exports = nodeClass; // Export the class for Node-RED to use
module.exports = nodeClass;

File diff suppressed because it is too large Load Diff