This guide shows how to use NeqSim safely from Python, wrap Java objects by composition, implement only narrow Java interfaces when a proxy is genuinely required, and integrate calculations with scientific Python. Core thermodynamic and physical-property models should be implemented and tested in Java rather than approximated with incomplete Python proxies.

Table of Contents

  1. Overview
  2. Setting Up JPype with NeqSim
  3. Calling Java from Python
  4. Python Proxies and Extension Boundaries
  5. Custom Process Equipment Wrappers
  6. Custom Thermodynamic Calculations
  7. Batch Processing and Optimization
  8. Integration with Scientific Python
  9. Performance Considerations
  10. Best Practices

Overview

NeqSim Python integration has four distinct boundaries:

Approach Use case Boundary
Direct Java access Use released NeqSim classes Preferred for standard simulations
Python composition wrapper Add Python ergonomics, validation, or tabular results around Java objects Does not become a NeqSim process unit or thermodynamic model
Narrow Java-interface proxy Supply callbacks to an API that explicitly accepts a small interface Must implement every abstract method; validate lifetime and threading
Java extension Add process equipment, thermodynamic models, or physical-property methods Implement and test in src/main/java and src/test/java

When to use each approach

Setting Up JPype with NeqSim

# Standard neqsim-python import
from neqsim import jneqsim

# This auto-starts the JVM and provides access to all NeqSim classes
SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
ThermodynamicOperations = jneqsim.thermodynamicoperations.ThermodynamicOperations
Stream = jneqsim.process.equipment.stream.Stream

Direct JPype Access (Advanced)

import jpype
import jpype.imports
from jpype.types import JDouble, JArray, JString

# Only needed if not using neqsim package
if not jpype.isJVMStarted():
    # Path to neqsim JAR
    jar_path = "/path/to/neqsim.jar"
    jpype.startJVM(classpath=[jar_path])

# Import Java packages
from neqsim.thermo.system import SystemSrkEos
from neqsim.process.equipment.stream import Stream

Type Conversion Reference

Python Type Java Type Notes
float double Automatic
int int/long Automatic
str String Automatic
bool boolean Automatic
list ArrayList or array Use JArray for typed arrays
dict HashMap Manual conversion needed
numpy.ndarray double[][] Use JArray(JDouble, 2)(arr)

Calling Java from Python

Basic Fluid Creation

from neqsim import jneqsim

# Create thermodynamic system (Temperature in Kelvin, Pressure in bara)
SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
fluid = SystemSrkEos(273.15 + 25.0, 60.0)

# Add components
fluid.addComponent("methane", 0.85)
fluid.addComponent("ethane", 0.10)
fluid.addComponent("propane", 0.05)

# REQUIRED: Set mixing rule
fluid.setMixingRule("classic")

# Run flash calculation
ThermodynamicOperations = jneqsim.thermodynamicoperations.ThermodynamicOperations
ops = ThermodynamicOperations(fluid)
ops.TPflash()
fluid.initProperties()

# Access results
print(f"Density: {fluid.getDensity('kg/m3'):.2f} kg/m³")
print(f"Z-factor: {fluid.getPhase('gas').getZ():.4f}")

Working with Process Equipment

from neqsim import jneqsim

# Import classes
ProcessSystem = jneqsim.process.processmodel.ProcessSystem
Stream = jneqsim.process.equipment.stream.Stream
Separator = jneqsim.process.equipment.separator.Separator
Compressor = jneqsim.process.equipment.compressor.Compressor

# Create fluid
SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
feed_fluid = SystemSrkEos(273.15 + 30.0, 60.0)
feed_fluid.addComponent("methane", 0.85)
feed_fluid.addComponent("n-pentane", 0.15)
feed_fluid.setMixingRule("classic")

# Build process
process = ProcessSystem("Gas Processing")

feed = Stream("Feed", feed_fluid)
feed.setFlowRate(50000.0, "kg/hr")
process.add(feed)

separator = Separator("HP Sep", feed)
process.add(separator)

# Get outlet and compress
gas = separator.getGasOutStream()
compressor = Compressor("Comp", gas)
compressor.setOutletPressure(100.0)
process.add(compressor)

# Run simulation
process.run()

# Results
print(f"Compressor power: {compressor.getPower('kW'):.1f} kW")
print(f"Outlet T: {compressor.getOutletStream().getTemperature() - 273.15:.1f}°C")

Handling Java Arrays

import jpype
from jpype.types import JDouble, JArray

# Create Java double array
java_array = JArray(JDouble)([1.0, 2.0, 3.0, 4.0, 5.0])

# Convert NumPy array to Java
import numpy as np
numpy_array = np.array([1.0, 2.0, 3.0])
java_from_numpy = JArray(JDouble)(numpy_array.tolist())

# 2D arrays
numpy_2d = np.array([[1.0, 2.0], [3.0, 4.0]])
java_2d = JArray(JDouble, 2)(numpy_2d.tolist())

# Convert Java array back to Python
python_list = list(java_array)
numpy_back = np.array(list(java_array))

Python Proxies and Extension Boundaries

JPype can expose a Python object as an implementation of an existing Java interface. A proxy must implement every abstract method with compatible argument and return types. It does not create a Java class, cannot extend a concrete Java class, and does not automatically acquire NeqSim lifecycle, cloning, serialization, calculation-identity, or thread-safety behavior.

Do not proxy core thermodynamic or property-model interfaces

The following shortcuts are invalid extension patterns:

Implement reusable components, phases, equations of state, physical-property methods, and process equipment in Java with focused Java tests. Use Python composition for study-specific calculations that do not need to enter NeqSim’s Java object graph.

Complete narrow-interface proxy example

NamedInterface is a deliberately small example with four abstract methods. Importing jneqsim starts the JVM before JPype resolves the interface name.

from jpype import JImplements
from jpype import JOverride
from neqsim import jneqsim


@JImplements("neqsim.util.NamedInterface")
class PythonTag:
    """Complete proxy for NeqSim's narrow NamedInterface contract."""

    def __init__(self, name, tag_number):
        self._name = str(name)
        self._tag_number = str(tag_number)

    @JOverride
    def getName(self):
        return self._name

    @JOverride
    def setName(self, name):
        self._name = str(name)

    @JOverride
    def getTagNumber(self):
        return self._tag_number

    @JOverride
    def setTagNumber(self, tag_number):
        self._tag_number = str(tag_number)


python_tag = PythonTag("inlet separator", "20-VG-001")
assert python_tag.getName() == "inlet separator"
assert python_tag.getTagNumber() == "20-VG-001"

This proxy only demonstrates the JPype boundary. It is not a process unit and cannot be added to a ProcessSystem. Before using any proxy in production, read the current NamedInterface JavaDoc and the accepting Java API, then test callback lifetime, exceptions, Java threads, and repeated execution.

Custom Process Equipment Wrappers

Creating Pythonic Wrappers

from neqsim import jneqsim
from dataclasses import dataclass
from typing import Optional, Dict, Any
import json

@dataclass
class SeparatorResult:
    """Results from separator calculation."""
    gas_flow_rate: float  # kg/hr
    liquid_flow_rate: float  # kg/hr
    gas_density: float  # kg/m³
    liquid_density: float  # kg/m³
    temperature: float  # °C
    pressure: float  # bara

class PythonSeparator:
    """Pythonic wrapper for NeqSim Separator."""

    def __init__(self, name: str, inlet_stream=None):
        """
        Create separator.

        Args:
            name: Equipment name
            inlet_stream: NeqSim Stream object or dict with fluid spec
        """
        self.name = name
        self._java_sep = None
        self._inlet = inlet_stream

        Separator = jneqsim.process.equipment.separator.Separator

        if inlet_stream is not None:
            if isinstance(inlet_stream, dict):
                # Create stream from dict specification
                self._inlet = self._create_stream_from_dict(inlet_stream)
            self._java_sep = Separator(name, self._inlet)

    def _create_stream_from_dict(self, spec: Dict[str, Any]):
        """Create stream from dictionary specification."""
        SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
        Stream = jneqsim.process.equipment.stream.Stream

        T = spec.get('temperature_C', 25.0) + 273.15
        P = spec.get('pressure_bara', 1.0)

        fluid = SystemSrkEos(T, P)
        for comp, frac in spec.get('composition', {}).items():
            fluid.addComponent(comp, frac)
        fluid.setMixingRule(spec.get('mixing_rule', 'classic'))

        stream = Stream(f"{self.name}_feed", fluid)
        stream.setFlowRate(spec.get('flow_rate', 1000.0),
                         spec.get('flow_unit', 'kg/hr'))

        return stream

    def run(self) -> SeparatorResult:
        """Run separator and return results."""
        if self._java_sep is None:
            raise ValueError("No inlet stream connected")

        self._java_sep.run()

        gas = self._java_sep.getGasOutStream()
        liquid = self._java_sep.getLiquidOutStream()

        return SeparatorResult(
            gas_flow_rate=gas.getFlowRate("kg/hr"),
            liquid_flow_rate=liquid.getFlowRate("kg/hr"),
            gas_density=gas.getFluid().getDensity("kg/m3"),
            liquid_density=liquid.getFluid().getDensity("kg/m3"),
            temperature=self._java_sep.getTemperature() - 273.15,
            pressure=self._java_sep.getPressure()
        )

    def get_gas_stream(self):
        """Get gas outlet stream."""
        return self._java_sep.getGasOutStream()

    def get_liquid_stream(self):
        """Get liquid outlet stream."""
        return self._java_sep.getLiquidOutStream()

    def to_json(self) -> str:
        """Export results to JSON."""
        result = self.run()
        return json.dumps({
            'name': self.name,
            'gas_flow_rate_kg_hr': result.gas_flow_rate,
            'liquid_flow_rate_kg_hr': result.liquid_flow_rate,
            'gas_density_kg_m3': result.gas_density,
            'liquid_density_kg_m3': result.liquid_density,
            'temperature_C': result.temperature,
            'pressure_bara': result.pressure
        }, indent=2)

# Usage
sep = PythonSeparator("HP-SEP", {
    'temperature_C': 30.0,
    'pressure_bara': 60.0,
    'composition': {'methane': 0.85, 'n-pentane': 0.15},
    'flow_rate': 50000.0,
    'flow_unit': 'kg/hr'
})

result = sep.run()
print(f"Gas flow: {result.gas_flow_rate:.1f} kg/hr")
print(f"Liquid flow: {result.liquid_flow_rate:.1f} kg/hr")

Process Builder Pattern

class ProcessBuilder:
    """Fluent API for building NeqSim processes."""

    def __init__(self, name: str):
        self.name = name
        self._process = jneqsim.process.processmodel.ProcessSystem(name)
        self._current_stream = None
        self._equipment = {}

    def feed(self, name: str, fluid_spec: dict, flow_rate: float,
             flow_unit: str = "kg/hr"):
        """Add feed stream."""
        SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
        Stream = jneqsim.process.equipment.stream.Stream

        T = fluid_spec.get('T_C', 25.0) + 273.15
        P = fluid_spec.get('P_bara', 1.0)

        fluid = SystemSrkEos(T, P)
        for comp, frac in fluid_spec.get('composition', {}).items():
            fluid.addComponent(comp, frac)
        fluid.setMixingRule('classic')

        stream = Stream(name, fluid)
        stream.setFlowRate(flow_rate, flow_unit)

        self._process.add(stream)
        self._current_stream = stream
        self._equipment[name] = stream

        return self

    def separator(self, name: str):
        """Add separator using current stream."""
        Separator = jneqsim.process.equipment.separator.Separator
        sep = Separator(name, self._current_stream)
        self._process.add(sep)
        self._current_stream = sep.getGasOutStream()
        self._equipment[name] = sep
        return self

    def compressor(self, name: str, outlet_pressure: float):
        """Add compressor."""
        Compressor = jneqsim.process.equipment.compressor.Compressor
        comp = Compressor(name, self._current_stream)
        comp.setOutletPressure(outlet_pressure)
        self._process.add(comp)
        self._current_stream = comp.getOutletStream()
        self._equipment[name] = comp
        return self

    def cooler(self, name: str, outlet_temp_C: float):
        """Add cooler."""
        Cooler = jneqsim.process.equipment.heatexchanger.Cooler
        cooler = Cooler(name, self._current_stream)
        cooler.setOutTemperature(outlet_temp_C + 273.15)
        self._process.add(cooler)
        self._current_stream = cooler.getOutletStream()
        self._equipment[name] = cooler
        return self

    def use_liquid_from(self, separator_name: str):
        """Switch to liquid outlet of a separator."""
        sep = self._equipment.get(separator_name)
        if sep:
            self._current_stream = sep.getLiquidOutStream()
        return self

    def build(self):
        """Build and return the process."""
        return self._process

    def run(self):
        """Build and run the process."""
        self._process.run()
        return self

# Usage with fluent API
process = (ProcessBuilder("Gas Plant")
    .feed("Feed", {
        'T_C': 30, 'P_bara': 60,
        'composition': {'methane': 0.8, 'ethane': 0.1, 'propane': 0.1}
    }, flow_rate=100000)
    .separator("HP-Sep")
    .compressor("Comp-1", outlet_pressure=100.0)
    .cooler("After-Cooler", outlet_temp_C=40.0)
    .run()
)

# Access results
comp = process._equipment["Comp-1"]
print(f"Compressor power: {comp.getPower('MW'):.2f} MW")

Custom Thermodynamic Calculations

Property Calculator

import numpy as np
from typing import List, Dict, Tuple

class PropertyCalculator:
    """Calculate thermodynamic properties over ranges."""

    def __init__(self, composition: Dict[str, float],
                 eos: str = "SRK"):
        """
        Initialize calculator.

        Args:
            composition: Component name to mole fraction mapping
            eos: Equation of state ("SRK", "PR", "CPA")
        """
        self.composition = composition
        self.eos = eos
        self._fluid = self._create_fluid()

    def _create_fluid(self):
        """Create base fluid."""
        eos_classes = {
            "SRK": jneqsim.thermo.system.SystemSrkEos,
            "PR": jneqsim.thermo.system.SystemPrEos,
            "CPA": jneqsim.thermo.system.SystemSrkCPAstatoil
        }

        FluidClass = eos_classes.get(self.eos,
                                    jneqsim.thermo.system.SystemSrkEos)
        fluid = FluidClass(300.0, 50.0)

        for comp, frac in self.composition.items():
            fluid.addComponent(comp, frac)

        mixing_rule = "classic" if self.eos != "CPA" else 2
        fluid.setMixingRule(mixing_rule)

        return fluid

    def calculate_properties(self, T_C: float, P_bara: float) -> Dict:
        """Calculate properties at single point."""
        fluid = self._fluid.clone()
        fluid.setTemperature(T_C + 273.15)
        fluid.setPressure(P_bara)

        ops = jneqsim.thermodynamicoperations.ThermodynamicOperations(fluid)
        ops.TPflash()
        fluid.initProperties()

        results = {
            'temperature_C': T_C,
            'pressure_bara': P_bara,
            'number_of_phases': fluid.getNumberOfPhases(),
        }

        # Gas phase properties (if exists)
        if fluid.hasPhaseType("gas"):
            gas = fluid.getPhase("gas")
            results['gas'] = {
                'Z': gas.getZ(),
                'density_kg_m3': gas.getDensity("kg/m3"),
                'viscosity_cP': gas.getViscosity("cP"),
                'molar_mass_kg_mol': gas.getMolarMass("kg/mol"),
                'phase_fraction': fluid.getBeta(fluid.getPhaseIndex("gas"))
            }

        # Liquid phase properties (if exists)
        if fluid.hasPhaseType("oil") or fluid.hasPhaseType("aqueous"):
            phase_type = "oil" if fluid.hasPhaseType("oil") else "aqueous"
            liq = fluid.getPhase(phase_type)
            results['liquid'] = {
                'density_kg_m3': liq.getDensity("kg/m3"),
                'viscosity_cP': liq.getViscosity("cP"),
                'phase_fraction': fluid.getBeta(fluid.getPhaseIndex(phase_type))
            }

        return results

    def property_table(self, T_range: Tuple[float, float],
                       P_range: Tuple[float, float],
                       n_T: int = 10, n_P: int = 10) -> 'pd.DataFrame':
        """Generate property table over T-P range."""
        import pandas as pd

        T_vals = np.linspace(T_range[0], T_range[1], n_T)
        P_vals = np.linspace(P_range[0], P_range[1], n_P)

        data = []
        for T in T_vals:
            for P in P_vals:
                try:
                    props = self.calculate_properties(T, P)
                    row = {
                        'T_C': T, 'P_bara': P,
                        'n_phases': props['number_of_phases']
                    }
                    if 'gas' in props:
                        row['gas_Z'] = props['gas']['Z']
                        row['gas_rho'] = props['gas']['density_kg_m3']
                    if 'liquid' in props:
                        row['liq_rho'] = props['liquid']['density_kg_m3']
                    data.append(row)
                except Exception as e:
                    print(f"Failed at T={T}, P={P}: {e}")

        return pd.DataFrame(data)

    def phase_envelope(self) -> Dict[str, List[float]]:
        """Calculate phase envelope."""
        fluid = self._fluid.clone()
        fluid.setTemperature(250.0)
        fluid.setPressure(1.0)

        ops = jneqsim.thermodynamicoperations.ThermodynamicOperations(fluid)
        ops.calcPTphaseEnvelope()

        envelope = fluid.getPhaseEnvelope()

        return {
            'T_dew': list(envelope.getDewPointLine().get("temperature")),
            'P_dew': list(envelope.getDewPointLine().get("pressure")),
            'T_bubble': list(envelope.getBubblePointLine().get("temperature")),
            'P_bubble': list(envelope.getBubblePointLine().get("pressure")),
            'cricondentherm_T': envelope.getCricondenthermTemperature(),
            'cricondenbar_P': envelope.getCricondenbarPressure()
        }

# Usage
calc = PropertyCalculator({
    'methane': 0.8,
    'ethane': 0.1,
    'propane': 0.1
}, eos="SRK")

# Single point
props = calc.calculate_properties(T_C=25, P_bara=50)
print(f"Gas Z-factor: {props['gas']['Z']:.4f}")

# Generate table
df = calc.property_table(
    T_range=(-20, 100),
    P_range=(10, 200),
    n_T=5, n_P=5
)
print(df)

# Phase envelope
envelope = calc.phase_envelope()
print(f"Cricondentherm: {envelope['cricondentherm_T']:.1f} K")

Batch Processing and Optimization

Parallel Property Calculations

from concurrent.futures import ThreadPoolExecutor, as_completed
import numpy as np

def calculate_point(args):
    """Calculate properties at single T,P point."""
    T, P, composition = args

    # Create fresh fluid (thread-safe)
    SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
    fluid = SystemSrkEos(T + 273.15, P)

    for comp, frac in composition.items():
        fluid.addComponent(comp, frac)
    fluid.setMixingRule("classic")

    ops = jneqsim.thermodynamicoperations.ThermodynamicOperations(fluid)

    try:
        ops.TPflash()
        fluid.initProperties()
        return {
            'T': T, 'P': P,
            'Z': fluid.getPhase(0).getZ(),
            'rho': fluid.getDensity("kg/m3"),
            'success': True
        }
    except Exception as e:
        return {'T': T, 'P': P, 'success': False, 'error': str(e)}

def batch_calculate(composition, T_range, P_range, n_workers=4):
    """Calculate properties over grid in parallel."""
    T_vals = np.linspace(*T_range, 20)
    P_vals = np.linspace(*P_range, 20)

    # Create work items
    work = [(T, P, composition) for T in T_vals for P in P_vals]

    results = []
    with ThreadPoolExecutor(max_workers=n_workers) as executor:
        futures = [executor.submit(calculate_point, w) for w in work]

        for future in as_completed(futures):
            results.append(future.result())

    return results

# Usage
composition = {'methane': 0.9, 'ethane': 0.1}
results = batch_calculate(composition, T_range=(-50, 100), P_range=(1, 200))

# Filter successful results
successful = [r for r in results if r['success']]
print(f"Completed {len(successful)}/{len(results)} calculations")

Optimization with SciPy

from scipy.optimize import minimize, differential_evolution
import numpy as np

def optimize_compressor_stages(feed_spec, target_pressure,
                               max_stages=5, max_temp_C=150):
    """
    Optimize compressor staging for minimum power.

    Args:
        feed_spec: Feed stream specification dict
        target_pressure: Target discharge pressure (bara)
        max_stages: Maximum number of stages
        max_temp_C: Maximum discharge temperature per stage

    Returns:
        Optimal pressure ratios and total power
    """

    def calculate_power(pressure_ratios):
        """Calculate total power for given pressure ratios."""
        # Create process
        ProcessSystem = jneqsim.process.processmodel.ProcessSystem
        Stream = jneqsim.process.equipment.stream.Stream
        Compressor = jneqsim.process.equipment.compressor.Compressor
        Cooler = jneqsim.process.equipment.heatexchanger.Cooler
        SystemSrkEos = jneqsim.thermo.system.SystemSrkEos

        process = ProcessSystem("Compressor Optimization")

        # Create feed
        fluid = SystemSrkEos(feed_spec['T_C'] + 273.15, feed_spec['P_bara'])
        for comp, frac in feed_spec['composition'].items():
            fluid.addComponent(comp, frac)
        fluid.setMixingRule("classic")

        feed = Stream("Feed", fluid)
        feed.setFlowRate(feed_spec['flow_rate'], "kg/hr")
        process.add(feed)

        current_stream = feed
        current_pressure = feed_spec['P_bara']
        total_power = 0.0

        # Add compressor stages
        for i, ratio in enumerate(pressure_ratios):
            if ratio <= 1.0:
                continue

            outlet_P = current_pressure * ratio

            comp = Compressor(f"Comp-{i+1}", current_stream)
            comp.setOutletPressure(outlet_P)
            process.add(comp)

            # Check if cooling needed
            cooler = Cooler(f"Cooler-{i+1}", comp.getOutletStream())
            cooler.setOutTemperature(feed_spec['T_C'] + 273.15)  # Cool to feed T
            process.add(cooler)

            current_stream = cooler.getOutletStream()
            current_pressure = outlet_P

        try:
            process.run()

            # Sum compressor power
            for i in range(len(pressure_ratios)):
                comp_name = f"Comp-{i+1}"
                try:
                    comp = process.getUnit(comp_name)
                    total_power += comp.getPower("kW")
                except:
                    pass

            # Penalty if target not reached
            if current_pressure < target_pressure * 0.99:
                total_power += 1e6 * (target_pressure - current_pressure)

            return total_power

        except Exception as e:
            return 1e9  # Large penalty for failed calculation

    # Optimize using differential evolution
    n_stages = 3  # Start with 3 stages

    def objective(x):
        ratios = x.tolist()
        return calculate_power(ratios)

    # Calculate total ratio needed
    total_ratio = target_pressure / feed_spec['P_bara']
    avg_ratio = total_ratio ** (1/n_stages)

    # Bounds: each ratio between 1 and 4
    bounds = [(1.0, 4.0)] * n_stages

    result = differential_evolution(
        objective, bounds,
        maxiter=50, seed=42,
        workers=1  # Single worker due to JVM
    )

    return {
        'optimal_ratios': result.x.tolist(),
        'total_power_kW': result.fun,
        'success': result.success
    }

# Usage
feed = {
    'T_C': 30,
    'P_bara': 5,
    'composition': {'methane': 0.95, 'ethane': 0.05},
    'flow_rate': 100000
}

result = optimize_compressor_stages(feed, target_pressure=100)
print(f"Optimal ratios: {result['optimal_ratios']}")
print(f"Total power: {result['total_power_kW']:.1f} kW")

Integration with Scientific Python

Visualization with Matplotlib

import matplotlib.pyplot as plt
import numpy as np
from neqsim import jneqsim

def plot_phase_diagram(composition, T_range, P_range, n_points=50):
    """Plot phase diagram showing number of phases."""
    SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
    ThermodynamicOperations = jneqsim.thermodynamicoperations.ThermodynamicOperations

    T_vals = np.linspace(*T_range, n_points)
    P_vals = np.linspace(*P_range, n_points)
    T_grid, P_grid = np.meshgrid(T_vals, P_vals)

    n_phases = np.zeros_like(T_grid)

    for i, P in enumerate(P_vals):
        for j, T in enumerate(T_vals):
            fluid = SystemSrkEos(T + 273.15, P)
            for comp, frac in composition.items():
                fluid.addComponent(comp, frac)
            fluid.setMixingRule("classic")

            ops = ThermodynamicOperations(fluid)
            try:
                ops.TPflash()
                n_phases[i, j] = fluid.getNumberOfPhases()
            except:
                n_phases[i, j] = np.nan

    fig, ax = plt.subplots(figsize=(10, 8))

    # Plot phase regions
    contour = ax.contourf(T_grid, P_grid, n_phases, levels=[0.5, 1.5, 2.5, 3.5],
                          colors=['lightblue', 'lightgreen', 'lightyellow'])

    # Add phase envelope
    fluid = SystemSrkEos(250.0, 1.0)
    for comp, frac in composition.items():
        fluid.addComponent(comp, frac)
    fluid.setMixingRule("classic")

    ops = ThermodynamicOperations(fluid)
    ops.calcPTphaseEnvelope()

    envelope = fluid.getPhaseEnvelope()
    T_dew = np.array(list(envelope.getDewPointLine().get("temperature"))) - 273.15
    P_dew = np.array(list(envelope.getDewPointLine().get("pressure")))
    T_bub = np.array(list(envelope.getBubblePointLine().get("temperature"))) - 273.15
    P_bub = np.array(list(envelope.getBubblePointLine().get("pressure")))

    ax.plot(T_dew, P_dew, 'b-', linewidth=2, label='Dew line')
    ax.plot(T_bub, P_bub, 'r-', linewidth=2, label='Bubble line')

    ax.set_xlabel('Temperature (°C)')
    ax.set_ylabel('Pressure (bar)')
    ax.set_title('Phase Diagram')
    ax.legend()
    ax.grid(True, alpha=0.3)

    plt.colorbar(contour, label='Number of phases')
    plt.tight_layout()
    return fig

# Usage
fig = plot_phase_diagram(
    {'methane': 0.8, 'n-pentane': 0.2},
    T_range=(-50, 200),
    P_range=(1, 100)
)
plt.show()

Integration with Pandas

import pandas as pd
from neqsim import jneqsim

def fluid_to_dataframe(fluid) -> pd.DataFrame:
    """Convert fluid composition to DataFrame."""
    data = []

    for i in range(fluid.getNumberOfComponents()):
        comp = fluid.getComponent(i)
        row = {
            'Component': comp.getName(),
            'Mole Fraction': comp.getz(),
            'Molar Mass (g/mol)': comp.getMolarMass() * 1000,
            'Tc (K)': comp.getTC(),
            'Pc (bar)': comp.getPC(),
            'Acentric Factor': comp.getAcentricFactor()
        }

        # Add phase compositions if available
        for phase_idx in range(fluid.getNumberOfPhases()):
            phase = fluid.getPhase(phase_idx)
            phase_name = str(phase.getType())
            row[f'x_{phase_name}'] = phase.getComponent(i).getx()

        data.append(row)

    return pd.DataFrame(data)

def process_results_to_dataframe(process) -> pd.DataFrame:
    """Extract process results to DataFrame."""
    data = []

    # Iterate through units
    for i in range(process.getNumberOfUnits()):
        unit = process.getUnit(i)

        row = {
            'Unit': unit.getName(),
            'Type': type(unit).__name__,
        }

        # Get outlet conditions if stream-like
        try:
            outlet = unit.getOutletStream() if hasattr(unit, 'getOutletStream') else unit
            row['T_out (°C)'] = outlet.getTemperature() - 273.15
            row['P_out (bar)'] = outlet.getPressure()
            row['Flow (kg/hr)'] = outlet.getFlowRate("kg/hr")
        except:
            pass

        # Get power if applicable
        try:
            row['Power (kW)'] = unit.getPower("kW")
        except:
            pass

        # Get duty if applicable
        try:
            row['Duty (kW)'] = unit.getDuty() / 1000
        except:
            pass

        data.append(row)

    return pd.DataFrame(data)

# Usage
SystemSrkEos = jneqsim.thermo.system.SystemSrkEos
fluid = SystemSrkEos(300.0, 50.0)
fluid.addComponent("methane", 0.8)
fluid.addComponent("ethane", 0.15)
fluid.addComponent("propane", 0.05)
fluid.setMixingRule("classic")

ops = jneqsim.thermodynamicoperations.ThermodynamicOperations(fluid)
ops.TPflash()
fluid.initProperties()

df = fluid_to_dataframe(fluid)
print(df.to_string())

Performance Considerations

Minimize Java-Python Boundary Crossings

# BAD: Many small calls
for i in range(1000):
    T = fluid.getTemperature()  # Crosses JNI boundary each time
    P = fluid.getPressure()

# GOOD: Batch operations
# Get all data at once using toJson() or similar bulk method
json_str = fluid.toJson()
data = json.loads(json_str)

Reuse Fluid Objects

# BAD: Creating new fluid each time
def bad_calculate(T, P, composition):
    fluid = SystemSrkEos(T, P)  # New object each call
    # ... add components
    ops = ThermodynamicOperations(fluid)
    ops.TPflash()
    fluid.initProperties()
    return fluid.getDensity()

# GOOD: Clone from template
template_fluid = None

def good_calculate(T, P, composition):
    global template_fluid

    if template_fluid is None:
        template_fluid = SystemSrkEos(T, P)
        for comp, frac in composition.items():
            template_fluid.addComponent(comp, frac)
        template_fluid.setMixingRule("classic")

    fluid = template_fluid.clone()  # Reuse structure
    fluid.setTemperature(T)
    fluid.setPressure(P)

    ops = ThermodynamicOperations(fluid)
    ops.TPflash()
    fluid.initProperties()
    return fluid.getDensity()

Use Appropriate Data Types

# Explicit type conversion for better performance
from jpype.types import JDouble

# When passing many values, pre-convert
temperature = JDouble(300.0)  # Explicit conversion
fluid.setTemperature(temperature)

Best Practices

1. Error Handling

from neqsim import jneqsim

def safe_flash(fluid, T=None, P=None):
    """Safe flash calculation with error handling."""
    try:
        if T is not None:
            fluid.setTemperature(T)
        if P is not None:
            fluid.setPressure(P)

        ops = jneqsim.thermodynamicoperations.ThermodynamicOperations(fluid)
        ops.TPflash()

        if fluid.getNumberOfPhases() == 0:
            raise ValueError("Flash calculation returned no phases")

        return True

    except Exception as e:
        # Java exceptions wrapped by JPype
        error_msg = str(e)
        if "convergence" in error_msg.lower():
            print(f"Convergence failed at T={T}, P={P}")
        else:
            print(f"Flash failed: {error_msg}")
        return False

2. Resource Management

# Clean up large objects when done
import gc

def intensive_calculation():
    results = []

    for i in range(1000):
        fluid = SystemSrkEos(300.0, 50.0)
        # ... calculations
        results.append(some_result)

        # Periodically trigger garbage collection
        if i % 100 == 0:
            gc.collect()

    return results

3. Logging and Debugging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def debug_flash(fluid):
    """Flash calculation with detailed logging."""
    logger.info(f"Starting flash: T={fluid.getTemperature():.1f}K, "
                f"P={fluid.getPressure():.1f}bar")

    # Log composition
    for i in range(fluid.getNumberOfComponents()):
        comp = fluid.getComponent(i)
        logger.debug(f"  {comp.getName()}: z={comp.getz():.4f}")

    ops = jneqsim.thermodynamicoperations.ThermodynamicOperations(fluid)
    ops.TPflash()

    logger.info(f"Flash complete: {fluid.getNumberOfPhases()} phases")

    return fluid

4. Unit Conversion Helper

class UnitConverter:
    """Helper for unit conversions."""

    @staticmethod
    def C_to_K(T_C: float) -> float:
        return T_C + 273.15

    @staticmethod
    def K_to_C(T_K: float) -> float:
        return T_K - 273.15

    @staticmethod
    def bara_to_psia(P_bara: float) -> float:
        return P_bara * 14.5038

    @staticmethod
    def psia_to_bara(P_psia: float) -> float:
        return P_psia / 14.5038

    @staticmethod
    def kg_m3_to_lb_ft3(rho: float) -> float:
        return rho * 0.062428

    @staticmethod
    def cP_to_Pa_s(mu: float) -> float:
        return mu * 0.001

# Usage
uc = UnitConverter()
T_K = uc.C_to_K(25)  # 298.15 K

See Also

Document last updated: February 2026