Files
machineGroupControl/src/specificClass.js
Rene De Ren 96b84d3124 Revert: handleInput unchanged-demand short-circuit
Reverts a14aa0d. The "skip when demand unchanged" optimisation broke
the live demo: in some real conditions (basin transitions, safety
controller activations) PS sends repeated demand=0 and the optimisation
correctly turned pumps off the first time but then declined to re-act
when conditions changed in a way the test suite didn't cover. Live
result: pumps stayed off even when basin filled to overflow.

The original symptom (pumps stuck mid-ramp under saturated demand) needs
a different approach — likely a pump-side guard rather than an MGC-side
demand filter. Investigating in a follow-up.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-08 20:55:41 +02:00

1739 lines
77 KiB
JavaScript

//load local dependencies
const EventEmitter = require("events");
const {logger,configUtils,configManager, MeasurementContainer, interpolation , childRegistrationUtils, convert, POSITIONS} = require('generalFunctions');
const CANONICAL_UNITS = Object.freeze({
pressure: 'Pa',
flow: 'm3/s',
power: 'W',
temperature: 'K',
});
const DEFAULT_IO_UNITS = Object.freeze({
pressure: 'mbar',
flow: 'm3/h',
power: 'kW',
temperature: 'C',
});
/**
* Machine group controller domain model.
* Aggregates multiple rotating machines and coordinates group-level optimization/control.
*/
class MachineGroup {
constructor(machineGroupConfig = {}) {
this.emitter = new EventEmitter(); // Own EventEmitter
this.configManager = new configManager(); // Config manager to handle dynamic config loading
this.defaultConfig = this.configManager.getConfig('machineGroupControl'); // Load default config for rotating machine ( use software type name ? )
this.configUtils = new configUtils(this.defaultConfig);// this will handle the config endpoints so we can load them dynamically
this.config = this.configUtils.initConfig(machineGroupConfig); // verify and set the config for the machine group
this.unitPolicy = this._buildUnitPolicy(this.config);
this.config = this.configUtils.updateConfig(this.config, {
general: {
unit: this.unitPolicy.output.flow,
}
});
// Init after config is set
this.logger = new logger(this.config.general.logging.enabled,this.config.general.logging.logLevel, this.config.general.name);
// Initialize measurements
this.measurements = new MeasurementContainer({
autoConvert: true,
windowSize: 50,
defaultUnits: {
pressure: this.unitPolicy.output.pressure,
flow: this.unitPolicy.output.flow,
power: this.unitPolicy.output.power,
temperature: this.unitPolicy.output.temperature
},
preferredUnits: {
pressure: this.unitPolicy.output.pressure,
flow: this.unitPolicy.output.flow,
power: this.unitPolicy.output.power,
temperature: this.unitPolicy.output.temperature
},
canonicalUnits: this.unitPolicy.canonical,
storeCanonical: true,
strictUnitValidation: true,
throwOnInvalidUnit: true,
requireUnitForTypes: ['pressure', 'flow', 'power', 'temperature']
});
this.interpolation = new interpolation();
// Machines and child data
this.machines = {};
this.child = {};
this.scaling = this.config.scaling.current;
this.mode = this.config.mode.current;
this.absDistFromPeak = 0 ;
this.relDistFromPeak = 0;
// Combination curve data
this.dynamicTotals = { flow: { min: Infinity, max: 0 }, power: { min: Infinity, max: 0 } , NCog : 0};
this.absoluteTotals = { flow: { min: Infinity, max: 0 }, power: { min: Infinity, max: 0 }};
//this always last in the constructor
this.childRegistrationUtils = new childRegistrationUtils(this);
this.logger.info("MachineGroup initialized.");
}
registerChild(child,softwareType) {
this.logger.debug('Setting up childs specific for this class');
// Prefer functionality-scoped position metadata; keep general fallback for legacy nodes.
const position = child.config?.functionality?.positionVsParent || child.config?.general?.positionVsParent;
if(softwareType == "machine"){
// Check if the machine is already registered
this.machines[child.config.general.id] === undefined ? this.machines[child.config.general.id] = child : this.logger.warn(`Machine ${child.config.general.id} is already registered.`);
//listen for machine pressure changes
this.logger.debug(`Listening for pressure changes from machine ${child.config.general.id}`);
child.measurements.emitter.on("pressure.measured.differential", (eventData) => {
this.logger.debug(`Pressure update from ${child.config.general.id}: ${eventData.value} ${eventData.unit}`);
this.handlePressureChange();
});
child.measurements.emitter.on("pressure.measured.downstream", (eventData) => {
this.logger.debug(`Pressure update from ${child.config.general.id}: ${eventData.value} ${eventData.unit}`);
this.handlePressureChange();
});
child.measurements.emitter.on("flow.predicted.downstream", (eventData) => {
this.logger.debug(`Flow prediction update from ${child.config.general.id}: ${eventData.value} ${eventData.unit}`);
//later change to this.handleFlowPredictionChange();
this.handlePressureChange();
});
} else if (softwareType === "measurement") {
// Header-side measurement (e.g. discharge-manifold pressure
// sensor at MGC's downstream, suction-manifold sensor at
// upstream). Subscribed at the group level so optimalControl
// can use ONE header operating point for all pumps instead of
// each pump's individual reading. Without this, small per-pump
// pressure differences make the BEP-Gravitation optimum flip
// between near-equivalent combinations every tick → flap.
const measurementType = child.config?.asset?.type;
if (!measurementType || !position) {
this.logger.warn(`Measurement child ${child.config?.general?.id} missing asset.type or positionVsParent — skipping`);
return;
}
const eventName = `${measurementType}.measured.${position}`;
this.logger.debug(`Listening for ${eventName} from measurement ${child.config.general.id}`);
child.measurements.emitter.on(eventName, (eventData = {}) => {
this.measurements
.type(measurementType)
.variant("measured")
.position(position)
.value(eventData.value, eventData.timestamp, eventData.unit);
// Header pressure changes are operating-point inputs to
// optimalControl — recompute combinations.
if (measurementType === "pressure") this.handlePressureChange();
});
}
}
calcAbsoluteTotals() {
const absoluteTotals = { flow: { min: Infinity, max: 0 }, power: { min: Infinity, max: 0 } };
Object.values(this.machines).forEach(machine => {
const totals = { flow: { min: Infinity, max: 0 }, power: { min: Infinity, max: 0 } };
//fetch min flow ever seen over all machines
Object.entries(machine.predictFlow.inputCurve).forEach(([pressure, xyCurve], _index) => {
const minFlow = Math.min(...xyCurve.y);
const maxFlow = Math.max(...xyCurve.y);
const minPower = Math.min(...machine.predictPower.inputCurve[pressure].y);
const maxPower = Math.max(...machine.predictPower.inputCurve[pressure].y);
// min ever seen for 1 machine
if (minFlow < totals.flow.min) { totals.flow.min = minFlow; }
if (minPower < totals.power.min) { totals.power.min = minPower; }
if( maxFlow > totals.flow.max ){ totals.flow.max = maxFlow; }
if( maxPower > totals.power.max ){ totals.power.max = maxPower; }
});
//surplus machines for max flow and power
if( totals.flow.min < absoluteTotals.flow.min ){ absoluteTotals.flow.min = totals.flow.min; }
if( totals.power.min < absoluteTotals.power.min ){ absoluteTotals.power.min = totals.power.min; }
absoluteTotals.flow.max += totals.flow.max;
absoluteTotals.power.max += totals.power.max;
});
if(absoluteTotals.flow.min === Infinity) {
this.logger.warn(`Flow min ${absoluteTotals.flow.min} is Infinity. Setting to 0.`);
absoluteTotals.flow.min = 0;
}
if(absoluteTotals.power.min === Infinity) {
this.logger.warn(`Power min ${absoluteTotals.power.min} is Infinity. Setting to 0.`);
absoluteTotals.power.min = 0;
}
if(absoluteTotals.flow.max === -Infinity) {
this.logger.warn(`Flow max ${absoluteTotals.flow.max} is -Infinity. Setting to 0.`);
absoluteTotals.flow.max = 0;
}
if(absoluteTotals.power.max === -Infinity) {
this.logger.warn(`Power max ${absoluteTotals.power.max} is -Infinity. Setting to 0.`);
absoluteTotals.power.max = 0;
}
// Place data in object for external use
this.absoluteTotals = absoluteTotals;
return absoluteTotals;
}
//max and min current flow and power based on their actual pressure curve
calcDynamicTotals() {
const dynamicTotals = { flow: { min: Infinity, max: 0, act: 0 }, power: { min: Infinity, max: 0, act: 0 }, NCog : 0 };
this.logger.debug(`\n --------- Calculating dynamic totals for ${Object.keys(this.machines).length} machines. @ current pressure settings : ----------`);
Object.values(this.machines).forEach(machine => {
//skip machines without valid curve
if(!machine.hasCurve){
this.logger.error(`Machine ${machine.config.general.id} does not have a valid curve. Skipping in dynamic totals calculation.`);
return;
}
this.logger.debug(`Processing machine with id: ${machine.config.general.id}`);
const gpf = this._groupFlow(machine);
const gpp = this._groupPower(machine);
this.logger.debug(`Group operating point: ${JSON.stringify(gpf.currentF)}`);
//fetch min flow ever seen over all machines (at the group operating point)
const minFlow = gpf.currentFxyYMin;
const maxFlow = gpf.currentFxyYMax;
const minPower = gpp.currentFxyYMin;
const maxPower = gpp.currentFxyYMax;
const actFlow = this._readChildMeasurement(machine, "flow", "predicted", POSITIONS.DOWNSTREAM, this.unitPolicy.canonical.flow) || 0;
const actPower = this._readChildMeasurement(machine, "power", "predicted", POSITIONS.AT_EQUIPMENT, this.unitPolicy.canonical.power) || 0;
this.logger.debug(`Machine ${machine.config.general.id} - Min Flow: ${minFlow}, Max Flow: ${maxFlow}, Min Power: ${minPower}, Max Power: ${maxPower}, NCog: ${this._groupNCog(machine)}`);
if( minFlow < dynamicTotals.flow.min ){ dynamicTotals.flow.min = minFlow; }
if( minPower < dynamicTotals.power.min ){ dynamicTotals.power.min = minPower; }
dynamicTotals.flow.max += maxFlow;
dynamicTotals.power.max += maxPower;
dynamicTotals.flow.act += actFlow;
dynamicTotals.power.act += actPower;
//fetch total Normalized Cog over all machines (group operating point)
dynamicTotals.NCog += this._groupNCog(machine);
});
// Place data in object for external use
this.dynamicTotals = dynamicTotals;
return dynamicTotals;
}
activeTotals() {
const totals = { flow: { min: 0, max: 0 }, power: { min: 0, max: 0 }, countActiveMachines: 0 };
Object.entries(this.machines).forEach(([id, machine]) => {
this.logger.debug(`Processing machine with id: ${id}`);
if(this.isMachineActive(id)){
//fetch min flow ever seen over all machines (group operating point)
const minFlow = this._groupFlow(machine).currentFxyYMin;
const maxFlow = this._groupFlow(machine).currentFxyYMax;
const minPower = this._groupPower(machine).currentFxyYMin;
const maxPower = this._groupPower(machine).currentFxyYMax;
totals.flow.min += minFlow;
totals.flow.max += maxFlow;
totals.power.min += minPower;
totals.power.max += maxPower;
totals.countActiveMachines++;
}
});
return totals;
}
handlePressureChange() {
this.logger.debug("Pressure change detected.");
// Equalize before computing dynamicTotals so the cached value (read
// by optimalControl) reflects the consistent header operating point,
// not whichever per-pump sensor fired last.
this._equalizeOperatingPoint();
// Recalculate totals
const { flow, power } = this.calcDynamicTotals();
this.logger.debug(`Dynamic Totals after pressure change - Flow: Min ${flow.min}, Max ${flow.max}, Act ${flow.act} | Power: Min ${power.min}, Max ${power.max}, Act ${power.act}`);
this._writeMeasurement("flow", "predicted", POSITIONS.AT_EQUIPMENT, flow.act, this.unitPolicy.canonical.flow);
// Mirror the aggregate flow onto DOWNSTREAM as well. PS subscribes to
// flow.predicted.downstream from MGC and uses it as the outflow
// estimate for net-flow computation. Without this mirror, the only
// place DOWNSTREAM gets written is optimalControl's bestFlow (the
// optimizer's TARGET, not the achieved aggregate). During transients
// — e.g. demand dropping to dead-band keep-alive while pumps are
// still ramping down from full throttle — PS would see a stale
// 25 m³/h target while pumps are physically delivering 500+ m³/h,
// making netFlow look small and stable when the basin is actually
// draining fast. flow.act here is the sum of every pump's current
// predicted output, so it IS the achieved aggregate.
this._writeMeasurement("flow", "predicted", POSITIONS.DOWNSTREAM, flow.act, this.unitPolicy.canonical.flow);
this._writeMeasurement("power", "predicted", POSITIONS.AT_EQUIPMENT, power.act, this.unitPolicy.canonical.power);
const { maxEfficiency, lowestEfficiency } = this.calcGroupEfficiency(this.machines);
const efficiency = this.measurements.type("efficiency").variant("predicted").position(POSITIONS.AT_EQUIPMENT).getCurrentValue();
this.calcDistanceBEP(efficiency,maxEfficiency,lowestEfficiency);
}
calcDistanceFromPeak(currentEfficiency,peakEfficiency){
return Math.abs(currentEfficiency - peakEfficiency);
}
calcRelativeDistanceFromPeak(currentEfficiency,maxEfficiency,minEfficiency){
let distance = 1;
if(currentEfficiency != null && maxEfficiency !== minEfficiency){
distance = this.interpolation.interpolate_lin_single_point(currentEfficiency,maxEfficiency, minEfficiency, 0, 1);
}
return distance;
}
calcDistanceBEP(efficiency,maxEfficiency,minEfficiency){
const absDistFromPeak = this.calcDistanceFromPeak(efficiency,maxEfficiency);
const relDistFromPeak = this.calcRelativeDistanceFromPeak(efficiency,maxEfficiency,minEfficiency);
//store internally
this.absDistFromPeak = absDistFromPeak ;
this.relDistFromPeak = relDistFromPeak;
return { absDistFromPeak: absDistFromPeak, relDistFromPeak: relDistFromPeak };
}
checkSpecialCases(machines, Qd) {
Object.values(machines).forEach(machine => {
const state = machine.state.getCurrentState();
const mode = machine.currentMode;
//add special cases
if( state === "operational" && ( mode == "virtualControl" || mode === "fysicalControl") ){
let flow = 0;
const measuredFlow = this._readChildMeasurement(machine, "flow", "measured", POSITIONS.DOWNSTREAM, this.unitPolicy.canonical.flow);
const predictedFlow = this._readChildMeasurement(machine, "flow", "predicted", POSITIONS.DOWNSTREAM, this.unitPolicy.canonical.flow);
if (Number.isFinite(measuredFlow) && measuredFlow !== 0) {
flow = measuredFlow;
}
else if (Number.isFinite(predictedFlow) && predictedFlow !== 0) {
flow = predictedFlow;
}
else{
this.logger.error("Dont perform calculation at all seeing that there is a machine working but we dont know the flow its producing");
//abort the calculation
return false;
}
//Qd is less because we allready have machines delivering flow on manual control
Qd = Qd - flow;
}
});
return Qd ;
}
validPumpCombinations(machines, Qd, PowerCap = Infinity) {
let subsets = [[]];
// adjust demand flow when there are machines being controlled by a manual source
Qd = this.checkSpecialCases(machines, Qd);
// Generate all possible subsets of machines (power set)
Object.keys(machines).forEach(machineId => {
const state = machines[machineId].state.getCurrentState();
const validActionForMode = machines[machineId].isValidActionForMode("execsequence", "auto");
// Reasons why a machine is not valid for the combination
if( state === "off" || state === "coolingdown" || state === "stopping" || state === "emergencystop" || !validActionForMode){
return;
}
// go through each machine and add it to the subsets
let newSubsets = subsets.map(set => [...set, machineId]);
subsets = subsets.concat(newSubsets);
});
// Filter for non-empty subsets that can meet or exceed demand flow
const combinations = subsets.filter(subset => {
if (subset.length === 0) return false;
// Calculate total and minimum flow for the subset in one pass
// (uses group operating point — see _groupFlow/_groupPower)
const { maxFlow, minFlow, maxPower } = subset.reduce(
(acc, machineId) => {
const machine = machines[machineId];
const minFlow = this._groupFlow(machine).currentFxyYMin;
const maxFlow = this._groupFlow(machine).currentFxyYMax;
const maxPower = this._groupPower(machine).currentFxyYMax;
return {
maxFlow: acc.maxFlow + maxFlow,
minFlow: acc.minFlow + minFlow,
maxPower: acc.maxPower + maxPower
};
},
{ maxFlow: 0, minFlow: 0 , maxPower: 0 }
);
// If total flow can deliver the demand
if(maxFlow >= Qd && minFlow <= Qd && maxPower <= PowerCap){
return true;
}
else{
return false;
}
});
return combinations;
}
calcBestCombination(combinations, Qd) {
let bestCombination = null;
let bestPower = Infinity;
let bestFlow = 0;
let bestCog = 0;
combinations.forEach(combination => {
let flowDistribution = [];
let totalCoG = 0;
let totalPower = 0;
// Sum normalized CoG for the combination (group operating point)
combination.forEach(machineId => {
totalCoG += Math.round((this._groupNCog(this.machines[machineId]) || 0) * 100) / 100;
});
// Initial CoG-based distribution
combination.forEach(machineId => {
let flow = 0;
if (totalCoG === 0) {
flow = Qd / combination.length;
} else {
flow = ((this._groupNCog(this.machines[machineId]) || 0) / totalCoG) * Qd;
this.logger.debug(`Machine Normalized CoG-based distribution ${machineId} flow: ${flow}`);
}
flowDistribution.push({ machineId, flow });
});
// Clamp to min/max and spill leftover once (group operating point)
const clamped = flowDistribution.map(entry => {
const machine = this.machines[entry.machineId];
const min = this._groupFlow(machine).currentFxyYMin;
const max = this._groupFlow(machine).currentFxyYMax;
const clampedFlow = Math.min(max, Math.max(min, entry.flow));
return { ...entry, flow: clampedFlow, min, max, desired: entry.flow };
});
let remainder = Qd - clamped.reduce((sum, entry) => sum + entry.flow, 0);
if (Math.abs(remainder) > 1e-6) {
const adjustable = clamped.filter(entry =>
remainder > 0 ? entry.flow < entry.max : entry.flow > entry.min
);
const weightSum = adjustable.reduce((sum, entry) => sum + entry.desired, 0) || adjustable.length;
adjustable.forEach(entry => {
const weight = entry.desired / weightSum || 1 / adjustable.length;
const delta = remainder * weight;
const next = remainder > 0
? Math.min(entry.max, entry.flow + delta)
: Math.max(entry.min, entry.flow + delta);
remainder -= (next - entry.flow);
entry.flow = next;
});
}
flowDistribution = clamped;
let totalFlow = 0;
flowDistribution.forEach(({ machineId, flow }) => {
totalFlow += flow;
totalPower += this._groupCalcPower(this.machines[machineId], flow);
});
if (totalPower < bestPower) {
this.logger.debug(`New best combination found: ${totalPower} < ${bestPower}`);
this.logger.debug(`combination ${JSON.stringify(flowDistribution)}`);
bestPower = totalPower;
bestFlow = totalFlow;
bestCog = totalCoG;
bestCombination = flowDistribution;
}
});
return { bestCombination, bestPower, bestFlow, bestCog };
}
// Estimate the local dP/dQ slopes around the BEP for the provided machine.
estimateSlopesAtBEP(machine, Q_BEP, delta = 1.0) {
const fallback = {
slopeLeft: 0,
slopeRight: 0,
alpha: 1,
Q_BEP: Q_BEP || 0,
P_BEP: 0
};
// Group operating point — slopes around BEP must use the same op-point
// the optimizer evaluates at, otherwise gravitation pulls toward an
// off-by-one BEP target.
const minFlow = this._groupFlow(machine).currentFxyYMin;
const maxFlow = this._groupFlow(machine).currentFxyYMax;
const span = Math.max(0, maxFlow - minFlow);
const normalizedCog = Math.max(0, Math.min(1, this._groupNCog(machine) || 0));
const targetBEP = Q_BEP ?? (minFlow + span * normalizedCog);
const clampFlow = (flow) => Math.min(maxFlow, Math.max(minFlow, flow)); // ensure within bounds using small helper function
const center = clampFlow(targetBEP);
const deltaSafe = Math.max(delta, 0.01);
const leftFlow = clampFlow(center - deltaSafe);
const rightFlow = clampFlow(center + deltaSafe);
const powerAt = (flow) => this._groupCalcPower(machine, flow); // helper to get power at a given flow
const P_center = powerAt(center);
const P_left = powerAt(leftFlow);
const P_right = powerAt(rightFlow);
const slopeLeft = (P_center - P_left) / Math.max(1e-6, center - leftFlow);
const slopeRight = (P_right - P_center) / Math.max(1e-6, rightFlow - center);
const alpha = Math.max(1e-6, (Math.abs(slopeLeft) + Math.abs(slopeRight)) / 2);
return {
slopeLeft,
slopeRight,
alpha,
Q_BEP: center,
P_BEP: P_center
};
}
//Redistribute remaining demand using slope-based weights so flatter curves attract more flow.
redistributeFlowBySlope(pumpInfos, flowDistribution, delta, directional = true) {
const tolerance = 1e-3; // Small tolerance to avoid infinite loops
let remaining = delta; // Remaining flow to distribute
const entryMap = new Map(flowDistribution.map(entry => [entry.machineId, entry])); // Map for quick access
// Loop until remaining flow is within tolerance
while (Math.abs(remaining) > tolerance) {
const increasing = remaining > 0; // Determine if we are increasing or decreasing flow
// Build candidates with capacity and weight
const candidates = pumpInfos.map(info => {
const entry = entryMap.get(info.id);
if (!entry) { return null; }
const capacity = increasing ? info.maxFlow - entry.flow : entry.flow - info.minFlow; // Calculate available capacity based on direction
if (capacity <= tolerance) { return null; }
const slope = increasing
? (directional ? info.slopes.slopeRight : info.slopes.alpha)
: (directional ? info.slopes.slopeLeft : info.slopes.alpha);
const weight = 1 / Math.max(1e-6, Math.abs(slope) || info.slopes.alpha || 1);
return { entry, capacity, weight };
}).filter(Boolean);
if (!candidates.length) { break; } // No candidates available, exit loop
const weightSum = candidates.reduce((sum, candidate) => sum + candidate.weight * candidate.capacity, 0); // weighted sum of capacities
if (weightSum <= 0) { break; } // Avoid division by zero
let progress = 0;
// Distribute remaining flow among candidates based on their weights and capacities
candidates.forEach(candidate => {
let share = (candidate.weight * candidate.capacity / weightSum) * Math.abs(remaining);
share = Math.min(share, candidate.capacity); // Ensure we don't exceed capacity
if (share <= 0) { return; } // Skip if no share to allocate
if (increasing) {
candidate.entry.flow += share;
} else {
candidate.entry.flow -= share;
}
progress += share; // Track total progress made in this iteration
});
if (progress <= tolerance) { break; }
remaining += increasing ? -progress : progress; // Update remaining flow to distribute
}
}
// BEP-gravitation based combination finder that biases allocation around each pump's BEP.
calcBestCombinationBEPGravitation(combinations, Qd, method = "BEP-Gravitation-Directional") {
let bestCombination = null;
let bestPower = Infinity;
let bestFlow = 0;
let bestCog = 0;
let bestDeviation = Infinity;
const directional = method === "BEP-Gravitation-Directional";
combinations.forEach(combination => {
const pumpInfos = combination.map(machineId => {
const machine = this.machines[machineId];
// Group operating point — BEP and curve envelope must come
// from the same view the optimizer evaluates power on.
const minFlow = this._groupFlow(machine).currentFxyYMin;
const maxFlow = this._groupFlow(machine).currentFxyYMax;
const span = Math.max(0, maxFlow - minFlow);
const NCog = Math.max(0, Math.min(1, this._groupNCog(machine) || 0));
const estimatedBEP = minFlow + span * NCog; // Estimated BEP flow based on current curve
const slopes = this.estimateSlopesAtBEP(machine, estimatedBEP);
return {
id: machineId,
machine,
minFlow,
maxFlow,
NCog,
Q_BEP: slopes.Q_BEP,
slopes
};
});
// Skip if no pumps in combination
if (pumpInfos.length === 0) { return; }
// Start at BEP flows
const flowDistribution = pumpInfos.map(info => ({
machineId: info.id,
flow: Math.min(info.maxFlow, Math.max(info.minFlow, info.Q_BEP))
}));
let totalFlow = flowDistribution.reduce((sum, entry) => sum + entry.flow, 0); // Initial total flow
const delta = Qd - totalFlow; // Difference to target demand
if (Math.abs(delta) > 1e-6) {
this.redistributeFlowBySlope(pumpInfos, flowDistribution, delta, directional);
}
// Clamp and compute initial power
flowDistribution.forEach(entry => {
const info = pumpInfos.find(info => info.id === entry.machineId);
entry.flow = Math.min(info.maxFlow, Math.max(info.minFlow, entry.flow));
});
// Marginal-cost refinement: shift flow from most expensive to cheapest
// pump using actual power evaluations on the group operating
// point. Converges regardless of curve convexity.
const mcDelta = Math.max(1e-6, (Qd / pumpInfos.length) * 0.005);
for (let refineIter = 0; refineIter < 50; refineIter++) {
const mcEntries = flowDistribution.map(entry => {
const info = pumpInfos.find(i => i.id === entry.machineId);
const pNow = this._groupCalcPower(info.machine, entry.flow);
const pUp = this._groupCalcPower(info.machine, Math.min(info.maxFlow, entry.flow + mcDelta));
return { entry, info, mc: (pUp - pNow) / mcDelta };
});
let expensive = null, cheap = null;
for (const e of mcEntries) {
if (e.entry.flow > e.info.minFlow + mcDelta) { if (!expensive || e.mc > expensive.mc) expensive = e; }
if (e.entry.flow < e.info.maxFlow - mcDelta) { if (!cheap || e.mc < cheap.mc) cheap = e; }
}
if (!expensive || !cheap || expensive === cheap) break;
if (expensive.mc - cheap.mc < expensive.mc * 0.001) break;
const before = this._groupCalcPower(expensive.info.machine, expensive.entry.flow) + this._groupCalcPower(cheap.info.machine, cheap.entry.flow);
const after = this._groupCalcPower(expensive.info.machine, expensive.entry.flow - mcDelta) + this._groupCalcPower(cheap.info.machine, cheap.entry.flow + mcDelta);
if (after < before) { expensive.entry.flow -= mcDelta; cheap.entry.flow += mcDelta; } else { break; }
}
let totalPower = 0;
totalFlow = 0;
flowDistribution.forEach(entry => {
totalFlow += entry.flow;
const info = pumpInfos.find(i => i.id === entry.machineId);
totalPower += this._groupCalcPower(info.machine, entry.flow);
});
const totalCog = pumpInfos.reduce((sum, info) => sum + info.NCog, 0);
const deviation = pumpInfos.reduce((sum, info) => {
const entry = flowDistribution.find(item => item.machineId === info.id);
const deltaFlow = entry ? (entry.flow - info.Q_BEP) : 0;
return sum + (deltaFlow * deltaFlow) * (info.slopes.alpha || 1);
}, 0);
const shouldUpdate = totalPower < bestPower ||
(totalPower === bestPower && deviation < bestDeviation);
if (shouldUpdate) {
bestCombination = flowDistribution.map(entry => ({ ...entry }));
bestPower = totalPower;
bestFlow = totalFlow;
bestCog = totalCog;
bestDeviation = deviation;
}
});
return {
bestCombination,
bestPower,
bestFlow,
bestCog,
bestDeviation,
method
};
}
// -------- Mode and Input Management -------- //
isValidActionForMode(action, mode) {
const allowedActionsSet = this.config.mode.allowedActions[mode] || [];
return allowedActionsSet.has(action);
}
setScaling(scaling) {
const scalingSet = new Set(this.defaultConfig.scaling.current.rules.values.map( (value) => value.value));
scalingSet.has(scaling)? this.scaling = scaling : this.logger.warn(`${scaling} is not a valid scaling option.`);
this.logger.debug(`Scaling set to: ${scaling}`);
}
async abortActiveMovements(reason = "new demand") {
await Promise.all(Object.values(this.machines).map(async machine => {
this.logger.warn(`Aborting active movements for machine ${machine.config.general.id} due to: ${reason}`);
if (typeof machine.abortMovement === "function") {
await machine.abortMovement(reason);
}
}));
}
//handle input from parent / user / UI
async optimalControl(Qd, powerCap = Infinity) {
try{
if (Object.keys(this.machines).length === 0) {
this.logger.warn("No machines registered. Cannot execute optimal control.");
return;
}
this._equalizeOperatingPoint();
//fetch dynamic totals
const dynamicTotals = this.dynamicTotals;
const machineStates = Object.entries(this.machines).reduce((acc, [machineId, machine]) => {
acc[machineId] = machine.state.getCurrentState();
return acc;
}, {});
if( Qd <= 0 ) {
this.logger.debug("Flow demand <= 0, turning all machines off.");
await this.turnOffAllMachines();
return;
}
if( Qd < dynamicTotals.flow.min && Qd > 0 ){
//Capping Qd to lowest possible value
this.logger.warn(`Flow demand ${Qd} is below minimum possible flow ${dynamicTotals.flow.min}. Capping to minimum flow.`);
Qd = dynamicTotals.flow.min;
}
else if( Qd > dynamicTotals.flow.max ){
//Capping Qd to highest possible value
this.logger.warn(`Flow demand ${Qd} is above maximum possible flow ${dynamicTotals.flow.max}. Capping to maximum flow.`);
Qd = dynamicTotals.flow.max;
}
// fetch all valid combinations that meet expectations
const combinations = this.validPumpCombinations(this.machines, Qd, powerCap);
if (!combinations || combinations.length === 0) {
this.logger.warn(`Demand: ${Qd.toFixed(2)} -> No valid combination found (empty set).`);
return;
}
// Decide which optimization routine we run. Defaults to BEP-based gravitation with directionality.
const optimizationMethod = this.config.optimization?.method || "BEP-Gravitation-Directional";
let bestResult;
if (optimizationMethod === "NCog") {
bestResult = this.calcBestCombination(combinations, Qd);
} else if (
optimizationMethod === "BEP-Gravitation" ||
optimizationMethod === "BEP-Gravitation-Directional"
) {
bestResult = this.calcBestCombinationBEPGravitation(combinations, Qd, optimizationMethod);
} else {
this.logger.warn(`Unknown optimization method '${optimizationMethod}', falling back to BEP-Gravitation-Directional.`);
bestResult = this.calcBestCombinationBEPGravitation(combinations, Qd, "BEP-Gravitation-Directional");
}
if(bestResult.bestCombination === null){
this.logger.warn(`Demand: ${Qd.toFixed(2)} -> No valid combination found => not updating control `);
return;
}
const debugInfo = bestResult.bestCombination.map(({ machineId, flow }) => `${machineId}: ${flow.toFixed(2)} units`).join(" | ");
this.logger.debug(`Moving to demand: ${Qd.toFixed(2)} -> Pumps: [${debugInfo}] => Total Power: ${bestResult.bestPower.toFixed(2)}`);
// Store the optimizer's INTENT on AT_EQUIPMENT (what we
// commanded). DOWNSTREAM is reserved for the live aggregate
// written by handlePressureChange — PS subscribes to that
// for net-flow computation and must see what pumps are
// actually delivering, not the planned target. Writing
// bestFlow to DOWNSTREAM here would clobber the live value
// every handleInput tick (see ps-mgc-flow-contract test).
this._writeMeasurement("power", "predicted", POSITIONS.AT_EQUIPMENT, bestResult.bestPower, this.unitPolicy.canonical.power);
this._writeMeasurement("flow", "predicted", POSITIONS.AT_EQUIPMENT, bestResult.bestFlow, this.unitPolicy.canonical.flow);
this.measurements.type("efficiency").variant("predicted").position(POSITIONS.AT_EQUIPMENT).value(bestResult.bestFlow / bestResult.bestPower);
this.measurements.type("Ncog").variant("predicted").position(POSITIONS.AT_EQUIPMENT).value(bestResult.bestCog);
await Promise.all(Object.entries(this.machines).map(async ([machineId, machine]) => {
// Find the flow for this machine in the best combination
this.logger.debug(`Searching for machine ${machineId} with state ${machineStates[machineId]} in best combination.`);
const pumpInfo = bestResult.bestCombination.find(item => item.machineId == machineId);
let flow;
if(pumpInfo !== undefined){
flow = pumpInfo.flow;
} else {
this.logger.debug(`Machine ${machineId} not in best combination, setting flow control to 0`);
flow = 0;
}
// Dispatch policy: send the setpoint to ANY pump that
// should be running (flow > 0), not just operational
// ones. rotatingMachine.state.moveTo handles queueing:
// - operational → execute immediately
// - accelerating /
// decelerating → unpark post-abort residue
// and execute (state.js fix)
// - idle / starting /
// warmingup / stopping /
// coolingdown → save as delayedMove,
// auto-fires on next
// transition to operational
//
// CRUCIAL ORDERING: flowmovement BEFORE execsequence
// startup. If we awaited startup first (~3 s), other
// concurrent MGC.handleInput calls would update this
// pump's delayedMove during the startup window. When
// startup completes, transitionToState('operational')
// correctly fires the LATEST delayedMove. But then this
// call's chained `await flowmovement(stale)` would run
// on an already-operational pump and overwrite the
// correct position with the stale snapshot value.
//
// By sending flowmovement first, the setpoint lands in
// delayedMove while the pump is still idle. Concurrent
// calls overwrite delayedMove with newer setpoints. The
// final transitionToState('operational') at the end of
// startup fires whichever delayedMove is current — the
// genuinely latest demand wins.
//
// See test/integration/idle-startup-deadlock.integration.test.js
// Scenario 4 for the deterministic reproducer.
const state = machineStates[machineId];
if (flow > 0) {
await machine.handleInput("parent", "flowmovement", this._canonicalToOutputFlow(flow));
if (state === "idle") {
await machine.handleInput("parent", "execsequence", "startup");
}
} else if (state === "operational" || state === "accelerating" || state === "decelerating") {
await machine.handleInput("parent", "execsequence", "shutdown");
}
// flow ≤ 0 AND state already in shutdown chain (idle/
// stopping/coolingdown/off/emergencystop) → nothing
// to do, preserve previous behaviour.
}));
}
catch(err){
this.logger.error(err);
}
}
// Equalize all machines (running + idle) to the group's header
// operating point so dynamicTotals + combination optimization see one
// consistent operating point. See _equalizeOperatingPoint for the
// implementation rationale.
equalizePressure(){
this._equalizeOperatingPoint();
}
// Force every machine's predict-curve interpolators to use the same
// (header) differential pressure for the duration of MGC's optimization.
//
// Why direct fDimension assignment, not measurement writes:
// rotatingMachine._getPreferredPressureValue reads from each pressure
// sensor child (keyed by child id) BEFORE falling back to the position-
// level measurement. MGC has no way to know which child id a pump's
// sensor uses, so writes via _writeChildMeasurement land at the
// "default" child key and are never consulted by getMeasuredPressure().
// Setting fDimension directly is the same effect getMeasuredPressure()
// would have produced if its read had succeeded.
//
// Per-pump diagnostics are unaffected: this only mutates the predict
// objects' interpolation parameter, NOT the pump's measurement container.
// The pump's own emitted upstream/downstream measurements (and the
// differential they imply) keep their real sensor values.
//
// Header source order:
// 1. MGC's own header measurement (a measurement child registered at
// DOWNSTREAM / UPSTREAM with MGC as parent). Authoritative manifold
// reading when present.
// 2. Worst-case envelope across pump-side sensors —
// downstream = max (highest discharge load),
// upstream = min of POSITIVE values (lowest suction = highest
// required head). Zeros are filtered to skip pumps
// that haven't emitted yet.
_equalizeOperatingPoint(){
if (Object.keys(this.machines).length === 0) return;
const groupHeaderDown = this.measurements
.type("pressure").variant("measured").position(POSITIONS.DOWNSTREAM)
.getCurrentValue(this.unitPolicy.canonical.pressure);
const groupHeaderUp = this.measurements
.type("pressure").variant("measured").position(POSITIONS.UPSTREAM)
.getCurrentValue(this.unitPolicy.canonical.pressure);
const childDown = [];
const childUp = [];
Object.values(this.machines).forEach(machine => {
const d = this._readChildMeasurement(machine, "pressure", "measured", POSITIONS.DOWNSTREAM, this.unitPolicy.canonical.pressure);
const u = this._readChildMeasurement(machine, "pressure", "measured", POSITIONS.UPSTREAM, this.unitPolicy.canonical.pressure);
if (Number.isFinite(d) && d > 0) childDown.push(d);
if (Number.isFinite(u) && u > 0) childUp.push(u);
});
const headerDownSrc = Number.isFinite(groupHeaderDown) && groupHeaderDown > 0 ? "header" : "max-child";
const headerUpSrc = Number.isFinite(groupHeaderUp) && groupHeaderUp > 0 ? "header" : "min-child";
const headerDownstream = headerDownSrc === "header" ? groupHeaderDown : (childDown.length ? Math.max(...childDown) : 0);
const headerUpstream = headerUpSrc === "header" ? groupHeaderUp : (childUp.length ? Math.min(...childUp) : 0);
const headerDiff = headerDownstream - headerUpstream;
if (!Number.isFinite(headerDiff) || headerDiff <= 0) {
this.logger.debug(`Skipping equalization: invalid header diff ${headerDiff} (down=${headerDownstream}, up=${headerUpstream})`);
return;
}
this.logger.debug(`Equalizing operating point: down=${headerDownstream} (${headerDownSrc}), up=${headerUpstream} (${headerUpSrc}), diff=${headerDiff}`);
// Push the header operating point onto each pump's group-scope
// predicts. The pump's individual predicts (driven by its own
// sensors) are untouched; only the group view used by this MGC
// is shifted. See rotatingMachine.setGroupOperatingPoint().
Object.values(this.machines).forEach(machine => {
if (typeof machine.setGroupOperatingPoint === "function") {
machine.setGroupOperatingPoint(headerDownstream, headerUpstream);
} else {
// Older rotatingMachine without the group API — fall back
// to direct fDimension write so the demo still works while
// submodules are rolled forward.
if (machine.predictFlow) machine.predictFlow.fDimension = headerDiff;
if (machine.predictPower) machine.predictPower.fDimension = headerDiff;
if (machine.predictCtrl) machine.predictCtrl.fDimension = headerDiff;
}
});
}
// ---------- Group-scope read helpers ----------
// Optimization paths read pump curves at the GROUP operating point,
// not the pump's individual sensor-driven point. These helpers fall
// back to the individual predicts if a pump hasn't been initialised
// for group operation yet (first tick after registration).
_groupFlow(machine) { return machine.groupPredictFlow ?? machine.predictFlow; }
_groupPower(machine) { return machine.groupPredictPower ?? machine.predictPower; }
_groupNCog(machine) { return machine.groupPredictFlow ? (machine.groupNCog ?? 0) : (machine.NCog ?? 0); }
_groupCalcPower(machine, flow) {
return typeof machine.groupCalcPower === "function"
? machine.groupCalcPower(flow)
: machine.inputFlowCalcPower(flow);
}
isMachineActive(machineId){
if(this.machines[machineId].state.getCurrentState() === "operational" || this.machines[machineId].state.getCurrentState() === "accelerating" || this.machines[machineId].state.getCurrentState() === "decelerating"){
return true;
}
return false;
}
capFlowDemand(Qd,dynamicTotals){
if (Qd < dynamicTotals.flow.min && Qd > 0) {
this.logger.warn(`Flow demand ${Qd} is below minimum possible flow ${dynamicTotals.flow.min}. Capping to minimum flow.`);
Qd = dynamicTotals.flow.min;
} else if (Qd > dynamicTotals.flow.max) {
this.logger.warn(`Flow demand ${Qd} is above maximum possible flow ${dynamicTotals.flow.max}. Capping to maximum flow.`);
Qd = dynamicTotals.flow.max;
}
return Qd;
}
sortMachinesByPriority(priorityList) {
let machinesInPriorityOrder;
if (priorityList && Array.isArray(priorityList)) {
machinesInPriorityOrder = priorityList
.filter(id => this.machines[id])
.map(id => ({ id, machine: this.machines[id] }));
} else {
machinesInPriorityOrder = Object.entries(this.machines)
.map(([id, machine]) => ({ id: id, machine }))
.sort((a, b) => a.id - b.id);
}
return machinesInPriorityOrder;
}
filterOutUnavailableMachines(list) {
const newList = list.filter(({ machine }) => {
const state = machine.state.getCurrentState();
const validActionForMode = machine.isValidActionForMode("execsequence", "auto");
return !(state === "off" || state === "coolingdown" || state === "stopping" || state === "emergencystop" || !validActionForMode);
});
return newList;
}
calcGroupEfficiency(machines){
let cumEfficiency = 0;
let machineCount = 0;
let lowestEfficiency = Infinity;
// Calculate the average efficiency of all machines -> peak is the average of them all
Object.entries(machines).forEach(([_machineId, machine]) => {
cumEfficiency += machine.cog;
if(machine.cog < lowestEfficiency){
lowestEfficiency = machine.cog;
}
machineCount++;
});
const maxEfficiency = cumEfficiency / machineCount;
return { maxEfficiency, lowestEfficiency };
}
//move machines assuming equal control in flow and a priority list
async equalFlowControl(Qd, _powerCap = Infinity, priorityList = null) {
try {
// equalize pressure across all machines
this.equalizePressure();
// Update dynamic totals
const dynamicTotals = this.calcDynamicTotals();
// Cap flow demand to min/max possible values
Qd = this.capFlowDemand(Qd,dynamicTotals);
// Get machines sorted by priority
let machinesInPriorityOrder = this.sortMachinesByPriority(priorityList);
// Filter out machines that are unavailable for control
machinesInPriorityOrder = this.filterOutUnavailableMachines(machinesInPriorityOrder);
// Initialize flow distribution
let flowDistribution = [];
let totalFlow = 0;
let totalPower = 0;
let totalCog = 0;
const activeTotals = this.activeTotals();
// Distribute flow equally among all available machines
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 machine = machinesInPriorityOrder[i];
if (this.isMachineActive(machine.id)) {
flowDistribution.push({ machineId: machine.id, flow: 0 });
availableFlow -= this._groupFlow(machine.machine).currentFxyYMin;
}
}
// Determine remaining active machines (not shut down).
const remainingMachines = machinesInPriorityOrder.filter(
({ id }) =>
this.isMachineActive(id) &&
!flowDistribution.some(item => item.machineId === id)
);
// Evenly distribute Qd among the remaining machines.
const distributedFlow = Qd / remainingMachines.length;
for (let machine of remainingMachines) {
flowDistribution.push({ machineId: machine.id, flow: distributedFlow });
totalFlow += distributedFlow;
totalPower += this._groupCalcPower(machine.machine, distributedFlow);
}
break;
}
case (Qd > activeTotals.flow.max): {
// Case 2: Demand is above the maximum available flow.
// Start the non-active machine with the highest priority and distribute Qd over all available machines.
let i = 1;
while (totalFlow < Qd && i <= machinesInPriorityOrder.length) {
Qd = Qd / i;
if(this._groupFlow(machinesInPriorityOrder[i-1].machine).currentFxyYMax >= Qd){
for ( let i2 = 0; i2 < i ; i2++){
if(! this.isMachineActive(machinesInPriorityOrder[i2].id)){
flowDistribution.push({ machineId: machinesInPriorityOrder[i2].id, flow: Qd });
totalFlow += Qd;
totalPower += this._groupCalcPower(machinesInPriorityOrder[i2].machine, Qd);
}
}
}
i++;
}
break;
}
default: {
// Default case: Demand is within the active range.
const countActiveMachines = machinesInPriorityOrder.filter(({ id }) => this.isMachineActive(id)).length;
Qd /= countActiveMachines;
// Simply distribute the demand equally among all available machines.
for ( let i = 0 ; i < countActiveMachines ; i++){
flowDistribution.push({ machineId: machinesInPriorityOrder[i].id, flow: Qd});
totalFlow += Qd ;
totalPower += this._groupCalcPower(machinesInPriorityOrder[i].machine, Qd);
}
break;
}
}
// Log information about flow distribution
const debugInfo = flowDistribution
.filter(({ flow }) => flow > 0)
.map(({ machineId, flow }) => `${machineId}: ${flow.toFixed(2)} units`)
.join(" | ");
this.logger.debug(`Priority control for demand: ${totalFlow.toFixed(2)} -> Active pumps: [${debugInfo}] => Total Power: ${totalPower.toFixed(2)}`);
// Store the planned distribution as INTENT on AT_EQUIPMENT.
// DOWNSTREAM (live aggregate) is owned by handlePressureChange.
// Writing the plan here would clobber PS's outflow signal.
this._writeMeasurement("power", "predicted", POSITIONS.AT_EQUIPMENT, totalPower, this.unitPolicy.canonical.power);
this._writeMeasurement("flow", "predicted", POSITIONS.AT_EQUIPMENT, totalFlow, this.unitPolicy.canonical.flow);
this.measurements.type("efficiency").variant("predicted").position(POSITIONS.AT_EQUIPMENT).value(totalFlow / totalPower);
this.measurements.type("Ncog").variant("predicted").position(POSITIONS.AT_EQUIPMENT).value(totalCog);
this.logger.debug(`Flow distribution: ${JSON.stringify(flowDistribution)}`);
// Apply the flow distribution to machines
await Promise.all(flowDistribution.map(async ({ machineId, flow }) => {
const machine = this.machines[machineId];
this.logger.debug(this.machines[machineId].state);
const currentState = this.machines[machineId].state.getCurrentState();
// Same dispatch shape as optimalControl — see the comment
// there for the rationale. flowmovement BEFORE startup so
// concurrent retargets can update delayedMove without a
// stale chained flowmovement overwriting it after startup.
if (flow > 0) {
await machine.handleInput("parent", "flowmovement", this._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) {
this.logger.error(err);
}
}
//only valid with equal machines
async prioPercentageControl(input, priorityList = null) {
try{
// stop all machines if input is negative
if(input < 0 ){
await this.turnOffAllMachines();
return;
}
//capp input to 100
if (input > 100) { input = 100; }
const numOfMachines = Object.keys(this.machines).length;
const procentTotal = numOfMachines * input;
const machinesNeeded = Math.ceil(procentTotal/100);
const activeTotals = this.activeTotals();
const machinesActive = activeTotals.countActiveMachines;
// Get machines sorted by priority
let machinesInPriorityOrder = this.sortMachinesByPriority(priorityList);
const ctrlDistribution = []; //{machineId : 0, flow : 0} push for each machine
if(machinesNeeded > machinesActive){
//start extra machine and put all active machines at min control
machinesInPriorityOrder.forEach(({ id }, index) => {
if(index < machinesNeeded){
ctrlDistribution.push({machineId : id, ctrl : 0});
}
});
}
if(machinesNeeded < machinesActive){
machinesInPriorityOrder.forEach(({ id }, index) => {
if(this.isMachineActive(id)){
if(index < machinesNeeded){
ctrlDistribution.push({machineId : id, ctrl : 100});
}
else{
//turn machine off
ctrlDistribution.push({machineId : id, ctrl : -1});
}
}
});
}
if (machinesNeeded === machinesActive) {
// distribute input equally among active machines (0 - 100%)
const ctrlPerMachine = procentTotal / machinesActive;
machinesInPriorityOrder.forEach(({ id }) => {
if (this.isMachineActive(id)) {
// ensure ctrl is capped between 0 and 100%
const ctrlValue = Math.max(0, Math.min(ctrlPerMachine, 100));
ctrlDistribution.push({ machineId: id, ctrl: ctrlValue });
}
});
}
const debugInfo = ctrlDistribution.map(({ machineId, ctrl }) => `${machineId}: ${ctrl.toFixed(2)}%`).join(" | ");
this.logger.debug(`Priority control for input: ${input.toFixed(2)} -> Active pumps: [${debugInfo}]`);
// Apply the ctrl distribution to machines
await Promise.all(ctrlDistribution.map(async ({ machineId, ctrl }) => {
const machine = this.machines[machineId];
const currentState = this.machines[machineId].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 = [];
// fetch and store measurements
Object.entries(this.machines).forEach(([_machineId, machine]) => {
const powerValue = this._readChildMeasurement(machine, "power", "predicted", POSITIONS.AT_EQUIPMENT, this.unitPolicy.canonical.power);
const flowValue = this._readChildMeasurement(machine, "flow", "predicted", POSITIONS.DOWNSTREAM, this.unitPolicy.canonical.flow);
if (powerValue !== null) {
totalPower.push(powerValue);
}
if (flowValue !== null) {
totalFlow.push(flowValue);
}
});
// Write to AT_EQUIPMENT not DOWNSTREAM. handlePressureChange
// is the canonical writer of DOWNSTREAM (the live aggregate
// that PS subscribes to for outflow). See optimalControl
// comment above.
this._writeMeasurement("power", "predicted", POSITIONS.AT_EQUIPMENT, totalPower.reduce((a, b) => a + b, 0), this.unitPolicy.canonical.power);
this._writeMeasurement("flow", "predicted", POSITIONS.AT_EQUIPMENT, totalFlow.reduce((a, b) => a + b, 0), this.unitPolicy.canonical.flow);
if(totalPower.reduce((a, b) => a + b, 0) > 0){
this.measurements.type("efficiency").variant("predicted").position(POSITIONS.AT_EQUIPMENT).value(totalFlow.reduce((a, b) => a + b, 0) / totalPower.reduce((a, b) => a + b, 0));
}
}
catch(err){
this.logger.error(err);
}
}
async handleInput(source, demand, powerCap = Infinity, priorityList = null) {
const demandQ = parseFloat(demand);
if(!Number.isFinite(demandQ)){
this.logger.error(`Invalid flow demand input: ${demand}. Must be a finite number.`);
return;
}
//abort current movements
await this.abortActiveMovements("new demand received");
const scaling = this.scaling;
const mode = this.mode;
const dynamicTotals = this.calcDynamicTotals();
let demandQout = 0; // keep output Q by default 0 for safety
this.logger.debug(`Handling input from ${source}: Demand = ${demand}, Power Cap = ${powerCap}, Priority List = ${priorityList}`);
switch (scaling) {
case "absolute":
if (isNaN(demandQ)) {
this.logger.warn(`Invalid absolute flow demand: ${demand}. Must be a number.`);
demandQout = 0;
return;
}
if (demandQ <= 0) {
this.logger.debug(`Turning machines off`);
demandQout = 0;
await this.turnOffAllMachines();
return;
} else if (demandQ < this.absoluteTotals.flow.min) {
this.logger.warn(`Flow demand ${demandQ} is below minimum possible flow ${this.absoluteTotals.flow.min}. Capping to minimum flow.`);
demandQout = this.absoluteTotals.flow.min;
} else if (demandQ > this.absoluteTotals.flow.max) {
this.logger.warn(`Flow demand ${demandQ} is above maximum possible flow ${this.absoluteTotals.flow.max}. Capping to maximum flow.`);
demandQout = this.absoluteTotals.flow.max;
} else {
demandQout = demandQ;
}
break;
case "normalized":
this.logger.debug(`Normalizing flow demand: ${demandQ} with min: ${dynamicTotals.flow.min} and max: ${dynamicTotals.flow.max}`);
// demand <= 0 → off. Previously only `< 0` triggered off,
// so demand=0 fell through to interpolate(0, 0..100, min..max)
// which returns flow.min — i.e., a pumpingStation dead-zone
// (level in [stopLevel, startLevel] sending percControl=0)
// would silently keep a pump running at min flow,
// balancing inflow and pinning the basin in the dead band.
if (demandQ <= 0) {
this.logger.debug(`Demand ≤ 0 — turning all machines off`);
demandQout = 0;
await this.turnOffAllMachines();
return;
}
// Scale demand to flow range. interpolate_lin_single_point
// maps demandQ (0..100) onto (flow.min..flow.max) linearly.
demandQout = this.interpolation.interpolate_lin_single_point(demandQ, 0, 100, dynamicTotals.flow.min, dynamicTotals.flow.max );
this.logger.debug(`Normalized flow demand ${demandQ}% to: ${demandQout} Q units`);
break;
}
// Execute control based on mode
switch(mode) {
case "prioritycontrol":
this.logger.debug(`Calculating prio control. Input flow demand: ${demandQ} scaling : ${scaling} -> ${demandQout}`);
await this.equalFlowControl(demandQout,powerCap,priorityList);
break;
case "prioritypercentagecontrol":
this.logger.debug(`Calculating prio percentage control. Input flow demand: ${demandQ} scaling : ${scaling} -> ${demandQout}`);
if(scaling !== "normalized"){
this.logger.warn("Priority percentage control is only valid with normalized scaling.");
return;
}
await this.prioPercentageControl(demandQout,priorityList);
break;
case "optimalcontrol":
this.logger.debug(`Calculating optimal control. Input flow demand: ${demandQ} scaling : ${scaling} -> ${demandQout}`);
await this.optimalControl(demandQout,powerCap);
break;
default:
this.logger.warn(`${mode} is not a valid mode.`);
break;
}
//recalc distance from BEP
const { maxEfficiency, lowestEfficiency } = this.calcGroupEfficiency(this.machines);
const efficiency = this.measurements.type("efficiency").variant("predicted").position(POSITIONS.AT_EQUIPMENT).getCurrentValue();
this.calcDistanceBEP(efficiency,maxEfficiency,lowestEfficiency);
}
async turnOffAllMachines(){
await Promise.all(Object.entries(this.machines).map(async ([machineId, machine]) => {
if (this.isMachineActive(machineId)) { await machine.handleInput("parent", "execsequence", "shutdown"); }
}));
// Update measurements to zero so the parent (PS) sees the
// outflow drop immediately — without this the PS keeps the
// last active flow value cached and computes wrong net flow.
this._writeMeasurement("flow", "predicted", POSITIONS.DOWNSTREAM, 0, this.unitPolicy.canonical.flow);
this._writeMeasurement("flow", "predicted", POSITIONS.AT_EQUIPMENT, 0, this.unitPolicy.canonical.flow);
this._writeMeasurement("power", "predicted", POSITIONS.AT_EQUIPMENT, 0, this.unitPolicy.canonical.power);
}
_buildUnitPolicy(config = {}) {
const flowUnit = this._resolveUnitOrFallback(
config?.general?.unit,
'volumeFlowRate',
DEFAULT_IO_UNITS.flow
);
const pressureUnit = this._resolveUnitOrFallback(
config?.general?.pressureUnit,
'pressure',
DEFAULT_IO_UNITS.pressure
);
const powerUnit = this._resolveUnitOrFallback(
config?.general?.powerUnit,
'power',
DEFAULT_IO_UNITS.power
);
return {
canonical: { ...CANONICAL_UNITS },
output: {
flow: flowUnit,
pressure: pressureUnit,
power: powerUnit,
temperature: DEFAULT_IO_UNITS.temperature,
},
};
}
_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;
}
}
_canonicalToOutputFlow(value) {
const from = this.unitPolicy.canonical.flow;
const to = this.unitPolicy.output.flow;
if (!from || !to || from === to) return value;
return convert(value).from(from).to(to);
}
_outputUnitForType(type) {
switch (String(type || '').toLowerCase()) {
case 'flow':
return this.unitPolicy.output.flow;
case 'power':
return this.unitPolicy.output.power;
case 'pressure':
return this.unitPolicy.output.pressure;
case 'temperature':
return this.unitPolicy.output.temperature;
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()) {
if (!Number.isFinite(value)) {
return;
}
this.measurements
.type(type)
.variant(variant)
.position(position)
.value(value, timestamp, unit || undefined);
}
_readChildMeasurement(machine, type, variant, position, unit = null) {
return machine?.measurements
?.type(type)
?.variant(variant)
?.position(position)
?.getCurrentValue(unit || undefined);
}
_writeChildMeasurement(machine, type, variant, position, value, unit = null, timestamp = Date.now()) {
if (!machine?.measurements || !Number.isFinite(value)) {
return;
}
machine.measurements
.type(type)
.variant(variant)
.position(position)
.value(value, timestamp, unit || undefined);
}
setMode(mode) {
this.mode = mode;
}
getOutput() {
// Improved output object generation
const output = {};
//build the output object
this.measurements.getTypes().forEach(type => {
this.measurements.getVariants(type).forEach(variant => {
const unit = this._outputUnitForType(type);
const downstreamVal = this._readMeasurement(type, variant, POSITIONS.DOWNSTREAM, unit);
const atEquipmentVal = this._readMeasurement(type, variant, POSITIONS.AT_EQUIPMENT, unit);
const upstreamVal = this._readMeasurement(type, variant, POSITIONS.UPSTREAM, unit);
if (downstreamVal != null) {
output[`downstream_${variant}_${type}`] = downstreamVal;
}
if (upstreamVal != null) {
output[`upstream_${variant}_${type}`] = upstreamVal;
}
if (atEquipmentVal != null) {
output[`atEquipment_${variant}_${type}`] = atEquipmentVal;
}
if (downstreamVal != null && upstreamVal != null) {
const diff = this.measurements
.type(type)
.variant(variant)
.difference({ from: POSITIONS.DOWNSTREAM, to: POSITIONS.UPSTREAM, unit });
if (diff?.value != null) {
output[`differential_${variant}_${type}`] = diff.value;
}
}
});
});
//fill in the rest of the output object
output["mode"] = this.mode;
output["scaling"] = this.scaling;
output["flow"] = this.flow;
output["power"] = this.power;
output["NCog"] = this.NCog; // normalized cog
output["absDistFromPeak"] = this.absDistFromPeak;
output["relDistFromPeak"] = this.relDistFromPeak;
//this.logger.debug(`Output: ${JSON.stringify(output)}`);
return output;
}
}
module.exports = MachineGroup;
/*
const {coolprop} = require('generalFunctions');
const Machine = require('../../rotatingMachine/src/specificClass');
const Measurement = require('../../measurement/src/specificClass');
const specs = require('../../generalFunctions/datasets/assetData/curves/hidrostal-H05K-S03R.json');
const { max } = require("mathjs");
function createBaseMachineConfig(machineNum, name,specs) {
return {
general: {
logging: { enabled: true, logLevel: "debug" },
name: name,
id: machineNum,
unit: "m3/h"
},
functionality: {
softwareType: "machine",
role: "rotationaldevicecontroller"
},
asset: {
category: "pump",
type: "centrifugal",
model: "hidrostal-h05k-s03r",
supplier: "hydrostal",
machineCurve: specs
},
mode: {
current: "auto",
allowedActions: {
auto: ["execsequence", "execmovement", "statuscheck"],
virtualControl: ["execmovement", "statuscheck"],
fysicalControl: ["statuscheck"]
},
allowedSources: {
auto: ["parent", "GUI"],
virtualControl: ["GUI"],
fysicalControl: ["fysical"]
}
},
sequences: {
startup: ["starting", "warmingup", "operational"],
shutdown: ["stopping", "coolingdown", "idle"],
emergencystop: ["emergencystop", "off"],
boot: ["idle", "starting", "warmingup", "operational"]
}
};
}
function createStateConfig(){
return {
time:{
starting: 1,
stopping: 1,
warmingup: 1,
coolingdown: 1,
emergencystop: 1
},
movement:{
mode:"dynspeed",
speed:100,
maxSpeed: 1000
}
}
};
function createBaseMachineGroupConfig(name) {
return {
general: {
logging: { enabled: true, logLevel: "debug" },
name: name
},
functionality: {
softwareType: "machinegroup",
role: "groupcontroller"
},
scaling: {
current: "normalized"
},
mode: {
current: "optimalControl"
}
};
}
const machineGroupConfig = createBaseMachineGroupConfig("testmachinegroup");
const stateConfigs = {};
const machineConfigs = {};
stateConfigs[1] = createStateConfig();
stateConfigs[2] = createStateConfig();
machineConfigs[1]= createBaseMachineConfig("asdfkj;asdf","testmachine",specs);
machineConfigs[2] = createBaseMachineConfig("asdfkj;asdf2","testmachine2",specs);
const ptConfig = {
general: {
logging: { enabled: true, logLevel: "debug" },
name: "testpt",
id: "0",
unit: "mbar",
},
functionality: {
softwareType: "measurement",
role: "sensor"
},
asset: {
category: "sensor",
type: "pressure",
model: "testmodel",
supplier: "vega"
},
scaling:{
absMin:0,
absMax: 4000,
}
}
async function makeMachines(){
const mg = new MachineGroup(machineGroupConfig);
const pt1 = new Measurement(ptConfig);
const numofMachines = 2;
for(let i = 1; i <= numofMachines; i++){
const machine = new Machine(machineConfigs[i],stateConfigs[i]);
//mg.machines[i] = machine;
mg.childRegistrationUtils.registerChild(machine, "downstream");
}
Object.keys(mg.machines).forEach(machineId => {
mg.machines[machineId].childRegistrationUtils.registerChild(pt1, "downstream");
});
mg.setMode("prioritycontrol");
mg.setScaling("normalized");
const absMax = mg.dynamicTotals.flow.max;
const absMin = mg.dynamicTotals.flow.min;
const percMin = 0;
const percMax = 100;
try{
for(let demand = mg.dynamicTotals.flow.min ; demand <= mg.dynamicTotals.flow.max ; demand += 2){
//set pressure
console.log("------------------------------------");
await mg.handleInput("parent",demand);
pt1.calculateInput(1400);
//await new Promise(resolve => setTimeout(resolve, 200));
console.log("------------------------------------");
}
for(let demand = 240 ; demand >= mg.dynamicTotals.flow.min ; demand -= 40){
//set pressure
console.log("------------------------------------");
await mg.handleInput("parent",demand);
pt1.calculateInput(1400);
//await new Promise(resolve => setTimeout(resolve, 200));
console.log("------------------------------------");
}
//*//*
for(let demand = 0 ; demand <= 50 ; demand += 1){
//set pressure
console.log(`TESTING: processing demand of ${demand}`);
await mg.handleInput("parent",demand);
Object.keys(mg.machines).forEach(machineId => {
console.log(mg.machines[machineId].state.getCurrentState());
});
console.log(`updating pressure to 1400 mbar`);
pt1.calculateInput(1400);
console.log("------------------------------------");
}
}
catch(err){
console.log(err);
}
}
if (require.main === module) {
makeMachines();
}
//*/