@@ -48,7 +48,17 @@ class Machine {
this . errorMetrics = new nrmse ( errorMetricsConfig , this . logger ) ;
// Initialize measurements
this . measurements = new MeasurementContainer ( ) ;
this . measurements = new MeasurementContainer ( {
autoConvert : true ,
windowSize : 50 ,
defaultUnits : {
pressure : 'mbar' ,
flow : this . config . general . unit ,
power : 'kW' ,
temperature : 'C'
}
} ) ;
this . interpolation = new interpolation ( ) ;
this . flowDrift = null ;
@@ -68,48 +78,114 @@ class Machine {
this . updatePosition ( ) ;
} ) ;
// used for holding the source and sink unit operations or other object with setInfluent / getEffluent method for e.g. recirculation.
this . upstreamSource = null ;
this . downstreamSink = null ;
//When state changes look if we need to do other updates
this . state . emitter . on ( "stateChange" , ( newState ) => {
this . logger . debug ( ` State change detected: ${ newState } ` ) ;
this . _updateState ( ) ;
} ) ;
this . child = { } ; // object to hold child information so we know on what to subscribe
this . childRegistrationUtils = new childRegistrationUtils ( this ) ; // Child registration utility
// --- KPI tracking ---
this . kpi = {
failures : 0 ,
totalRuntimeHours : 0 ,
totalDowntimeHours : 0 ,
lastFailureTime : null ,
lastRepairTime : null ,
MTBF : 0 ,
MTTR : 0 ,
availability : 0
} ;
this . assetHealth = {
index : 0 // 0 = optimal, 5 = failure
} ;
this . state . emitter . on ( 'stateChange' , ( payload ) => {
const stateStr = typeof payload === 'string'
? payload
: ( payload ? . state ? ? payload ? . newState ? ? payload ) ;
if ( typeof stateStr !== 'string' ) {
this . logger . warn ( ` stateChange event without parsable state: ${ JSON . stringify ( payload ) } ` ) ;
return ;
}
this . _handleStateChangeForKPI ( stateStr ) ;
} ) ;
// --- KPI tracking ---
this . kpi = {
failures : 0 ,
totalRuntimeHours : 0 ,
totalDowntimeHours : 0 ,
lastFailureTime : null ,
lastRepairTime : null ,
MTBF : 0 ,
MTTR : 0 ,
availability : 0
} ;
this . assetHealth = {
index : 0 // 0 = optimal, 5 = failure
} ;
this . state . emitter . on ( 'stateChange' , ( payload ) => {
const stateStr = typeof payload === 'string'
? payload
: ( payload ? . state ? ? payload ? . newState ? ? payload ) ;
if ( typeof stateStr !== 'string' ) {
this . logger . warn ( ` stateChange event without parsable state: ${ JSON . stringify ( payload ) } ` ) ;
return ;
}
this . _handleStateChangeForKPI ( stateStr ) ;
} ) ;
}
_updateState ( ) {
const isOperational = this . _isOperationalState ( ) ;
if ( ! isOperational ) {
//overrule the last prediction this should be 0 now
this . measurements . type ( "flow" ) . variant ( "predicted" ) . position ( "downstream" ) . value ( 0 ) ;
}
}
_updateState ( ) {
const isOperational = this . _isOperationalState ( ) ;
if ( ! isOperational ) {
//overrule the last prediction this should be 0 now
this . measurements . type ( "flow" ) . variant ( "predicted" ) . position ( "downstream" ) . value ( 0 ) ;
}
}
/*------------------- Register child events -------------------*/
registerChild ( child , softwareType ) {
if ( ! child ) {
this . logger . error ( ` Invalid ${ softwareType } child provided. ` ) ;
return ;
}
this . logger . debug ( 'Setting up child event for softwaretype ' + softwareType ) ;
switch ( softwareType ) {
case "measurement" :
this . logger . debug ( ` Registering measurement child... ` ) ;
this . _connectMeasurement ( child ) ;
break ;
case "reactor" :
this . logger . debug ( ` Registering reactor child... ` ) ;
this . _connectReactor ( child ) ;
break ;
default :
this . logger . error ( ` Unrecognized softwareType: ${ softwareType } ` ) ;
}
}
_connectMeasurement ( measurementChild ) {
const position = measurementChild . config . functionality . positionVsParent ;
const distance = measurementChild . config . functionality . distanceVsParent || 0 ;
const measurementType = measurementChild . config . asset . type ;
if ( softwareType === "measurement" ) {
const position = child . config . functionality . positionVsParent ;
const distance = child . config . functionality . distanceVsParent || 0 ;
const measurementType = child . config . asset . type ;
const key = ` ${ measurementType } _ ${ position } ` ;
//rebuild to measurementype.variant no position and then switch based on values not strings or names.
const eventName = ` ${ measurementType } .measured. ${ position } ` ;
this . logger . debug ( ` Setting up listener for ${ eventName } from child ${ measurementC hild. config . general . name } ` ) ;
this . logger . debug ( ` Setting up listener for ${ eventName } from child ${ c hild. config . general . name } ` ) ;
// Register event listener for measurement updates
measurementC hild. measurements . emitter . on ( eventName , ( eventData ) => {
c hild . measurements . emitter . on ( eventName , ( eventData ) => {
this . logger . debug ( ` 🔄 ${ position } ${ measurementType } from ${ eventData . childName } : ${ eventData . value } ${ eventData . unit } ` ) ;
console . log ( ` Emitting... ${ eventName } with data: ` ) ;
// Store directly in parent's measurement container
this . measurements
. type ( measurementType )
@@ -118,28 +194,31 @@ class Machine {
. value ( eventData . value , eventData . timestamp , eventData . unit ) ;
// Call the appropriate handler
this . _callMeasurementHandler ( measurementType , eventData . value , position , eventData ) ;
} ) ;
}
}
// Centralized handler dispatcher
_callMeasurementHandler ( measurementType , value , position , context ) {
switch ( measurementType ) {
case 'pressure' :
this . updateMeasuredPressure ( eventData . value, position , eventData ) ;
this . updateMeasuredPressure ( value , position , context ) ;
break ;
case 'flow' :
this . updateMeasuredFlow ( eventData . value, position , eventData ) ;
this . updateMeasuredFlow ( value , position , context ) ;
break ;
default :
this . logger . warn ( ` No handler for measurement type: ${ measurementType } ` ) ;
// Generic handler - just update position
this . updatePosition ( ) ;
break ;
}
} ) ;
}
}
_connectReactor ( reactorChild ) {
this . downstreamSink = reactorChild ; // downstream from the pumps perpective
}
//---------------- END child stuff -------------//
//---------------- END child stuff -------------//
// Method to assess drift using errorMetrics
assessDrift ( measurement , processMin , processMax ) {
@@ -176,43 +255,68 @@ class Machine {
// -------- Mode and Input Management -------- //
isValidSourceForMode ( source , mode ) {
const allowedSourcesSet = this . config . mode . allowedSources [ mode ] || [ ] ;
return allowedSourcesSet . has ( source ) ;
const allowed = allowedSourcesSet . has ( source ) ;
allowed ?
this . logger . debug ( ` source is allowed proceeding with ${ source } for mode ${ mode } ` ) :
this . logger . warn ( ` ${ source } is not allowed in mode ${ mode } ` ) ;
return allowed ;
}
isValidActionForMode ( action , mode ) {
const allowedActionsSet = this . config . mode . allowedActions [ mode ] || [ ] ;
return allowedActionsSet . has ( action ) ;
const allowed = allowedActionsSet . has ( action ) ;
allowed ?
this . logger . debug ( ` Action is allowed proceeding with ${ action } for mode ${ mode } ` ) :
this . logger . warn ( ` ${ action } is not allowed in mode ${ mode } ` ) ;
return allowed ;
}
async handleInput ( source , action , parameter ) {
if ( ! this . isValidSourceForMode ( source , this . currentMode ) ) {
let warningTxt = ` Source ' ${ source } ' is not valid for mode ' ${ this . currentMode } '. ` ;
this . logger . warn ( warningTxt ) ;
return { status : false , feedback : warningTxt } ;
}
this . logger . debug ( "hello" ) ;
//sanitize input
if ( typeof action !== 'string' ) { this . logger . error ( ` Action must be string ` ) ; return ; }
//convert to lower case to avoid to many mistakes in commands
action = action . toLowerCase ( ) ;
// check for validity of the request
if ( ! this . isValidActionForMode ( action , this . currentMode ) ) { return ; }
if ( ! this . isValidSourceForMode ( source , this . currentMode ) ) { return ; }
this . logger . debug ( "hello2" ) ;
this . logger . info ( ` Handling input from source ' ${ source } ' with action ' ${ action } ' in mode ' ${ this . currentMode } '. ` ) ;
try {
switch ( action ) {
case "execSequence" :
case "execsequence" :
return await this . executeSequence ( parameter ) ;
case "execM ovement" :
case "execm ovement" :
return await this . setpoint ( parameter ) ;
case "flowMovement " :
case "entermaintenance " :
return await this . executeSequence ( parameter ) ;
case "exitmaintenance" :
return await this . executeSequence ( parameter ) ;
case "flowmovement" :
// Calculate the control value for a desired flow
const pos = this . calcCtrl ( parameter ) ;
// Move to the desired setpoint
return await this . setpoint ( pos ) ;
case "emergencyS top" :
case "emergencys top" :
this . logger . warn ( ` Emergency stop activated by ' ${ source } '. ` ) ;
return await this . executeSequence ( "emergencyStop" ) ;
case "statusC heck" :
case "statusc heck" :
this . logger . info ( ` Status Check: Mode = ' ${ this . currentMode } ', Source = ' ${ source } '. ` ) ;
break ;
@@ -500,7 +604,6 @@ class Machine {
// NEW: Flow handler
updateMeasuredFlow ( value , position , context = { } ) {
if ( ! this . _isOperationalState ( ) ) {
this . logger . warn ( ` Machine not operational, skipping flow update from ${ context . childName || 'unknown' } ` ) ;
return ;
@@ -508,28 +611,19 @@ class Machine {
this . logger . debug ( ` Flow update: ${ value } at ${ position } from ${ context . childName || 'child' } ` ) ;
if ( this . upstreamSource && this . downstreamSink ) {
this . _updateSourceSink ( ) ;
}
// Store in parent's measurement container
this . measurements . type ( "flow" ) . variant ( "measured" ) . position ( position ) . value ( value , context . timestamp , context . unit ) ;
// Update predicted flow if you have prediction capability
if ( this . predictFlow ) {
this . measurements . type ( "flow" ) . variant ( "predicted" ) . position ( "atEquipment " ) . value ( this . predictFlow . outputY || 0 ) ;
this . measurements . type ( "flow" ) . variant ( "predicted" ) . position ( "downstream " ) . value ( this . predictFlow . outputY || 0 ) ;
}
}
_updateSourceSink ( ) {
// Handles flow according to the configured "flow number"
this . logger . debug ( ` Updating source-sink pair: ${ this . upstreamSource . config . functionality . softwareType } - ${ this . downstreamSink . config . functionality . softwareType } ` ) ;
this . downstreamSink . setInfluent = this . upstreamSource . getEffluent [ this . config . flowNumber ] ;
}
// Helper method for operational state check
_isOperationalState ( ) {
const state = this . state . getCurrentState ( ) ;
this . logger . debug ( ` Checking operational state ${ this . state . getCurrentState ( ) } ? ${ [ "operational" , "accelerating" , "decelerating" ] . includes ( state ) } ` ) ;
return [ "operational" , "accelerating" , "decelerating" ] . includes ( state ) ;
}
@@ -555,6 +649,136 @@ class Machine {
}
}
/////////////////////////////
/**
* Compute a single drift score in [0..1] using predicted vs measured series.
* Uses min/max of the *predicted* window as normalization range.
* If no usable data -> returns 0 (neutral).
*/
_computeDriftScore ( ) {
try {
const metrics = [
{ key : "pressure" , pos : "downstream" } ,
{ key : "flow" , pos : "downstream" } ,
{ key : "power" , pos : "atEquipment" }
] ;
const values = [ ] ;
for ( const m of metrics ) {
const pred = this . measurements . type ( m . key ) . variant ( "predicted" ) . position ( m . pos ) . getAllValues ( ) ? . values ;
const meas = this . measurements . type ( m . key ) . variant ( "measured" ) . position ( m . pos ) . getAllValues ( ) ? . values ;
if ( ! Array . isArray ( pred ) || ! Array . isArray ( meas ) || pred . length < 2 || meas . length < 2 ) continue ;
const expectedMin = Math . min ( ... pred ) ;
const expectedMax = Math . max ( ... pred ) ;
if ( ! Number . isFinite ( expectedMin ) || ! Number . isFinite ( expectedMax ) || expectedMax === expectedMin ) continue ;
const drift = this . errorMetrics . assessDrift ( pred , meas , expectedMin , expectedMax ) ;
if ( Number . isFinite ( drift ) ) {
// assessDrift is already normalized; keep it in [0..1]
values . push ( Math . max ( 0 , Math . min ( 1 , Math . abs ( drift ) ) ) ) ;
}
}
if ( values . length === 0 ) return 0 ; // neutral if no data
const avg = values . reduce ( ( s , v ) => s + v , 0 ) / values . length ;
return Math . max ( 0 , Math . min ( 1 , avg ) ) ;
} catch ( e ) {
this . logger ? . warn ? . ( ` Drift score error: ${ e . message } ` ) ;
return 0 ;
}
}
_calculateAssetHealthIndex ( ) {
try {
// 1) Hard fail -> worst health
// if (this.state?.getCurrentState && this.state.getCurrentState() === "failed")
if ( [ "off" ] . includes ( this . state ? . getCurrentState ? . ( ) ) ) {
this . assetHealth . index = 5 ;
return 5 ;
}
// 2) Inputs (clamped to 0..1)
const availability = typeof this . kpi ? . availability === 'number' ? this . kpi . availability : 1 ;
const unavailability = 1 - Math . max ( 0 , Math . min ( 1 , availability ) ) ;
const effPenalty = Math . max ( 0 , Math . min ( 1 , typeof this . relDistFromPeak === 'number' ? this . relDistFromPeak : 0 ) ) ;
const driftScore = this . _computeDriftScore ( ) ; // 0..1
// 3) Blend (weights sum to 1.0)
// Tweak these if you like: e.g. make drift more/less important.
const wAvail = 0.4 ; // unavailability weight
const wDrift = 0.4 ; // drift weight
const wEff = 0.2 ; // efficiency distance weight
const score01 = ( wAvail * unavailability ) + ( wDrift * driftScore ) + ( wEff * effPenalty ) ;
// 4) Scale to 0..5 integer, clamp
const index = Math . max ( 0 , Math . min ( 5 , Math . round ( score01 * 5 ) ) ) ;
this . assetHealth . index = index ;
return index ;
} catch ( err ) {
this . logger ? . error ? . ( ` AHI calc error: ${ err . message } ` ) ;
this . assetHealth . index = 0 ;
return 0 ;
}
}
_handleStateChangeForKPI ( newState ) {
const now = Date . now ( ) ;
const runtime = this . state . getRunTimeHours ( ) ;
const lastState = this . state . getPreviousState ? . ( ) || "unknown" ;
// --- Treat OFF as failure and start of downtime ---
if ( newState === "off" ) {
this . kpi . failures ++ ; // always count a new failure when OFF
this . kpi . lastFailureTime = now ; // mark the start of downtime
this . logger . warn ( ` Machine OFF (counted as failure). Total failures: ${ this . kpi . failures } ` ) ;
}
// --- When we leave OFF and become OPERATIONAL, book downtime ---
if ( newState === "operational" ) {
// Only calculate downtime if we had an OFF period before
if ( this . kpi . lastFailureTime != null ) {
const downtimeHours = ( now - this . kpi . lastFailureTime ) / 3600000 ;
this . kpi . totalDowntimeHours += downtimeHours ;
this . kpi . lastRepairTime = now ; // moment of "repaired"
this . kpi . lastFailureTime = null ; // close downtime window
this . logger . info ( ` OFF → OPERATIONAL. Added ${ downtimeHours . toFixed ( 2 ) } h downtime. ` ) ;
}
}
// --- Compute KPI Metrics ---
const failures = this . kpi . failures ;
const downtime = this . kpi . totalDowntimeHours ;
// If no failures yet: MTBF = total runtime; MTTR = 0
this . kpi . MTBF = failures > 0 ? runtime / failures : runtime ;
this . kpi . MTTR = failures > 0 ? downtime / failures : 0 ;
// --- Compute Availability ---
const mtbf = this . kpi . MTBF ? ? 0 ;
const mttr = this . kpi . MTTR ? ? 0 ;
if ( mtbf <= 0 && mttr <= 0 ) {
this . kpi . availability = 1 ; // Default: 100% if no data
} else {
const availability = mtbf / ( mtbf + mttr ) ;
this . kpi . availability = Math . min ( 1 , Math . max ( 0 , availability ) ) ; // clamp 0– 1
}
this . logger . debug (
` KPI updated — MTBF: ${ this . kpi . MTBF . toFixed ( 2 ) } h, MTTR: ${ this . kpi . MTTR . toFixed ( 2 ) } h, ` +
` Availability: ${ ( this . kpi . availability * 100 ) . toFixed ( 2 ) } % `
) ;
}
//////////////////////////////////////////////
calcDistanceFromPeak ( currentEfficiency , peakEfficiency ) {
return Math . abs ( currentEfficiency - peakEfficiency ) ;
}
@@ -583,7 +807,6 @@ class Machine {
} ;
}
// Calculate the center of gravity for current pressure
calcCog ( ) {
@@ -699,23 +922,12 @@ class Machine {
// Improved output object generation
const output = { } ;
//build the output object
this . measurements . getTypes ( ) . forEach ( type => {
this . measurements . getVariants ( type ) . forEach ( variant => {
const downstreamVal = this . measurements . type ( type ) . variant ( variant ) . position ( "downstream" ) . getCurrentValue ( ) ;
const upstreamVal = thi s . measurements . type ( type ) . variant ( variant ) . position( "upstream" ) . getCurrentValue ( ) ;
if ( downstreamVal ! = null ) {
output [ ` downstream_ ${ variant } _ ${ type } ` ] = downstreamVal ;
}
if ( upstreamVal != null ) {
output [ ` upstream_ ${ variant } _ ${ type } ` ] = upstreamVal ;
}
if ( downstreamVal != null && upstreamVal != null ) {
const diffVal = this . measurements . type ( type ) . variant ( variant ) . difference ( ) . value ;
output [ ` differential_ ${ variant } _ ${ type } ` ] = diffVal ;
}
Object . entries ( this . measurements . measurements ) . forEach ( ( [ type , variants ] ) => {
Object . entrie s( variants ) . forEach ( ( [ variant , positions ] ) => {
Object . entries ( positions ) . forEach ( ( [ position , measurement ] ) => {
output [ ` ${ type } . ${ variant } . ${ position } ` ] = measurement . getCurrentValue ( ) ;
} ) ;
} ) ;
} ) ;
@@ -728,6 +940,14 @@ class Machine {
output [ "cog" ] = this . cog ; // flow / power efficiency
output [ "NCog" ] = this . NCog ; // normalized cog
output [ "NCogPercent" ] = Math . round ( this . NCog * 100 * 100 ) / 100 ;
output [ "kpi_MTBF" ] = this . kpi . MTBF ;
output [ "kpi_MTTR" ] = this . kpi . MTTR ;
output [ "kpi_assetAvailability" ] = Math . round ( this . kpi . availability * 100 * 100 ) / 100 ;
output [ "kpi_totalFailuresCount" ] = this . kpi . failures ;
output [ "asset_tag_number" ] = 'L001' ;
// output["asset_tag_number"] = this.assetTagNumber;
output [ "maintenanceTime" ] = this . state . getMaintenanceTimeHours ( ) ;
if ( this . flowDrift != null ) {
const flowDrift = this . flowDrift ;
@@ -742,6 +962,21 @@ class Machine {
output [ "effRelDistFromPeak" ] = this . relDistFromPeak ;
//this.logger.debug(`Output: ${JSON.stringify(output)}`);
/////////////////////////////////
// this._calculateAssetHealthIndex();
// output["assetHealthIndex"] = this.assetHealth.index;
this . _calculateAssetHealthIndex ( ) ;
output [ "assetHealthIndex" ] = this . assetHealth . index ;
// 0 = darkgreen, 1 = green, 2 = yellow, 3 = orange, 4 = red, 5 = darkred
// const healthColors = ["darkgreen", "green", "yellow", "orange", "red", "darkred"];
const healthColors = [ "#006400" , "#008000" , "#FFFF00" , "#FFA500" , "#FF0000" , "#8B0000" ] ;
output [ "assetHealthColor" ] = healthColors [ this . assetHealth . index ] || "unknown" ;
//////////////////////////
return output ;
}
@@ -751,9 +986,9 @@ class Machine {
module . exports = Machine ;
/*------------------- Testing -------------------*/
/*
curve = require('C:/Users/zn375/.node-red/public/fallbackData.json');
/*
//curve = require('C:/Users/zn375/.node-red/public/fallbackData.json');
//import a child
const Child = require('../../measurement/src/specificClass');
@@ -769,7 +1004,6 @@ const PT1 = new Child(config={
},
functionality:{
softwareType:"measurement",
positionVsParent:"upstream",
},
asset:{
supplier:"Vega",
@@ -791,7 +1025,6 @@ const PT2 = new Child(config={
},
functionality:{
softwareType:"measurement",
positionVsParent:"upstream",
},
asset:{
supplier:"Vega",
@@ -807,18 +1040,17 @@ console.log(`Creating machine...`);
const machineConfig = {
general: {
name: "Hy drostal",
name: "Hi drostal",
logging: {
enabled: true,
logLevel: "debug",
}
},
asset: {
supplier: "Hy drostal",
supplier: "Hi drostal",
type: "pump",
category: "centrifugal",
model: "H05K-S03R+HGM1X-X280KO ", // Ensure this field is present.
machineCurve: curve["machineCurves"]["Hydrostal"]["H05K-S03R+HGM1X-X280KO"],
model: "hidrostal-H05K-S03R ", // Ensure this field is present.
}
}
@@ -847,18 +1079,17 @@ const machine = new Machine(machineConfig, stateConfig);
machine.logger.info(`Registering child...`);
machine.childRegistrationUtils.registerChild(PT1, "upstream");
machine.childRegistrationUtils.registerChild(PT2, "downstream");
//feed curve to the machine class
//machine.updateCurve(curve["machineCurves"]["Hydrostal"]["H05K-S03R+HGM1X-X280KO"]);
/*
PT1.logger.info(`Enable sim...`);
PT1.toggleSimulation();
PT2.logger.info(`Enable sim...`);
PT2.toggleSimulation();
machine.getOutput();
*/
//manual test
//machine.handleInput("parent", "execSequence", "startup");
/*
machine.measurements.type("pressure").variant("measured").position('upstream').value(-200);
machine.measurements.type("pressure").variant("measured").position('downstream').value(1000);
@@ -868,8 +1099,8 @@ const tickLoop = setInterval(changeInput,1000);
function changeInput(){
PT1.logger.info(`tick...`);
PT1.tick();
PT2.tick();
// PT1.tick();
// PT2.tick();
}
async function testingSequences(){