// PumpingStation — S88 Process Cell orchestrator. // // Wires the basin / measurement / control / safety modules in configure() // and runs them in tick(). All real work lives in the modules; this file // only stitches them together. See wiki/functional-description.md for the // behaviour spec. const { BaseDomain, UnitPolicy, statusBadge } = require('generalFunctions'); const BasinGeometry = require('./basin/BasinGeometry'); const { validateThresholdOrdering } = require('./basin/thresholdValidator'); const FlowAggregator = require('./measurement/flowAggregator'); const MeasurementRouter = require('./measurement/measurementRouter'); const calibration = require('./measurement/calibration'); const control = require('./control'); const SafetyController = require('./safety/safetyController'); class PumpingStation extends BaseDomain { static name = 'pumpingStation'; // Internal math runs in m3/s for flow and m for level so the volume // integrator (flow × dt) is unit-consistent. Strict canonicals make // unit drift in child-fed measurements an explicit error. static unitPolicy = UnitPolicy.declare({ canonical: { flow: 'm3/s', pressure: 'Pa', power: 'W', temperature: 'K' }, output: { flow: 'm3/s', netFlowRate: 'm3/s', level: 'm', volume: 'm3' }, requireUnitForTypes: [], }); configure() { this.basin = new BasinGeometry(this.config.basin, this.config.hydraulics); this.flowVariants = ['measured', 'predicted']; this.levelVariants = ['measured', 'predicted']; this.volVariants = ['measured', 'predicted']; this.flowPositions = { inflow: ['in', 'upstream'], outflow: ['out', 'downstream'] }; this.mode = this.config.control.mode; this.controlState = { percControl: 0 }; this.state = { direction: 'steady', netFlow: 0, flowSource: null, seconds: null, remainingSource: null }; // FlowAggregator owns the predicted-volume integrator + net-flow + ETA. this.flowAggregator = new FlowAggregator({ measurements: this.measurements, basin: this.basin, config: this.config, logger: this.logger, flowVariants: this.flowVariants, levelVariants: this.levelVariants, flowPositions: this.flowPositions, }); this.measurementRouter = new MeasurementRouter({ measurements: this.measurements, basin: this.basin, logger: this.logger, }); // Threshold ordering is non-fatal — log + surface for tests/status. this.thresholdIssues = validateThresholdOrdering(this.basin, this.config.control?.levelbased, this.config.safety); for (const issue of this.thresholdIssues) this.logger.warn(issue.msg); // Seed predicted volume at the operational floor — without it the // integrator starts from null and the first tick has no anchor. this.measurements.type('volume').variant('predicted').position('atequipment') .value(this.basin.minVol, Date.now(), 'm3').unit('m3'); // Plain id-keyed maps. Tests assign into them directly (legacy contract); // ChildRouter onRegister handlers below also populate them. this.machines = {}; this.stations = {}; this.machineGroups = {}; this.predictedFlowChildren = new Map(); // SafetyController constructed after child maps so its captured ctx // references the live dicts rather than undefined. this.safety = new SafetyController(this.context()); this.router .onRegister('measurement', (child) => this._subscribeMeasurement(child)) .onRegister('machine', (child) => { this.machines[child.config.general.id] = child; // Skip individual machines when a machineGroup parent is present — // the group's flow.predicted already aggregates child machines. if (Object.keys(this.machineGroups).length === 0) { this._subscribePredictedFlow(child); } }) .onRegister('machinegroup', (child) => { this.machineGroups[child.config.general.id] = child; this._subscribePredictedFlow(child); }) .onRegister('pumpingstation', (child) => { this.stations[child.config.general.id] = child; this._subscribePredictedFlow(child); }); this.logger.debug('PumpingStation initialized'); } // Frozen view passed to control strategies + safety. context() { return Object.freeze({ ...super.context(), basin: this.basin, flowAggregator: this.flowAggregator, machines: this.machines, machineGroups: this.machineGroups, stations: this.stations, mode: this.mode, flowVariants: this.flowVariants, levelVariants: this.levelVariants, volVariants: this.volVariants, }); } tick() { const { netFlow, remaining } = this.flowAggregator.tick(); const safe = this.safety.evaluate({ direction: netFlow.direction, secondsRemaining: remaining.seconds }); this.safetyControllerActive = safe.blocked; if (!safe.blocked) { Promise.resolve(control.dispatch(this.mode, this.context(), this.controlState)) .catch((err) => this.logger.error(`control dispatch failed: ${err.message}`)); } this.state = { direction: netFlow.direction, netFlow: netFlow.value, flowSource: netFlow.source, seconds: remaining.seconds, remainingSource: remaining.source, }; this.notifyOutputChanged(); } changeMode(newMode) { if (this.config.control.allowedModes?.has?.(newMode)) { this.logger.info(`Control mode changing from ${this.mode} to ${newMode}`); this.mode = newMode; } else { this.logger.warn(`Attempted to change to unsupported control mode: ${newMode}`); } } // Calibration — public methods preserved for tests + commands registry. calibratePredictedVolume(vol, ts = Date.now()) { calibration.calibratePredictedVolume(this, vol, ts); } calibratePredictedLevel(lvl, ts = Date.now(), unit = 'm') { calibration.calibratePredictedLevel(this, lvl, ts, unit); } setManualInflow(value, ts = Date.now(), unit) { calibration.setManualInflow(this, value, ts, unit); } forwardDemandToChildren(demand) { return control.manual.forwardDemand(this.context(), demand); } // Direct delegations preserved so existing tests can drive the strategy // without re-mocking the dispatch layer. async _controlLevelBased() { return control.strategies.levelbased.run(this.context(), this.controlState); } // Public getter so legacy tests + getOutput keep reading the live demand. get percControl() { return this.controlState.percControl; } set percControl(v) { this.controlState.percControl = v; } getOutput() { const out = this.measurements.getFlattenedOutput(); Object.assign(out, this.basin.snapshot()); out.direction = this.state.direction; out.flowSource = this.state.flowSource; out.timeleft = this.state.seconds; out.percControl = this.controlState.percControl; return out; } getStatusBadge() { const STYLES = { filling: { arrow: '⬆️', fill: 'blue' }, draining: { arrow: '⬇️', fill: 'orange' }, steady: { arrow: '⏸️', fill: 'green' }, }; const { arrow = '❔', fill = 'grey' } = STYLES[this.state?.direction] || {}; const vol = this.measurements.type('volume').variant('predicted').position('atequipment').getCurrentValue('m3') ?? 0; const pct = this.measurements.type('volumePercent').variant('predicted').position('atequipment').getCurrentValue() ?? 0; const maxVol = this.basin?.maxVolAtOverflow ?? 0; const netFlowM3h = (this.state?.netFlow ?? 0) * 3600; const seconds = this.state?.seconds; const tStr = seconds != null ? `t≈${Math.round(seconds / 60)} min` : null; return statusBadge.compose( [`${arrow} ${pct.toFixed(1)}%`, `V=${vol.toFixed(2)} / ${maxVol.toFixed(2)} m³`, `net: ${netFlowM3h.toFixed(0)} m³/h`, tStr], { fill, shape: 'dot' } ); } // ── Direction helper kept for tests pinning the dead-band semantics ── _deriveDirection(netFlow) { return this.flowAggregator.deriveDirection(netFlow); } // ── Volume/level conversions kept for tests + back-compat ────────────── _calcVolumeFromLevel(level) { return this.basin.volumeFromLevel(level); } _calcLevelFromVolume(volume) { return this.basin.levelFromVolume(volume); } _subscribeMeasurement(child) { const position = child.config.functionality.positionVsParent; const measurementType = child.config.asset.type; const eventName = `${measurementType}.measured.${position}`; child.measurements.emitter.on(eventName, (eventData = {}) => { this.logger.debug( `Measurement update ${eventName} <- ${eventData.childName || child.config.general.name}: ${eventData.value} ${eventData.unit}` ); this.measurements.type(measurementType).variant('measured').position(position) .value(eventData.value, eventData.timestamp, eventData.unit); this.measurementRouter.route(measurementType, eventData.value, position, eventData); }); } _subscribePredictedFlow(child) { // Map the child's position to the orchestrator's posKey + the most // specific aggregator event. 'downstream' is preferred over 'atequipment' // because they carry the same total — subscribing to both double-counts. const POS_MAP = { downstream: ['out', 'flow.predicted.downstream'], out: ['out', 'flow.predicted.downstream'], atequipment:['out', 'flow.predicted.downstream'], upstream: ['in', 'flow.predicted.upstream'], in: ['in', 'flow.predicted.upstream'], }; const position = (child.config.functionality.positionVsParent || '').toLowerCase(); const mapped = POS_MAP[position]; if (!mapped) { this.logger.warn(`Unsupported predicted flow position "${position}" from ${child.config.general.name}`); return; } const [posKey, eventName] = mapped; const childId = child.config.general.id ?? child.config.general.name; if (!this.predictedFlowChildren.has(childId)) { this.predictedFlowChildren.set(childId, { in: 0, out: 0 }); } child.measurements.emitter.on(eventName, (eventData = {}) => { const unit = eventData.unit || child.config?.general?.unit; const ts = eventData.timestamp || Date.now(); this.measurements.type('flow').variant('predicted').position(posKey).child(childId) .value(eventData.value, ts, unit); }); } } module.exports = PumpingStation;