Data Reconciliation and Steady-State Detection

The neqsim.process.util.reconciliation package provides:

A weighted least squares (WLS) data reconciliation engine that adjusts plant measurements so that mass (and optionally energy) balance constraints are exactly satisfied.
A steady-state detector (SSD) based on the R-statistic method that monitors process variables and determines when the plant has reached steady state — a prerequisite for meaningful reconciliation.

Running the Examples

Python blocks labeled complete examples run independently with neqsim installed. The live demonstration uses 35 synthetic scans and terminates. Helper-function blocks in the four-stage pipeline must be run in order before calling the helpers. They do not contact a historian or update plant controls.

Java snippets use neqsim.process.util.reconciliation.*, java.util.*, neqsim.thermo.system.*, neqsim.process.equipment.stream.Stream, neqsim.process.equipment.separator.Separator, and neqsim.process.processmodel.ProcessSystem. Put imports above the class and statements inside main(String[] args). Run alternatives in separate scopes. The named and coefficient-array constraints are alternative definitions.

Steady-State Detection
Data Reconciliation
Overview
When to Use Data Reconciliation
Mathematical Background
Architecture
Step-by-Step Usage
Working with Model (Tuned) Variables
Complete Java Example
Complete Python Example
API Reference
Uncertainty Guidelines
Multi-Node Network Example
Gross Error Elimination
Integration with ProcessSystem
Building Live NeqSim Models (Python)
JSON and Text Reports
Troubleshooting
Related Documentation

Steady-State Detection

Why Detect Steady State?

Data reconciliation assumes that the measured values represent a single operating point governed by conservation laws. If the plant is actively transitioning (e.g., a rate change, startup, or upset), reconciling transient data produces meaningless results. Steady-state detection (SSD) answers the question: Are the current readings stable enough to reconcile?

The R-Statistic Method

The detector uses the Cao-Rhinehart R-statistic (Cao & Rhinehart, 1995) — a ratio of the filtered variance to the unfiltered variance:

\[R = \frac{\sigma^2_f}{\sigma^2_u}\]

Where:

Filtered variance $\sigma^2f$ — computed from successive differences: $\frac{1}{2(n-1)} \sum{i=2}^{n} (x_i - x_{i-1})^2$. This captures sample-to-sample noise.
Unfiltered variance $\sigma^2u$ — the ordinary sample variance: $\frac{1}{n-1} \sum{i=1}^{n} (x_i - \bar{x})^2$. This captures both noise and any trend.

Interpretation:

R value	Meaning
R close to 1.0	White noise only — steady state
R much less than 1.0	Trend, drift, or step change — transient
R greater than 1.0	Oscillation or alternating pattern

The default threshold is R ≥ 0.5. Optional supplementary tests:

Test	Purpose	Default
Slope test	Catches slow monotonic drift that R might miss	Disabled (threshold = 0)
Std.dev test	Rejects signals that are technically “steady” but too noisy	Disabled (threshold = 0)

SSD Step-by-Step Usage

Step 1 — Create the detector

Java output uses Log4j2. Declare this field inside your example class: private static final org.apache.logging.log4j.Logger logger = org.apache.logging.log4j.LogManager.getLogger("OptimizationExample");.

// Window of 30 samples, R-threshold 0.5
SteadyStateDetector detector = new SteadyStateDetector(30);
detector.setRThreshold(0.5);

Step 2 — Register variables

// By name (uses default window size)
detector.addVariable("TI-2001");

// Flow variable with explicit window and uncertainty
SteadyStateVariable v = new SteadyStateVariable("FI-1001", 30);
v.setUnit("kg/hr");
v.setUncertainty(20.0); // needed if bridging to reconciliation
detector.addVariable(v);

Step 3 — Feed data (streaming loop)

// Synthetic scan; replace these values with your historian readings:
detector.updateVariable("FI-1001", 1000.0);
detector.updateVariable("TI-2001", 25.0);

Or update all at once:

Map<String, Double> snapshot = new LinkedHashMap<String, Double>();
snapshot.put("FI-1001", 1000.0);
snapshot.put("TI-2001", 25.0);
detector.updateAll(snapshot);

Step 4 — Evaluate

SteadyStateResult result = detector.evaluate();

if (result.isAtSteadyState()) {
    logger.info("Plant is at steady state — eligible for reconciliation checks");
} else {
    logger.info("Transient variables: " + result.getTransientVariables());
}

Or combine update + evaluate:

SteadyStateResult result = detector.updateAndEvaluate(snapshot);

Step 5 — Read results

// Per-variable diagnostics
for (SteadyStateVariable v : result.getVariables()) {
    logger.info(String.format("%-12s  R=%.3f  mean=%.1f  steady=%s%n",
        v.getName(), v.getRStatistic(), v.getMean(), v.isAtSteadyState()));
}

// Reports
logger.info(result.toReport()); // formatted table
String json = result.toJson();          // machine-readable

SSD Java Example

import neqsim.process.util.reconciliation.*;

// Create detector
SteadyStateDetector ssd = new SteadyStateDetector(30);
ssd.setRThreshold(0.5);
ssd.setSlopeThreshold(0.5);   // Optional: catch slow drifts

// Register variables with uncertainties
SteadyStateVariable feed = new SteadyStateVariable("feed_flow", 30);
feed.setUnit("kg/hr").setUncertainty(20.0);
ssd.addVariable(feed);

SteadyStateVariable gas = new SteadyStateVariable("gas_flow", 30);
gas.setUnit("kg/hr").setUncertainty(15.0);
ssd.addVariable(gas);

SteadyStateVariable liquid = new SteadyStateVariable("liquid_flow", 30);
liquid.setUnit("kg/hr").setUncertainty(10.0);
ssd.addVariable(liquid);

// Simulate 30 readings at steady state
java.util.Random noise = new java.util.Random(42);
for (int i = 0; i < 30; i++) {
    ssd.updateVariable("feed_flow", 1000.0 + (noise.nextDouble() - 0.5) * 2);
    ssd.updateVariable("gas_flow", 605.0 + (noise.nextDouble() - 0.5) * 2);
    ssd.updateVariable("liquid_flow", 398.0 + (noise.nextDouble() - 0.5) * 2);
}

SteadyStateResult ssResult = ssd.evaluate();
logger.info(ssResult.toReport());

if (ssResult.isAtSteadyState()) {
    // Bridge directly to reconciliation
    DataReconciliationEngine engine = ssd.createReconciliationEngine();
    engine.addMassBalanceConstraint("separator",
        new String[]{"feed_flow"},
        new String[]{"gas_flow", "liquid_flow"});
    ReconciliationResult recResult = engine.reconcile();
    logger.info(recResult.toReport());
}

SSD Python Example

from neqsim import jneqsim

SteadyStateDetector = jneqsim.process.util.reconciliation.SteadyStateDetector
SteadyStateVariable = jneqsim.process.util.reconciliation.SteadyStateVariable

# Create detector with window=30
ssd = SteadyStateDetector(30)
ssd.setRThreshold(0.5)

# Register variables
feed = SteadyStateVariable("feed_flow", 30)
feed.setUnit("kg/hr").setUncertainty(20.0)
ssd.addVariable(feed)

gas = SteadyStateVariable("gas_flow", 30)
gas.setUnit("kg/hr").setUncertainty(15.0)
ssd.addVariable(gas)

# Push 30 constant-ish readings
import random
random.seed(42)
for i in range(30):
    ssd.updateVariable("feed_flow", 1000.0 + random.uniform(-1, 1))
    ssd.updateVariable("gas_flow", 600.0 + random.uniform(-1, 1))

result = ssd.evaluate()
print(result.toReport())

if result.isAtSteadyState():
    engine = ssd.createReconciliationEngine()
    # This detector monitors feed and gas only; a liquid meter is needed for a
    # separator mass balance. The complete four-flow example below adds all outlets.

SSD API Reference

SteadyStateDetector

Method	Description
`SteadyStateDetector(int windowSize)`	Create detector with given default window size
`addVariable(SteadyStateVariable v)`	Register a pre-configured variable
`addVariable(String name)`	Register by name using default window; returns created variable
`removeVariable(String name)`	Unregister a variable; returns true if found
`getVariable(String name)`	Get variable by name (null if not found)
`getVariableCount()`	Number of registered variables
`updateVariable(String name, double value)`	Push a new sample for one variable
`updateAll(Map name, Double value)`	Push new samples for all variables
`evaluate()`	Evaluate all variables; returns `SteadyStateResult`
`updateAndEvaluate(Map)`	Convenience: updateAll + evaluate
`createReconciliationEngine()`	Build a `DataReconciliationEngine` from steady-state variables
`setRThreshold(double)`	Set R-statistic threshold (default 0.5)
`setSlopeThreshold(double)`	Set max absolute slope (0 = disabled)
`setStdDevThreshold(double)`	Set max standard deviation (0 = disabled)
`setRequiredFraction(double)`	Fraction of variables that must be steady (default 1.0)
`setRequireFullWindow(boolean)`	Require window to be full before evaluating (default true)
`clear()`	Remove all variables and reset

SteadyStateVariable

Method	Description
`SteadyStateVariable(String name, int windowSize)`	Create with name and sliding window size (min 3)
`addValue(double value)`	Add a sample; recomputes statistics
`clear()`	Clear all samples
`getMean()`	Window mean
`getStandardDeviation()`	Window standard deviation
`getRStatistic()`	Cao-Rhinehart R-statistic
`getSlope()`	Linear regression slope (per sample)
`isAtSteadyState()`	Whether last evaluation flagged as steady
`getCount()`	Number of samples in window
`getWindowSize()`	Configured window size
`setUnit(String)`	Set engineering unit (fluent)
`setUncertainty(double)`	Set measurement uncertainty for reconciliation (fluent)

SteadyStateResult

Method	Description
`isAtSteadyState()`	Overall SSD verdict
`getSteadyCount()`	Number of steady variables
`getTransientCount()`	Number of transient variables
`getVariables()`	All variables with their per-variable statistics
`getTransientVariables()`	Only the variables that failed the SSD test
`toReport()`	Human-readable formatted text report
`toJson()`	Machine-readable JSON

Tuning the Detector

Parameter	Typical range	Guidance
Window size	20-60	Larger = more stable but slower response. 30 is a good default for 10-second scan intervals (~5 min window)
R-threshold	0.3-0.8	Lower = more tolerant of trends. 0.5 works well for most process variables
Slope threshold	0-1.0	Enable only if you need to catch very slow drifts. Units depend on your variable’s scale
Std.dev threshold	0-inf	Enable to reject signals that are “steady” but too noisy to be useful
Required fraction	0.5-1.0	Set below 1.0 to allow reconciliation even if some variables are still settling

Bridging to Data Reconciliation

The createReconciliationEngine() method creates a DataReconciliationEngine pre-populated with variables that:

Are at steady state (per R-statistic evaluation)
Have defined uncertainty (from setUncertainty())

The bridge uses each variable’s window mean as the measurement value and the configured uncertainty as sigma. Transient variables and variables without uncertainty are excluded.

// Continue from SSD Java Example: the detector already has all three meters.
if (ssd.evaluate().isAtSteadyState()) {
    DataReconciliationEngine engine = ssd.createReconciliationEngine();
    engine.addMassBalanceConstraint("node1", new String[] {"feed_flow"},
        new String[] {"gas_flow", "liquid_flow"});
    ReconciliationResult result = engine.reconcile();
    logger.info(result.toReport());
}

Data Reconciliation

Overview

Plant instruments (flow meters, pressure transmitters, temperature sensors) always have measurement errors. Raw readings almost never satisfy the fundamental conservation laws — mass in rarely equals mass out when you add up the meter tags. Data reconciliation corrects these readings by finding the smallest statistically-weighted adjustments that make all balances close exactly.

Key Capabilities

Feature	Description
Weighted Least Squares	Adjustments weighted by 1/sigma² — uncertain meters move more
Linear Constraints	Mass balance, energy balance, or any linear relation A·x = 0
Gross Error Detection	Per-variable normalized residual test flags faulty sensors
Iterative Elimination	Automatically removes worst sensor and re-reconciles
Chi-Square Global Test	Detects if overall measurement quality is acceptable
JSON / Text Reports	Machine-readable and human-readable output formats
EJML Matrix Engine	Uses the EJML `SimpleMatrix` library already in NeqSim

Typical Workflow

Plant DCS/Historian ──► Python (collect tags) ──► Set measurements on Engine
                                                          │
                                                          ▼
                                                   Define constraints
                                                          │
                                                          ▼
                                                   engine.reconcile()
                                                          │
                                                          ▼
                                               Read reconciled values
                                                   Detect bad sensors
                                                   Update ProcessSystem

Online loop: The data collection and scheduling happen externally, typically in Python. The Java engine provides the reconciliation math — you feed it measurements via setters and read back reconciled values.

When to Use Data Reconciliation

Scenario	Recommendation
Mass balance doesn’t close across separator/mixer	Use reconciliation
Need to identify a faulty flow meter	Use gross error detection
Calibrating model parameters to plant data	Use reconciliation first, then BatchParameterEstimator
Adjusting a single variable for a target	Use `Adjuster` instead
Streaming real-time data at high frequency	Collect externally, call `reconcile()` at each interval

Mathematical Background

Given $n$ measurements $\mathbf{y}$ with diagonal covariance $\mathbf{V} = \text{diag}(\sigma_1^2, \ldots, \sigma_n^2)$ and $m$ linear constraints $\mathbf{A} \cdot \mathbf{x} = \mathbf{0}$, the WLS solution is:

\[\hat{\mathbf{x}} = \mathbf{y} - \mathbf{V} \mathbf{A}^T (\mathbf{A} \mathbf{V} \mathbf{A}^T)^{-1} \mathbf{A} \mathbf{y}\]

Objective minimized:

\[J = \sum_{i=1}^{n} \left(\frac{\hat{x}_i - y_i}{\sigma_i}\right)^2\]

Normalized residual for gross error detection:

\[r_i = \frac{\hat{x}_i - y_i}{\sqrt{V_{ii} - V_{ii}^{adj}}}\]

where $V^{adj} = V - V A^T (A V A^T)^{-1} A V$. If $

r_i

$ exceeds a z-threshold (default 1.96 for 95% confidence), the measurement is flagged as a gross error.

Global test: The objective $J$ follows a chi-square distribution with $m$ degrees of freedom under the null hypothesis of no gross errors.

Architecture

neqsim.process.util.reconciliation
├── ReconciliationVariable    — One measured variable (value + sigma + result)
├── ReconciliationResult      — Full result with statistics, JSON, report
├── DataReconciliationEngine  — WLS solver, gross error detection
└── package-info.java         — Package documentation

Class	Responsibility
`ReconciliationVariable`	Holds a single measurement: name, measured value, uncertainty (sigma), reconciled value, unit, optional equipment/property link, normalized residual, gross error flag
`DataReconciliationEngine`	Builds the problem (variables + constraints), solves the WLS system using EJML, runs statistical tests
`ReconciliationResult`	Immutable result container: all variables, objective value, chi-square statistic, degrees of freedom, global test, gross errors list, constraint residuals before/after, compute time, JSON and text report output

Step-by-Step Usage

Step 1 — Define Measurements

Create a ReconciliationVariable for each plant measurement. The constructor takes (name, measuredValue, uncertainty):

// Each variable is one plant tag
ReconciliationVariable feed = new ReconciliationVariable("feed_flow", 1000.0, 20.0);
ReconciliationVariable gas  = new ReconciliationVariable("gas_flow",   620.0, 15.0);
ReconciliationVariable liq  = new ReconciliationVariable("liq_flow",   370.0, 10.0);

Parameters:

name — unique string identifier (e.g. the DCS tag name)
measuredValue — the raw meter reading in engineering units
uncertainty — standard deviation (sigma) of the measurement error in the same units

You can also link a variable to a specific equipment property in a ProcessSystem:

ReconciliationVariable feed = new ReconciliationVariable(
    "feed_flow",          // name
    "HP_Separator",       // equipmentName in ProcessSystem
    "massFlowRate",       // property name
    1000.0,               // measured value
    20.0                  // uncertainty (sigma)
);
feed.setUnit("kg/hr");

Step 2 — Set Measurement Uncertainties

The uncertainty (sigma) is the standard deviation of the measurement error. It controls how much each measurement is allowed to move during reconciliation:

Low sigma → high confidence in reading → small adjustment allowed
High sigma → low confidence → variable absorbs more of the imbalance

// Precise Coriolis meter: sigma = 0.5% of reading
ReconciliationVariable precise = new ReconciliationVariable("coriolis_flow", 1000.0, 5.0);

// Less precise orifice meter: sigma = 2% of reading
ReconciliationVariable rough = new ReconciliationVariable("orifice_flow", 600.0, 12.0);

// Calculated/estimated value: sigma = 5% of reading
ReconciliationVariable estimated = new ReconciliationVariable("estimated_flow", 400.0, 20.0);

Typical uncertainty guidelines (see Uncertainty Guidelines for detailed values).

Step 3 — Define Balance Constraints

Add linear constraints of the form: $\sum_i a_i \cdot x_i = 0$

Option A — Raw coefficient array:

DataReconciliationEngine engine = new DataReconciliationEngine();
engine.addVariable(feed);   // index 0
engine.addVariable(gas);    // index 1
engine.addVariable(liq);    // index 2

// Constraint: feed - gas - liq = 0  (mass balance around separator)
engine.addConstraint(new double[]{1.0, -1.0, -1.0});

The coefficient array has one entry per variable, in the order they were added. Use +1 for inlets, -1 for outlets.

Option B — Named mass balance (recommended):

DataReconciliationEngine namedEngine = new DataReconciliationEngine();
namedEngine.addVariable(feed);
namedEngine.addVariable(gas);
namedEngine.addVariable(liq);

namedEngine.addMassBalanceConstraint("Separator balance",
    new String[]{"feed_flow"},                    // inlet names
    new String[]{"gas_flow", "liq_flow"});        // outlet names

Use either Option A (engine) or Option B (namedEngine); do not add the same variables or balance twice. Option B is equivalent to {1.0, -1.0, -1.0} but self-documenting and less error-prone for large networks.

Step 4 — Run Reconciliation

ReconciliationResult result = engine.reconcile();

if (result.isConverged()) {
    logger.info("Reconciliation successful");
    logger.info("Objective (weighted SSQ): " + result.getObjectiveValue());
} else {
    logger.info("Failed: " + result.getErrorMessage());
}

Step 5 — Read Reconciled Values

After reconciliation, each variable holds its adjusted value:

for (ReconciliationVariable v : result.getVariables()) {
    logger.info(String.format("%-15s  meas=%.1f  rec=%.1f  adj=%.2f %s%n",
        v.getName(),
        v.getMeasuredValue(),
        v.getReconciledValue(),
        v.getAdjustment(),
        v.getUnit()));
}

You can also look up individual variables by name:

double reconciledFeed = engine.getVariable("feed_flow").getReconciledValue();
double feedAdjustment = engine.getVariable("feed_flow").getAdjustment();

Step 6 — Detect Gross Errors

The engine computes a normalized residual for each variable. If $

r_i

> \text{threshold}$ (default 1.96), the variable is flagged:

for (ReconciliationVariable v : result.getVariables()) {
    if (v.isGrossError()) {
        logger.info("GROSS ERROR: " + v.getName()
            + " |r|=" + Math.abs(v.getNormalizedResidual()));
    }
}

// Or check the global test
if (!result.isGlobalTestPassed()) {
    logger.info("WARNING: Global chi-square test failed — possible gross errors");
}

Working with Model (Tuned) Variables

In many online optimization workflows, you compare the reconciled plant values against model-predicted values from a tuned process simulation. The difference between reconciled and model-predicted values highlights where the simulation deviates from reality.

Setting Model Values

After running a NeqSim ProcessSystem simulation, set its predicted values on each variable:

// Build the model explicitly; metadata alone does not create measurement devices.
SystemSrkEos fluid = new SystemSrkEos(298.15, 60.0);
fluid.addComponent("methane", 0.70);
fluid.addComponent("n-decane", 0.30);
fluid.setMixingRule("classic");
Stream modelFeed = new Stream("Feed", fluid);
modelFeed.setFlowRate(1000.0, "kg/hr");
Separator modelSeparator = new Separator("HP Sep", modelFeed);
ProcessSystem process = new ProcessSystem();
process.add(modelFeed);
process.add(modelSeparator);
process.run();
engine.getVariable("feed_flow").setModelValue(modelFeed.getFlowRate("kg/hr"));
engine.getVariable("gas_flow").setModelValue(
    modelSeparator.getGasOutStream().getFlowRate("kg/hr"));
engine.getVariable("liq_flow").setModelValue(
    modelSeparator.getLiquidOutStream().getFlowRate("kg/hr"));

Reading Model vs Reconciled Comparison

for (ReconciliationVariable v : result.getVariables()) {
    if (v.hasModelValue()) {
        double modelDelta = v.getReconciledValue() - v.getModelValue();
        logger.info(String.format("%-15s  reconciled=%.1f  model=%.1f  delta=%.2f%n",
            v.getName(), v.getReconciledValue(), v.getModelValue(), modelDelta));
    }
}

Using Reconciled Values to Tune the Process Model

After reconciliation gives you balanced measurements that have passed the configured quality tests, use those values to update the simulation model parameters:

// After reconciliation:
double validatedFeed = engine.getVariable("feed_flow").getReconciledValue();
double validatedGas  = engine.getVariable("gas_flow").getReconciledValue();

// Update simulation inputs with reconciled values
Stream feedStream = (Stream) process.getUnit("Feed");
feedStream.setFlowRate(validatedFeed, "kg/hr");

// Re-run the simulation with corrected inputs
process.run();

// Compare model outputs to reconciled values to identify where model needs tuning
double modelLiqOut = ((Separator) process.getUnit("HP Sep")).getLiquidOutStream()
    .getFlowRate("kg/hr");
double reconciledLiq = engine.getVariable("liq_flow").getReconciledValue();
double modelError = reconciledLiq - modelLiqOut;
// A large gap may indicate fluid-assay, metering, phase-model, or boundary errors; diagnose before tuning

Full Reconciliation-then-Calibration Workflow

For a complete loop that reconciles measurements and then tunes model parameters, combine with BatchParameterEstimator:

 ┌─────────────────────────────────────────────────┐
 │ 1. Collect plant measurements (Python/DCS)      │
 │ 2. Set measurements on DataReconciliationEngine │
 │ 3. engine.reconcile()                           │
 │ 4. Check gross errors, remove bad sensors       │
 │ 5. Use reconciled values as "truth"             │
 │ 6. Feed into BatchParameterEstimator            │
 │    to tune model parameters (UA, efficiency,    │
 │    k-values, etc.)                              │
 │ 7. Update ProcessSystem with tuned parameters   │
 │ 8. Repeat at next time interval                 │
 └─────────────────────────────────────────────────┘

Complete Java Example

import neqsim.process.util.reconciliation.*;

public class SeparatorReconciliation {
    private static final org.apache.logging.log4j.Logger logger =
        org.apache.logging.log4j.LogManager.getLogger(SeparatorReconciliation.class);
    public static void main(String[] args) {
        // Create engine
        DataReconciliationEngine engine = new DataReconciliationEngine();

        // Add plant measurements: (name, measuredValue, uncertainty)
        engine.addVariable(
            new ReconciliationVariable("feed", 10000.0, 200.0).setUnit("kg/hr"));
        engine.addVariable(
            new ReconciliationVariable("gas", 3500.0, 100.0).setUnit("kg/hr"));
        engine.addVariable(
            new ReconciliationVariable("oil", 4800.0, 150.0).setUnit("kg/hr"));
        engine.addVariable(
            new ReconciliationVariable("water", 1900.0, 80.0).setUnit("kg/hr"));

        // Measurement imbalance: 10000 - 3500 - 4800 - 1900 = -200 kg/hr

        // Define mass balance: feed - gas - oil - water = 0
        engine.addMassBalanceConstraint("3-Phase Separator",
            new String[]{"feed"},
            new String[]{"gas", "oil", "water"});

        // Reconcile
        ReconciliationResult result = engine.reconcile();

        // Print text report
        logger.info(result.toReport());

        // Access individual reconciled values
        double recFeed = engine.getVariable("feed").getReconciledValue();
        double recGas  = engine.getVariable("gas").getReconciledValue();
        double recOil  = engine.getVariable("oil").getReconciledValue();
        double recWater = engine.getVariable("water").getReconciledValue();

        logger.info(String.format("Balance check: %.2f - %.2f - %.2f - %.2f = %.6f%n",
            recFeed, recGas, recOil, recWater,
            recFeed - recGas - recOil - recWater));

        // Check for gross errors
        if (result.hasGrossErrors()) {
            logger.info("*** Gross errors detected in: ");
            for (ReconciliationVariable ge : result.getGrossErrors()) {
                logger.info("  " + ge.getName());
            }
        }

        // Machine-readable output
        logger.info(result.toJson());
    }
}

Complete Python Example

from neqsim import jneqsim

# Import reconciliation classes
ReconciliationVariable = jneqsim.process.util.reconciliation.ReconciliationVariable
DataReconciliationEngine = jneqsim.process.util.reconciliation.DataReconciliationEngine

# Create engine
engine = DataReconciliationEngine()

# Add measurements from plant DCS/historian
engine.addVariable(ReconciliationVariable("feed", 10000.0, 200.0).setUnit("kg/hr"))
engine.addVariable(ReconciliationVariable("gas", 3500.0, 100.0).setUnit("kg/hr"))
engine.addVariable(ReconciliationVariable("oil", 4800.0, 150.0).setUnit("kg/hr"))
engine.addVariable(ReconciliationVariable("water", 1900.0, 80.0).setUnit("kg/hr"))

# Mass balance: feed - gas - oil - water = 0
engine.addMassBalanceConstraint("3-Phase Sep",
    ["feed"], ["gas", "oil", "water"])

# Run reconciliation
result = engine.reconcile()
print(result.toReport())

# Read reconciled values
for v in result.getVariables():
    print(f"{str(v.getName()):15s}  meas={v.getMeasuredValue():10.1f}  "
          f"rec={v.getReconciledValue():10.1f}  "
          f"adj={v.getAdjustment():+8.2f}  "
          f"|r|={abs(v.getNormalizedResidual()):6.3f}  "
          f"{'**GE**' if v.isGrossError() else 'ok'}")

# Check if measurements are globally consistent
if result.isGlobalTestPassed():
    print("All measurements consistent (chi-square test passed)")
else:
    print("WARNING: measurement quality issue detected")

Python Online Loop Pattern

import time
from neqsim import jneqsim

ReconciliationVariable = jneqsim.process.util.reconciliation.ReconciliationVariable
DataReconciliationEngine = jneqsim.process.util.reconciliation.DataReconciliationEngine

def get_plant_measurements():
    """Read current measurements from DCS/historian (user implementation)."""
    # Example: read from OPC-UA, PI, IP.21, or CSV
    return {
        "feed": (10050.0, 200.0),   # (value, sigma)
        "gas": (3520.0, 100.0),
        "oil": (4780.0, 150.0),
        "water": (1880.0, 80.0),
    }

# Periodic reconciliation loop
for scan in range(3):
    measurements = get_plant_measurements()

    engine = DataReconciliationEngine()
    for name, (value, sigma) in measurements.items():
        engine.addVariable(ReconciliationVariable(name, float(value), float(sigma)))

    engine.addMassBalanceConstraint("Sep",
        ["feed"], ["gas", "oil", "water"])

    result = engine.reconcile()

    if result.isConverged():
        print(f"OK  obj={result.getObjectiveValue():.3f}  "
              f"gross_errors={result.hasGrossErrors()}")
        # Use reconciled values downstream...
    else:
        print(f"FAILED: {result.getErrorMessage()}")

    # In a scheduled service, call one iteration per historian scan.

API Reference

ReconciliationVariable

Method	Returns	Description
`ReconciliationVariable(name, value, sigma)`	—	Constructor: name, measured value, uncertainty
`ReconciliationVariable(name, equip, prop, value, sigma)`	—	Constructor with equipment/property link
`getName()`	`String`	Variable identifier
`getMeasuredValue()`	`double`	Raw plant reading
`getUncertainty()`	`double`	Standard deviation (sigma)
`getReconciledValue()`	`double`	Adjusted value after reconciliation
`getAdjustment()`	`double`	reconciledValue − measuredValue
`getNormalizedResidual()`	`double`	Statistical test value for gross error detection
`isGrossError()`	`boolean`	True if flagged by normalized residual test
`getModelValue()`	`double`	Model-predicted value (NaN if not set)
`setModelValue(double)`	`void`	Set model prediction for comparison
`hasModelValue()`	`boolean`	Whether a model value was set
`setUnit(String)`	`this`	Engineering unit (fluent)
`setEquipmentName(String)`	`this`	Link to ProcessSystem equipment (fluent)
`setPropertyName(String)`	`this`	Property being measured (fluent)

DataReconciliationEngine

Method	Returns	Description
`addVariable(var)`	`this`	Register a measurement variable
`addConstraint(double[])`	`this`	Add unnamed linear constraint A·x = 0
`addConstraint(double[], name)`	`this`	Add named linear constraint
`addMassBalanceConstraint(name, inlets, outlets)`	`this`	Named mass balance by variable names
`reconcile()`	`ReconciliationResult`	Run WLS reconciliation with gross error detection
`reconcileWithGrossErrorElimination(max)`	`ReconciliationResult`	Iterative elimination of worst sensor
`getVariable(name)`	`ReconciliationVariable`	Look up variable by name
`getVariableCount()`	`int`	Number of registered variables
`getConstraintCount()`	`int`	Number of registered constraints
`setGrossErrorThreshold(z)`	`this`	Set z-value (1.96=95%, 2.576=99%)
`clear()`	`void`	Remove all variables and constraints
`clearConstraints()`	`void`	Remove constraints only, keep variables

ReconciliationResult

Method	Returns	Description
`isConverged()`	`boolean`	Whether reconciliation succeeded
`getObjectiveValue()`	`double`	Weighted sum of squared adjustments
`getChiSquareStatistic()`	`double`	Same as objective — compared against chi-square distribution
`getDegreesOfFreedom()`	`int`	Number of constraints (redundancy)
`isGlobalTestPassed()`	`boolean`	True if objective within chi-square critical value
`getVariables()`	`List`	All variables with reconciled values
`getGrossErrors()`	`List`	Variables flagged as gross errors
`hasGrossErrors()`	`boolean`	Whether any gross errors were detected
`getConstraintResidualsBefore()`	`double[]`	Constraint residuals using raw measurements
`getConstraintResidualsAfter()`	`double[]`	Constraint residuals after reconciliation (near-zero)
`getComputeTimeMs()`	`long`	Execution time in milliseconds
`getErrorMessage()`	`String`	Error description if not converged
`toJson()`	`String`	Full JSON output
`toReport()`	`String`	Formatted text table

Uncertainty Guidelines

Choosing the right uncertainty (sigma) is critical. Here are typical values for common instrument types:

Instrument Type	Typical Accuracy	Sigma as % of Reading
Coriolis flow meter	±0.1–0.5%	0.2–0.5%
Ultrasonic flow meter	±0.5–1.0%	0.5–1.0%
Orifice plate	±1.0–2.0%	1.0–2.0%
Vortex flow meter	±0.5–1.5%	0.5–1.5%
Turbine meter	±0.25–0.5%	0.25–0.5%
Level-inferred flow	±3–5%	3–5%
RTD temperature	±0.1–0.3 °C	Use absolute value
Thermocouple	±1.0–2.5 °C	Use absolute value
Pressure transmitter	±0.1–0.25% FS	0.1–0.25% of full scale
Calculated/estimated	±5–10%	5–10%

Example: computing sigma from instrument spec:

double flowReading = 5000.0;  // kg/hr
double accuracy = 0.01;       // 1% orifice plate
double sigma = flowReading * accuracy;  // 50.0 kg/hr

ReconciliationVariable v = new ReconciliationVariable("FT-101", flowReading, sigma);

Rules of thumb:

If a meter is known to be drifting, increase its sigma (2–3× normal)
For calculated values (not directly measured), use sigma = 5–10% of value
All sigma values must be in the same engineering units as the measured value

Multi-Node Network Example

For process networks with multiple balance points, add one constraint per node:

DataReconciliationEngine engine = new DataReconciliationEngine();

// Node 1: Feed separator
engine.addVariable(new ReconciliationVariable("well_flow",  5000.0, 100.0));
engine.addVariable(new ReconciliationVariable("sep_gas",    2100.0,  50.0));
engine.addVariable(new ReconciliationVariable("sep_liquid", 2850.0,  70.0));

// Node 2: Compressor (gas path)
engine.addVariable(new ReconciliationVariable("comp_out",   2080.0,  50.0));

// Node 3: Pump (liquid path)
engine.addVariable(new ReconciliationVariable("pump_out",   2870.0,  70.0));

// Constraints: one per balance node
engine.addMassBalanceConstraint("Separator",
    new String[]{"well_flow"},
    new String[]{"sep_gas", "sep_liquid"});

engine.addMassBalanceConstraint("Compressor",
    new String[]{"sep_gas"},
    new String[]{"comp_out"});

engine.addMassBalanceConstraint("Pump",
    new String[]{"sep_liquid"},
    new String[]{"pump_out"});

// Solve all balances simultaneously
ReconciliationResult result = engine.reconcile();

// Degrees of freedom = 3 (three constraints, 5 variables → 2 DoF)
logger.info("DoF: " + result.getDegreesOfFreedom());
logger.info(result.toReport());

Important: The system must be over-determined (more variables than constraints) for reconciliation to work. If $n \leq m$, the system is exactly determined or under-determined and the engine returns an error.

Gross Error Elimination

When one sensor has a large systematic bias, simple reconciliation distorts all other readings. Use iterative elimination to automatically identify and downweight the faulty sensor:

// Set a stricter threshold (99% confidence)
engine.setGrossErrorThreshold(2.576);

// Iteratively eliminate up to 2 gross errors
ReconciliationResult result = engine.reconcileWithGrossErrorElimination(2);

// Check which sensors were flagged
for (ReconciliationVariable ge : result.getGrossErrors()) {
    logger.info("Faulty sensor: " + ge.getName()
        + " (normalized residual: " + ge.getNormalizedResidual() + ")");
}

How it works:

Run standard reconciliation
Find the variable with the largest normalized residual exceeding the threshold
Set that variable’s uncertainty to a very large number (1×10¹²) — effectively removing it
Re-reconcile
Repeat until no more gross errors or max eliminations reached
Restore original uncertainties and report all identified gross errors

Integration with ProcessSystem

The reconciliation engine is designed to work alongside a NeqSim process simulation. A typical pattern:

// 1. Build and run process simulation
SystemInterface fluid = new SystemSrkEos(273.15 + 25.0, 60.0);
fluid.addComponent("methane", 0.85);
fluid.addComponent("ethane", 0.10);
fluid.addComponent("propane", 0.05);
fluid.setMixingRule("classic");

Stream feed = new Stream("Feed", fluid);
feed.setFlowRate(10000.0, "kg/hr");

Separator sep = new Separator("HP Sep", feed);

ProcessSystem process = new ProcessSystem();
process.add(feed);
process.add(sep);
process.run();

// 2. Get model predictions
double modelFeed = feed.getFlowRate("kg/hr");
double modelGas  = sep.getGasOutStream().getFlowRate("kg/hr");
double modelLiq  = sep.getLiquidOutStream().getFlowRate("kg/hr");

// 3. Reconcile plant measurements
DataReconciliationEngine engine = new DataReconciliationEngine();
ReconciliationVariable vFeed = new ReconciliationVariable(
    "feed", "Feed", "massFlowRate", 10050.0, 200.0);
ReconciliationVariable vGas = new ReconciliationVariable(
    "gas", "HP Sep gas", "massFlowRate", 6180.0, 120.0);
ReconciliationVariable vLiq = new ReconciliationVariable(
    "liquid", "HP Sep liq", "massFlowRate", 3720.0, 80.0);

engine.addVariable(vFeed);
engine.addVariable(vGas);
engine.addVariable(vLiq);

engine.addMassBalanceConstraint("HP Sep balance",
    new String[]{"feed"}, new String[]{"gas", "liquid"});

ReconciliationResult result = engine.reconcile();

// 4. Compare reconciled vs model
vFeed.setModelValue(modelFeed);
vGas.setModelValue(modelGas);
vLiq.setModelValue(modelLiq);

for (ReconciliationVariable v : result.getVariables()) {
    if (v.hasModelValue()) {
        double gap = v.getReconciledValue() - v.getModelValue();
        logger.info(String.format("%s: reconciled=%.1f, model=%.1f, gap=%.1f%n",
            v.getName(), v.getReconciledValue(), v.getModelValue(), gap));
    }
}

// 5. Update simulation with reconciled inputs for model tuning
feed.setFlowRate(engine.getVariable("feed").getReconciledValue(), "kg/hr");
process.run();

Building Live NeqSim Models (Python)

A “live model” (or digital twin) continuously reads plant data, validates it, and keeps a NeqSim simulation synchronized with reality. In NeqSim the computation engine is Java, but the orchestration layer is Python — Python owns the scheduling, data acquisition (OPC-UA, PI, CSV, database), visualization, and alarm logic, while Java handles the thermodynamics, process simulation, and numerical optimization.

This section describes the optimal architecture for combining the SteadyStateDetector, DataReconciliationEngine, and ProcessSystem in a live Python application.

Architecture — Python Orchestrator, Java Engine

┌───────────────────────────────────────────────────────────────────┐
│  Python Application (scheduling, I/O, dashboards, alerts)        │
│                                                                   │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────────┐  │
│  │  Data     │──▶│  Steady  │──▶│  Data    │──▶│ NeqSim Model │  │
│  │  Source   │   │  State   │   │  Recon   │   │ Update +     │  │
│  │ (OPC/PI)  │   │  Detect  │   │  Engine  │   │ Optimizer    │  │
│  └──────────┘   └──────────┘   └──────────┘   └──────────────┘  │
│       ▲                                              │           │
│       │              via jneqsim (JPype JVM)         ▼           │
│       │                                       Reconciled +       │
│       │                                       Optimized Results  │
│       └──────────────────────────────────────────────────────────┘
│                                                                   │
│  Python libraries: pandas, schedule/APScheduler, opcua, matplotlib│
└───────────────────────────────────────────────────────────────────┘

Why this split?

Layer	Python	Java (via jneqsim)
Data acquisition	OPC-UA, PI SDK, REST APIs, CSV/DB	—
Scheduling	`schedule`, `APScheduler`, `asyncio`	—
SSD	Pushes values to Java detector	`SteadyStateDetector` (R-statistic)
Reconciliation	Reads results as dicts	`DataReconciliationEngine` (WLS)
Process simulation	Sets inputs, calls `run()`	`ProcessSystem`, EOS solvers
Optimization	SciPy, or calls Java optimizer	`ProcessSensitivityAnalyzer`, LM
Dashboards	Plotly, Streamlit, Grafana	—
Alerting	Email, Teams, PagerDuty	—

The Four-Stage Pipeline

Every scan cycle (typically 30-120 seconds) executes four stages:

  ┌───────────┐     ┌──────────┐     ┌──────────────┐     ┌────────────┐
  │ 1. COLLECT │────▶│ 2. SSD   │────▶│ 3. RECONCILE │────▶│ 4. UPDATE  │
  │   plant    │     │  gate    │     │   balance    │     │   model +  │
  │   tags     │     │ (R-stat) │     │   enforce    │     │  optimize  │
  └───────────┘     └──────────┘     └──────────────┘     └────────────┘
       │                 │                  │                    │
    raw tags       steady/transient   reconciled vals      model predictions
    + timestamps    per variable       + gross errors       + KPIs

Stage 2 is the gate — if the process is not at steady state, stages 3 and 4 are skipped and the previous good model state is retained. This prevents the model from chasing transients.

Stage 1 — Collect and Buffer Measurements

import time
from collections import OrderedDict

def read_plant_tags():
    """Read current tag values from your data source.
    Replace this with your OPC-UA / PI / historian reader."""
    return OrderedDict([
        ("FI-1001", 10050.0),   # feed flow, kg/hr
        ("FI-2001",  3520.0),   # gas out flow
        ("FI-3001",  4780.0),   # oil out flow
        ("FI-4001",  1880.0),   # water out flow
        ("TI-1001",    82.3),   # separator temperature, C
        ("PI-1001",    65.2),   # separator pressure, bara
    ])

The data source is entirely Python — OPC-UA (opcua or asyncua), OSIsoft PI (PIconnect), CSV polling, or a database query. NeqSim never touches the I/O layer directly.

Stage 2 — Steady-State Gate

Push each new reading into the SteadyStateDetector and evaluate:

from neqsim import jneqsim

SteadyStateDetector = jneqsim.process.util.reconciliation.SteadyStateDetector
SteadyStateVariable = jneqsim.process.util.reconciliation.SteadyStateVariable

# One-time setup (keep alive across scan cycles)
def create_ssd():
    ssd = SteadyStateDetector(30)   # 30-sample sliding window
    ssd.setRThreshold(0.5)

    # Register all monitored tags with instrument uncertainties
    tags = {
        "FI-1001": {"unit": "kg/hr", "sigma": 200.0},
        "FI-2001": {"unit": "kg/hr", "sigma": 100.0},
        "FI-3001": {"unit": "kg/hr", "sigma": 150.0},
        "FI-4001": {"unit": "kg/hr", "sigma":  80.0},
        "TI-1001": {"unit": "C",     "sigma":   0.5},
        "PI-1001": {"unit": "bara",  "sigma":   0.2},
    }
    for name, info in tags.items():
        v = SteadyStateVariable(name, 30)
        v.setUnit(info["unit"]).setUncertainty(info["sigma"])
        ssd.addVariable(v)

    return ssd

# Per-cycle call
def check_steady_state(ssd, plant_tags):
    """Push new readings and evaluate.
    Returns (is_steady, result_object)."""
    java_map = __import__("jpype").JClass("java.util.LinkedHashMap")()
    for tag, value in plant_tags.items():
        java_map.put(tag, float(value))
    result = ssd.updateAndEvaluate(java_map)
    return result.isAtSteadyState(), result

Key point: The SteadyStateDetector instance is long-lived — it accumulates history across scan cycles. Do not recreate it every cycle.

Stage 3 — Data Reconciliation

Once the SSD gate passes, bridge directly to reconciliation:

def reconcile_measurements(ssd):
    """Bridge SSD to reconciliation engine and solve."""
    engine = ssd.createReconciliationEngine()

    # Add mass balance constraints (separator: feed = gas + oil + water)
    engine.addMassBalanceConstraint(
        "3-Phase Sep",
        ["FI-1001"],                            # inlets
        ["FI-2001", "FI-3001", "FI-4001"]       # outlets
    )

    result = engine.reconcileWithGrossErrorElimination(2)

    if not result.isConverged():
        print(f"Reconciliation failed: {result.getErrorMessage()}")
        return None

    if not result.isGlobalTestPassed() or result.hasGrossErrors():
        print("Measurement quality gate failed; hold the previous model")
        return None
    return result

The createReconciliationEngine() bridge automatically:

Uses window means as measured values (noise filtered)
Carries the configured uncertainties (sigma)
Excludes any variable that was transient or has no uncertainty

Stage 4 — Model Update and Optimization

With reconciled (balanced) values, update the NeqSim process model:

from neqsim import jneqsim

SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
ProcessSystem = jneqsim.process.processmodel.ProcessSystem
Stream = jneqsim.process.equipment.stream.Stream
Separator = jneqsim.process.equipment.separator.ThreePhaseSeparator

# One-time model build
def build_model():
    fluid = SystemSrkEos(273.15 + 80.0, 65.0)
    fluid.addComponent("methane", 0.70)
    fluid.addComponent("ethane", 0.10)
    fluid.addComponent("propane", 0.05)
    fluid.addComponent("n-decane", 0.10)
    fluid.addComponent("water", 0.05)
    fluid.setMixingRule("classic")
    fluid.setMultiPhaseCheck(True)

    feed = Stream("Feed", fluid)
    feed.setFlowRate(10000.0, "kg/hr")
    feed.setTemperature(80.0, "C")
    feed.setPressure(65.0, "bara")

    sep = Separator("HP Sep", feed)

    process = ProcessSystem()
    process.add(feed)
    process.add(sep)
    return process, feed, sep

# Per-cycle update
def update_model(process, feed, sep, rec_result):
    """Push reconciled values into the simulation and re-run."""
    rec_vars = {str(v.getName()): v for v in rec_result.getVariables()}

    # Get reconciled flows
    rec_feed  = rec_vars["FI-1001"].getReconciledValue()
    rec_temp  = rec_vars["TI-1001"].getReconciledValue()
    rec_press = rec_vars["PI-1001"].getReconciledValue()

    # Update simulation inputs
    feed.setFlowRate(float(rec_feed), "kg/hr")
    feed.setTemperature(float(rec_temp), "C")
    feed.setPressure(float(rec_press), "bara")

    # Re-run the process model
    process.run()

    # Extract model predictions for comparison
    model_gas  = sep.getGasOutStream().getFlowRate("kg/hr")
    model_oil  = sep.getOilOutStream().getFlowRate("kg/hr")
    model_water = sep.getWaterOutStream().getFlowRate("kg/hr")

    return {
        "model_gas": model_gas,
        "model_oil": model_oil,
        "model_water": model_water,
    }

Optimization (optional, Stage 4b): If model predictions diverge from reconciled plant values, use a parameter tuning step:

from scipy.optimize import minimize

def tune_model(process, feed, sep, rec_result):
    """Fit two composition fractions under mole-fraction and mass-flow constraints.

    This is a synthetic identifiability example, not proof of a unique fluid assay.
    The component order is methane, ethane, propane, n-decane, water.
    """
    rec_vars = {str(v.getName()): v for v in rec_result.getVariables()}
    original_fluid = feed.getFluid().clone()
    original_mass = feed.getFlowRate("kg/hr")
    rec_mass = rec_vars["FI-1001"].getReconciledValue()
    gas_variable = rec_vars["FI-2001"]
    oil_variable = rec_vars["FI-3001"]

    def objective(params):
        water_fraction = 0.85 - params[0] - params[1]
        if water_fraction < 0.0:
            return 1.0e20
        feed.getFluid().setMolarComposition(
            [params[0], 0.10, 0.05, params[1], water_fraction])
        # Changing molar composition changes molar mass; restore the measured mass rate.
        feed.setFlowRate(float(rec_mass), "kg/hr")
        process.run()
        gas_error = (sep.getGasOutStream().getFlowRate("kg/hr")
                     - gas_variable.getReconciledValue()) / gas_variable.getUncertainty()
        oil_error = (sep.getOilOutStream().getFlowRate("kg/hr")
                     - oil_variable.getReconciledValue()) / oil_variable.getUncertainty()
        return gas_error**2 + oil_error**2

    result = minimize(
        objective, x0=[0.70, 0.10], method="SLSQP",
        bounds=[(0.50, 0.80), (0.05, 0.20)],
        constraints=[{"type": "ineq", "fun": lambda x: 0.85 - x[0] - x[1]}],
        options={"maxiter": 30, "ftol": 1.0e-7})
    if result.success:
        objective(result.x)  # Restore the accepted optimum after numerical probes.
    else:
        feed.setThermoSystem(original_fluid)
        feed.setFlowRate(original_mass, "kg/hr")
        process.run()
    return result

Complete Python Live-Loop Example

This is the recommended end-to-end pattern for a live NeqSim model:

"""
Live NeqSim digital twin — complete four-stage pipeline.

This finite demonstration processes 35 synthetic scans with no delay.
For a service, replace the data reader and schedule one scan at a time.
"""
import time
import json
import logging
from collections import OrderedDict
from neqsim import jneqsim

# ---------- Java imports via jneqsim ----------
SteadyStateDetector = jneqsim.process.util.reconciliation.SteadyStateDetector
SteadyStateVariable = jneqsim.process.util.reconciliation.SteadyStateVariable
DataReconciliationEngine = jneqsim.process.util.reconciliation.DataReconciliationEngine
ReconciliationVariable = jneqsim.process.util.reconciliation.ReconciliationVariable
SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
ProcessSystem = jneqsim.process.processmodel.ProcessSystem
Stream = jneqsim.process.equipment.stream.Stream
Separator = jneqsim.process.equipment.separator.ThreePhaseSeparator

log = logging.getLogger("live_model")
SCAN_INTERVAL = 0  # No delay for the finite demonstration; service interval is site-specific
logging.basicConfig(level=logging.INFO)

# --------- 1. TAG CONFIGURATION ---------
TAG_CONFIG = OrderedDict([
    ("FI-1001", {"desc": "Feed flow",  "unit": "kg/hr", "sigma": 200.0}),
    ("FI-2001", {"desc": "Gas out",    "unit": "kg/hr", "sigma": 100.0}),
    ("FI-3001", {"desc": "Oil out",    "unit": "kg/hr", "sigma": 150.0}),
    ("FI-4001", {"desc": "Water out",  "unit": "kg/hr", "sigma":  80.0}),
    ("TI-1001", {"desc": "Sep temp",   "unit": "C",     "sigma":   0.5}),
    ("PI-1001", {"desc": "Sep press",  "unit": "bara",  "sigma":   0.2}),
])

# --------- 2. BUILD OBJECTS (once) ---------

# SSD detector
ssd = SteadyStateDetector(30)
ssd.setRThreshold(0.5)
for tag, cfg in TAG_CONFIG.items():
    v = SteadyStateVariable(tag, 30)
    v.setUnit(cfg["unit"]).setUncertainty(cfg["sigma"])
    ssd.addVariable(v)

# Process model
fluid = SystemSrkEos(273.15 + 80.0, 65.0)
fluid.addComponent("methane", 0.70)
fluid.addComponent("ethane", 0.10)
fluid.addComponent("propane", 0.05)
fluid.addComponent("n-decane", 0.10)
fluid.addComponent("water", 0.05)
fluid.setMixingRule("classic")
fluid.setMultiPhaseCheck(True)

feed = Stream("Feed", fluid)
feed.setFlowRate(10000.0, "kg/hr")
feed.setTemperature(80.0, "C")
feed.setPressure(65.0, "bara")

sep = Separator("HP Sep", feed)

process = ProcessSystem()
process.add(feed)
process.add(sep)
process.run()  # initial steady-state solve

log.info("Live model initialized")

# --------- 3. MAIN LOOP ---------

synthetic_tags = {
    "FI-1001": feed.getFlowRate("kg/hr") * 1.005,
    "FI-2001": sep.getGasOutStream().getFlowRate("kg/hr") * 1.003,
    "FI-3001": sep.getOilOutStream().getFlowRate("kg/hr") * 0.997,
    "FI-4001": sep.getWaterOutStream().getFlowRate("kg/hr") * 1.006,
    "TI-1001": feed.getTemperature("C"),
    "PI-1001": feed.getPressure("bara"),
}

def read_plant_tags():
    """Synthetic readings derived once from the initial model, with meter offsets."""
    return dict(synthetic_tags)

last_good_result = None
successful_updates = 0

for scan in range(35):
    try:
        # Stage 1: Collect
        tags = read_plant_tags()

        # Stage 2: SSD gate
        java_map = __import__("jpype").JClass("java.util.LinkedHashMap")()
        for tag, value in tags.items():
            java_map.put(tag, float(value))
        ss_result = ssd.updateAndEvaluate(java_map)

        if not ss_result.isAtSteadyState():
            transient_names = [v.getName()
                               for v in ss_result.getTransientVariables()]
            log.info("Transient — skipping (%s)", ", ".join(str(name) for name in transient_names))
            time.sleep(SCAN_INTERVAL)
            continue

        # Stage 3: Reconcile
        engine = ssd.createReconciliationEngine()
        engine.addMassBalanceConstraint(
            "3-Phase Sep",
            ["FI-1001"],
            ["FI-2001", "FI-3001", "FI-4001"]
        )
        rec_result = engine.reconcileWithGrossErrorElimination(2)

        if not rec_result.isConverged():
            log.warning("Reconciliation failed: %s",
                        rec_result.getErrorMessage())
            time.sleep(SCAN_INTERVAL)
            continue

        if rec_result.hasGrossErrors():
            for ge in rec_result.getGrossErrors():
                log.warning("Gross error: %s |r|=%.2f",
                            ge.getName(),
                            abs(ge.getNormalizedResidual()))

        if not rec_result.isGlobalTestPassed() or rec_result.hasGrossErrors():
            log.warning("Measurement quality gate failed; holding the previous model")
            continue

        rec_vars = {str(v.getName()): v for v in rec_result.getVariables()}

        # Stage 4: Solve a copy; publish it only after conservation checks.
        candidate = process.copy()
        candidate_feed = candidate.getUnit("Feed")
        candidate_sep = candidate.getUnit("HP Sep")
        candidate_feed.setFlowRate(
            float(rec_vars["FI-1001"].getReconciledValue()),
            "kg/hr")
        candidate_feed.setTemperature(
            float(rec_vars["TI-1001"].getReconciledValue()),
            "C")
        candidate_feed.setPressure(
            float(rec_vars["PI-1001"].getReconciledValue()),
            "bara")
        candidate.run()
        candidate_out = sum(stream.getFlowRate("kg/hr") for stream in [
            candidate_sep.getGasOutStream(), candidate_sep.getOilOutStream(),
            candidate_sep.getWaterOutStream()])
        candidate_in = candidate_feed.getFlowRate("kg/hr")
        import math
        if not math.isfinite(candidate_out) or abs(candidate_in - candidate_out) > 1e-6 * max(1.0, candidate_in):
            raise RuntimeError("Candidate model mass balance failed")
        process, feed, sep = candidate, candidate_feed, candidate_sep

        # Compare model vs reconciled
        model_gas = sep.getGasOutStream().getFlowRate("kg/hr")
        rec_gas = rec_vars["FI-2001"].getReconciledValue()
        gap_pct = abs(model_gas - rec_gas) / max(abs(rec_gas), 1.0e-9) * 100

        log.info("OK  feed=%.0f  gas=%.0f (model=%.0f, gap=%.1f%%)",
                 rec_vars["FI-1001"].getReconciledValue(),
                 rec_gas, model_gas, gap_pct)

        last_good_result = rec_result
        successful_updates += 1

    except Exception as e:
        log.exception("Scan cycle error: %s", e)

    time.sleep(SCAN_INTERVAL)

assert successful_updates > 0, "The synthetic example must pass the steady-state gate"
print(f"Completed {successful_updates} validated model updates")

Design Guidelines for Live Models

1. Object lifetime

Object	Lifetime	Rationale
`SteadyStateDetector`	Application lifetime	Accumulates sliding window history across scans
`ProcessSystem`	Application lifetime	Expensive to build; re-run with updated inputs each cycle
`DataReconciliationEngine`	Per-cycle (disposable)	Created fresh from SSD bridge each cycle
`ReconciliationResult`	Per-cycle	Store `last_good_result` for fallback

2. Scan interval selection

The scan interval determines how often you push a new sample to the SSD and (if steady) reconcile + re-run the model.

Scenario	Scan interval	SSD window	Effective detection window
Fast-changing platform	10 s	30	5 min
Typical offshore separator	30-60 s	30	15-30 min
Slow pipeline or storage	5 min	20	100 min

Rule of thumb: The SSD window should cover 3-5 process time constants to reliably detect transitions.

3. Keep the model simple

A live model should converge in under 2 seconds per cycle. Avoid:

Deep distillation columns (many stages)
Multiple nested recycles
Full multi-phase flash with many components

If the model is complex, consider running the heavy simulation on a coarser schedule (every 5 min) and using a simplified proxy for the fast cycle.

4. Separate flow variables from condition variables

In the reconciliation step, only flow-rate tags participate in mass balance constraints. Temperature and pressure tags are “condition” variables — they do not enter the balance but are still useful for:

SSD evaluation (is the process stable?)
Model input update (feed T and P)

You can either reconcile them with separate energy balance constraints, or simply use their raw (or SSD-filtered mean) values as direct model inputs.

Choosing the Right Update Strategy

NeqSim provides several layers that can be combined. Choose based on your needs:

Strategy	When to use	NeqSim classes
SSD + Reconciliation only	You trust the model structure; just need balanced inputs	`SteadyStateDetector` + `DataReconciliationEngine`
SSD + Reconciliation + Model re-run	Balanced inputs, then predict unmeasured outputs	Above + `ProcessSystem.run()`
SSD + Reconciliation + Parameter tuning	Model predictions diverge; tune composition, UA, etc.	Above + SciPy `minimize` or `ProcessSensitivityAnalyzer`
SSD + Reconciliation + LM optimizer	Formal model calibration with uncertainty	Above + `LevenbergMarquardtOptimizer` (batch)
Direct EnKF (no SSD)	Streaming updates without explicit SSD gate	`EnKFParameterEstimator` handles both detection and update

Recommended starting point: SSD + Reconciliation + Model re-run. This gives you balanced measurements, a validated model, and predicted KPIs with minimal complexity. Add parameter tuning only when the model-vs-plant gap consistently exceeds 5-10%.

Failure Handling and Fallback

def run_cycle(ssd, process, feed, sep, tags):
    """Return a candidate model only after all data and simulation gates pass.

    Reuses check_steady_state, reconcile_measurements, and update_model above.
    The caller replaces its active process only for status == "ok".
    """
    is_steady, ss_result = check_steady_state(ssd, tags)
    if not is_steady:
        return {"status": "transient", "action": "hold_previous_model"}
    rec_result = reconcile_measurements(ssd)
    if rec_result is None or not rec_result.isGlobalTestPassed() or rec_result.hasGrossErrors():
        return {"status": "recon_failed", "action": "hold_previous_model"}

    # Never modify the active model before a successful candidate simulation.
    candidate = process.copy()
    candidate_feed = candidate.getUnit("Feed")
    candidate_sep = candidate.getUnit("HP Sep")
    try:
        outputs = update_model(candidate, candidate_feed, candidate_sep, rec_result)
        import math
        if not all(math.isfinite(value) for value in outputs.values()):
            raise ValueError("Non-finite model output")
        mass_out = sum(outputs.values())
        mass_in = candidate_feed.getFlowRate("kg/hr")
        if abs(mass_out - mass_in) > 1.0e-6 * max(1.0, abs(mass_in)):
            raise ValueError("Model mass balance did not close")
    except Exception:
        return {"status": "model_failed", "action": "hold_previous_model"}
    return {"status": "ok", "result": rec_result, "process": candidate,
            "feed": candidate_feed, "separator": candidate_sep, "outputs": outputs}

The golden rule: If any stage fails, hold the previous good model state. Never push a diverged or unconverged model to downstream consumers (dashboards, optimizers, MPC). Log the failure, alert if it persists for N consecutive cycles, and re-try next scan.

JSON and Text Reports

JSON Output

String json = result.toJson();
// Returns:
// {
//   "converged": true,
//   "objectiveValue": 0.4123,
//   "chiSquareStatistic": 0.4123,
//   "degreesOfFreedom": 1,
//   "globalTestPassed": true,
//   "computeTimeMs": 3,
//   "variables": [
//     {"name": "feed", "measured": 10000.0, "reconciled": 9985.2, ...},
//     {"name": "gas",  "measured": 3500.0,  "reconciled": 3507.1, ...},
//     ...
//   ]
// }

Text Report

String report = result.toReport();
// Returns:
// === Data Reconciliation Report ===
// Converged: true
// Objective (weighted SSQ): 0.4123
// Chi-square statistic: 0.4123 (df=1)
// Global test passed: true
// Compute time: 3 ms
//
// Variable               Measured   Reconciled   Adjustment |r_norm|     Flag
// --------               --------   ----------   ---------- --------     ----
// feed                 10000.0000   9985.2000     -14.8000    0.234       ok
// gas                   3500.0000   3507.1000       7.1000    0.156       ok
// oil                   4800.0000   4795.3000      -4.7000    0.098       ok
// water                 1900.0000   1882.8000     -17.2000    0.312       ok

Troubleshooting

Problem	Cause	Solution
“No variables added”	Called `reconcile()` before adding any variables	Add variables with `addVariable()` first
“No constraints added”	Called `reconcile()` without constraints	Add at least one constraint with `addConstraint()` or `addMassBalanceConstraint()`
“Need more variables than constraints”	More constraints than variables (under-determined)	Add more measurements or remove redundant constraints
All adjustments are zero	Measurements already satisfy constraints exactly	This is correct — no adjustment needed
One variable gets all adjustment	Its uncertainty is much larger than others	Review sigma values — ensure they reflect actual instrument accuracy
Global test fails	Systematic bias or faulty sensor present	Use `reconcileWithGrossErrorElimination()` to identify the culprit
`IllegalArgumentException: Uncertainty must be positive`	Sigma ≤ 0	All uncertainties must be strictly positive
`IllegalArgumentException: Constraint length does not match`	Coefficient array size ≠ number of variables	Ensure constraint array has exactly one entry per registered variable
`IllegalArgumentException: Variable not found`	Name in `addMassBalanceConstraint` doesn’t match any variable	Check variable names match exactly (case-sensitive)
Matrix inversion fails	Singular constraint matrix (redundant constraints)	Check that constraints are linearly independent

Process Optimization Framework — Batch parameter estimation with Levenberg-Marquardt
Optimization Overview — When to use which optimizer
Constraint Framework — Unified ProcessConstraint interface
Batch Studies — Sensitivity analysis with parameter sweeps
External Optimizer Integration — Python/SciPy integration patterns

References

Cao, S. & Rhinehart, R.R. (1995). “An efficient method for on-line identification of steady state.” Journal of Process Control, 5(6), 363-374.
Jiang, T., Chen, B. & He, X. (2003). “Industrial application of wavelet transform to the on-line prediction of side draw qualities of crude unit.” Computers & Chemical Engineering, 27(4), 519-527.
Narasimhan, S. & Jordache, C. (2000). Data Reconciliation and Gross Error Detection. Gulf Publishing.