RnD/machineGroupControl
3
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mgc): demand telemetry + movement gate (demand debounce)

Browse Source 
- Movement gate: hold non-urgent demand while the group is 'working'
  (mid-ramp/sequencing) and flush it once 'ready', instead of aborting
  in-flight ramps on every incoming demand — which could freeze pumps at 0.
  Urgent demand (stop, mode/priority change, large step) still pre-empts.
- getMovementState()/_isUrgentDemand()/_maybeFlushPendingDemand() helpers.
- Demand telemetry: emit demandFlow (m³/h) and demandPct (0..100 of envelope)
  resolved by the last dispatch; omitted before the first demand (degraded).
- Capacity envelope now emitted in output flow unit (m³/h) not raw m³/s.
- Manifest + populated/degraded tests for the new outputs.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
znetsixe
2026-05-27 16:09:18 +02:00
parent e1e1977139
commit b59d8e60f7
5 changed files with 311 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/io/output.js
View File

@@ -58,15 +58,48 @@ function getOutput(mgc) {
// Group capacity + active-machine counts. Surfaced so dashboards can
// show the same numbers the status badge does without subscribing to
// every child node individually.
out.flowCapacityMax = mgc.dynamicTotals?.flow?.max ?? 0;
out.flowCapacityMin = mgc.dynamicTotals?.flow?.min ?? 0;
// every child node individually. Emitted in the output flow unit (m³/h)
// so the dashed capacity envelope lands on the SAME axis as the predicted-
// flow series — dynamicTotals is canonical m³/s, so convert here. (Both
// telemetry consumers — the Grafana flow panel and the FlowFuse fanout —
// assume m³/h; emitting raw m³/s made the capacity lines render as ~0.)
const fUnit = unitPolicy.output.flow;
const capMax = mgc.dynamicTotals?.flow?.max;
const capMin = mgc.dynamicTotals?.flow?.min;
out.flowCapacityMax = Number.isFinite(capMax)
? unitPolicy.convert(capMax, 'm3/s', fUnit, 'MGC flow capacity max') : 0;
out.flowCapacityMin = Number.isFinite(capMin)
? unitPolicy.convert(capMin, 'm3/s', fUnit, 'MGC flow capacity min') : 0;
// Operator demand resolved by the last dispatch. Surfaced so the dashboard
// can overlay "what was asked" against the achieved total flow:
// - demandFlow: resolved flow setpoint (post-envelope-clamp) in the output
// flow unit (m³/h), same scale as the total-flow series.
// - demandPct: that setpoint as 0..100 % of the live capacity envelope
// (flow.min..flow.max), so a % demand entered by the operator round-trips
// regardless of whether they asked in % or absolute flow.
// Omitted entirely before the first demand arrives (degraded state).
if (mgc._lastDemand) {
const clampedCanonical = mgc._lastDemand.clamped;
out.demandFlow = unitPolicy.convert(clampedCanonical, 'm3/s', fUnit, 'MGC demand setpoint');
const span = Number.isFinite(capMin) && Number.isFinite(capMax) ? capMax - capMin : 0;
out.demandPct = span > 0
? Math.max(0, Math.min(100, ((clampedCanonical - capMin) / span) * 100))
: 0;
}
out.machineCount = Object.keys(mgc.machines || {}).length;
out.machineCountActive = Object.values(mgc.machines || {}).filter((m) => {
const s = m?.state?.getCurrentState?.();
const md = m?.currentMode;
return s && s !== 'off' && s !== 'maintenance' && md !== 'maintenance';
}).length;
// Group movement status: 'working' while any child is still ramping /
// sequencing toward its dispatched setpoint, 'ready' once all have settled.
// The dispatch gate holds non-urgent demand until 'ready'; surfacing it lets
// a dashboard show why a fresh setpoint hasn't been applied yet.
out.movementState = typeof mgc.getMovementState === 'function' ? mgc.getMovementState() : 'ready';
return out;
}

110
src/specificClass.js
View File

@@ -27,6 +27,13 @@ const MovementExecutor = require('./movement/movementExecutor');
const ACTIVE_STATES = new Set(['operational', 'accelerating', 'decelerating']);
// A machine in one of these states has settled — it is not mid-ramp and is
// not stepping through a start/stop sequence. Anything else (starting,
// warmingup, accelerating, decelerating, stopping, coolingdown) means the
// group is still converging on its last dispatched intent. Drives
// getMovementState(): 'ready' when every machine is settled, else 'working'.
const SETTLED_STATES = new Set(['operational', 'idle', 'off', 'maintenance', 'emergencystop']);
// Canonical mode names (camelCase). The dispatcher already lowercases for its
// switch, but we normalise at setMode so this.mode is always in the canonical
// form — keeps allowedActions/allowedSources lookups (which key on the
@@ -63,6 +70,19 @@ class MachineGroup extends BaseDomain {
this.mode = _normaliseMode(this.config.mode.current) || 'optimalControl';
this.absDistFromPeak = 0;
this.relDistFromPeak = 0;
// Last operator demand resolved by _runDispatch. `null` until the first
// demand arrives so getOutput can omit the demand telemetry (the
// degraded / pre-first-demand state) rather than emit a misleading 0.
// { canonical: m³/s requested, clamped: m³/s after envelope clamp }.
this._lastDemand = null;
// Demand held by the movement gate while the group is 'working'. Latest
// wins; flushed by _maybeFlushPendingDemand once the group is 'ready'.
this._pendingDemand = null;
// Intent of the last dispatch that actually proceeded — used by the
// movement gate to treat a mode/priority change as urgent (a new
// intent), not a hold-worthy nudge.
this._lastDispatchedMode = null;
this._lastPriorityKey = JSON.stringify(null);
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 } };
@@ -213,6 +233,68 @@ class MachineGroup extends BaseDomain {
const eff = this.measurements.type('efficiency').variant('predicted').position(POSITIONS.AT_EQUIPMENT).getCurrentValue() ?? null;
this.calcDistanceBEP(eff, maxEfficiency, lowestEfficiency);
this.notifyOutputChanged();
// Group may have just settled — release any demand the gate is holding.
this._maybeFlushPendingDemand();
}
// Aggregate movement status of the group:
// 'working' — at least one machine is mid-ramp, has a queued setpoint
// (delayedMove), still has move time left, OR the executor
// has scheduled commands that haven't fired yet.
// 'ready' — every machine has settled; a fresh demand can be dispatched
// cleanly without interrupting an in-flight move.
// Surfaced as telemetry (out.movementState) and used by the dispatch gate
// to hold non-urgent demand until the group is ready, instead of aborting
// ramps on every incoming demand (which froze pumps at 0 — connected
// devices must never be able to do that). Urgent demand still pre-empts.
getMovementState() {
const machines = Object.values(this.machines);
if (machines.length === 0) return 'ready';
if (typeof this.movementExecutor?.pending === 'function' && this.movementExecutor.pending() > 0) {
return 'working';
}
for (const m of machines) {
const st = m?.state?.getCurrentState?.();
if (st && !SETTLED_STATES.has(st)) return 'working';
if (m?.state?.delayedMove != null) return 'working';
if ((m?.state?.getMoveTimeLeft?.() ?? 0) > 0) return 'working';
}
return 'ready';
}
// Is this demand urgent enough to pre-empt an in-flight group movement?
// • a stop (≤0) is always urgent — never make the operator wait to stop;
// • the first demand (no prior) dispatches immediately;
// • a control-mode switch or a changed priority order is a new intent,
// not a nudge — dispatch it now rather than holding it;
// • otherwise a step larger than `planner.urgentDemandFraction` of the
// capacity envelope (default 25%) pre-empts; smaller nudges wait for
// the group to be 'ready' so they don't thrash the current ramp.
_isUrgentDemand(demandQ, priorityList) {
if (!(demandQ > 0)) return true;
if (this._lastDemand?.canonical == null) return true;
if (this.mode !== this._lastDispatchedMode) return true;
if (JSON.stringify(priorityList ?? null) !== this._lastPriorityKey) return true;
const dt = (typeof this.calcDynamicTotals === 'function' ? this.calcDynamicTotals() : this.dynamicTotals) || {};
const span = Number(dt?.flow?.max) || 0;
if (span <= 0) return true;
const frac = Math.abs(demandQ - this._lastDemand.canonical) / span;
const thr = Number(this.config?.planner?.urgentDemandFraction);
return frac >= (Number.isFinite(thr) ? thr : 0.25);
}
// Dispatch a demand held by the movement gate, once the group has settled.
// Driven off handlePressureChange (fires several times/s), so a held demand
// is applied promptly when the last ramp completes. Routed back through the
// latest-wins dispatcher so a demand arriving in the same window still wins.
_maybeFlushPendingDemand() {
if (!this._pendingDemand) return;
if (this.getMovementState() !== 'ready') return;
const p = this._pendingDemand;
this._pendingDemand = null;
this.logger.debug(`Group 'ready' — dispatching held demand ${Number(p.demand).toFixed(3)}.`);
Promise.resolve(this._demandDispatcher.fireAndWait(p))
.catch((e) => this.logger?.error?.(`deferred dispatch failed: ${e?.message || e}`));
}
async abortActiveMovements(reason = 'new demand') {
@@ -402,6 +484,26 @@ class MachineGroup extends BaseDomain {
// The handler routes negatives directly to turnOffAllMachines, but
// keep a defensive check in case turnOff-state arrives some other way.
if (demandQ <= 0) { await this.turnOffAllMachines(); return; }
// Movement gate. If the group is still converging on its previous
// intent ('working') and this demand is NOT urgent, hold it instead of
// aborting the in-flight ramps. The held demand (latest wins) is
// dispatched the moment the group reports 'ready'
// (_maybeFlushPendingDemand, off handlePressureChange). This is what
// stops a fast-re-commanding parent from freezing pumps at 0 by
// aborting every ramp before it can progress. Urgent demand (shutdown,
// or a large step) still pre-empts and dispatches immediately.
if (this.getMovementState() === 'working' && !this._isUrgentDemand(demandQ, priorityList)) {
this._pendingDemand = { source, demand: demandQ, powerCap, priorityList };
this.logger.debug(`Demand ${demandQ.toFixed(3)} held — group 'working'; will dispatch when 'ready'.`);
return;
}
this._pendingDemand = null;
// Record the intent now driving the group, so a later same-magnitude
// demand in the same mode/priority is correctly seen as a nudge.
this._lastDispatchedMode = this.mode;
this._lastPriorityKey = JSON.stringify(priorityList ?? null);
await this.abortActiveMovements('new demand received');
const dt = this.calcDynamicTotals();
// Clamp against the current-pressure envelope.
@@ -409,6 +511,12 @@ class MachineGroup extends BaseDomain {
if (demandQout < dt.flow.min) demandQout = dt.flow.min;
else if (demandQout > dt.flow.max) demandQout = dt.flow.max;
// Record what the operator asked for (canonical) and the setpoint we
// actually drive after the current-pressure envelope clamp. getOutput
// turns this into the demand telemetry the dashboard overlays on the
// total-flow graph (resolved flow setpoint + % of group capacity).
this._lastDemand = { canonical: demandQ, clamped: demandQout };
// Normalize for the switch — schema enum values use camelCase
// (optimalControl, priorityControl) while legacy callers send
// lowercase. Accept both rather than silently falling through.
@@ -429,6 +537,8 @@ class MachineGroup extends BaseDomain {
// Cancel any parked demand — turnOff is latest user intent so a
// pending fireAndWait must not re-engage pumps post-shutdown.
this._demandDispatcher.cancelPending();
// Demand resolved to "stop": reflect 0 setpoint in the telemetry.
this._lastDemand = { canonical: 0, clamped: 0 };
await Promise.all(Object.entries(this.machines).map(async ([id, machine]) => {
if (this._shutdownInFlight.has(id)) return;
if (this.isMachineActive(id)) {

7
test/_output-manifest.md
View File

@@ -26,10 +26,13 @@ Built by `src/io/output.js :: getOutput(mgc)`. Delta-compressed by
| `scaling` | `mgc.scaling` | string ∈ {`absolute`, `normalized`} or undefined | commands.basic.test.js | dashboard-fanout (undefined → raw-rows shows '—') |
| `absDistFromPeak` | `groupEfficiency.calcDistanceFromPeak` (specificClass.js:132) | number ≥ 0 (η-points) | bep-distance-demand-sweep, group-bep-cascade, groupEfficiency.basic | groupEfficiency.basic test 7 (undefined when current = null) |
| `relDistFromPeak` | `groupEfficiency.calcRelativeDistanceFromPeak` | number ∈ [0,1] **OR `undefined`** for degenerate (homogeneous pumps) | bep-distance-demand-sweep, group-bep-cascade | groupEfficiency.basic tests 5/6/7 (undefined cases), dashboard-fanout test 11 (undefined → '—' display) |
| `flowCapacityMax` | `mgc.dynamicTotals.flow.max` (totalsCalculator) | number m³/s ≥ 0 | totalsCalculator.basic, dashboard-fanout (post-setup) | absent until first equalize; dashboard-fanout (state A) |
| `flowCapacityMin` | `mgc.dynamicTotals.flow.min` | number m³/s ≥ 0 | totalsCalculator.basic | same as above |
| `flowCapacityMax` | `mgc.dynamicTotals.flow.max` (totalsCalculator), **converted to `unitPolicy.output.flow` (m³/h)** in output.js:62 | number m³/h ≥ 0; `0` when envelope unresolved (Infinity/NaN) | totalsCalculator.basic, dashboard-fanout (post-setup), demand-telemetry.basic | absent until first equalize; dashboard-fanout (state A); demand-telemetry (Infinity → 0) |
| `flowCapacityMin` | `mgc.dynamicTotals.flow.min`, **converted to output flow unit (m³/h)** | number m³/h ≥ 0; `0` when unresolved | totalsCalculator.basic, demand-telemetry.basic | same as above |
| `demandFlow` | `mgc._lastDemand.clamped` (set in `_runDispatch`, output.js:62) | number, canonical m³/s clamped to envelope, converted to `unitPolicy.output.flow` | demand-telemetry.basic (populated) | demand-telemetry.basic (absent before first demand); turnOff → 0 |
| `demandPct` | derived `(clamped − flow.min)/(flow.max − flow.min)·100` (output.js:62) | number ∈ [0,100], `0` when capacity span ≤ 0 | demand-telemetry.basic (populated) | demand-telemetry.basic (absent before first demand) |
| `machineCount` | `Object.keys(mgc.machines).length` | integer ≥ 0 | demand-cycle-walkthrough, ncog-distribution | n/a — always reflects current registration count |
| `machineCountActive` | filtered count excluding `off`/`maintenance` states | integer ≥ 0 | demand-cycle-walkthrough, ncog-distribution | dashboard-fanout (state A: 0 active) |
| `movementState` | `mgc.getMovementState()` (specificClass) — `'working'` while any child is ramping/sequencing or the executor has pending commands, else `'ready'` | string `'working'`\|`'ready'`, never null | movement-gate.basic (working: accelerating/warmingup/delayedMove/moveTimeLeft/executor-pending) | movement-gate.basic (ready: no machines, all settled) |
### Conditional pressure-header fields (emitted only when equalize resolved a positive ΔP)

83
test/basic/demand-telemetry.basic.test.js Normal file
View File

@@ -0,0 +1,83 @@
'use strict';
const test = require('node:test');
const assert = require('node:assert/strict');
const { getOutput } = require('../../src/io/output.js');
const MachineGroup = require('../../src/specificClass.js');
// Real declared unit policy so the m³/s → m³/h conversion is the production one.
const unitPolicy = MachineGroup.unitPolicy;
// Minimal MGC stand-in exposing exactly the surface getOutput reads. The
// measurement loop is short-circuited with an empty type list so the test
// isolates the demand telemetry without needing curves / CoolProp.
function mockMgc(overrides = {}) {
return {
measurements: { getTypes: () => [] },
unitPolicy,
mode: 'optimalControl',
scaling: 'absolute',
absDistFromPeak: 0,
relDistFromPeak: 0,
dynamicTotals: { flow: { min: 0.05, max: 0.25 } }, // m³/s
machines: {},
operatingPoint: {},
_lastDemand: null,
...overrides,
};
}
test('demandFlow + demandPct emitted once a demand is resolved', () => {
// Demand resolved to 0.15 m³/s inside a 0.05..0.25 envelope → midpoint = 50%.
const out = getOutput(mockMgc({ _lastDemand: { canonical: 0.15, clamped: 0.15 } }));
// m³/s → m³/h is ×3600. 0.15 m³/s = 540 m³/h.
assert.equal(out.demandFlow, 540);
assert.ok(Math.abs(out.demandPct - 50) < 1e-9, `expected ~50%, got ${out.demandPct}`);
});
test('demandPct reflects the clamped setpoint, not the raw request', () => {
// Operator asked for 0.40 m³/s but the envelope caps at 0.25 → 100%.
const out = getOutput(mockMgc({ _lastDemand: { canonical: 0.40, clamped: 0.25 } }));
assert.equal(out.demandFlow, 900); // 0.25 m³/s = 900 m³/h
assert.equal(out.demandPct, 100);
});
test('demandPct is 0 (never NaN) when the capacity span is zero', () => {
const out = getOutput(mockMgc({
dynamicTotals: { flow: { min: 0.1, max: 0.1 } },
_lastDemand: { canonical: 0.1, clamped: 0.1 },
}));
assert.equal(out.demandPct, 0);
assert.ok(Number.isFinite(out.demandFlow));
});
test('turnOff demand (0) emits a zero setpoint, not absent', () => {
const out = getOutput(mockMgc({ _lastDemand: { canonical: 0, clamped: 0 } }));
assert.equal(out.demandFlow, 0);
assert.equal(out.demandPct, 0);
});
test('demand telemetry is absent before the first demand (degraded state)', () => {
const out = getOutput(mockMgc({ _lastDemand: null }));
assert.ok(!('demandFlow' in out), 'demandFlow must be absent pre-first-demand');
assert.ok(!('demandPct' in out), 'demandPct must be absent pre-first-demand');
// The always-on capacity fields are still present, converted to the output
// flow unit (m³/h): 0.05 m³/s → 180, 0.25 m³/s → 900.
assert.equal(out.flowCapacityMin, 180);
assert.equal(out.flowCapacityMax, 900);
});
test('flow capacity is emitted in the output unit (m³/h), matching the flow series', () => {
const out = getOutput(mockMgc({ dynamicTotals: { flow: { min: 0.1, max: 0.3 } } }));
assert.equal(out.flowCapacityMin, 360); // 0.1 m³/s × 3600
assert.equal(out.flowCapacityMax, 1080); // 0.3 m³/s × 3600
});
test('flow capacity falls back to 0 when the envelope is unresolved (Infinity)', () => {
// Pre-first-equalize: dynamicTotals seeds min=Infinity, max=0.
const out = getOutput(mockMgc({ dynamicTotals: { flow: { min: Infinity, max: 0 } } }));
assert.equal(out.flowCapacityMin, 0);
assert.equal(out.flowCapacityMax, 0);
});

77
test/basic/movement-gate.basic.test.js Normal file
View File

@@ -0,0 +1,77 @@
// Unit tests for the MGC movement state + dispatch-gate helpers
// (getMovementState / _isUrgentDemand). Exercised via prototype.call with a
// minimal fake `this` so no Node-RED runtime or full MachineGroup boot is
// needed. See project rule .claude/rules/testing.md (basic = pure logic).
const test = require('node:test');
const assert = require('node:assert/strict');
const MachineGroup = require('../../src/specificClass');
function machine(state, { delayedMove = null, moveTimeLeft = 0 } = {}) {
return { state: { getCurrentState: () => state, delayedMove, getMoveTimeLeft: () => moveTimeLeft } };
}
function movementStateOf(machines, pending = 0) {
return MachineGroup.prototype.getMovementState.call({
machines,
movementExecutor: { pending: () => pending },
});
}
test('movementState: ready when no machines are registered', () => {
assert.equal(movementStateOf({}), 'ready');
});
test('movementState: ready when every machine is settled and nothing is pending', () => {
assert.equal(movementStateOf({ a: machine('operational'), b: machine('idle') }), 'ready');
});
test('movementState: working while a machine is mid-ramp', () => {
assert.equal(movementStateOf({ a: machine('operational'), b: machine('accelerating') }), 'working');
});
test('movementState: working during a start/stop sequence step', () => {
assert.equal(movementStateOf({ a: machine('warmingup') }), 'working');
});
test('movementState: working when a setpoint is queued (delayedMove)', () => {
assert.equal(movementStateOf({ a: machine('operational', { delayedMove: 50 }) }), 'working');
});
test('movementState: working while move time remains', () => {
assert.equal(movementStateOf({ a: machine('operational', { moveTimeLeft: 1.2 }) }), 'working');
});
test('movementState: working when the executor still has scheduled commands', () => {
assert.equal(movementStateOf({ a: machine('operational') }, 2), 'working');
});
function urgent(demandQ, {
mode = 'optimalControl', lastMode = 'optimalControl',
last = 10, priorityList = null, lastPriorityKey = 'null', span = 100, thr,
} = {}) {
return MachineGroup.prototype._isUrgentDemand.call({
_lastDemand: last == null ? null : { canonical: last },
mode, _lastDispatchedMode: lastMode, _lastPriorityKey: lastPriorityKey,
calcDynamicTotals: () => ({ flow: { max: span } }),
config: { planner: thr == null ? {} : { urgentDemandFraction: thr } },
}, demandQ, priorityList);
}
test('urgent: a stop (≤0) always pre-empts', () => {
assert.equal(urgent(0), true);
assert.equal(urgent(-5), true);
});
test('urgent: the first demand (no prior) dispatches immediately', () => {
assert.equal(urgent(50, { last: null }), true);
});
test('urgent: a control-mode switch is a new intent', () => {
assert.equal(urgent(10, { mode: 'priorityControl', lastMode: 'optimalControl' }), true);
});
test('urgent: a changed priority order is a new intent', () => {
assert.equal(urgent(10, { priorityList: ['eff', 'std'], lastPriorityKey: 'null' }), true);
});
test('urgent: a small same-mode nudge is held (not urgent)', () => {
assert.equal(urgent(12, { last: 10, span: 100 }), false); // 2% of span < 25%
});
test('urgent: a large same-mode step pre-empts', () => {
assert.equal(urgent(60, { last: 10, span: 100 }), true); // 50% of span ≥ 25%
});
test('urgent: threshold is configurable via planner.urgentDemandFraction', () => {
assert.equal(urgent(15, { last: 10, span: 100, thr: 0.02 }), true); // 5% ≥ 2%
assert.equal(urgent(15, { last: 10, span: 100, thr: 0.5 }), false); // 5% < 50%
});