updates

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();
@@ -37,13 +38,14 @@ class nodeClass {
   _loadConfig(uiConfig, node) {
     const cfgMgr = new configManager();
     this.defaultConfig = cfgMgr.getConfig(this.name);
+    const flowUnit = this._resolveUnitOrFallback(uiConfig.unit, 'volumeFlowRate', 'm3/h', 'flow');
 
     // 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)
+        unit: flowUnit,
         logging: {
           enabled: uiConfig.enableLog,
           logLevel: uiConfig.logLevel,
@@ -57,29 +59,44 @@ class nodeClass {
     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 +108,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 +156,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 +169,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 +204,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 +258,7 @@ class nodeClass {
     this.node.on("close", (done) => {
       clearInterval(this._tickInterval);
       clearInterval(this._statusInterval);
+      this.source?.destroy?.();
       if (typeof done === 'function') done();
     });
   }

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

@@ -38,7 +38,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 +55,7 @@
         },
         oneditsave: function(){
             const node = this;
+            let success = true;
 
             // Validate logger properties using the logger menu
             if (window.EVOLV?.nodes?.valveGroupControl?.loggerMenu?.saveEditor) {
@@ -65,6 +66,8 @@
             if (window.EVOLV?.nodes?.valveGroupControl?.positionMenu?.saveEditor) {
                 window.EVOLV.nodes.valveGroupControl.positionMenu.saveEditor(this);
             }
+
+            return success;
            
         }
     });