git:20260714.06a800d to git:20260714.fb85e05

13 added, 0 removed. Audit A to A.

---

name: neqsim-platform-modeling

description: "Production platform process modeling patterns for NeqSim. USE WHEN: building full topside process models for oil & gas platforms (FPSO, fixed, semi-sub) from design documents, P&IDs, or operational data. Covers fluid creation with TBP fractions, multi-stage separation with recycles, recompression trains with compressor curves and anti-surge, export/injection compression, oil stabilization, scrubber liquid recovery, iteration strategies, and structured result extraction. Derived from 15+ production NCS platform models."

last_verified: "2026-07-04"

---

# 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)

```

| Component | Purpose | Pattern |

|-----------|---------|---------|

| `ProcessInput` | All operating conditions as typed fields | Pydantic BaseModel with 50-100+ fields |

| `simulate()` | Builds and runs the ProcessSystem | Single function, 500-1500 lines |

| `calc_result()` | Extracts structured results from solved system | Calls `simulate()`, builds ProcessOutput |

| `ProcessOutput` | All results in typed, serializable format | Pydantic BaseModel with response objects |

| `neqsim_responses.py` | Helper functions to extract equipment data | `get_separator_response()`, `get_compressor_response()`, etc. |

| `pydantic_classes.py` | Shared response types | `CompressorResponse`, `SeparatorResponse`, `HeaterResponse`, etc. |

### 1.2 Two Execution Strategies

**Strategy A — Manual Iteration (ASGA pattern):**

```python

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):**

```python

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.

---

## 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:

```python

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

```python

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,

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,

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:

```python

# 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:

```python

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).

```python

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:

```python

# === 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.

```python

# 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):

```python

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:

| Scrubber | Suction pressure | Return destination |

|----------|------------------|--------------------|

| Export / injection scrubber | HP / stage-1 header | Stage-1 (HP) separator inlet |

| MP recompression scrubber | MP / stage-2 header | Stage-2 (MP) separator inlet |

| LP recompression scrubber | LP / stage-3 header | Stage-3 (LP) separator inlet |

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:

```python

# 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:

```python

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:

```python

# 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:

```python

# === 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)

```python

# 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):

```python

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:

+ ```python

+ 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:

```python

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):

```python

# 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:

```python

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:

```python

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:

```python

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)

```

---

## 7. Measurement Devices

### 7.1 Hydrate Temperature Monitoring

Add at every cooler outlet to check hydrate risk:

```python

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:

```python

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:

```python

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

```python

# 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.

```python

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)

```

For a multi-area `ProcessModel`, also confirm convergence before the balance check:

```python

assert plant.run() or plant.solved(), "ProcessModel did not converge"

```

---

## 9. ProcessInput Configuration Pattern

### 9.1 Pydantic Model for All Operating Conditions

```python

from pydantic import BaseModel, Field

class ProcessInput(BaseModel):

"""All operating conditions for a platform process simulation."""

# Feed conditions

flow_rate_prod: float = Field(description="Production flow rate [kg/hr]")

flow_rate_test: float = Field(0.0, description="Test separator flow [kg/hr]")

# First stage separation

pressure_prod_separator: float = Field(description="HP separator pressure [bara]")

temperature_prod_separator: float = Field(description="HP separator temp [C]")

# Second stage separation

second_stage_pressure: float = Field(description="MP separator pressure [bara]")

second_stage_temperature: float = Field(description="MP separator temp [C]")

# Third stage separation

third_stage_pressure: float = Field(description="LP separator pressure [bara]")

third_stage_temperature: float = Field(description="LP separator temp [C]")

# Recompression (R1)

first_stage_recompressor_out_pressure: float = Field(description="R1 outlet P [bara]")

first_stage_recompressor_out_temperature: float = Field(description="R1 outlet T [C]")

first_stage_recompressor_scrubber_pressure: float = Field(description="R1 scrubber P [bara]")

first_stage_recompressor_scrubber_temperature: float = Field(description="R1 scrubber T [C]")

first_stage_recompressor_cooler_pressure_drop: float = Field(description="R1 cooler dP [bar]")

first_stage_recompressor_speed: float = Field(description="R1 compressor speed [rpm]")

# Anti-surge valves

antisurge_valve_opening_r1: float = Field(0.0, description="R1 ASV opening [%]")

Cv_value_antisurge_valve_r1: float = Field(description="R1 ASV Cv value")

# ... repeat for R2, R3, booster, export, injection stages

# Export

export_compressor_outlet_pressure: float = Field(description="Export comp P [bara]")

export_compressor_outlet_temperature: float = Field(description="Export comp T [C]")

export_speed: float = Field(description="Export comp speed [rpm]")

# Oil export

export_oil_pressure: float = Field(description="Oil export P [barg]")

# Production split

split_export: float = Field(1.0, description="Fraction of gas to export [0-1]")

split_injection: float = Field(0.0, description="Fraction of gas to injection [0-1]")

```

---

## 10. Common Constants

```python

MIN_VALVE_OPENING = 10 # % — below this, valve is treated as closed

MIN_BOOSTER_SPEED = 1000 # rpm — below this, booster is bypassed

MIN_SPLIT_PRODUCTION = 0.05 # fraction — below this, export/injection path skipped

NUMBER_OF_ITERATIONS = 25 # run_step iterations for convergence

SYNC_REQUEST_TIMEOUT_MS = 120_000 # ms timeout for blocking run

```

---

## 11. Checklist: Building a Platform Model

When building a new platform model from design documents:

- [ ] **Identify separation stages**: HP, MP, LP pressures and temperatures

- [ ] **Identify compression trains**: recompression (LP→HP), export, injection, booster

- [ ] **Map out all recycle loops**: scrubber liquids, export liquids, anti-surge valves

- [ ] **Pre-create all StaticMixers** before building equipment

- [ ] **Pre-create all recycle seed streams** (cloned, small flow rate, correct T/P)

- [ ] **Use Heater as T/P setter** before scrubbers and at recycle return points

- [ ] **Add hydrate temperature monitors** after every cooler

- [ ] **Add PressureDrop** elements before coolers (piping/shell losses)

- [ ] **Compressor curves**: dual object (control + curves) per stage

- [ ] **Anti-surge**: Splitter(2) → valve → Recycle back to suction seed

- [ ] **Conditional sections**: check booster speed, export/injection split fractions

- [ ] **Three-phase separators** for HP/MP (water), two-phase Separator for LP scrubbers

- [ ] **Oil TV P measurement**: `stream.TVP(20.0, "C")` for true vapor pressure at 20°C

- [ ] **Run iterations**: 25 `run_step()` calls or single threaded `runAsThread()` with timeout

- [ ] **Extract results**: structured response helpers for every equipment type

- [ ] **Validate mass balance FIRST (acceptance gate)**: sum feed `kg/hr` vs all

product/export `kg/hr`; closure error must be `< 0.1 %` before accepting the

solution (see Section 8.3). An imbalance means a dropped stream, non-converged

recycle, or bad split — fix and re-run.

- [ ] **Validate**: energy balance, hydrate temperatures above dewpoint

---

## 12. Notebook Template

For a Jupyter notebook implementation, see the complete starter in the

`neqsim-notebook-patterns` skill. The platform model follows the same

dual-boot setup cell pattern. Key additional imports:

```python

from neqsim import jneqsim

# Standard equipment

ProcessSystem = jneqsim.process.processmodel.ProcessSystem

Stream = jneqsim.process.equipment.stream.Stream

ThreePhaseSeparator = jneqsim.process.equipment.separator.ThreePhaseSeparator

Separator = jneqsim.process.equipment.separator.Separator

Compressor = jneqsim.process.equipment.compressor.Compressor

Heater = jneqsim.process.equipment.heatexchanger.Heater

ThrottlingValve = jneqsim.process.equipment.valve.ThrottlingValve

Splitter = jneqsim.process.equipment.splitter.Splitter

StaticMixer = jneqsim.process.equipment.mixer.StaticMixer

Mixer = jneqsim.process.equipment.mixer.Mixer

Pump = jneqsim.process.equipment.pump.Pump

Recycle = jneqsim.process.equipment.util.Recycle

PressureDrop = jneqsim.process.equipment.util.PressureDrop

# Measurement devices

HydrateAnalyser = jneqsim.process.measurementdevice.HydrateEquilibriumTemperatureAnalyser

WellAllocator = jneqsim.process.measurementdevice.WellAllocator

```