Neqsim Platform Modeling

NeqSim Production Platform Modeling

Comprehensive patterns for building complete topside process simulations from platform design documents. Derived from production-grade models of 15+ NCS platforms (Åsgard A/B, Troll A/B, Grane, Castberg, Martin Linge, Gudrun, etc.) in the NeqSim-dev-environment.

1. Architecture Overview

1.1 Standard Model Structure

Every production platform model follows the same architecture:

ProcessInput (pydantic) → simulate() function → ProcessSystem → calc_result() → ProcessOutput (pydantic)

1.2 Two Execution Strategies

Strategy A — Manual Iteration (ASGA pattern):

NUMBER_OF_ITERATIONS = 25

operations = ProcessSystem()
# ... build entire flowsheet ...
for _ in range(NUMBER_OF_ITERATIONS):
    operations.run_step()

Best for: models with many recycles that need controlled convergence.

Strategy B — Single Blocking Run (Martin Linge pattern):

operations = ProcessSystem()
# ... build entire flowsheet ...
thread = operations.runAsThread()
thread.join(config.SYNC_REQUEST_TIMEOUT_MS)  # e.g., 120000 ms

if thread.isAlive():
    thread.interrupt()
    raise CalculationTimeout("Timed out")

Best for: models with few recycles where NeqSim's internal solver handles convergence.

Strategy C — ProcessModel.runUntilConverged with flow filters (recommended for multi-area plants with stagnant legs):

plant = ProcessModel()
plant.add("sep train A", sep_a)
# ... add all areas ...

# Stagnant sections (e.g. an HT injection train at zero rate) are not solved at
# all. Manifold, ThrottlingValve, PipeBeggsAndBrills and MultiStreamHeatExchanger
# all honour this and still publish their outlet pressure at zero flow, so a
# downstream mixer keeps a valid pressure boundary.
if inp.injection_gas_rate_ht <= 0.0:
    for area in ("HT injection process A", "HT injection process B"):
        plant.get(area).setSectionLowFlowThreshold(50.0, "kg/hr")

# The plant gate is a MAX over RELATIVE boundary-stream errors, which a dead leg
# dominates (0.007 kg/hr on 0.1 kg/hr = 6.6e-02, vs a real 443 kg/hr on
# 138 t/hr = 3.2e-03). Filter it at the source:
plant.setBoundaryFlowFloor(1.0)                     # drop sub-1 kg/hr streams
converged = plant.runUntilConverged(15, 1e-3, 1.0)  # rel 1e-3 OR abs 1 kg/hr

print(plant.getConvergenceSummary())                # prints absolute Δflow too
for e in plant.getNonConvergedBoundaryStreamErrors():
    print(e.getStreamName(), e.getFlowError(), e.getAbsoluteFlowChange())

Best for: full-platform ProcessModels where some trains are seasonally or scenario-wise inactive.

Gotcha: setSectionLowFlowThreshold() deactivates units for the rest of the solve pass. Never apply it to a section that is only dry on the first recycle iteration (e.g. a JT valve on a separator liquid outlet).

2. Fluid Creation

2.1 Composition with TBP Fractions (Recommended)

Production fluids almost always include heavy ends characterized as TBP (True Boiling Point) fractions. The fluid_creator() pattern accepts a composition dictionary with optional molar mass and density for TBP fractions:

from neqsim import jneqsim

def fluid_creator(composition: dict) -> "SystemInterface":
    """Create a NeqSim fluid from a composition dictionary.

    Args:
        composition: dict with keys:
            - "component_name": list of component names
            - "molar_composition[-]": list of mole fractions
            - "molar_mass[kg/mol]": list (None for defined components)
            - "relative_density[-]": list (None for defined components)
    """
    fluid = jneqsim.thermo.system.SystemSrkEos(273.15 + 15.0, 1.01325)

    names = composition["component_name"]
    molfracs = composition["molar_composition[-]"]
    molar_masses = composition["molar_mass[kg/mol]"]
    rel_densities = composition["relative_density[-]"]

    for i, name in enumerate(names):
        if molar_masses[i] is not None and rel_densities[i] is not None:
            # TBP fraction — heavy hydrocarbon pseudo-component
            fluid.addTBPfraction(
                name, molfracs[i], molar_masses[i], rel_densities[i]
            )
        else:
            # Defined component (methane, ethane, CO2, water, etc.)
            fluid.addComponent(name, molfracs[i])

    fluid.setMixingRule("classic")
    fluid.setMultiPhaseCheck(True)
    fluid.useVolumeCorrection(True)
    return fluid

2.2 Typical NCS Gas Condensate Composition

composition = {
    "component_name": [
        "water", "nitrogen", "CO2", "methane", "ethane",
        "propane", "i-butane", "n-butane", "i-pentane", "n-pentane",
        "2-methylpentane",  # proxy for nC6 fraction
        # TBP fractions for C10+ (named by carbon range)
        "nC10-nC12", "nC13-nC15", "nC16-nC18", "nC19-nC22",
        "nC23-nC26", "nC27-nC30", "nC31-nC34", "nC35-nC38", "nC39+"
    ],
    "molar_composition[-]": [
        0.058, 0.005, 0.037, 0.664, 0.086,
        0.050, 0.008, 0.016, 0.005, 0.006,
        0.006,
        0.010, 0.009, 0.006, 0.011,
        0.005, 0.004, 0.004, 0.003, 0.002
    ],
    "molar_mass[kg/mol]": [
        None, None, None, None, None,
        None, None, None, None, None,
        None,
        0.091, 0.103, 0.117, 0.146,
        0.181, 0.212, 0.248, 0.289, 0.330
    ],
    "relative_density[-]": [
        None, None, None, None, None,
        None, None, None, None, None,
        None,
        0.741, 0.769, 0.789, 0.804,
        0.825, 0.838, 0.849, 0.863, 0.875
    ],
}

2.3 Multi-Well Composition (Martin Linge Pattern)

When a platform receives fluid from multiple wells/fields, define a base fluid template with ALL possible TBP fractions, then clone and adjust per well:

# Base template with all TBP fraction definitions
base_fluid = jneqsim.thermo.system.SystemPrEos(273.15 + 30.0, 65.0)
base_fluid.addComponent("nitrogen", 0.08)
base_fluid.addComponent("CO2", 3.38)
base_fluid.addComponent("methane", 69.83)
# ... light HCs ...
base_fluid.addTBPfraction("C6_FieldA", 0.24, 84.99/1000, 695/1000)
base_fluid.addTBPfraction("C6_FieldB", 0.0, 84.0/1000, 684/1000)
# ... more TBP fractions for each field source ...
base_fluid.setMixingRule("classic")

# Per-well fluids via clone + setMolarComposition
well_fluid_A = base_fluid.clone()
well_fluid_A.setMolarComposition([0.108, 3.379, 69.5, ...])  # FieldA composition

well_fluid_B = base_fluid.clone()
well_fluid_B.setMolarComposition([0.095, 2.100, 72.1, ...])  # FieldB composition

Key rule: All wells must share the same component list (same TBP fraction definitions). Use setMolarComposition() to set zero fractions for components not present in a particular well.

Prefer a shared E300 characterization for multi-reservoir mixing. When two or more reservoirs (e.g. an oil and a gas-condensate field) feed a common topside, build BOTH fluids from a SINGLE PVTsim/E300 characterization file and apply different compositions via setMolarComposition(), rather than hand-building two separate addTBPfraction templates:

e300 = jneqsim.thermo.util.readwrite.EclipseFluidReadWrite
oil = e300.read("shared_characterization.e300")
oil.setMolarComposition([...oil-heavy fractions...])   # same component order as the E300 CNAMES
oil.setMixingRule(2); oil.setMultiPhaseCheck(True)

gas = e300.read("shared_characterization.e300")
gas.setMolarComposition([...condensate-light fractions...])
gas.setMixingRule(2); gas.setMultiPhaseCheck(True)

Why this matters: well-defined pseudo-components from a real characterization carry full ideal-gas Cp coefficients, so they keep finite enthalpy after the clone() + re-flash that happens inside WellFlow/PipeBeggsAndBrills and isenthalpic chokes. Hand-built addTBPfraction pseudo-components can have degenerate Cp coefficients that lose enthalpy on re-flash and produce NaN downstream (e.g. NaN compressor power). A shared file also makes topside mixing consistent because every stream uses the same component basis. Set a nonzero water mole fraction directly in the composition (H2O is a CNAMES entry) instead of calling addWater().

3. Process Building Patterns

3.1 Pre-create Mixers Before Adding Streams

A critical pattern in production models: create StaticMixer objects at the start, then add streams to them as equipment is built. This solves the forward-reference problem (downstream mixer needs to exist before upstream equipment creates its outlet stream).

operations = ProcessSystem()

# Pre-create all mixers/manifolds FIRST (no streams yet)
inlet_oil_mixer = jneqsim.process.equipment.mixer.StaticMixer("Inlet oil mixer")
recomp_gas_mixer = jneqsim.process.equipment.mixer.StaticMixer("Recomp gas mixer")
# A production/gathering/export MANIFOLD is modelled with the Manifold class,
# NOT a Mixer/StaticMixer: it carries header/branch diameters and ALWAYS routes
# downstream through a split stream. For a single destination give it one split
# (setSplitFactors([1.0])) and route getSplitStream(0); for several destinations
# set setSplitFactors([...]) and read getSplitStream(i). getMixedStream() is the
# internal commingled stream (before the split) - for inspection only.
export_manifold = jneqsim.process.equipment.manifold.Manifold("Export manifold")
export_manifold.setHeaderInnerDiameter(0.9)
export_manifold.setSplitFactors([1.0])
recycle_oil_mixer = jneqsim.process.equipment.mixer.StaticMixer("Recycle from export")

# Build upstream equipment...
# Then add their outlets to the pre-created mixers:
inlet_oil_mixer.addStream(oil_heater_test.getOutStream())
inlet_oil_mixer.addStream(oil_heater_prod.getOutStream())
# ... later, when you're ready to use the mixer:
operations.add(inlet_oil_mixer)

Important: Add the mixer to ProcessSystem AFTER all its inlet streams are connected. All addStream() calls must happen before operations.add(mixer).

3.2 Multi-Stage Separation Train

Standard NCS platform separation: HP → MP → LP with oil heating/letdown between stages:

# === First Stage (HP) Separator ===
well_stream = Stream("Well Stream", well_fluid)
well_stream.setFlowRate(process_input.flow_rate, "kg/hr")
well_stream.setTemperature(process_input.inlet_temp, "C")
well_stream.setPressure(process_input.hp_pressure, "bara")
operations.add(well_stream)

hp_separator = ThreePhaseSeparator("HP Separator", well_stream)
operations.add(hp_separator)

# === Oil letdown to Second Stage ===
oil_heater_to_mp = Heater("Oil heater to MP", hp_separator.getLiquidOutStream())
oil_heater_to_mp.setOutTemperature(process_input.mp_temperature, "C")
oil_heater_to_mp.setOutPressure(process_input.mp_pressure, "bara")
operations.add(oil_heater_to_mp)

# Collect oil from multiple sources (see pre-created mixer pattern §3.1)
inlet_oil_mixer.addStream(oil_heater_to_mp.getOutStream())

# === Second Stage (MP) Separator ===
mp_separator = ThreePhaseSeparator("MP Separator", inlet_oil_mixer.getOutletStream())
operations.add(mp_separator)

# === Oil letdown to Third Stage ===
oil_to_lp = Heater("Oil to LP", mp_separator.getLiquidOutStream())
oil_to_lp.setOutTemperature(process_input.lp_temperature, "C")
oil_to_lp.setOutPressure(process_input.lp_pressure, "bara")
operations.add(oil_to_lp)

# === Third Stage (LP) Separator ===
lp_separator = Separator("LP Separator", oil_to_lp.getOutletStream())
operations.add(lp_separator)

# Oil export pump
oil_pump = Pump("Oil Export Pump", lp_separator.getLiquidOutStream())
oil_pump.setOutletPressure(process_input.oil_export_pressure + 1.01325)  # barg to bara
operations.add(oil_pump)

3.2.1 Separator Physical Configuration via MechanicalDesign

Physical dimensions (vessel ID, nozzle sizes), internals (demister type, inlet device), and design parameters (K-factor, retention time) are set via SeparatorMechanicalDesign — NOT directly on the Separator. This follows the same pattern used for wells, pipelines, and compressors in NeqSim.

Call initMechanicalDesign() after the process has been run() so the design calculation has access to process conditions.

# After operations.run():
hp_separator.initMechanicalDesign()
hp_design = hp_separator.getMechanicalDesign()
hp_design.setMaxOperationPressure(85.0)
hp_design.setMaxOperationTemperature(273.15 + 80.0)
hp_design.setGasLoadFactor(0.107)       # K-factor [m/s]
hp_design.setRetentionTime(180.0)       # Liquid retention [s]
hp_design.setInletNozzleID(0.356)       # 14-inch inlet [m]
hp_design.setDemisterType("wire_mesh")
hp_design.readDesignSpecifications()
hp_design.calcDesign()

# Repeat for other separators
mp_separator.initMechanicalDesign()
mp_design = mp_separator.getMechanicalDesign()
mp_design.setMaxOperationPressure(25.0)
mp_design.setGasLoadFactor(0.107)
mp_design.setRetentionTime(120.0)
mp_design.readDesignSpecifications()
mp_design.calcDesign()

3.3 Heater as T/P Setter

A common pattern uses a Heater with both outlet T and P set. This acts as a T/P setter — useful for ensuring streams enter equipment at the correct conditions (especially after mixing or before scrubbers):

tp_setter = Heater("TP Setter 3rd Stage", mixed_stream)
tp_setter.setOutTemperature(process_input.lp_temperature, "C")
tp_setter.setOutPressure(process_input.lp_pressure)
operations.add(tp_setter)

# Use a *GasScrubber class (vertical by default) for scrubbers, NOT Separator
# (horizontal by default). A horizontal vessel derates the gas area by the design
# liquid level (~80%), over-reading getGasLoadFactor() by ~5x for a vertical scrubber.
scrubber = GasScrubber("LP Scrubber", tp_setter.getOutletStream())   # vertical
# (or, if you keep Separator/ThreePhaseSeparator: scrubber.setOrientation("vertical"))
operations.add(scrubber)

Separator class ↔ orientation: Separator / ThreePhaseSeparator default to horizontal; GasScrubber / GasScrubberSimple / NeqGasScrubber (2-phase) and ThreePhaseGasScrubber (3-phase) default to vertical. Pick the class to match the datasheet (VA-tag horizontal, VG-tag vertical) or override with setOrientation(...). Wrong orientation only affects gas-capacity / getGasLoadFactor(), not the flash split, but it silently skews utilisation studies.

4. Recycle Patterns

DEFAULT (do this automatically for every recompression setup): whenever you build a recompression and/or export-compression train, you MUST close the liquid recycle from every suction/export/injection scrubber back to a separator — never leave a scrubber.getLiquidOutStream() unconnected. A dropped scrubber-liquid stream silently discards knocked-out condensate, which under-counts oil/condensate recovery and mis-states the compressor gas load. Apply this without being asked; only deviate if the user explicitly says the scrubber liquid is routed elsewhere (e.g. to a dedicated condensate handling system).

Matching-pressure routing rule (default): route each scrubber's liquid back to the separator whose operating pressure matches that scrubber's suction pressure, via a TP-setter Heater + Recycle tear. This mirrors real plants, where each knock-out drum drains to the stage operating at the same pressure:

Returning every scrubber to the MP separator (as some legacy models do) is only acceptable when all suction scrubbers share that pressure; otherwise use the matching-pressure mapping above so the recycled liquid re-flashes at the correct conditions.

4.1 Scrubber Liquid Recycle (Oil Recovery)

Liquid knocked out in recompression scrubbers is recycled back to the separation train at the matching pressure. The standard pattern creates clone seed streams:

# Pre-create seed streams (cloned from the separator liquid for same composition)
recycle_seed_1 = mp_separator.getLiquidOutStream().clone()
recycle_seed_1.setName("Recycle from 1st stage scrubber")
recycle_seed_1.setFlowRate(1, "kg/hr")  # Small initial flow for convergence
recycle_seed_1.setPressure(process_input.lp_pressure)
recycle_seed_1.setTemperature(process_input.lp_temperature, "C")

recycle_seed_2 = mp_separator.getLiquidOutStream().clone()
recycle_seed_2.setName("Recycle from 2nd stage scrubber")
recycle_seed_2.setFlowRate(1, "kg/hr")
recycle_seed_2.setPressure(process_input.lp_pressure)
recycle_seed_2.setTemperature(process_input.lp_temperature, "C")

# Add seeds to operations AND to the mixer that feeds the LP separator
operations.add(recycle_seed_1)
operations.add(recycle_seed_2)

rec_oil_mixer = StaticMixer("Recycle oil mixer")
rec_oil_mixer.addStream(oil_to_lp.getOutletStream())  # Main oil flow
rec_oil_mixer.addStream(recycle_seed_1)
rec_oil_mixer.addStream(recycle_seed_2)
operations.add(rec_oil_mixer)

# ... later, after building the recompression scrubbers:
# Wire actual scrubber liquid to a Heater (TP setter), then to Recycle object

tp_set_scrub_liq_1 = Heater("TP set scrub liq 1", first_scrubber.getLiquidOutStream())
tp_set_scrub_liq_1.setOutTemperature(process_input.lp_temperature, "C")
tp_set_scrub_liq_1.setOutPressure(process_input.lp_pressure)
operations.add(tp_set_scrub_liq_1)

recycle_oil_1 = Recycle("Recycle oil 1")
recycle_oil_1.addStream(tp_set_scrub_liq_1.getOutletStream())
recycle_oil_1.setOutletStream(recycle_seed_1)
operations.add(recycle_oil_1)

Critical details:

• Seed streams need small but non-zero flow (1 kg/hr) for numerical stability
• Seed T, P must match the mixer/separator they feed into
• Clone the composition from the appropriate location in the process
• The Heater before the Recycle object acts as a T/P equalizer

Reusable helper (recommended): wrap the tear logic so every scrubber is closed the same way. Pre-create one seed per separator stage, mix each seed into that stage's separator inlet, then call the helper for each scrubber:

def close_scrubber_recycle(ops, scrubber, seed_stream, pressure, temperature_C, tag):
    """Return scrubber liquid to the matching-pressure separator via a Recycle tear."""
    tp_setter = Heater("%s scrubber liq TP set" % tag, scrubber.getLiquidOutStream())
    tp_setter.setOutTemperature(temperature_C, "C")
    tp_setter.setOutPressure(pressure, "bara")
    ops.add(tp_setter)
    recycle = Recycle("%s scrubber liq recycle" % tag)
    recycle.addStream(tp_setter.getOutletStream())
    recycle.setOutletStream(seed_stream)   # the pre-created seed mixed into the stage inlet
    recycle.setTolerance(1.0e-4)
    ops.add(recycle)
    return recycle

# Matching-pressure mapping (HP=stage1, MP=stage2, LP=stage3):
close_scrubber_recycle(ops, lp_scrubber,     seed_stage3, lp_pressure, lp_temperature, "recomp LP")
close_scrubber_recycle(ops, mp_scrubber,     seed_stage2, mp_pressure, mp_temperature, "recomp MP")
close_scrubber_recycle(ops, export_scrubber, seed_stage1, hp_pressure, hp_temperature, "export")

4.2 Export/Injection Scrubber Liquid Recycle

By the matching-pressure routing rule above, export/injection scrubber liquids are returned to the stage-1 (HP) separator (their suction sits at the HP header). The shared-mixer-to-MP variant below is a legacy simplification — use it only when the export/injection scrubbers genuinely operate at the MP pressure. For the default behavior, point the TP-setter and the seed stream at the HP/stage-1 separator conditions instead of MP:

# Pre-create mixer for all high-pressure scrubber liquids
mixer_recycle_from_export = StaticMixer("Recycle from export line")

# ... later, add scrubber liquid streams from export/injection/booster:
if has_booster:
    mixer_recycle_from_export.addStream(booster_scrubber.getLiquidOutStream())
if has_export:
    mixer_recycle_from_export.addStream(export_scrubber.getLiquidOutStream())
if has_injection:
    mixer_recycle_from_export.addStream(inj_scrubber_1.getLiquidOutStream())
    mixer_recycle_from_export.addStream(inj_scrubber_2.getLiquidOutStream())

operations.add(mixer_recycle_from_export)

# Recycle back to MP separator via TP setter
tp_set_export_rec = Heater("TP set export rec", mixer_recycle_from_export.getOutletStream())
tp_set_export_rec.setOutTemperature(process_input.mp_temperature, "C")
tp_set_export_rec.setOutPressure(process_input.mp_pressure)
operations.add(tp_set_export_rec)

# Seed stream was added to inlet_oil_mixer earlier
recycle_from_export = Recycle("Recycle from export")
recycle_from_export.addStream(tp_set_export_rec.getOutletStream())
recycle_from_export.setOutletStream(export_recycle_seed)  # Pre-created seed
operations.add(recycle_from_export)

4.3 Recycle Topology Summary

Typical NCS platform has 4-6 recycle loops:

5. Recompression Train Pattern

5.1 Standard Stage (Cooler → Scrubber → Compressor → Anti-surge)

Each recompression stage follows this repeating pattern:

# === Anti-surge seed stream (pre-created for Recycle object) ===
asv_seed = lp_separator.getGasOutStream().clone()
asv_seed.setName("ASV seed R1")
asv_seed.setFlowRate(0.1, "kg/hr")  # Tiny flow for anti-surge recycle
asv_seed.setPressure(process_input.lp_pressure)
operations.add(asv_seed)

# === Mixer: main gas + anti-surge recycle ===
inlet_mixer = StaticMixer("Inlet cooler R1")
inlet_mixer.addStream(asv_seed)
inlet_mixer.addStream(lp_separator.getGasOutStream())
operations.add(inlet_mixer)

# === Pressure drop (piping + cooler shell-side) ===
pdrop = PressureDrop("PD R1 Cooler")
pdrop.setInletStream(inlet_mixer.getOutletStream())
pdrop.setPressureDrop(process_input.r1_cooler_dp, "bara")
operations.add(pdrop)

# === Aftercooler ===
cooler = Heater("R1 Cooler", pdrop.getOutletStream())
cooler.setOutTemperature(process_input.r1_scrubber_temp, "C")
operations.add(cooler)

# === Hydrate temperature measurement ===
hydrate_analyser = HydrateEquilibriumTemperatureAnalyser("Hydrate R1", cooler.getOutletStream())
operations.add(hydrate_analyser)

# === TP Setter before scrubber (ensures correct inlet conditions) ===
tp_set = Heater("TP set R1", cooler.getOutletStream())
tp_set.setOutTemperature(process_input.r1_scrubber_temp, "C")
tp_set.setOutPressure(process_input.r1_scrubber_pressure)
operations.add(tp_set)

# === Scrubber ===
scrubber = Separator("R1 Scrubber", tp_set.getOutletStream())
scrubber.setInternalDiameter(1.9)
operations.add(scrubber)

# === Compressor (T/P control version) ===
compressor = Compressor("R1 Compressor", scrubber.getGasOutStream())
compressor.setUsePolytropicCalc(True)
compressor.setOutletPressure(process_input.r1_outlet_pressure)
compressor.setOutTemperature(process_input.r1_outlet_temp + 273.15)  # Kelvin!
operations.add(compressor)

# === Anti-surge split + valve + recycle ===
asv_split = Splitter("ASV split R1", compressor.getOutletStream())
asv_split.setSplitNumber(2)
asv_split.setFlowRates([-1, asv_mass_flow], "kg/hr")  # -1 = remainder
operations.add(asv_split)

asv_valve = ThrottlingValve("ASV R1", asv_split.getSplitStream(1))
asv_valve.setOutletPressure(process_input.lp_pressure)
operations.add(asv_valve)

recycle_asv = Recycle("Recycle ASV R1")
recycle_asv.addStream(asv_valve.getOutletStream())
recycle_asv.setOutletStream(asv_seed)
operations.add(recycle_asv)

5.2 Compressor Performance Curves (Dual Object Pattern)

Production models use two compressor objects per stage:

1. Control compressor: sets outlet P and T (for converging the process)
2. Curve compressor: uses actual performance map (for monitoring/reporting)

# Performance curves — separate object using the SAME inlet stream
comp_curves = Compressor("R1 Compressor Curves", scrubber.getGasOutStream())
comp_curves.setUsePolytropicCalc(True)

chart_conditions = [1.0, 1.0, 1.0, 1.0]  # Reference conditions multiplier
speeds = [4421, 5684, 6632]               # RPM

# flow[speed_index][point_index] in m3/hr (actual volume flow at suction)
flow = [
    [5758, 6429, 6679],
    [7616, 10079, 11344],
    [10722, 13295, 15046],
]

# head[speed_index][point_index] in kJ/kg (polytropic head)
head = [
    [51.4, 48.5, 47.7],
    [90.5, 84, 75],
    [123, 115, 94],
]

# efficiency[speed_index][point_index] in % (polytropic efficiency)
poly_eff = [
    [80, 80, 79.7],
    [77.5, 80.3, 77.91],
    [75.3, 78.24, 73],
]

comp_curves.getCompressorChart().setCurves(
    chart_conditions, speeds, flow, head, poly_eff
)
comp_curves.setSpeed(process_input.r1_speed)
comp_curves.getCompressorChart().setHeadUnit("kJ/kg")
operations.add(comp_curves)

Optional: Surge curve definition (Martin Linge pattern):

surge_flow = [2770.39, 3199.03, 4395.44]   # m3/hr at each speed
surge_head = [97.63, 135.65, 235.06]        # kJ/kg at surge
comp_curves.getCompressorChart().getSurgeCurve().setCurve(
    chart_conditions, surge_flow, surge_head
)
comp_curves.getAntiSurge().setActive(True)
comp_curves.getAntiSurge().setSurgeControlFactor(1.05)  # 5% safety margin

Optional: Multiple named charts per compressor (vendor vs as-tested vs field-fitted). A Compressor can carry several charts in a CompressorChartLibrary and switch the active one with a single call — useful for revamp/what-if studies and digital twins that keep both the datasheet curve and a historian-fitted curve on the same machine:

comp_curves.addChart("R1-design", design_chart)
comp_curves.addChart("R1-fieldfit", fieldfit_chart)
comp_curves.selectChart("R1-fieldfit")   # sets + enables it, turns on polytropic calc
# comp_curves.getAvailableCharts(); comp_curves.getSelectedChartName()
# comp_curves.getChartLibrary().saveToFile("R1_charts.json")  # shared vendor-curve DB

5.3 Anti-Surge Valve Flow Calculation (Cv-based)

Compute anti-surge valve mass flow from Cv and valve opening:

import math

def get_gas_valve_mass_flow(
    p_upstream_pa: float,
    p_downstream_pa: float,
    density_kgm3: float,
    cv_value: float,
    valve_opening_pct: float,
) -> float:
    """Gas valve mass flow using ISA/IEC valve sizing equation.

    Returns mass flow in kg/hr.
    """
    if valve_opening_pct < 10:  # MIN_VALVE_OPENING
        return 0.1  # Tiny seed flow

    dp = abs(p_upstream_pa - p_downstream_pa)
    n8 = 94.8  # ISA constant for mass flow

    mass_flow = (
        n8 * (valve_opening_pct / 100.0) * cv_value
        * math.sqrt(dp * density_kgm3)
    )
    return max(mass_flow, 0.1)

Apply after first operations.run_step() call (needs actual pressures/densities):

# After initial run, calculate ASV flows from actual conditions
mass_r1 = get_gas_valve_mass_flow(
    operations.getUnit("ASV R1").getInletStream().getPressure("Pa"),
    operations.getUnit("ASV R1").getOutletStream().getPressure("Pa"),
    operations.getUnit("ASV R1").getInletStream().getFluid().getDensity("kg/m3"),
    process_input.cv_asv_r1,
    process_input.asv_opening_r1,
)
operations.getUnit("ASV split R1").setFlowRates([-1, mass_r1], "kg/hr")
operations.getUnit("ASV seed R1").setFlowRate(mass_r1, "kg/hr")

6. Export and Injection Gas Processing

6.1 Production Split

Gas from the recompression train goes to export and/or injection via a splitter:

export_manifold.addStream(r3_output)  # All sources to manifold
operations.add(export_manifold)

# Manifold commingled outlet is getMixedStream() (not getOutletStream()).
production_split = Splitter("Prod split", export_manifold.getSplitStream(0))
production_split.setSplitFactors([split_export, split_injection])
operations.add(production_split)

6.2 Conditional Export/Injection Sections

Production models typically support switching export/injection on/off:

MIN_SPLIT = 0.05

if split_export > MIN_SPLIT:
    # Export cooler → scrubber → compressor → aftercooler
    # Same pattern as recompression stage (§5.1) with anti-surge
    ...

if split_injection > MIN_SPLIT:
    # Multi-stage injection compression (2+ stages, same pattern)
    ...

6.3 Booster Compressor (Optional)

Some platforms have a booster between HP separation and the export manifold:

MIN_BOOSTER_SPEED = 1000  # rpm threshold

if process_input.booster_speed > MIN_BOOSTER_SPEED:
    booster_mixer.addStream(hp_gas)
    operations.add(booster_mixer)
    # Same cooler → scrubber → compressor → ASV pattern
    ...
    export_manifold.addStream(booster_output)
else:
    # HP gas goes directly to export manifold
    export_manifold.addStream(hp_gas)

6.4 Multiple parallel trains, 2-stage machines, dehydration, shared shafts

Real platforms rarely have one machine per duty. Match the P&ID / STID exactly:

• Two full separation trains (A/B). Many platforms run two parallel 1st/2nd/3rd stage trains (e.g. 20-VA01/02/03 A and B), not two 1st-stage separators feeding a shared 2nd/3rd stage. Build a sep_train(train, feed) helper and commingle the final-stage oil from both trains before the crude cooler.
• Two recompression trains (A/B) — one 3-body cascade each. Route each train's HP discharge back to the gas header, then split the header to export/fuel/ injection/lift. Do NOT commingle a recompressor interstage straight into the export suction (a common wiring bug — the recompressed gas must rejoin the header first).
• Parallel export / injection machines. Split the feed 50/50 into parallel bodies (27-KA01A/B), each its own anti-surge, then a Mixer to commingle. Splitting one large machine into two halves the per-machine gas load — e.g. a suction scrubber that reads ~168 % of design on one body drops to ~80 % on two.
• Multi-stage injection — parallel 1st-stage bodies (26-KA01A/B) → commingle → a common 2nd stage (26-KA02) to the reservoir pressure, with an interstage pressure.
• Dehydration (TEG) — a cooler + scrubber (24-VG03) before export knocks out free water; a full TEG contactor with glycol recycle is a further step.
• Shared-shaft feasibility. For a shared-shaft recompression string, converge the one common speed with CompressorShaftCalculator (process-integrated, added after the bodies) or CompressorShaft.solveSpeed. Both report a feasibility result — shaft.isFeasible() / getLastSolveResult() gives the status (PRESSURE_ABOVE_MAX_SPEED / PRESSURE_BELOW_MIN_SPEED / OVER_POWER / …) and the min-/max-achievable discharge — so a maximise-throughput sweep can gate on it. A below-min-speed target is handled by setPressureControl(DOWNSTREAM_CHOKE); the downstream Mixer.isPressureMismatch() flag shows where an unmet pressure collapses a commingling node to the lowest inlet.
• Fixed-speed IGV control. For a constant-speed (motor-driven) machine, control pressure/capacity with inlet guide vanes: comp.setInletGuideVaneOpening(f) / setGuideVaneAngle(deg) applies a parametric InletGuideVaneModel (lowers head, efficiency and surge flow at fixed speed); attach vendor per-vane maps with comp.setInletGuideVaneChart(CompressorChartIGV) for interpolated IGV-position charts.

# Parallel A/B export bodies -> commingle
exp_split = Splitter("export split A/B", dry_gas, 2)
exp_split.setSplitFactors([0.5, 0.5]); p.add(exp_split)
exp_mix = Mixer("gas export commingle")
for i, tr in enumerate(("A", "B")):
    st = antisurge_stage(p, f"export compressor 27-KA01{tr}", exp_split.getSplitStream(i),
                         EXPORT_P, SUCTION_P, 0.78, RATED_KW, feed_fluid)
    exp_mix.addStream(st["fwd"])
p.add(exp_mix)

Shared driver shaft (common speed). Bodies on one gas-turbine shaft must turn at one speed. Group them with neqsim.process.equipment.compressor.CompressorShaft and solveSpeed(reference, targetP, unit, runnable) to iterate the single common speed to the final discharge (intermediates float). See the neqsim-compressor-antisurge-recycle skill and docs/process/equipment/compressor_shaft.md. Each shaft solve costs several full-field solves, so on a big multi-train plant make it an opt-in step and solve each train's shaft in turn.

Fuel gas from power demand. Size the fuel-gas offtake from the actual compression shaft power with a GasTurbine in power-demand mode: sum every compressor's getPower() plus a lumped utility load, then gt.setRequiredPower(W) + gt.getFuelFlowRate(...). Iterate 2–3 times so the small fuel offtake and the compressor power are consistent.

Vessel utilization (Souders-Brown). Report separator/scrubber gas load K = v_gas·sqrt(rho_g/(rho_l-rho_g)) vs a design K (~0.107 m/s) using the vessel internal diameter, so over-loaded vessels (a real bottleneck) are flagged.

7. Measurement Devices

7.1 Hydrate Temperature Monitoring

Add at every cooler outlet to check hydrate risk:

HydrateAnalyser = jneqsim.process.measurementdevice.HydrateEquilibriumTemperatureAnalyser

hydrate_mon = HydrateAnalyser("Hydrate R1 cooler", cooler.getOutletStream())
operations.add(hydrate_mon)

# Read after running:
hydrate_temp_C = operations.getMeasurementDevice("Hydrate R1 cooler").getMeasuredValue("C")

7.2 Well Allocators

For multi-well platforms, track each well's contribution to exports:

WellAllocator = jneqsim.process.measurementdevice.WellAllocator

allocator = WellAllocator("Well A-3", well_stream_a3)
allocator.setExportGasStream(export_gas)
allocator.setExportOilStream(stable_oil)
operations.add(allocator)

# Read allocated rates per well
gas_alloc = allocator.getMeasuredValue("gas export rate", "kg/hr")
oil_alloc = allocator.getMeasuredValue("oil export rate", "kg/hr")

8. Result Extraction

8.1 Structured Response Helpers

Use standardized helper functions to extract equipment results into typed objects:

def get_separator_response(separator) -> dict:
    """Extract separator state into structured dict."""
    result = {
        "name": str(separator.getName()),
        "pressure_bara": float(separator.getPressure("bara")),
        "temperature_C": float(separator.getTemperature("C")),
        "mass_flow_kghr": float(separator.getFluid().getFlowRate("kg/hr")),
        "gas_load_factor": float(separator.getGasLoadFactor()),
    }
    if separator.getThermoSystem().hasPhaseType("gas"):
        result["gas_flow_kghr"] = float(separator.getGasOutStream().getFlowRate("kg/hr"))
    if separator.getThermoSystem().hasPhaseType("oil"):
        result["oil_flow_kghr"] = float(
            separator.getThermoSystem().phaseToSystem("oil").getFlowRate("kg/hr")
        )
    return result


def get_compressor_response(compressor, asv_valve=None, curves=None) -> dict:
    """Extract compressor state with anti-surge and curve data."""
    result = {
        "name": str(compressor.getName()),
        "suction_P_bara": float(compressor.getInletStream().getPressure("bara")),
        "discharge_P_bara": float(compressor.getOutletStream().getPressure("bara")),
        "suction_T_C": float(compressor.getInletStream().getTemperature("C")),
        "discharge_T_C": float(compressor.getOutletStream().getTemperature("C")),
        "power_kW": float(compressor.getPower("kW")),
        "polytropic_head": float(compressor.getPolytropicFluidHead()),
        "polytropic_efficiency": float(compressor.getPolytropicEfficiency()),
        "mass_flow_kghr": float(compressor.getInletStream().getFlowRate("kg/hr")),
        "suction_vol_flow_m3hr": float(compressor.getInletStream().getFlowRate("m3/hr")),
    }
    if asv_valve:
        result["asv_flow_kghr"] = float(asv_valve.getOutletStream().getFlowRate("kg/hr"))
        result["net_flow_kghr"] = result["mass_flow_kghr"] - result["asv_flow_kghr"]
    if curves:
        result["curve_head"] = float(curves.getPolytropicHead())
        result["curve_efficiency"] = float(curves.getPolytropicEfficiency())
        result["speed"] = float(curves.getSpeed())
    return result

8.2 Key Output Extraction

# Oil export
oil_rate_m3day = operations.getUnit("LP Separator").getLiquidOutStream().getFlowRate("m3/hr") * 24
oil_tvp = operations.getUnit("LP Separator").getLiquidOutStream().TVP(20.0, "C")
oil_density = operations.getUnit("LP Separator").getLiquidOutStream().getFluid().getDensity("kg/m3")

# Gas export
gas_rate_MSm3day = operations.getUnit("Export aftercooler").getOutletStream().getFlowRate("MSm3/day")

# Total power
total_power_kW = sum(
    operations.getUnit(name).getPower("kW")
    for name in ["R1 Compressor", "R2 Compressor", "R3 Compressor", "Export compressor"]
)

# Total cooling duty
total_cooling_kW = sum(
    operations.getUnit(name).getDuty() / 1000
    for name in ["R1 Cooler", "R2 Cooler", "R3 Cooler", "Export cooler"]
)

8.3 Mass-Balance Acceptance Gate (MANDATORY)

Never accept or report a platform-model solution until the overall mass balance closes. Sum the kg/hr of every feed entering the model and every product/export stream leaving it; the closure error must be below 0.1 %. A larger imbalance almost always means a stream was silently dropped (an unconnected scrubber liquid out-stream is the most common cause — see Section 4), a Recycle tear did not converge, or a Splitter fraction is wrong.

def check_mass_balance(feeds, products, tol=1.0e-3):
    """Verify overall mass balance before accepting the solution.

    feeds, products: lists of StreamInterface (all model inlets / all model outlets).
    Returns (ok, closure_error_fraction). Raises if the model is unbalanced.
    """
    m_in = sum(float(s.getFlowRate("kg/hr")) for s in feeds)
    m_out = sum(float(s.getFlowRate("kg/hr")) for s in products)
    closure = abs(m_in - m_out) / m_in if m_in > 0 else float("inf")
    if closure > tol:
        raise AssertionError(
            "Mass balance not closed: in=%.3f out=%.3f kg/hr (%.3f%% error). "
            "Check for dropped scrubber liquid, non-converged recycle, or bad split."
            % (m_in, m_out, 100.0 * closure)
        )
    return True, closure

# feeds = [well-stream(s) / reservoir feed(s)]
# products = [oil export, gas export, gas injection, produced water, fuel/flare, ...]
check_mass_balance(feeds, products

…(truncated)