P6: convert monster to BaseDomain + BaseNodeAdapter + concern split

Refactor of monster to use the platform infrastructure (BaseDomain, BaseNodeAdapter,
ChildRouter, commandRegistry, statusBadge). Extracts concerns into
focused modules per .claude/refactor/MODULE_SPLIT.md generic template.
Tests stay green; CONTRACT.md generated; legacy aliases preserved.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CONTRACT.md

@@ -0,0 +1,49 @@
+# monster — Contract
+
+Hand-maintained for Phase 6; the `## Inputs` table is generated from
+`src/commands/index.js` (see Phase 9 generator). Keep ≤ 80 lines.
+
+## Inputs (msg.topic on Port 0)
+
+| Canonical | Aliases (deprecated) | Payload | Effect |
+|---|---|---|---|
+| `cmd.start` | `i_start` | truthy/falsy | Sets `source.i_start`. On the next tick a sampling run begins if flow bounds validate. |
+| `set.schedule` | `monsternametijden` | array of AQUON rows (`SAMPLE_NAME`, `DESCRIPTION`, `SAMPLED_DATE`, `START_DATE`, `END_DATE`) | Stores the schedule and recomputes `nextDate` + `daysPerYear` for the configured `aquonSampleName`. |
+| `set.rain` | `rain_data` | per-location rain forecast (Open-Meteo shape) | Aggregates hourly precipitation into `sumRain` / `avgRain`; feeds the rain-scaled flow prediction. |
+| `data.flow` | `input_q` | `{ value: number, unit: string }` | Converts to m³/h and pushes into `flow.manual.atequipment`. Blends with measured-child flow in `getEffectiveFlow()`. |
+| `set.mode` | `setMode` | string | Delegated to `source.setMode()` if defined. Reserved for future use. |
+| `set.model-prediction` | `model_prediction` | numeric | Delegated to `source.setModelPrediction()` if defined. Reserved for future use. |
+
+Aliases log a one-time deprecation warning the first time they fire.
+
+## Outputs (msg.topic on Port 0/1/2)
+
+- **Port 0 (process):** `msg.topic = config.general.name`. Payload built
+  by `outputUtils.formatMsg(..., 'process')` from `getOutput()`. Delta-
+  compressed — only changed fields are emitted. Carries `pulse`, `running`,
+  `bucketVol`, `sumPuls`, `predFlow`, `m3PerPuls`, `q`, `timeLeft`,
+  `targetVolumeM3`, `targetProgressPct`, `targetDeltaL`, `predictedRateM3h`,
+  `sumRain`, `avgRain`, `nextDate`, plus the flat measurements snapshot.
+- **Port 1 (InfluxDB telemetry):** same shape as Port 0, formatted with the
+  `'influxdb'` formatter.
+- **Port 2 (registration):** at startup the node sends one
+  `{ topic: 'child.register', payload: <node.id>, positionVsParent, distance }`
+  to its parent.
+
+## Events emitted by `source.measurements.emitter`
+
+The `MeasurementContainer` fires `<type>.measured.<position>` whenever a
+matching series receives a new value. monster writes:
+
+- `flow.manual.atequipment` — operator-supplied manual flow.
+- `flow.measured.<position>` — re-emitted when a child measurement fires
+  (one of `flow.measured.upstream`, `flow.measured.downstream`,
+  `flow.measured.atequipment`).
+
+## Children accepted
+
+`measurement` only. The router subscribes to a child's
+`flow.measured.<position>` events when the child's `config.asset.type` is
+`'flow'` (or missing). Other asset types are ignored. monster has no
+position-based filtering — all three positions are wired and the latest
+value wins for each.

src/commands/handlers.js

@@ -0,0 +1,42 @@
+'use strict';
+
+// Handlers for monster input topics. Each is a pure function over the
+// domain (source). Unit conversion for incoming flow happens in the
+// handler (the legacy nodeClass did it inline) — anything else inbound
+// is passed straight through to source.handleInput.
+
+const { convert } = require('generalFunctions');
+
+exports.cmdStart = (source, msg) => {
+  source.handleInput('i_start', Boolean(msg.payload));
+};
+
+exports.setSchedule = (source, msg) => {
+  source.handleInput('monsternametijden', msg.payload);
+};
+
+exports.setRain = (source, msg) => {
+  source.handleInput('rain_data', msg.payload);
+};
+
+exports.dataFlow = (source, msg, ctx) => {
+  const log = ctx?.logger || source.logger;
+  const value = Number(msg.payload?.value);
+  const unit = msg.payload?.unit;
+  if (!Number.isFinite(value) || !unit) {
+    log?.warn?.('data.flow payload must include numeric value and unit.');
+    return;
+  }
+  let converted = value;
+  try { converted = convert(value).from(unit).to('m3/h'); }
+  catch (err) { log?.warn?.(`data.flow unit conversion failed: ${err.message}`); return; }
+  source.handleInput('input_q', { value: converted, unit: 'm3/h' });
+};
+
+exports.setMode = (source, msg) => {
+  if (typeof source.setMode === 'function') source.setMode(msg.payload);
+};
+
+exports.setModelPrediction = (source, msg) => {
+  if (typeof source.setModelPrediction === 'function') source.setModelPrediction(msg.payload);
+};

src/commands/index.js

@@ -0,0 +1,47 @@
+'use strict';
+
+// monster command registry. Canonical names follow CONTRACTS.md §1.
+// Legacy names (i_start, monsternametijden, rain_data, input_q, setMode,
+// model_prediction) are surfaced as aliases — they log a one-time
+// deprecation warning on first use and are removed in Phase 7.
+
+const handlers = require('./handlers');
+
+module.exports = [
+  {
+    topic: 'cmd.start',
+    aliases: ['i_start'],
+    payloadSchema: { type: 'any' },
+    handler: handlers.cmdStart,
+  },
+  {
+    topic: 'set.schedule',
+    aliases: ['monsternametijden'],
+    payloadSchema: { type: 'any' },
+    handler: handlers.setSchedule,
+  },
+  {
+    topic: 'set.rain',
+    aliases: ['rain_data'],
+    payloadSchema: { type: 'any' },
+    handler: handlers.setRain,
+  },
+  {
+    topic: 'data.flow',
+    aliases: ['input_q'],
+    payloadSchema: { type: 'object' },
+    handler: handlers.dataFlow,
+  },
+  {
+    topic: 'set.mode',
+    aliases: ['setMode'],
+    payloadSchema: { type: 'any' },
+    handler: handlers.setMode,
+  },
+  {
+    topic: 'set.model-prediction',
+    aliases: ['model_prediction'],
+    payloadSchema: { type: 'any' },
+    handler: handlers.setModelPrediction,
+  },
+];

src/flow/flowTracker.js

@@ -0,0 +1,59 @@
+'use strict';
+
+// Flow tracking — manual override + measured-child fan-in + effective
+// blend (mean of the two when both present). Wraps the
+// MeasurementContainer so the domain stays read-only on flow state.
+
+const POSITIONS = ['upstream', 'downstream', 'atequipment'];
+
+class FlowTracker {
+  constructor({ measurements, logger }) {
+    this.measurements = measurements;
+    this.logger = logger;
+    this.manualFlow = null;
+  }
+
+  updateManualFlow(payload = {}) {
+    const value = Number(payload.value);
+    if (!Number.isFinite(value)) return;
+    const unit = payload.unit || 'm3/h';
+    this.manualFlow = value;
+    this.measurements.type('flow').variant('manual').position('atequipment')
+      .value(value, Date.now(), unit);
+  }
+
+  handleMeasuredFlow(eventData) {
+    const value = Number(eventData?.value);
+    if (!Number.isFinite(value)) return;
+    const position = String(eventData.position || 'atequipment').toLowerCase();
+    const unit = eventData.unit || 'm3/h';
+    this.measurements.type('flow').variant('measured').position(position)
+      .value(value, eventData.timestamp || Date.now(), unit);
+  }
+
+  getMeasuredFlow() {
+    const values = [];
+    for (const pos of POSITIONS) {
+      const v = this.measurements.type('flow').variant('measured').position(pos).getCurrentValue();
+      if (Number.isFinite(v)) values.push(v);
+    }
+    if (!values.length) return null;
+    return values.reduce((s, c) => s + c, 0) / values.length;
+  }
+
+  getManualFlow() {
+    const v = this.measurements.type('flow').variant('manual').position('atequipment').getCurrentValue();
+    return Number.isFinite(v) ? v : null;
+  }
+
+  getEffectiveFlow() {
+    const measured = this.getMeasuredFlow();
+    const manual = this.getManualFlow();
+    if (measured != null && manual != null) return (measured + manual) / 2;
+    if (measured != null) return measured;
+    if (manual != null) return manual;
+    return 0;
+  }
+}
+
+module.exports = FlowTracker;

src/io/output.js

@@ -0,0 +1,70 @@
+'use strict';
+
+// Output formatter — assembles the snapshot shape getOutput returns each
+// tick. Heavy on derived fields (timeToNextPulse, targetDelta, ...) but
+// every value is read-only on the domain, so this can stay a pure function.
+
+const params = require('../parameters/parameters');
+
+function buildOutput(m) {
+  const output = m.measurements.getFlattenedOutput();
+  const flowRate = Number(m.q) || 0;
+  const m3PerPulse = Number(m.m3PerPuls) || 0;
+  const pulseFraction = Number(m.temp_pulse) || 0;
+  const targetVolumeL = Number(m.targetVolume) > 0 ? m.targetVolume : 0;
+  const targetVolumeM3 = targetVolumeL > 0 ? targetVolumeL / 1000 : 0;
+  const flowToNextPulseM3 = m3PerPulse > 0 ? Math.max(0, (1 - pulseFraction) * m3PerPulse) : 0;
+  const timeToNextPulseSec = flowRate > 0 && flowToNextPulseM3 > 0
+    ? Math.round((flowToNextPulseM3 / (flowRate / 3600)) * 100) / 100
+    : 0;
+  const targetProgressPct = targetVolumeL > 0
+    ? Math.round((m.bucketVol / targetVolumeL) * 10000) / 100
+    : 0;
+  const targetDeltaL = targetVolumeL > 0
+    ? Math.round((m.bucketVol - targetVolumeL) * 100) / 100
+    : 0;
+  const targetDeltaM3 = targetVolumeL > 0
+    ? Math.round((targetDeltaL / 1000) * 10000) / 10000
+    : 0;
+
+  Object.assign(output, {
+    pulse: m.pulse,
+    running: m.running,
+    bucketVol: m.bucketVol,
+    bucketWeight: m.bucketWeight,
+    sumPuls: m.sumPuls,
+    predFlow: m.predFlow,
+    predM3PerSec: m.predM3PerSec,
+    timePassed: m.timePassed,
+    timeLeft: m.timeLeft,
+    m3Total: m.m3Total,
+    q: m.q,
+    nominalFlowMin: m.nominalFlowMin,
+    flowMax: m.flowMax,
+    invalidFlowBounds: m.invalidFlowBounds,
+    minSampleIntervalSec: m.minSampleIntervalSec,
+    missedSamples: m.missedSamples,
+    sampleCooldownMs: params.getSampleCooldownMs(m),
+    maxVolume: m.maxVolume,
+    minVolume: m.minVolume,
+    nextDate: m.nextDate,
+    daysPerYear: m.daysPerYear,
+    m3PerPuls: m.m3PerPuls,
+    m3PerPulse: m.m3PerPuls,
+    pulsesRemaining: Math.max(0, (m.targetPuls || 0) - (m.sumPuls || 0)),
+    pulseFraction,
+    flowToNextPulseM3,
+    timeToNextPulseSec,
+    targetVolumeM3,
+    targetProgressPct,
+    targetDeltaL,
+    targetDeltaM3,
+    predictedRateM3h: params.getPredictedFlowRate(m),
+    sumRain: m.rainAggregator?.sumRain ?? 0,
+    avgRain: m.rainAggregator?.avgRain ?? 0,
+  });
+
+  return output;
+}
+
+module.exports = { buildOutput };

src/io/statusBadge.js

@@ -0,0 +1,28 @@
+'use strict';
+
+// Status-badge composition. Three states the editor cares about:
+//   - red ring   : config error (flow bounds invalid)
+//   - yellow ring: sampling but cooldown is gating the next pulse
+//   - green dot  : sampling normally
+//   - grey ring  : idle
+// Shape mirrors the legacy nodeClass._updateNodeStatus output verbatim.
+
+const { statusBadge } = require('generalFunctions');
+const params = require('../parameters/parameters');
+
+function buildStatusBadge(m) {
+  if (m.invalidFlowBounds) {
+    return statusBadge.error(`Config error: nominalFlowMin (${m.nominalFlowMin}) >= flowMax (${m.flowMax})`);
+  }
+  if (m.running) {
+    const levelText = `${m.bucketVol}/${m.maxVolume} L`;
+    const cooldownMs = params.getSampleCooldownMs(m);
+    if (cooldownMs > 0) {
+      return statusBadge.compose([`SAMPLING (${Math.ceil(cooldownMs / 1000)}s)`, levelText], { fill: 'yellow', shape: 'ring' });
+    }
+    return statusBadge.compose([`AI: RUNNING`, levelText], { fill: 'green', shape: 'dot' });
+  }
+  return statusBadge.idle('AI: IDLE');
+}
+
+module.exports = { buildStatusBadge };

src/nodeClass.js

@@ -1,51 +1,19 @@
-/**
+'use strict';
- * node class.js
- *
- * Encapsulates all node logic in a reusable class. In future updates we can split this into multiple generic classes and use the config to specifiy which ones to use.
- * This allows us to keep the Node-RED node clean and focused on wiring up the UI and event handlers.
- */
-const { outputUtils, configManager, convert } = require('generalFunctions');
-const Specific = require('./specificClass');
 
-class nodeClass {
+const { BaseNodeAdapter } = require('generalFunctions');
-  /**
+const Monster = require('./specificClass');
-   * Create a Node.
+const commands = require('./commands');
-   * @param {object} uiConfig - Node-RED node configuration.
-   * @param {object} RED    - Node-RED runtime API.
-   * @param {object} nodeInstance - The Node-RED node instance.
-   * @param {string} nameOfNode - The name of the node.
-   */
-  constructor(uiConfig, RED, nodeInstance, nameOfNode) {
-    this.node = nodeInstance;
-    this.RED = RED;
-    this.name = nameOfNode;
-    this.source = null;
-    this.config = null;
 
-    // Load default & UI config
+class nodeClass extends BaseNodeAdapter {
-    this._loadConfig(uiConfig);
+  static DomainClass = Monster;
+  static commands = commands;
+  // Tick-driven: sampling integrator (m3PerTick → temp_pulse) needs
+  // wall-clock delta-time once per second.
+  static tickInterval = 1000;
+  static statusInterval = 1000;
 
-    // Instantiate core class
+  buildDomainConfig(uiConfig) {
-    this._setupSpecificClass(uiConfig);
+    return {
-
-    // Wire up event and lifecycle handlers
-    this._bindEvents();
-    this._registerChild();
-    this._startTickLoop();
-    this._attachInputHandler();
-    this._attachCloseHandler();
-  }
-
-  /**
-   * Load and merge default config with user-defined settings.
-   * Uses ConfigManager.buildConfig() for base sections, then adds monster-specific domain config.
-   * @param {object} uiConfig - Raw config from Node-RED UI.
-   */
-  _loadConfig(uiConfig) {
-    const cfgMgr = new configManager();
-
-    // Build config: base sections + monster-specific domain config
-    this.config = cfgMgr.buildConfig(this.name, uiConfig, this.node.id, {
       constraints: {
         samplingtime: Number(uiConfig.samplingtime) || 0,
         minVolume: Number(uiConfig.minvolume ?? uiConfig.minVolume) || 5,
@@ -55,186 +23,14 @@ class nodeClass {
         maxRainRef: Number(uiConfig.maxRainRef) || 10,
         minSampleIntervalSec: Number(uiConfig.minSampleIntervalSec) || 60,
       },
-    });
+      functionality: { role: 'samplingCabinet', aquonSampleName: uiConfig.aquon_sample_name || undefined },
-
+      asset: { emptyWeightBucket: Number(uiConfig.emptyWeightBucket) || 3 },
-    this.config.functionality = {
-      ...this.config.functionality,
-      role: 'samplingCabinet',
-      aquonSampleName: uiConfig.aquon_sample_name || undefined,
     };
-
-    this.config.asset = {
-      ...this.config.asset,
-      emptyWeightBucket: Number(uiConfig.emptyWeightBucket) || 3,
-    };
-
-    // Utility for formatting outputs
-    this._output = new outputUtils();
   }
 
-  /**
+  extraSetup() {
-   * Instantiate the core logic and store as source.
+    const uiSampleName = this.config?.functionality?.aquonSampleName;
-   */
+    if (uiSampleName) this.source.aquonSampleName = uiSampleName;
-  _setupSpecificClass(uiConfig) {
-    this.source = new Specific(this.config);
-
-    if (uiConfig?.aquon_sample_name) {
-      this.source.aquonSampleName = uiConfig.aquon_sample_name;
-    }
-
-    this.node.source = this.source;
-  }
-
-  /**
-   * Bind events to Node-RED status updates.
-   */
-  _bindEvents() {}
-
-  _updateNodeStatus() {
-    const m = this.source;
-    try {
-      const bucketVol = m.bucketVol;
-      const maxVolume = m.maxVolume;
-      const state = m.running;
-      const mode = 'AI';
-      const flowMin = m.nominalFlowMin;
-      const flowMax = m.flowMax;
-
-      if (m.invalidFlowBounds) {
-        return {
-          fill: 'red',
-          shape: 'ring',
-          text: `Config error: nominalFlowMin (${flowMin}) >= flowMax (${flowMax})`,
-        };
-      }
-
-      if (state) {
-        const levelText = `${bucketVol}/${maxVolume} L`;
-        const cooldownMs = typeof m.getSampleCooldownMs === 'function'
-          ? m.getSampleCooldownMs()
-          : 0;
-
-        if (cooldownMs > 0) {
-          const cooldownSec = Math.ceil(cooldownMs / 1000);
-          return { fill: 'yellow', shape: 'ring', text: `SAMPLING (${cooldownSec}s) ${levelText}` };
-        }
-
-        return { fill: 'green', shape: 'dot', text: `${mode}: RUNNING ${levelText}` };
-      }
-
-      return { fill: 'grey', shape: 'ring', text: `${mode}: IDLE` };
-    } catch (error) {
-      this.node.error(`Error in updateNodeStatus: ${error.message}`);
-      return { fill: 'red', shape: 'ring', text: 'Status Error' };
-    }
-  }
-
-  /**
-   * Register this node as a child upstream and downstream.
-   * Delayed to avoid Node-RED startup race conditions.
-   */
-  _registerChild() {
-    setTimeout(() => {
-      this.node.send([
-        null,
-        null,
-        { topic: 'registerChild', payload: this.config.general.id, positionVsParent: this.config?.functionality?.positionVsParent || 'atEquipment' },
-      ]);
-    }, 100);
-  }
-
-  /**
-   * Start the periodic tick loop.
-   */
-  _startTickLoop() {
-    setTimeout(() => {
-      this._tickInterval = setInterval(() => this._tick(), 1000);
-      this._statusInterval = setInterval(() => {
-        this.node.status(this._updateNodeStatus());
-      }, 1000);
-    }, 1000);
-  }
-
-  /**
-   * Execute a single tick: update measurement, format and send outputs.
-   */
-  _tick() {
-    this.source.tick();
-
-    const raw = this.source.getOutput();
-    const processMsg = this._output.formatMsg(raw, this.source.config, 'process');
-    const influxMsg = this._output.formatMsg(raw, this.source.config, 'influxdb');
-
-    this.node.send([processMsg, influxMsg]);
-  }
-
-  /**
-   * Attach the node's input handler, routing control messages to the class.
-   */
-  _attachInputHandler() {
-    this.node.on('input', (msg, send, done) => {
-      const m = this.source;
-      try {
-        switch (msg.topic) {
-          case 'input_q': {
-            const value = Number(msg.payload?.value);
-            const unit = msg.payload?.unit;
-            if (!Number.isFinite(value) || !unit) {
-              this.node.warn('input_q payload must include numeric value and unit.');
-              break;
-            }
-            let converted = value;
-            try {
-              converted = convert(value).from(unit).to('m3/h');
-            } catch (error) {
-              this.node.warn(`input_q unit conversion failed: ${error.message}`);
-              break;
-            }
-            m.handleInput('input_q', { value: converted, unit: 'm3/h' });
-            break;
-          }
-          case 'i_start':
-          case 'monsternametijden':
-          case 'rain_data':
-            m.handleInput(msg.topic, msg.payload);
-            break;
-          case 'registerChild': {
-            const childId = msg.payload;
-            const childObj = this.RED.nodes.getNode(childId);
-            if (childObj?.source) {
-              m.childRegistrationUtils.registerChild(childObj.source, msg.positionVsParent);
-            }
-            break;
-          }
-          case 'setMode':
-            m.setMode(msg.payload);
-            break;
-          case 'model_prediction':
-            if (typeof m.setModelPrediction === 'function') {
-              m.setModelPrediction(msg.payload);
-            }
-            break;
-          default:
-            m.logger?.warn(`Unknown topic: ${msg.topic}`);
-            break;
-        }
-      } catch (error) {
-        this.node.error(`Error handling input (${msg?.topic}): ${error?.message || error}`);
-      } finally {
-        if (typeof done === 'function') done();
-      }
-    });
-  }
-
-  /**
-   * Clean up timers and intervals when Node-RED stops the node.
-   */
-  _attachCloseHandler() {
-    this.node.on('close', (done) => {
-      clearInterval(this._tickInterval);
-      clearInterval(this._statusInterval);
-      if (typeof done === 'function') done();
-    });
   }
 }
 

src/parameters/parameters.js

@@ -0,0 +1,56 @@
+'use strict';
+
+// Sampling-cabinet boundary + target math + rain-scaled flow prediction.
+// All operations are pure given a domain handle — the domain owns the
+// mutable fields (maxVolume, targetPuls, …) so legacy tests that read
+// `monster.maxVolume` keep working.
+
+const RAIN_STALE_MS = 2 * 60 * 60 * 1000;
+
+function applyBoundsAndTargets(m) {
+  m.maxVolume = m.maxWeight - m.emptyWeightBucket;
+  m.minPuls = Math.round(m.minVolume / m.volume_pulse);
+  m.maxPuls = Math.round(m.maxVolume / m.volume_pulse);
+  m.absMaxPuls = Math.round(m.cap_volume / m.volume_pulse);
+  m.targetVolume = m.minVolume * Math.sqrt(m.maxVolume / m.minVolume);
+  m.targetPuls = Math.round(m.targetVolume / m.volume_pulse);
+}
+
+function validateFlowBounds(m) {
+  const min = Number(m.nominalFlowMin);
+  const max = Number(m.flowMax);
+  const valid = Number.isFinite(min) && Number.isFinite(max) && min >= 0 && max > 0 && min < max;
+  m.invalidFlowBounds = !valid;
+  if (!valid) m.logger.warn(`Invalid flow bounds. nominalFlowMin=${m.nominalFlowMin}, flowMax=${m.flowMax}`);
+  return valid;
+}
+
+function getRainIndex(m) {
+  if (!m.lastRainUpdate) return 0;
+  if (Date.now() - m.lastRainUpdate > RAIN_STALE_MS) return 0;
+  return Number.isFinite(m.avgRain) ? m.avgRain : 0;
+}
+
+function getPredictedFlowRate(m) {
+  const min = Number(m.nominalFlowMin);
+  const max = Number(m.flowMax);
+  if (!Number.isFinite(min) || !Number.isFinite(max) || min < 0 || max <= 0 || min >= max) return 0;
+  const rainIndex = getRainIndex(m);
+  const scale = Math.max(0, Math.min(1, m.rainMaxRef > 0 ? rainIndex / m.rainMaxRef : 0));
+  return min + (max - min) * scale;
+}
+
+function getSampleCooldownMs(m) {
+  if (!m.lastSampleTime) return 0;
+  const remaining = (m.minSampleIntervalSec * 1000) - (Date.now() - m.lastSampleTime);
+  return Math.max(0, remaining);
+}
+
+module.exports = {
+  applyBoundsAndTargets,
+  validateFlowBounds,
+  getRainIndex,
+  getPredictedFlowRate,
+  getSampleCooldownMs,
+  RAIN_STALE_MS,
+};

src/rain/rainAggregator.js

@@ -0,0 +1,58 @@
+'use strict';
+
+// Rain-data aggregator — sums per-location hourly precipitation, weighted
+// by per-hour probability, and stores both the raw and probability-weighted
+// values keyed by timestamp. sumRain/avgRain feed parameters.getRainIndex
+// which scales the predicted flow rate between nominalFlowMin and flowMax.
+
+class RainAggregator {
+  constructor({ logger } = {}) {
+    this.logger = logger;
+    this.aggregatedOutput = {};
+    this.sumRain = 0;
+    this.avgRain = 0;
+  }
+
+  // Returns the aggregated per-location object so callers can chain.
+  // Mutates this.aggregatedOutput / sumRain / avgRain in place.
+  update(value) {
+    if (!value) return this.aggregatedOutput;
+
+    const totalRaw = {};
+    const totalProb = {};
+    let numberOfLocations = 0;
+
+    Object.entries(value).forEach(([locationKey, location]) => {
+      numberOfLocations++;
+      const slot = (this.aggregatedOutput[locationKey] = {
+        tag: { latitude: location.latitude, longitude: location.longitude },
+        precipationRaw: {},
+        precipationProb: {},
+      });
+
+      Object.entries(location.hourly.time).forEach(([key, time]) => {
+        const currTimestamp = new Date(time).getTime();
+        let probability = 100;
+        if (typeof location.hourly.precipitation_probability !== 'undefined') {
+          probability = location.hourly.precipitation_probability[key];
+        }
+        if (probability > 0) probability /= 100;
+
+        if (totalRaw[currTimestamp] === undefined) totalRaw[currTimestamp] = 0;
+        if (totalProb[currTimestamp] === undefined) totalProb[currTimestamp] = 0;
+
+        totalRaw[currTimestamp] += location.hourly.precipitation[key];
+        totalProb[currTimestamp] += location.hourly.precipitation[key] * probability;
+
+        slot.precipationRaw[key] = { val: location.hourly.precipitation[key], time: currTimestamp };
+        slot.precipationProb[key] = { val: probability, time: currTimestamp };
+      });
+    });
+
+    this.sumRain = Object.values(totalProb).reduce((s, v) => s + v, 0);
+    this.avgRain = numberOfLocations > 0 ? this.sumRain / numberOfLocations : 0;
+    return this.aggregatedOutput;
+  }
+}
+
+module.exports = RainAggregator;

src/sampling/samplingProgram.js

@@ -0,0 +1,113 @@
+'use strict';
+
+// Sampling program — the time-driven core. Each tick:
+//   1. on i_start or scheduled date → init a sampling run (m3PerPuls, stop_time)
+//   2. while running: integrate m3PerTick into temp_pulse; emit a pulse when
+//      it crosses 1 unless the cooldown guard blocks it
+//   3. after stop_time: clear running state.
+// flowCalc derives m3PerTick from the latest q (m3/h) and the wall-clock
+// delta since the last call — runs once per tick before sampling_program.
+
+const params = require('../parameters/parameters');
+const { regNextDate } = require('../schedule/schedule');
+
+function getModelPrediction(m) {
+  const samplingHours = Number(m.sampling_time) || 0;
+  const predictedRate = params.getPredictedFlowRate(m);
+  const fallbackRate = m.flowTracker.getEffectiveFlow();
+  const flowM3PerHour = predictedRate > 0 ? predictedRate : fallbackRate;
+  m.predFlow = Math.max(0, flowM3PerHour * samplingHours);
+  return m.predFlow;
+}
+
+function flowCalc(m) {
+  const timePassed = m.flowTime > 0 ? (Date.now() - m.flowTime) / 1000 : 0;
+  m.m3PerTick = (m.q / 60 / 60) * timePassed;
+  m.flowTime = Date.now();
+}
+
+function _beginRun(m) {
+  m.running = true;
+  m.temp_pulse = 0;
+  m.pulse = false;
+  m.updateBucketVol(0);
+  m.sumPuls = 0;
+  m.m3Total = 0;
+  m.timePassed = 0;
+  m.timeLeft = 0;
+  m.predM3PerSec = 0;
+
+  getModelPrediction(m);
+  m.m3PerPuls = Math.round(m.predFlow / m.targetPuls);
+  m.predM3PerSec = m.predFlow / m.sampling_time / 60 / 60;
+  m.start_time = Date.now();
+  m.stop_time = Date.now() + (m.sampling_time * 60 * 60 * 1000);
+
+  regNextDate(m, m.monsternametijden);
+  m.i_start = false;
+}
+
+function _endRun(m) {
+  m.m3PerPuls = 0;
+  m.temp_pulse = 0;
+  m.pulse = false;
+  m.updateBucketVol(0);
+  m.sumPuls = 0;
+  m.timePassed = 0;
+  m.timeLeft = 0;
+  m.predFlow = 0;
+  m.predM3PerSec = 0;
+  m.m3Total = 0;
+  m.running = false;
+}
+
+function _maybeEmitPulse(m) {
+  if (!(m.temp_pulse >= 1 && m.sumPuls < m.absMaxPuls)) {
+    if (m.pulse) m.pulse = false;
+    return;
+  }
+
+  const now = Date.now();
+  const cooldownMs = m.minSampleIntervalSec * 1000;
+  const blocked = m.lastSampleTime && (now - m.lastSampleTime) < cooldownMs;
+
+  if (blocked) {
+    m.missedSamples++;
+    m.pulse = false;
+    m.temp_pulse = Math.min(m.temp_pulse, 1);
+    if (!m.lastSampleWarnTime || (now - m.lastSampleWarnTime) > cooldownMs) {
+      m.lastSampleWarnTime = now;
+      m.logger.warn(`Sampling too fast. Cooldown active for ${Math.ceil((cooldownMs - (now - m.lastSampleTime)) / 1000)}s.`);
+    }
+    return;
+  }
+
+  m.temp_pulse -= 1;
+  m.pulse = true;
+  m.lastSampleTime = now;
+  m.sumPuls++;
+  m.updateBucketVol(Math.round(m.sumPuls * m.volume_pulse * 100) / 100);
+}
+
+function samplingProgram(m) {
+  if (((m.i_start) || (Date.now() >= m.nextDate)) && !m.running) {
+    if (!params.validateFlowBounds(m)) {
+      m.running = false;
+      m.i_start = false;
+      return;
+    }
+    _beginRun(m);
+  }
+
+  if (m.stop_time > Date.now()) {
+    m.timePassed = Math.round((Date.now() - m.start_time) / 1000);
+    m.timeLeft = Math.round((m.stop_time - Date.now()) / 1000);
+    m.temp_pulse += m.m3PerTick / m.m3PerPuls;
+    m.m3Total += m.m3PerTick;
+    _maybeEmitPulse(m);
+  } else if (m.running) {
+    _endRun(m);
+  }
+}
+
+module.exports = { samplingProgram, flowCalc, getModelPrediction };

src/schedule/schedule.js

@@ -0,0 +1,42 @@
+'use strict';
+
+// AQUON sample schedule helpers. updateMonsternametijden validates the
+// row shape before storing; regNextDate walks the rows to find the next
+// future START_DATE for the configured aquonSampleName and counts how
+// many of those fall in the current calendar year.
+
+function updateMonsternametijden(m, value) {
+  if (!m.init || !value || Object.keys(value).length === 0) return;
+  if (
+    typeof value[0]?.SAMPLE_NAME !== 'undefined' &&
+    typeof value[0]?.DESCRIPTION !== 'undefined' &&
+    typeof value[0]?.SAMPLED_DATE !== 'undefined' &&
+    typeof value[0]?.START_DATE !== 'undefined' &&
+    typeof value[0]?.END_DATE !== 'undefined'
+  ) {
+    m.monsternametijden = value;
+    regNextDate(m, value);
+  }
+}
+
+function regNextDate(m, monsternametijden) {
+  let next_date = new Date(new Date().setFullYear(new Date().getFullYear() + 1));
+  let n_days_remaining = 0;
+
+  if (typeof monsternametijden !== 'undefined') {
+    Object.values(monsternametijden).forEach((line) => {
+      if (line.START_DATE === 'NULL') return;
+      const curr_date_conv = new Date(line.START_DATE);
+      const curr_date = curr_date_conv.getTime();
+      if (line.SAMPLE_NAME === m.aquonSampleName && curr_date > Date.now()) {
+        if (curr_date < next_date) next_date = curr_date;
+        if (new Date().getFullYear() === curr_date_conv.getFullYear()) n_days_remaining++;
+      }
+    });
+  }
+
+  m.daysPerYear = n_days_remaining;
+  m.nextDate = next_date;
+}
+
+module.exports = { updateMonsternametijden, regNextDate };