docs: add CLAUDE.md with S88 classification and superproject rule reference

References the flow-layout rule set in the EVOLV superproject
(.claude/rules/node-red-flow-layout.md) so Claude Code sessions working
in this repo know the S88 level, colour, and placement lane for this node.

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

CLAUDE.md

@@ -0,0 +1,23 @@
+# valveGroupControl — Claude Code context
+
+Coordinates multiple valve children.
+Part of the [EVOLV](https://gitea.wbd-rd.nl/RnD/EVOLV) wastewater-automation platform.
+
+## S88 classification
+
+| Level | Colour | Placement lane |
+|---|---|---|
+| **Unit** | `#50a8d9` | L4 |
+
+## Flow layout rules
+
+When wiring this node into a multi-node demo or production flow, follow the
+placement rule set in the **EVOLV superproject**:
+
+> `.claude/rules/node-red-flow-layout.md` (in the EVOLV repo root)
+
+Key points for this node:
+- Place on lane **L4** (x-position per the lane table in the rule).
+- Stack same-level siblings vertically.
+- Parent/children sit on adjacent lanes (children one lane left, parent one lane right).
+- Wrap in a Node-RED group box coloured `#50a8d9` (Unit).

src/nodeClass.js

@@ -1,4 +1,4 @@
-const { outputUtils, configManager } = require("generalFunctions");
+const { outputUtils, configManager, convert } = require("generalFunctions");
 const Specific = require("./specificClass");
 
 class nodeClass {
@@ -18,6 +18,7 @@ class nodeClass {
 
     // Load default & UI config
     this._loadConfig(uiConfig, this.node);
+    this._reconcileIntervalMs = this._resolveReconcileIntervalMs(uiConfig);
 
     // Instantiate core Measurement class
     this._setupSpecificClass();
@@ -38,48 +39,55 @@ class nodeClass {
     const cfgMgr = new configManager();
     this.defaultConfig = cfgMgr.getConfig(this.name);
 
-    // Merge UI config over defaults
-    this.config = {
-      general: {
-        name: uiConfig.name,
-        id: node.id, // node.id is for the child registration process
-        unit: uiConfig.unit, // add converter options later to convert to default units (need like a model that defines this which units we are going to use and then conver to those standards)
-        logging: {
-          enabled: uiConfig.enableLog,
-          logLevel: uiConfig.logLevel,
-        },
-      },
-      functionality: {
-        positionVsParent: uiConfig.positionVsParent || "atEquipment", // Default to 'atEquipment' if not set
-      },
-    };
+    // Resolve flow unit with validation before building config
+    const flowUnit = this._resolveUnitOrFallback(uiConfig.unit, 'volumeFlowRate', 'm3/h', 'flow');
+    const resolvedUiConfig = { ...uiConfig, unit: flowUnit };
+
+    // Build config: base sections (no domain-specific config for group controller)
+    this.config = cfgMgr.buildConfig(this.name, resolvedUiConfig, node.id);
+
     // Utility for formatting outputs
     this._output = new outputUtils();
   }
 
+  _resolveUnitOrFallback(candidate, expectedMeasure, fallbackUnit, label) {
+    const raw = typeof candidate === "string" ? candidate.trim() : "";
+    const fallback = String(fallbackUnit || "").trim();
+    if (!raw) {
+      return fallback;
+    }
+    try {
+      const desc = convert().describe(raw);
+      if (expectedMeasure && desc.measure !== expectedMeasure) {
+        throw new Error(`expected '${expectedMeasure}' but got '${desc.measure}'`);
+      }
+      return raw;
+    } catch (error) {
+      this.node?.warn?.(`Invalid ${label} unit '${raw}' (${error.message}). Falling back to '${fallback}'.`);
+      return fallback;
+    }
+  }
+
+  _resolveReconcileIntervalMs(uiConfig) {
+    const raw = Number(
+      uiConfig?.reconcileIntervalSeconds
+      ?? uiConfig?.reconcileIntervalSec
+      ?? uiConfig?.reconcileEverySeconds
+      ?? 1
+    );
+    const sec = Number.isFinite(raw) && raw > 0 ? raw : 1;
+    return Math.max(100, Math.round(sec * 1000));
+  }
+
   _updateNodeStatus() {
     const vg = this.source;
-    const mode = vg.mode;
-    const scaling = vg.scaling;
-    const totalFlow =
-      Math.round(
-        vg.measurements
-          .type("flow")
-          .variant("measured")
-          .position("downstream")
-          .getCurrentValue() * 1
-      ) / 1;
-
-    // Calculate total capacity based on available valves
-    const availableValves = Object.values(vg.valves).filter((valve) => {
-      const state = valve.state.getCurrentState();
-      const mode = valve.currentMode;
-      return !(
-        state === "off" ||
-        state === "maintenance" ||
-        mode === "maintenance"
-      );
-    });
+    const mode = vg.currentMode;
+    const flowUnit = vg?.unitPolicy?.output?.flow || this.config.general.unit || "m3/h";
+    const measuredFlow = vg.measurements.type("flow").variant("measured").position("atEquipment").getCurrentValue(flowUnit);
+    const predictedFlow = vg.measurements.type("flow").variant("predicted").position("atEquipment").getCurrentValue(flowUnit);
+    const totalFlowRaw = Number.isFinite(measuredFlow) ? measuredFlow : predictedFlow;
+    const totalFlow = Number.isFinite(totalFlowRaw) ? Math.round(totalFlowRaw) : 0;
+    const availableValves = Array.isArray(vg.getAvailableValves?.()) ? vg.getAvailableValves() : [];
 
     // const totalCapacity = Math.round(vg.dynamicTotals.flow.max * 1) / 1; ADD LATER?
 
@@ -91,7 +99,7 @@ class nodeClass {
 
 
     // Generate status text in a single line
-    const text = ` ${mode} | 💨=${totalFlow} | ${status}`;
+    const text = `${mode} | flow=${totalFlow} ${flowUnit} | ${status}`;
 
     return {
       fill: availableValves.length > 0 ? "green" : "red",
@@ -139,7 +147,7 @@ class nodeClass {
    */
   _startTickLoop() {
     setTimeout(() => {
-      this._tickInterval = setInterval(() => this._tick(), 1000);
+      this._tickInterval = setInterval(() => this._tick(), this._reconcileIntervalMs);
        // Update node status on nodered screen every second ( this is not the best way to do this, but it works for now)
       this._statusInterval = setInterval(() => {
         const status = this._updateNodeStatus();
@@ -152,6 +160,9 @@ class nodeClass {
    * Execute a single tick: update measurement, format and send outputs.
    */
   _tick() {
+    if (typeof this.source?.calcValveFlows === 'function') {
+      this.source.calcValveFlows();
+    }
     const raw = this.source.getOutput();
     const processMsg = this._output.formatMsg(raw, this.config, "process");
     const influxMsg = this._output.formatMsg(raw, this.config, "influxdb");
@@ -184,14 +195,39 @@ class nodeClass {
             case 'setMode':
               vg.setMode(msg.payload);
               break;
+            case 'setReconcileInterval': {
+              const nextSec = Number(msg.payload);
+              if (!Number.isFinite(nextSec) || nextSec <= 0) {
+                vg.logger.warn(`Invalid reconcile interval payload '${msg.payload}'. Expected seconds > 0.`);
+                break;
+              }
+              this._reconcileIntervalMs = Math.max(100, Math.round(nextSec * 1000));
+              clearInterval(this._tickInterval);
+              this._tickInterval = setInterval(() => this._tick(), this._reconcileIntervalMs);
+              vg.logger.info(`Flow reconciliation interval updated to ${nextSec}s (${this._reconcileIntervalMs}ms).`);
+              break;
+            }
             case 'execSequence': {
               const { source: seqSource, action: seqAction, parameter } = msg.payload;
               vg.handleInput(seqSource, seqAction, parameter);
               break;
             }
             case 'totalFlowChange': {
-              const { source: tfcSource, action: tfcAction, q} = msg.payload;
-              vg.handleInput(tfcSource, tfcAction, Number(q));
+              const payload = msg.payload || {};
+              if (payload && typeof payload === "object" && Object.prototype.hasOwnProperty.call(payload, "source")) {
+                const tfcSource = payload.source || "parent";
+                const tfcAction = payload.action || "totalFlowChange";
+                vg.handleInput(tfcSource, tfcAction, payload);
+              } else {
+                vg.handleInput("parent", "totalFlowChange", payload);
+              }
+              break;
+            }
+            case 'emergencystop':
+            case 'emergencyStop': {
+              const payload = msg.payload || {};
+              const esSource = payload.source || "parent";
+              vg.handleInput(esSource, "emergencystop");
               break;
             }
             default:
@@ -213,6 +249,7 @@ class nodeClass {
     this.node.on("close", (done) => {
       clearInterval(this._tickInterval);
       clearInterval(this._statusInterval);
+      this.source?.destroy?.();
       if (typeof done === 'function') done();
     });
   }

src/specificClass.js

@@ -1,111 +1,484 @@
 /**
  * @file valveGroupControl.js
- *
- * Permission is hereby granted to any person obtaining a copy of this software 
- * and associated documentation files (the "Software"), to use it for personal 
- * or non-commercial purposes, with the following restrictions:
- *
- * 1. **No Copying or Redistribution**: The Software or any of its parts may not 
- *    be copied, merged, distributed, sublicensed, or sold without explicit 
- *    prior written permission from the author.
- * 
- * 2. **Commercial Use**: Any use of the Software for commercial purposes requires 
- *    a valid license, obtainable only with the explicit consent of the author.
- * 
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 
- * FITNESS FOR A PARTICULAR PURPOSE, AND NONINFRINGEMENT. IN NO EVENT SHALL THE 
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES, OR OTHER 
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT, OR OTHERWISE, ARISING FROM, 
- * OUT OF, OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 
- * SOFTWARE.
- *
- * Ownership of this code remains solely with the original author. Unauthorized 
- * use of this Software is strictly prohibited.
- *
- * Author:
- * - Rene De Ren
- * Email:
- * - r.de.ren@brabantsedelta.nl
- *
- * Future Improvements:
- * - Time-based stability checks
- * - Warmup handling
- * - Dynamic outlier detection thresholds
- * - Dynamic smoothing window and methods
- * - Alarm and threshold handling
- * - Maintenance mode
- * - Historical data and trend analysis
  */
-/**
- * @file valveGroupControl.js
- *
- * Permission is hereby granted to any person obtaining a copy of this software 
- * and associated documentation files (the "Software"), to use it for personal 
-....
-*/
 
-//load local dependencies
 const EventEmitter = require('events');
-const {loadCurve,logger,configUtils,configManager,state, nrmse, MeasurementContainer, predict, interpolation , childRegistrationUtils} = require('generalFunctions');
+const { logger, configUtils, configManager, state, MeasurementContainer, childRegistrationUtils, convert } = require('generalFunctions');
+
+const CANONICAL_UNITS = Object.freeze({
+  pressure: 'Pa',
+  flow: 'm3/s',
+});
+
+const DEFAULT_IO_UNITS = Object.freeze({
+  pressure: 'mbar',
+  flow: 'm3/h',
+});
+
+const KNOWN_POSITIONS = new Set(['upstream', 'downstream', 'atEquipment']);
+const SERVICE_TYPES = new Set(['gas', 'liquid']);
+const SOURCE_SOFTWARE_TYPES = new Set([
+  'machine',
+  'rotatingmachine',
+  'machinegroup',
+  'machinegroupcontrol',
+  'pumpingstation',
+  'valvegroupcontrol',
+]);
+const SOURCE_FLOW_EVENTS = [
+  'flow.predicted.downstream',
+  'flow.predicted.atEquipment',
+  'flow.predicted.atequipment',
+  'flow.measured.downstream',
+  'flow.measured.atEquipment',
+  'flow.measured.atequipment',
+];
+const DEFAULT_SOURCE_SERVICE_TYPE = Object.freeze({
+  machine: 'liquid',
+  rotatingmachine: 'liquid',
+  machinegroup: 'liquid',
+  machinegroupcontrol: 'liquid',
+  pumpingstation: 'liquid',
+});
+const DEFAULT_FLOW_RECONCILIATION = Object.freeze({
+  maxPasses: 2,
+  residualTolerance: 0.001,
+});
 
 class ValveGroupControl {
   constructor(valveGroupControlConfig = {}) {
-        this.emitter = new EventEmitter(); // nodig voor ontvangen en uitvoeren van events emit() en on() --> Zien als internet berichten (niet bedraad in node-red)
+    this.emitter = new EventEmitter();
     this.configManager = new configManager();
-        this.defaultConfig = this.configManager.getConfig('valveGroupControl'); // Load default config for rotating machine ( use software type name ? )
+    this.defaultConfig = this.configManager.getConfig('valveGroupControl');
     this.configUtils = new configUtils(this.defaultConfig);
-        this.config = this.configUtils.initConfig(valveGroupControlConfig); // verify and set the config for the valve group
+    this.config = this.configUtils.initConfig(valveGroupControlConfig);
+    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();
+    this.measurements = new MeasurementContainer({
+      autoConvert: true,
+      defaultUnits: {
+        pressure: this.unitPolicy.output.pressure,
+        flow: this.unitPolicy.output.flow,
+      },
+      preferredUnits: {
+        pressure: this.unitPolicy.output.pressure,
+        flow: this.unitPolicy.output.flow,
+      },
+      canonicalUnits: this.unitPolicy.canonical,
+      storeCanonical: true,
+      strictUnitValidation: true,
+      throwOnInvalidUnit: true,
+      requireUnitForTypes: ['pressure', 'flow'],
+    }, this.logger);
+
     this.child = {};
-        this.valves = {}; // hold child object so we can get information from its child valves 
+    this.valves = {};
+    this._valveListeners = new Map();
+    this.sources = {};
+    this._sourceListeners = new Map();
+    this.fluidContract = {
+      status: 'unknown',
+      serviceType: null,
+      upstreamServiceTypes: [],
+      sourceCount: 0,
+      message: 'No upstream fluid sources registered.',
+    };
+    this.flowReconciliation = { ...DEFAULT_FLOW_RECONCILIATION };
+    this.lastFlowSolve = {
+      passes: 0,
+      residual: 0,
+      targetTotal: 0,
+      assignedTotal: 0,
+    };
 
-        
-
-        // Initialize variables
-        this.maxDeltaP = 0; // max deltaP is 0 als er geen child valves zijn
+    this.maxDeltaP = 0;
     this.currentMode = this.config.mode.current;
-        this.childRegistrationUtils = new childRegistrationUtils(this); // Child registration utility
+    this.childRegistrationUtils = new childRegistrationUtils(this);
 
+    this.state = new state({}, this.logger);
+    this.state.stateManager.currentState = 'operational';
   }
 
   registerOnChildEvents() {}
 
-    registerChild(child, positionVsParent) {
+  _resolveRegistrationContext(child, positionVsParentOrSoftwareType) {
+    const fromArg = String(positionVsParentOrSoftwareType || '').trim();
+    if (KNOWN_POSITIONS.has(fromArg)) {
+      return {
+        positionVsParent: fromArg,
+        softwareType: child?.config?.functionality?.softwareType || null,
+      };
+    }
+
+    return {
+      positionVsParent: child?.positionVsParent || 'atEquipment',
+      softwareType: fromArg || child?.config?.functionality?.softwareType || null,
+    };
+  }
+
+  _isValveLike(child) {
+    return Boolean(
+      child
+      && typeof child.updateFlow === 'function'
+      && child.state
+      && typeof child.state.getCurrentState === 'function'
+      && child.measurements
+    );
+  }
+
+  _isSourceLike(child, softwareType) {
+    const type = String(softwareType || child?.config?.functionality?.softwareType || '').trim().toLowerCase();
+    if (SOURCE_SOFTWARE_TYPES.has(type)) {
+      return true;
+    }
+    return typeof child?.getFluidContract === 'function';
+  }
+
+  registerChild(child, positionVsParentOrSoftwareType) {
+    const ctx = this._resolveRegistrationContext(child, positionVsParentOrSoftwareType);
+    const softwareType = String(ctx.softwareType || child?.config?.functionality?.softwareType || '').trim().toLowerCase();
+
+    if (softwareType === 'valve' || (!softwareType && this._isValveLike(child))) {
+      return this._registerValve(child, ctx.positionVsParent);
+    }
+
+    if (this._isSourceLike(child, softwareType)) {
+      return this._registerSource(child, ctx.positionVsParent, softwareType);
+    }
+
+    this.logger.warn(`registerChild skipped: unsupported child type '${softwareType || 'unknown'}'`);
+    return false;
+  }
+
+  _registerValve(child, positionVsParent) {
+    if (!this._isValveLike(child)) {
+      this.logger.warn('registerChild skipped: child is not valve-like');
+      return false;
+    }
+    const id = child.config?.general?.id || child.config?.general?.name || `valve-${Object.keys(this.valves).length + 1}`;
+    if (this.valves[id]) {
+      this.logger.debug(`registerChild skipped: valve ${id} already registered`);
+      return true;
+    }
+
+    child.positionVsParent = positionVsParent;
+    this.valves[id] = child;
+    this._bindValveEvents(id, child);
+
+    this.calcValveFlows();
+    this.calcMaxDeltaP();
+    this._refreshFluidContract();
+    this.logger.info(`Valve '${id}' registered at ${positionVsParent}.`);
+    return true;
+  }
+
+  _registerSource(child, positionVsParent, softwareType) {
+    const id = child?.config?.general?.id || child?.config?.general?.name || `source-${Object.keys(this.sources).length + 1}`;
+    if (this._sourceListeners.has(id)) {
+      this._unbindSourceEvents(id);
+    }
+    child.positionVsParent = positionVsParent;
+    this.sources[id] = child;
+
+    this._bindSourceEvents(id, child);
+    const contract = this._extractFluidContractFromChild(child, softwareType);
+    this.sources[id].fluidContract = contract;
+    this._refreshFluidContract();
+
+    this.logger.info(`Source '${id}' (${softwareType || 'unknown'}) registered at ${positionVsParent}.`);
+    return true;
+  }
+
+  _bindValveEvents(valveId, valve) {
+    const handlers = {
+      onPositionChange: () => {
+        this.logger.debug(`Valve ${valveId} position changed, recalculating flows.`);
+        this.calcValveFlows();
+      },
+      onDeltaPChange: () => {
+        this.logger.debug(`Valve ${valveId} deltaP changed, recalculating max deltaP.`);
+        this.calcMaxDeltaP();
+      },
+    };
+
+    if (valve.state?.emitter?.on) {
+      valve.state.emitter.on('positionChange', handlers.onPositionChange);
+    }
+    if (valve.emitter?.on) {
+      valve.emitter.on('deltaPChange', handlers.onDeltaPChange);
+    }
+
+    this._valveListeners.set(valveId, { valve, handlers });
+  }
+
+  _unbindValveEvents(valveId) {
+    const listener = this._valveListeners.get(valveId);
+    if (!listener) {
+      return;
+    }
+    const { valve, handlers } = listener;
+    if (handlers.onPositionChange && valve.state?.emitter?.off) {
+      valve.state.emitter.off('positionChange', handlers.onPositionChange);
+    }
+    if (handlers.onDeltaPChange && valve.emitter?.off) {
+      valve.emitter.off('deltaPChange', handlers.onDeltaPChange);
+    }
+    this._valveListeners.delete(valveId);
+  }
+
+  _bindSourceEvents(sourceId, source) {
+    const listeners = {
+      flow: [],
+      onFluidContractChange: null,
+    };
+    if (source?.measurements?.emitter?.on) {
+      SOURCE_FLOW_EVENTS.forEach((eventName) => {
+        const handler = (eventData = {}) => {
+          this._handleSourceFlowEvent(eventName, eventData);
+        };
+        source.measurements.emitter.on(eventName, handler);
+        listeners.flow.push({
+          emitter: source.measurements.emitter,
+          eventName,
+          handler,
+        });
+      });
+    }
+
+    if (source?.emitter?.on) {
+      listeners.onFluidContractChange = () => {
+        const contract = this._extractFluidContractFromChild(
+          source,
+          source?.config?.functionality?.softwareType
+        );
+        if (!this.sources[sourceId]) {
+          return;
+        }
+        this.sources[sourceId].fluidContract = contract;
+        this._refreshFluidContract();
+      };
+      source.emitter.on('fluidContractChange', listeners.onFluidContractChange);
+    }
+
+    this._sourceListeners.set(sourceId, { source, listeners });
+  }
+
+  _unbindSourceEvents(sourceId) {
+    const listener = this._sourceListeners.get(sourceId);
+    if (!listener) {
+      return;
+    }
+
+    const { source, listeners } = listener;
+    listeners.flow.forEach(({ emitter, eventName, handler }) => {
+      if (typeof emitter?.off === 'function') {
+        emitter.off(eventName, handler);
+      } else if (typeof emitter?.removeListener === 'function') {
+        emitter.removeListener(eventName, handler);
+      }
+    });
+
+    if (listeners.onFluidContractChange) {
+      if (typeof source?.emitter?.off === 'function') {
+        source.emitter.off('fluidContractChange', listeners.onFluidContractChange);
+      } else if (typeof source?.emitter?.removeListener === 'function') {
+        source.emitter.removeListener('fluidContractChange', listeners.onFluidContractChange);
+      }
+    }
+
+    this._sourceListeners.delete(sourceId);
+  }
+
+  _handleSourceFlowEvent(eventName, eventData = {}) {
+    const value = Number(eventData.value);
+    if (!Number.isFinite(value)) {
+      return;
+    }
+    const eventParts = String(eventName || '').split('.');
+    const variant = eventParts[1] === 'measured' ? 'measured' : 'predicted';
+    const unit = eventData.unit || this.unitPolicy.output.flow;
+    this.updateFlow(variant, value, 'atEquipment', unit);
+  }
+
+  _normalizeOptionalServiceType(value) {
+    const raw = String(value || '').trim().toLowerCase();
+    if (SERVICE_TYPES.has(raw)) {
+      return raw;
+    }
+    return null;
+  }
+
+  _deriveDefaultServiceTypeForSoftwareType(softwareType) {
+    const key = String(softwareType || '').trim().toLowerCase();
+    return DEFAULT_SOURCE_SERVICE_TYPE[key] || null;
+  }
+
+  _extractFluidContractFromChild(child, softwareType) {
+    let contract = null;
+    if (typeof child?.getFluidContract === 'function') {
+      try {
+        contract = child.getFluidContract();
+      } catch (error) {
+        this.logger.warn(`Failed to read child fluid contract: ${error.message}`);
+      }
+    }
+    const contractStatus = String(contract?.status || '').trim().toLowerCase();
+    if (contractStatus === 'conflict') {
+      return { status: 'conflict', serviceType: null };
+    }
+
+    const serviceTypeFromContract = this._normalizeOptionalServiceType(contract?.serviceType);
+    if (serviceTypeFromContract) {
+      return { status: 'resolved', serviceType: serviceTypeFromContract };
+    }
+
+    const directType = this._normalizeOptionalServiceType(
+      child?.serviceType
+      || child?.expectedServiceType
+      || child?.config?.asset?.serviceType
+    );
+    if (directType) {
+      return { status: 'resolved', serviceType: directType };
+    }
+
+    const fallbackType = this._deriveDefaultServiceTypeForSoftwareType(softwareType);
+    if (fallbackType) {
+      return { status: 'inferred', serviceType: fallbackType };
+    }
+
+    return { status: 'unknown', serviceType: null };
+  }
+
+  _refreshFluidContract() {
+    const contracts = Object.values(this.sources)
+      .map((source) => source?.fluidContract || null)
+      .filter(Boolean);
+    const serviceTypes = Array.from(new Set(
+      contracts
+        .map((contract) => this._normalizeOptionalServiceType(contract.serviceType))
+        .filter(Boolean)
+    ));
+    const hasConflict = contracts.some((contract) => String(contract.status || '').toLowerCase() === 'conflict');
+    let next = null;
+
+    if (hasConflict || serviceTypes.length > 1) {
+      next = {
+        status: 'conflict',
+        serviceType: null,
+        upstreamServiceTypes: serviceTypes,
+        sourceCount: Object.keys(this.sources).length,
+        message: `Conflicting upstream fluids detected: ${serviceTypes.join(', ') || 'unknown'}.`,
+      };
+    } else if (serviceTypes.length === 1) {
+      next = {
+        status: 'resolved',
+        serviceType: serviceTypes[0],
+        upstreamServiceTypes: serviceTypes,
+        sourceCount: Object.keys(this.sources).length,
+        message: `Upstream fluid resolved as ${serviceTypes[0]}.`,
+      };
+    } else {
+      next = {
+        status: 'unknown',
+        serviceType: null,
+        upstreamServiceTypes: [],
+        sourceCount: Object.keys(this.sources).length,
+        message: 'No upstream fluid sources registered.',
+      };
+    }
+
+    const prev = this.fluidContract || {};
+    const changed = (
+      prev.status !== next.status
+      || prev.serviceType !== next.serviceType
+      || prev.sourceCount !== next.sourceCount
+      || (prev.message || '') !== (next.message || '')
+    );
+    this.fluidContract = next;
+    if (changed) {
+      this.emitter.emit('fluidContractChange', this.getFluidContract());
+    }
+  }
+
+  getFluidContract() {
+    const state = this.fluidContract || {};
+    return {
+      status: state.status || 'unknown',
+      serviceType: state.serviceType || null,
+      upstreamServiceTypes: Array.isArray(state.upstreamServiceTypes) ? [...state.upstreamServiceTypes] : [],
+      sourceCount: Number(state.sourceCount) || 0,
+      message: state.message || '',
+      source: 'valvegroupcontrol',
+    };
+  }
+
+  destroy() {
+    for (const valveId of this._valveListeners.keys()) {
+      this._unbindValveEvents(valveId);
+    }
+    for (const sourceId of this._sourceListeners.keys()) {
+      this._unbindSourceEvents(sourceId);
+    }
+  }
+
+  _isValveAvailable(valve) {
+    const currentState = valve.state.getCurrentState();
+    const mode = valve.currentMode;
+    const kv = Number(valve.kv);
+    return (
+      currentState !== 'off'
+      && currentState !== 'maintenance'
+      && mode !== 'maintenance'
+      && Number.isFinite(kv)
+      && kv > 0
+    );
+  }
+
+  getAvailableValves() {
+    return Object.entries(this.valves)
+      .filter(([, valve]) => this._isValveAvailable(valve))
+      .map(([id, valve]) => ({ id, valve }));
   }
 
   isValidSourceForMode(source, mode) {
     const allowedSourcesSet = this.config.mode.allowedSources[mode] || [];
-        this.logger.info(`Allowed sources for mode '${mode}': ${allowedSourcesSet}`);
     return allowedSourcesSet.has(source);
   }
 
   async handleInput(source, action, parameter) {
     if (!this.isValidSourceForMode(source, this.currentMode)) {
-          let warningTxt = `Source '${source}' is not valid for mode '${this.currentMode}'.`;
+      const warningTxt = `Source '${source}' is not valid for mode '${this.currentMode}'.`;
       this.logger.warn(warningTxt);
-          return {status : false , feedback: warningTxt};
+      return { status: false, feedback: warningTxt };
     }
 
     this.logger.info(`Handling input from source '${source}' with action '${action}' in mode '${this.currentMode}'.`);
     try {
       switch (action) {
-            case "execSequence":
+        case 'execSequence':
           await this.executeSequence(parameter);
           break;
-            case "totalFlowChange":
-              await this.updateFlow(parameter);
+        case 'totalFlowChange': {
+          if (parameter && typeof parameter === 'object' && Object.prototype.hasOwnProperty.call(parameter, 'value')) {
+            await this.updateFlow(parameter.variant || 'measured', parameter.value, parameter.position || 'atEquipment', parameter.unit || this.unitPolicy.output.flow);
+          } else if (parameter && typeof parameter === 'object' && Object.prototype.hasOwnProperty.call(parameter, 'q')) {
+            await this.updateFlow('measured', Number(parameter.q), 'atEquipment', parameter.unit || this.unitPolicy.output.flow);
+          } else {
+            await this.updateFlow('measured', Number(parameter), 'atEquipment', this.unitPolicy.output.flow);
+          }
           break;
-            case "emergencyStop":
+        }
+        case 'emergencyStop':
+        case 'emergencystop':
           this.logger.warn(`Emergency stop activated by '${source}'.`);
-              await this.executeSequence("emergencyStop");
+          await this.executeSequence('emergencystop');
           break;
-            case "statusCheck":
+        case 'statusCheck':
           this.logger.info(`Status Check: Mode = '${this.currentMode}', Source = '${source}'.`);
           break;
         default:
@@ -113,15 +486,17 @@ class ValveGroupControl {
           break;
       }
       this.logger.debug(`Action '${action}' successfully executed`);
-          return {status : true , feedback: `Action '${action}' successfully executed.`};
+      return { status: true, feedback: `Action '${action}' successfully executed.` };
     } catch (error) {
       this.logger.error(`Error handling input: ${error}`);
+      return { status: false, feedback: `Error handling input: ${error.message || error}` };
     }
-        
   }
 
   setMode(newMode) {
-        const availableModes = defaultConfig.mode.current.rules.values.map(vgc => vgc.value);
+    const availableModes = Array.isArray(this.defaultConfig?.mode?.current?.rules?.values)
+      ? this.defaultConfig.mode.current.rules.values.map((vgc) => vgc.value)
+      : Object.keys(this.config?.mode?.allowedSources || {});
     if (!availableModes.includes(newMode)) {
       this.logger.warn(`Invalid mode '${newMode}'. Allowed modes are: ${availableModes.join(', ')}`);
       return;
@@ -131,11 +506,73 @@ class ValveGroupControl {
     this.logger.info(`Mode successfully changed to '${newMode}'.`);
   }
 
+  _buildUnitPolicy(config = {}) {
+    const flowUnit = this._resolveUnitOrFallback(
+      config?.general?.unit,
+      'volumeFlowRate',
+      DEFAULT_IO_UNITS.flow
+    );
 
+    return {
+      canonical: { ...CANONICAL_UNITS },
+      output: {
+        flow: flowUnit,
+        pressure: DEFAULT_IO_UNITS.pressure,
+      },
+    };
+  }
+
+  _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;
+    }
+  }
+
+  _outputUnitForType(type) {
+    switch (String(type || '').toLowerCase()) {
+      case 'flow':
+        return this.unitPolicy.output.flow;
+      case 'pressure':
+        return this.unitPolicy.output.pressure;
+      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()) {
+    const valueNum = Number(value);
+    if (!Number.isFinite(valueNum)) {
+      return;
+    }
+    this.measurements
+      .type(type)
+      .variant(variant)
+      .position(position)
+      .value(valueNum, timestamp, unit || undefined);
+  }
 
-    // -------- Sequence Handlers -------- //  
   async executeSequence(sequenceName) {
-
     const sequence = this.config.sequences[sequenceName];
 
     if (!sequence || sequence.size === 0) {
@@ -145,32 +582,33 @@ class ValveGroupControl {
 
     this.logger.info(` --------- Executing sequence: ${sequenceName} -------------`);
 
-    for (const state of sequence) {
+    for (const stateName of sequence) {
       try {
-        await this.state.transitionToState(state);
-        // Update measurements after state change
-
+        await this.state.transitionToState(stateName);
       } catch (error) {
         this.logger.error(`Error during sequence '${sequenceName}': ${error}`);
-        break; // Exit sequence execution on error
+        break;
       }
     }
   }
 
-updateFlow(variant,value,position) {
+  updateFlow(variant, value, position, unit = this.unitPolicy.output.flow) {
+    if (value === null || value === undefined) {
+      this.logger.warn(`Received null or undefined value for flow update. Variant: ${variant}, Position: ${position}`);
+      return;
+    }
 
     switch (variant) {
-      case ("measured"):
-        // put value in measurements container
+      case 'measured':
         this.logger.debug(`Updating measured flow for position ${position} with value ${value}`);
-        this.measurements.type("flow").variant("measured").position(position).value(value);
+        this._writeMeasurement('flow', 'measured', position, value, unit);
         this.calcValveFlows();
         break;
 
-      case ("predicted"):
+      case 'predicted':
         this.logger.debug(`Updating predicted flow for position ${position} with value ${value}`);
-        this.measurements.type("flow").variant("predicted").position(position).value(value);
-        this.calcValveFlows(); // Pass the value to calculate valve flows
+        this._writeMeasurement('flow', 'predicted', position, value, unit);
+        this.calcValveFlows();
         break;
 
       default:
@@ -179,85 +617,152 @@ updateFlow(variant,value,position) {
     }
   }
 
-  updateMeasurement(variant, subType, value, position) {
+  updateMeasurement(variant, subType, value, position, unit) {
     this.logger.debug(`---------------------- updating ${subType} ------------------ `);
     switch (subType) {
-      case "pressure":
-        // Update pressure measurement
-        //this.updatePressure(variant,value,position);
-        break;
-      case "flow":
-        this.updateFlow(variant,value,position);
-        break;
-      case "power":
-        // Update power measurement
+      case 'flow':
+        this.updateFlow(variant, value, position, unit || this.unitPolicy.output.flow);
         break;
       default:
         this.logger.error(`Type '${subType}' not recognized for measured update.`);
-        return;
+        break;
     }
   }
 
   calcValveFlows() {
-    const totalFlow = this.measurements.type("flow").variant("measured").position("atEquipment").getCurrentValue(); // get the total flow from the measurement container
-    let totalKv = 0; 
+    const totalFlowMeasured = this._readMeasurement('flow', 'measured', 'atEquipment', this.unitPolicy.output.flow);
+    const totalFlowPredicted = this._readMeasurement('flow', 'predicted', 'atEquipment', this.unitPolicy.output.flow);
+    const totalFlow = Number.isFinite(totalFlowMeasured) ? totalFlowMeasured : totalFlowPredicted;
 
-    this.logger.debug(`Calculating valve flows... ${totalFlow}`); //Checkpoint
-    
-    for (const key in this.valves){ //bereken sum kv values om verdeling total flow te maken
-        this.logger.info('kv: ' + this.valves[key].kv); //CHECKPOINT
-        if (this.valves[key].state.getCurrentPosition() != null) {
-            totalKv += this.valves[key].kv;
-            this.logger.info('Total Kv = ' + totalKv); //CHECKPOINT
-        }
-        if(totalKv === 0) {
-            this.logger.warn('Total Kv is 0, cannot calculate flow distribution.');
-            return; // Avoid division by zero
-        }
+    if (!Number.isFinite(totalFlow)) {
+      return;
     }
 
-    for (const key in this.valves){ 
-        const valve = this.valves[key]; 
-        this.logger.debug(`Calculating ratio for valve total: ${totalKv} valve.kv: ${valve.kv} ratio : ${valve.kv / totalKv}`); //Checkpoint
-        const ratio = valve.kv / totalKv; 
-        const flow = ratio * totalFlow; // bereken flow per valve 
-
-        //update flow per valve in de object zelf wat daar vervolgens weer de nieuwe deltaP berekent
-        valve.updateFlow("predicted", flow, "downstream"); 
-        this.logger.info(`--> Sending updated flow to valves --> ${flow} `); //Checkpoint 
+    const availableEntries = this.getAvailableValves();
+    const availableIds = new Set(availableEntries.map((entry) => entry.id));
+    const totalKv = availableEntries.reduce((sum, { valve }) => sum + Number(valve.kv), 0);
 
+    if (!availableEntries.length || !Number.isFinite(totalKv) || totalKv <= 0) {
+      this.logger.warn('No available valves with valid Kv, setting assigned flow to 0.');
+      for (const valve of Object.values(this.valves)) {
+        valve.updateFlow('predicted', 0, 'downstream', this.unitPolicy.output.flow);
       }
+      this._writeMeasurement('flow', 'predicted', 'atEquipment', 0, this.unitPolicy.output.flow);
+      this.lastFlowSolve = {
+        passes: 0,
+        residual: Number(totalFlow) || 0,
+        targetTotal: Number(totalFlow) || 0,
+        assignedTotal: 0,
+      };
+      return;
     }
 
-  calcMaxDeltaP() { // bereken de max deltaP van alle child valves
-    let maxDeltaP = 0; //max deltaP is 0 als er geen child valves zijn
-    this.logger.info('Calculating new max deltaP...'); 
-    for (const key in this.valves) {
-      const valve = this.valves[key]; //haal de child valve object op
-      const deltaP = valve.measurements.type("pressure").variant("predicted").position("delta").getCurrentValue(); //get delta P
-      this.logger.info(`Delta P for valve ${key}: ${deltaP}`); 
-      if (deltaP > maxDeltaP) { //als de deltaP van de child valve groter is dan de huidige maxDeltaP, dan update deze
+    const solve = this._solveFlowDistribution(totalFlow, availableEntries);
+    let assignedTotal = 0;
+    for (const [id, valve] of Object.entries(this.valves)) {
+      const flow = availableIds.has(id) ? (solve.flowsById[id] || 0) : 0;
+      valve.updateFlow('predicted', flow, 'downstream', this.unitPolicy.output.flow);
+      assignedTotal += flow;
+    }
+
+    this._writeMeasurement('flow', 'predicted', 'atEquipment', assignedTotal, this.unitPolicy.output.flow);
+    this.lastFlowSolve = {
+      passes: solve.passes,
+      residual: solve.residual,
+      targetTotal: totalFlow,
+      assignedTotal,
+    };
+    this.calcMaxDeltaP();
+  }
+
+  _readValveAcceptedFlow(valve) {
+    const accepted = Number(
+      valve?.measurements
+        ?.type('flow')
+        ?.variant('predicted')
+        ?.position('downstream')
+        ?.getCurrentValue(this.unitPolicy.output.flow)
+    );
+    return Number.isFinite(accepted) ? accepted : null;
+  }
+
+  _solveFlowDistribution(totalFlow, availableEntries) {
+    const totalKv = availableEntries.reduce((sum, { valve }) => sum + Number(valve.kv), 0);
+    if (!Number.isFinite(totalKv) || totalKv <= 0) {
+      return { flowsById: {}, residual: Number(totalFlow) || 0, passes: 0 };
+    }
+
+    const targetById = {};
+    availableEntries.forEach(({ id }) => {
+      targetById[id] = 0;
+    });
+
+    let residual = Number(totalFlow);
+    let passes = 0;
+    const maxPasses = Math.max(1, Number(this.flowReconciliation?.maxPasses) || DEFAULT_FLOW_RECONCILIATION.maxPasses);
+    const tolerance = Math.max(0, Number(this.flowReconciliation?.residualTolerance) || DEFAULT_FLOW_RECONCILIATION.residualTolerance);
+
+    while (passes < maxPasses && Number.isFinite(residual) && Math.abs(residual) > tolerance) {
+      availableEntries.forEach(({ id, valve }) => {
+        const kv = Number(valve.kv);
+        const share = (kv / totalKv) * residual;
+        const nextTarget = Number(targetById[id]) + share;
+        targetById[id] = nextTarget;
+        valve.updateFlow('predicted', nextTarget, 'downstream', this.unitPolicy.output.flow);
+      });
+
+      let acceptedTotal = 0;
+      availableEntries.forEach(({ id, valve }) => {
+        const accepted = this._readValveAcceptedFlow(valve);
+        if (Number.isFinite(accepted)) {
+          targetById[id] = accepted;
+          acceptedTotal += accepted;
+          return;
+        }
+        acceptedTotal += Number(targetById[id]) || 0;
+      });
+
+      residual = Number(totalFlow) - acceptedTotal;
+      passes += 1;
+    }
+
+    return {
+      flowsById: targetById,
+      residual: Number.isFinite(residual) ? residual : 0,
+      passes,
+    };
+  }
+
+  calcMaxDeltaP() {
+    let maxDeltaP = 0;
+    for (const [id, valve] of Object.entries(this.valves)) {
+      const deltaP = Number(
+        valve.measurements
+          .type('pressure')
+          .variant('predicted')
+          .position('delta')
+          .getCurrentValue(this.unitPolicy.output.pressure)
+      );
+
+      if (!Number.isFinite(deltaP)) {
+        continue;
+      }
+      this.logger.debug(`Delta P for valve ${id}: ${deltaP}`);
+      if (deltaP > maxDeltaP) {
         maxDeltaP = deltaP;
       }
     }
-    this.logger.info('Max Delta P updated to: ' + maxDeltaP); 
-
-    this.maxDeltaP = maxDeltaP; //update de max deltaP in de measurement container van de valveGroupControl class
-
-}
 
+    this.maxDeltaP = maxDeltaP;
+    this._writeMeasurement('pressure', 'predicted', 'deltaMax', maxDeltaP, this.unitPolicy.output.pressure);
+  }
 
   getOutput() {
-
-    // Improved output object generation
     const output = {};
-    //build the output object
-    this.measurements.getTypes().forEach(type => {
-      this.measurements.getVariants().forEach(variant => {
-        this.measurements.getPositions().forEach(position => {
-
-        const value = this.measurements.type(type).variant(variant).position(position).getCurrentValue(); //get the current value of the measurement
-        
+    Object.entries(this.measurements.measurements || {}).forEach(([type, variants]) => {
+      Object.entries(variants || {}).forEach(([variant, positions]) => {
+        Object.keys(positions || {}).forEach((position) => {
+          const value = this._readMeasurement(type, variant, position, this._outputUnitForType(type));
           if (value != null) {
             output[`${position}_${variant}_${type}`] = value;
           }
@@ -265,76 +770,11 @@ updateFlow(variant,value,position) {
       });
     });
 
-    //fill in the rest of the output object
-    output["mode"] = this.currentMode;
-    output["maxDeltaP"] = this.maxDeltaP; 
-
-    //this.logger.debug(`Output: ${JSON.stringify(output)}`);
+    output.mode = this.currentMode;
+    output.maxDeltaP = this.maxDeltaP;
 
     return output;
   }
-
 }
 
 module.exports = ValveGroupControl;
-
-
-const valve = require('../../valve/src/specificClass.js');
-const valveConfig = {
-      general: {
-        name: "valve",
-        logging: {
-          enabled: true,
-          logLevel: "debug"
-        }
-      },
-      asset: {
-        supplier: "binder",
-        category: "valve",
-        type: "control",
-        model: "ECDV",
-        unit: "m3/h"
-      },
-      functionality: {
-        positionVsParent: 'atEquipment', // Default to 'atEquipment' if not specified
-      }
-    };
-
-const stateConfig = {
-        general: {
-          logging: {
-            enabled: true,
-            logLevel: "debug"
-          }
-        },
-        movement: {
-          speed: 1
-        },
-        time: {
-          starting: 1,
-          warmingup: 1,
-          stopping: 1,
-          coolingdown: 1
-        }
-      };
-
-const valve1 = new valve(valveConfig, stateConfig);
-//const valve2 = new valve(valveConfig, stateConfig);
-//const valve3 = new valve(valveConfig, stateConfig);
-
-valve1.kv = 10; // Set Kv value for valve1
-//valve2.kv = 20; // Set Kv value for valve2  
-//valve3.kv = 30; // Set Kv value for valve3
-
-valve1.updateMeasurement("measured", "pressure" , 500, "downstream"); 
-//valve2.updateMeasurement("measured" , "pressure" , 500, "downstream"); 
-//valve3.updateMeasurement("measured" , "pressure" , 500, "downstream"); 
-
-const vgc = new ValveGroupControl(); 
-
-vgc.childRegistrationUtils.registerChild(valve1, "atEquipment");
-//vgc.childRegistrationUtils.registerChild(valve2, "atEquipment");
-//vgc.childRegistrationUtils.registerChild(valve3, "atEquipment");
-
-vgc.updateFlow("measured", 1000, "atEquipment"); // Update total flow to 100 m3/h
-

test/integration/flow-distribution.integration.test.js

@@ -0,0 +1,93 @@
+const test = require('node:test');
+const assert = require('node:assert/strict');
+
+const Valve = require('../../../valve/src/specificClass');
+const ValveGroupControl = require('../../src/specificClass');
+
+function buildValve(name) {
+  return new Valve(
+    {
+      general: {
+        name,
+        logging: { enabled: false, logLevel: 'error' },
+      },
+      asset: {
+        supplier: 'binder',
+        category: 'valve',
+        type: 'control',
+        model: 'ECDV',
+        unit: 'm3/h',
+      },
+      functionality: {
+        positionVsParent: 'atEquipment',
+      },
+    },
+    {
+      general: {
+        logging: { enabled: false, logLevel: 'error' },
+      },
+      movement: { speed: 1 },
+      time: { starting: 0, warmingup: 0, stopping: 0, coolingdown: 0 },
+    }
+  );
+}
+
+function primeValve(valve, position) {
+  valve.updatePressure('measured', 500, 'downstream', 'mbar');
+  valve.updateFlow('predicted', 100, 'downstream', 'm3/h');
+  valve.state.movementManager.currentPosition = position;
+  valve.updatePosition();
+}
+
+function buildGroup() {
+  return new ValveGroupControl({
+    general: {
+      name: 'vgc-test',
+      logging: { enabled: false, logLevel: 'error' },
+      unit: 'm3/h',
+    },
+    functionality: {
+      positionVsParent: 'atEquipment',
+    },
+  });
+}
+
+test('valveGroupControl distributes total flow according to supplier-curve Kv and keeps roundtrip balance', async () => {
+  const valve1 = buildValve('valve-1');
+  const valve2 = buildValve('valve-2');
+  primeValve(valve1, 50);
+  primeValve(valve2, 80);
+
+  const group = buildGroup();
+  assert.equal(await group.childRegistrationUtils.registerChild(valve1, 'atEquipment'), true);
+  assert.equal(await group.childRegistrationUtils.registerChild(valve2, 'atEquipment'), true);
+
+  group.updateFlow('measured', 1000, 'atEquipment', 'm3/h');
+
+  const q1 = valve1.measurements.type('flow').variant('predicted').position('downstream').getCurrentValue('m3/h');
+  const q2 = valve2.measurements.type('flow').variant('predicted').position('downstream').getCurrentValue('m3/h');
+  const distributedTotal = q1 + q2;
+  assert.ok(Math.abs(distributedTotal - 1000) < 0.001, `distributed flow mismatch: ${distributedTotal}`);
+
+  const expectedRatio = valve1.kv / (valve1.kv + valve2.kv);
+  const actualRatio = q1 / (q1 + q2);
+  assert.ok(Math.abs(expectedRatio - actualRatio) < 0.001, `expected ratio ${expectedRatio}, got ${actualRatio}`);
+
+  const expectedMaxDeltaP = Math.max(
+    valve1.measurements.type('pressure').variant('predicted').position('delta').getCurrentValue('mbar'),
+    valve2.measurements.type('pressure').variant('predicted').position('delta').getCurrentValue('mbar')
+  );
+  assert.ok(Math.abs(group.maxDeltaP - expectedMaxDeltaP) < 0.001, `expected max deltaP ${expectedMaxDeltaP}, got ${group.maxDeltaP}`);
+
+  group.destroy();
+  valve1.destroy();
+  valve2.destroy();
+});
+
+test('valveGroupControl rejects non-valve-like child payload', () => {
+  const group = buildGroup();
+  const result = group.registerChild({ config: { functionality: { softwareType: 'valve' } } }, 'atEquipment');
+  assert.equal(result, false);
+  assert.equal(Object.keys(group.valves).length, 0);
+  group.destroy();
+});

test/integration/source-topology.integration.test.js

@@ -0,0 +1,149 @@
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const EventEmitter = require('events');
+
+const Valve = require('../../../valve/src/specificClass');
+const ValveGroupControl = require('../../src/specificClass');
+
+function buildValve({ runtimeOptions = {} } = {}) {
+  return new Valve(
+    {
+      general: {
+        name: 'valve-topology-test',
+        logging: { enabled: false, logLevel: 'error' },
+      },
+      asset: {
+        supplier: 'binder',
+        category: 'valve',
+        type: 'control',
+        model: 'ECDV',
+        unit: 'm3/h',
+      },
+      functionality: {
+        positionVsParent: 'atEquipment',
+      },
+    },
+    {
+      general: {
+        logging: { enabled: false, logLevel: 'error' },
+      },
+      movement: { speed: 1 },
+      time: { starting: 0, warmingup: 0, stopping: 0, coolingdown: 0 },
+    },
+    runtimeOptions
+  );
+}
+
+function buildGroup() {
+  return new ValveGroupControl({
+    general: {
+      name: 'vgc-source-test',
+      logging: { enabled: false, logLevel: 'error' },
+      unit: 'm3/h',
+    },
+    functionality: {
+      positionVsParent: 'atEquipment',
+    },
+  });
+}
+
+function buildSource({
+  id,
+  softwareType,
+  serviceType = null,
+  status = 'resolved',
+}) {
+  const emitter = new EventEmitter();
+  let contract = { status, serviceType };
+  return {
+    emitter,
+    config: {
+      general: { id, name: id },
+      functionality: { softwareType },
+      asset: {
+        serviceType: serviceType || undefined,
+      },
+    },
+    measurements: {
+      emitter: new EventEmitter(),
+    },
+    getFluidContract() {
+      return { ...contract };
+    },
+    setFluidContract(next) {
+      contract = { ...contract, ...next };
+    },
+  };
+}
+
+test('valveGroupControl accepts machine source and syncs upstream flow events', () => {
+  const group = buildGroup();
+  const machine = buildSource({
+    id: 'machine-1',
+    softwareType: 'machine',
+    serviceType: 'liquid',
+  });
+
+  assert.equal(group.registerChild(machine, 'machine'), true);
+  machine.measurements.emitter.emit('flow.measured.downstream', {
+    value: 150,
+    unit: 'm3/h',
+  });
+
+  const totalMeasuredFlow = group.measurements
+    .type('flow')
+    .variant('measured')
+    .position('atEquipment')
+    .getCurrentValue('m3/h');
+  assert.ok(Math.abs(totalMeasuredFlow - 150) < 1e-9);
+
+  const contract = group.getFluidContract();
+  assert.equal(contract.status, 'resolved');
+  assert.equal(contract.serviceType, 'liquid');
+
+  group.destroy();
+});
+
+test('valveGroupControl exposes conflict when upstream sources mix fluid contracts', () => {
+  const group = buildGroup();
+  const machineLiquid = buildSource({
+    id: 'machine-liquid',
+    softwareType: 'machine',
+    serviceType: 'liquid',
+  });
+  const machineGas = buildSource({
+    id: 'machine-gas',
+    softwareType: 'machine',
+    serviceType: 'gas',
+  });
+
+  assert.equal(group.registerChild(machineLiquid, 'machine'), true);
+  assert.equal(group.registerChild(machineGas, 'machine'), true);
+
+  const contract = group.getFluidContract();
+  assert.equal(contract.status, 'conflict');
+  assert.deepEqual(new Set(contract.upstreamServiceTypes), new Set(['liquid', 'gas']));
+
+  group.destroy();
+});
+
+test('valve can validate fluid contract propagated by valveGroupControl', () => {
+  const group = buildGroup();
+  const machine = buildSource({
+    id: 'machine-1',
+    softwareType: 'machine',
+    serviceType: 'liquid',
+  });
+  const valve = buildValve({ runtimeOptions: { serviceType: 'gas' } });
+
+  assert.equal(group.registerChild(machine, 'machine'), true);
+  assert.equal(valve.registerChild(group, 'valvegroupcontrol'), true);
+
+  const compatibility = valve.getFluidCompatibility();
+  assert.equal(compatibility.status, 'mismatch');
+  assert.equal(compatibility.expectedServiceType, 'gas');
+  assert.equal(compatibility.receivedServiceType, 'liquid');
+
+  valve.destroy();
+  group.destroy();
+});

vgc.html

@@ -18,6 +18,8 @@
     defaults: {
       // Define default properties
       name: { value: "" },
+      processOutputFormat: { value: "process" },
+      dbaseOutputFormat: { value: "influxdb" },
 
       // Logger properties
       enableLog: { value: false },
@@ -38,7 +40,7 @@
         icon: "font-awesome/fa-tasks",
 
         label: function () {
-            return this.positionIcon + " " + "valveGroupControl";
+            return (this.positionIcon || "") + " valveGroupControl";
         },
         oneditprepare: function() {
             // Initialize the menu data for the node
@@ -55,6 +57,7 @@
         },
         oneditsave: function(){
             const node = this;
+            let success = true;
 
             // Validate logger properties using the logger menu
             if (window.EVOLV?.nodes?.valveGroupControl?.loggerMenu?.saveEditor) {
@@ -66,6 +69,8 @@
                 window.EVOLV.nodes.valveGroupControl.positionMenu.saveEditor(this);
             }
 
+            return success;
+           
         }
     });
     
@@ -73,6 +78,24 @@
 
 <script type="text/html" data-template-name="valveGroupControl">
 
+  <h3>Output Formats</h3>
+  <div class="form-row">
+    <label for="node-input-processOutputFormat"><i class="fa fa-random"></i> Process Output</label>
+    <select id="node-input-processOutputFormat" style="width:60%;">
+      <option value="process">process</option>
+      <option value="json">json</option>
+      <option value="csv">csv</option>
+    </select>
+  </div>
+  <div class="form-row">
+    <label for="node-input-dbaseOutputFormat"><i class="fa fa-database"></i> Database Output</label>
+    <select id="node-input-dbaseOutputFormat" style="width:60%;">
+      <option value="influxdb">influxdb</option>
+      <option value="json">json</option>
+      <option value="csv">csv</option>
+    </select>
+  </div>
+
   <!-- Logger fields injected here -->
   <div id="logger-fields-placeholder"></div>