diff --git a/examples/jupyter/examplesInPython.ipynb b/examples/jupyter/examplesInPython.ipynb index 87596bf8..8ef32e20 100644 --- a/examples/jupyter/examplesInPython.ipynb +++ b/examples/jupyter/examplesInPython.ipynb @@ -550,6 +550,125 @@ "fluid1.initPhysicalProperties()\n", "print(\"thermal conductivity \", fluid1.getPhase(\"oil\").getThermalConductivity('W/mK'))" ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Create Process from YAML Configuration\n", + "\n", + "The `create_process_from_config()` function allows you to build complete process simulations from YAML files. This is ideal for:\n", + "- Separating process configuration from code\n", + "- Version-controlling process designs\n", + "- Quickly modifying process parameters" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Example: Create a process from a YAML configuration string\n", + "# (In practice, you'd typically load this from a file)\n", + "\n", + "yaml_config = \"\"\"\n", + "name: Gas Compression Example\n", + "\n", + "fluids:\n", + " natural_gas:\n", + " model: srk\n", + " temperature: 303.15\n", + " pressure: 10.0\n", + " components:\n", + " - name: methane\n", + " moles: 0.90\n", + " - name: ethane\n", + " moles: 0.05\n", + " - name: propane\n", + " moles: 0.03\n", + " - name: n-butane\n", + " moles: 0.02\n", + "\n", + "equipment:\n", + " - type: stream\n", + " name: inlet\n", + " fluid: natural_gas\n", + " temperature: 303.15\n", + " pressure: 10.0\n", + " flow_rate: 5.0\n", + " flow_unit: MSm3/day\n", + "\n", + " - type: separator\n", + " name: inlet_sep\n", + " inlet: inlet\n", + "\n", + " - type: compressor\n", + " name: comp1\n", + " inlet: inlet_sep.gas\n", + " pressure: 50.0\n", + "\n", + " - type: cooler\n", + " name: cooler1\n", + " inlet: comp1\n", + " temperature: 303.15\n", + "\n", + " - type: splitter\n", + " name: splitter1\n", + " inlet: cooler1\n", + " split_factors: [0.8, 0.2]\n", + "\"\"\"\n", + "\n", + "# Parse YAML and create process\n", + "import yaml\n", + "from neqsim.process import create_process_from_config\n", + "\n", + "config = yaml.safe_load(yaml_config)\n", + "process = create_process_from_config(config)\n", + "\n", + "# Display results\n", + "print(\"=== Process Results ===\")\n", + "print(f\"Compressor power: {process.get('comp1').getPower()/1e6:.2f} MW\")\n", + "print(f\"Cooler duty: {process.get('cooler1').getDuty()/1e6:.2f} MW\")\n", + "\n", + "# Get results as DataFrame\n", + "df = process.results_dataframe()\n", + "print(\"\\n=== Stream Summary ===\")\n", + "print(df.head(10))" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Example: Create fluid from configuration\n", + "from neqsim.process import create_fluid_from_config\n", + "from neqsim.thermo import TPflash, dataFrame\n", + "\n", + "# Create a custom fluid from config\n", + "fluid_config = {\n", + " 'model': 'cpa', # Use CPA for polar components\n", + " 'temperature': 298.15,\n", + " 'pressure': 1.0,\n", + " 'components': [\n", + " {'name': 'methane', 'moles': 0.80},\n", + " {'name': 'CO2', 'moles': 0.10},\n", + " {'name': 'water', 'moles': 0.05},\n", + " {'name': 'MEG', 'moles': 0.05}\n", + " ],\n", + " 'multiphase': True\n", + "}\n", + "\n", + "fluid = create_fluid_from_config(fluid_config)\n", + "fluid.setTemperature(300.0)\n", + "fluid.setPressure(50.0)\n", + "TPflash(fluid)\n", + "\n", + "print(\"=== Fluid Properties ===\")\n", + "print(dataFrame(fluid))" + ] } ], "metadata": { @@ -568,7 +687,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.10.16" + "version": "3.11.6" } }, "nbformat": 4, diff --git a/examples/jupyter/processBuilderDemo.ipynb b/examples/jupyter/processBuilderDemo.ipynb new file mode 100644 index 00000000..fcaebf9f --- /dev/null +++ b/examples/jupyter/processBuilderDemo.ipynb @@ -0,0 +1,1095 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "27049d1c", + "metadata": {}, + "source": [ + "# ProcessBuilder Fluent API Demo\n", + "\n", + "This notebook demonstrates the **ProcessBuilder fluent API** for building oil and gas process simulations in NeqSim.\n", + "\n", + "The fluent API provides a chainable interface for creating complex processes, making code more readable and maintainable.\n", + "\n", + "## 1. Installation and Setup" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "cad4ceb4", + "metadata": {}, + "outputs": [], + "source": [ + "# Uncomment to install neqsim\n", + "# !pip install neqsim" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8b2a8a11", + "metadata": {}, + "outputs": [], + "source": [ + "import neqsim\n", + "from neqsim.thermo import fluid, printFrame\n", + "from neqsim.process import ProcessBuilder, clearProcess\n", + "print(f\"NeqSim version: {neqsim.__version__}\")" + ] + }, + { + "cell_type": "markdown", + "id": "b9015c75", + "metadata": {}, + "source": [ + "## 2. Define Process Input Parameters\n", + "\n", + "We define the input parameters for a typical oil/gas separation and compression process." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b6a9b033", + "metadata": {}, + "outputs": [], + "source": [ + "# Process input parameters\n", + "feed_rate = 10.0 # MSm3/day\n", + "feed_temperature = 55.0 # C\n", + "feed_pressure = 60.0 # bara\n", + "\n", + "# Separator conditions\n", + "sep1_pressure = 50.0 # bara (1st stage)\n", + "sep2_pressure = 15.0 # bara (2nd stage) \n", + "sep3_pressure = 2.5 # bara (3rd stage)\n", + "\n", + "# Compressor conditions\n", + "comp1_outlet_pressure = 15.0 # bara\n", + "comp2_outlet_pressure = 50.0 # bara\n", + "export_pressure = 120.0 # bara\n", + "\n", + "# Cooler outlet temperatures (Kelvin)\n", + "cooler_outlet_temp = 273.15 + 35.0 # 35C in Kelvin" + ] + }, + { + "cell_type": "markdown", + "id": "0e686eb3", + "metadata": {}, + "source": [ + "## 3. Create Well Fluid\n", + "\n", + "Define the composition of the well fluid (oil & gas mixture)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b86b5af8", + "metadata": {}, + "outputs": [], + "source": [ + "# Create well fluid with typical composition\n", + "wellFluid = fluid('pr') # Peng-Robinson equation of state\n", + "\n", + "# Add components with mole fractions\n", + "wellFluid.addComponent(\"nitrogen\", 0.5)\n", + "wellFluid.addComponent(\"CO2\", 2.5)\n", + "wellFluid.addComponent(\"methane\", 70.0)\n", + "wellFluid.addComponent(\"ethane\", 8.0)\n", + "wellFluid.addComponent(\"propane\", 5.0)\n", + "wellFluid.addComponent(\"i-butane\", 1.5)\n", + "wellFluid.addComponent(\"n-butane\", 2.5)\n", + "wellFluid.addComponent(\"i-pentane\", 1.0)\n", + "wellFluid.addComponent(\"n-pentane\", 1.0)\n", + "wellFluid.addComponent(\"n-hexane\", 2.0)\n", + "wellFluid.addComponent(\"n-heptane\", 3.0)\n", + "wellFluid.addComponent(\"n-octane\", 2.0)\n", + "\n", + "wellFluid.setMixingRule(2)\n", + "wellFluid.setMultiPhaseCheck(True)\n", + "\n", + "# Set feed conditions\n", + "wellFluid.setTemperature(feed_temperature, \"C\")\n", + "wellFluid.setPressure(feed_pressure, \"bara\")\n", + "wellFluid.setTotalFlowRate(feed_rate, \"MSm3/day\")\n", + "\n", + "print(\"Well fluid created with feed conditions:\")\n", + "print(f\"Temperature: {feed_temperature} C\")\n", + "print(f\"Pressure: {feed_pressure} bara\")\n", + "print(f\"Flow rate: {feed_rate} MSm3/day\")" + ] + }, + { + "cell_type": "markdown", + "id": "25e60e4d", + "metadata": {}, + "source": [ + "## 4. Build Complete Oil/Gas Separation Process Using Fluent API\n", + "\n", + "The ProcessBuilder provides a chainable API. Each `add_*` method returns the builder, allowing method chaining.\n", + "\n", + "We build a **3-stage separation process with recompression**." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "050b9472", + "metadata": {}, + "outputs": [], + "source": [ + "# Clear any existing process\n", + "clearProcess()\n", + "\n", + "# Build complete oil/gas separation and recompression process\n", + "builder = ProcessBuilder(\"Oil Gas Separation\")\n", + "\n", + "# ===== INLET AND 1ST STAGE SEPARATION =====\n", + "builder.add_stream(\"well stream\", wellFluid)\n", + "builder.add_valve(\"inlet choke\", inlet=\"well stream\", pressure=sep1_pressure)\n", + "builder.add_separator(\"1st stage separator\", inlet=\"inlet choke\", three_phase=True)\n", + "\n", + "# ===== 2ND STAGE SEPARATION (Oil from 1st stage) =====\n", + "builder.add_valve(\"HP-MP valve\", inlet=\"1st stage separator.oil\", pressure=sep2_pressure)\n", + "builder.add_separator(\"2nd stage separator\", inlet=\"HP-MP valve\")\n", + "\n", + "# ===== 3RD STAGE SEPARATION (Oil from 2nd stage) =====\n", + "builder.add_valve(\"MP-LP valve\", inlet=\"2nd stage separator.liquid\", pressure=sep3_pressure)\n", + "builder.add_separator(\"3rd stage separator\", inlet=\"MP-LP valve\")\n", + "\n", + "# ===== STABLE OIL PRODUCT =====\n", + "builder.add_pump(\"oil export pump\", inlet=\"3rd stage separator.liquid\", pressure=15.0)\n", + "\n", + "# ===== 1ST STAGE RECOMPRESSION (Gas from 3rd stage -> 2nd stage pressure) =====\n", + "builder.add_cooler(\"LP gas cooler\", inlet=\"3rd stage separator.gas\", temperature=cooler_outlet_temp)\n", + "builder.add_separator(\"LP scrubber\", inlet=\"LP gas cooler\")\n", + "builder.add_compressor(\"LP compressor\", inlet=\"LP scrubber.gas\", pressure=sep2_pressure)\n", + "\n", + "# ===== MIX WITH 2ND STAGE GAS =====\n", + "builder.add_mixer(\"MP gas mixer\", inlets=[\"LP compressor\", \"2nd stage separator.gas\"])\n", + "\n", + "# ===== 2ND STAGE RECOMPRESSION (Gas from mixer -> 1st stage pressure) =====\n", + "builder.add_cooler(\"MP gas cooler\", inlet=\"MP gas mixer\", temperature=cooler_outlet_temp)\n", + "builder.add_separator(\"MP scrubber\", inlet=\"MP gas cooler\")\n", + "builder.add_compressor(\"MP compressor\", inlet=\"MP scrubber.gas\", pressure=sep1_pressure)\n", + "\n", + "# ===== MIX WITH 1ST STAGE GAS =====\n", + "builder.add_mixer(\"HP gas mixer\", inlets=[\"MP compressor\", \"1st stage separator.gas\"])\n", + "\n", + "# ===== EXPORT COMPRESSION =====\n", + "builder.add_cooler(\"HP gas cooler\", inlet=\"HP gas mixer\", temperature=cooler_outlet_temp)\n", + "builder.add_separator(\"HP scrubber\", inlet=\"HP gas cooler\")\n", + "builder.add_compressor(\"1st stage export comp\", inlet=\"HP scrubber.gas\", pressure=80.0)\n", + "builder.add_cooler(\"1st stage export cooler\", inlet=\"1st stage export comp\", temperature=cooler_outlet_temp)\n", + "builder.add_compressor(\"2nd stage export comp\", inlet=\"1st stage export cooler\", pressure=export_pressure)\n", + "builder.add_cooler(\"export gas cooler\", inlet=\"2nd stage export comp\", temperature=cooler_outlet_temp)\n", + "\n", + "print(\"Process built with\", len(builder.equipment), \"equipment units\")" + ] + }, + { + "cell_type": "markdown", + "id": "d00d8718", + "metadata": {}, + "source": [ + "## 5. Run the Process Simulation" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f48020a1", + "metadata": {}, + "outputs": [], + "source": [ + "# Run the process simulation\n", + "builder.run()\n", + "print(\"Process simulation completed!\")" + ] + }, + { + "cell_type": "markdown", + "id": "cfcd2e2e", + "metadata": {}, + "source": [ + "## 6. Get Results - Compressor Powers" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d7891aa2", + "metadata": {}, + "outputs": [], + "source": [ + "# Get compressor powers\n", + "lp_comp = builder.get(\"LP compressor\")\n", + "mp_comp = builder.get(\"MP compressor\")\n", + "exp_comp1 = builder.get(\"1st stage export comp\")\n", + "exp_comp2 = builder.get(\"2nd stage export comp\")\n", + "\n", + "print(\"=\" * 50)\n", + "print(\"COMPRESSOR RESULTS\")\n", + "print(\"=\" * 50)\n", + "print(f\"LP Compressor Power: {lp_comp.getPower('kW'):>10.2f} kW\")\n", + "print(f\"LP Compressor Outlet Temp: {lp_comp.getOutStream().getTemperature('C'):>10.2f} C\")\n", + "print()\n", + "print(f\"MP Compressor Power: {mp_comp.getPower('kW'):>10.2f} kW\")\n", + "print(f\"MP Compressor Outlet Temp: {mp_comp.getOutStream().getTemperature('C'):>10.2f} C\")\n", + "print()\n", + "print(f\"1st Stage Export Comp Power: {exp_comp1.getPower('kW'):>10.2f} kW\")\n", + "print(f\"1st Stage Export Outlet Temp: {exp_comp1.getOutStream().getTemperature('C'):>10.2f} C\")\n", + "print()\n", + "print(f\"2nd Stage Export Comp Power: {exp_comp2.getPower('kW'):>10.2f} kW\")\n", + "print(f\"2nd Stage Export Outlet Temp: {exp_comp2.getOutStream().getTemperature('C'):>10.2f} C\")\n", + "print()\n", + "total_power = lp_comp.getPower('kW') + mp_comp.getPower('kW') + exp_comp1.getPower('kW') + exp_comp2.getPower('kW')\n", + "print(f\"TOTAL COMPRESSION POWER: {total_power:>10.2f} kW\")\n", + "print(f\" {total_power/1000:>10.2f} MW\")" + ] + }, + { + "cell_type": "markdown", + "id": "c736b2b5", + "metadata": {}, + "source": [ + "## 7. Get Results - Cooler Duties" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a006eb46", + "metadata": {}, + "outputs": [], + "source": [ + "# Get cooler duties (negative = cooling)\n", + "lp_cooler = builder.get(\"LP gas cooler\")\n", + "mp_cooler = builder.get(\"MP gas cooler\")\n", + "hp_cooler = builder.get(\"HP gas cooler\")\n", + "exp_cooler1 = builder.get(\"1st stage export cooler\")\n", + "exp_cooler = builder.get(\"export gas cooler\")\n", + "\n", + "print(\"=\" * 50)\n", + "print(\"COOLER DUTIES\")\n", + "print(\"=\" * 50)\n", + "print(f\"LP Gas Cooler: {lp_cooler.getDuty()/1000:>10.2f} kW\")\n", + "print(f\"MP Gas Cooler: {mp_cooler.getDuty()/1000:>10.2f} kW\")\n", + "print(f\"HP Gas Cooler: {hp_cooler.getDuty()/1000:>10.2f} kW\")\n", + "print(f\"1st Stage Export Cooler: {exp_cooler1.getDuty()/1000:>10.2f} kW\")\n", + "print(f\"Export Gas Cooler: {exp_cooler.getDuty()/1000:>10.2f} kW\")\n", + "print()\n", + "total_cooling = (lp_cooler.getDuty() + mp_cooler.getDuty() + hp_cooler.getDuty() + \n", + " exp_cooler1.getDuty() + exp_cooler.getDuty()) / 1000\n", + "print(f\"TOTAL COOLING DUTY: {total_cooling:>10.2f} kW\")" + ] + }, + { + "cell_type": "markdown", + "id": "c5497179", + "metadata": {}, + "source": [ + "## 8. Get Results - Product Streams" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1507042d", + "metadata": {}, + "outputs": [], + "source": [ + "# Get product stream info\n", + "oil_pump = builder.get(\"oil export pump\")\n", + "export_cooler = builder.get(\"export gas cooler\")\n", + "\n", + "print(\"=\" * 50)\n", + "print(\"PRODUCT STREAMS\")\n", + "print(\"=\" * 50)\n", + "print(\"\\nSTABLE OIL:\")\n", + "print(f\" Temperature: {oil_pump.getOutStream().getTemperature('C'):.2f} C\")\n", + "print(f\" Pressure: {oil_pump.getOutStream().getPressure('bara'):.2f} bara\")\n", + "print(f\" Flow rate: {oil_pump.getOutStream().getFlowRate('kg/hr'):.2f} kg/hr\")\n", + "\n", + "print(\"\\nEXPORT GAS:\")\n", + "print(f\" Temperature: {export_cooler.getOutStream().getTemperature('C'):.2f} C\")\n", + "print(f\" Pressure: {export_cooler.getOutStream().getPressure('bara'):.2f} bara\")\n", + "print(f\" Flow rate: {export_cooler.getOutStream().getFlowRate('MSm3/day'):.4f} MSm3/day\")" + ] + }, + { + "cell_type": "markdown", + "id": "492ee750", + "metadata": {}, + "source": [ + "## 9. Print Fluid Tables" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "2637a837", + "metadata": {}, + "outputs": [], + "source": [ + "# Print export gas composition\n", + "print(\"EXPORT GAS COMPOSITION:\")\n", + "printFrame(export_cooler.getOutStream().getFluid())" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f162fcd3", + "metadata": {}, + "outputs": [], + "source": [ + "# Print stable oil composition\n", + "print(\"STABLE OIL COMPOSITION:\")\n", + "printFrame(oil_pump.getOutStream().getFluid())" + ] + }, + { + "cell_type": "markdown", + "id": "23d2c530", + "metadata": {}, + "source": [ + "## 10. Parametric Study - Effect of 3rd Stage Pressure" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6b103ed6", + "metadata": {}, + "outputs": [], + "source": [ + "import matplotlib.pyplot as plt\n", + "import numpy as np\n", + "\n", + "# Study effect of 3rd stage separator pressure on total compression power\n", + "sep3_pressures = np.linspace(1.5, 5.0, 10)\n", + "total_powers = []\n", + "gas_rates = []\n", + "\n", + "for p3 in sep3_pressures:\n", + " # Create fresh fluid\n", + " test_fluid = fluid('pr')\n", + " test_fluid.addComponent(\"nitrogen\", 0.5)\n", + " test_fluid.addComponent(\"CO2\", 2.5)\n", + " test_fluid.addComponent(\"methane\", 70.0)\n", + " test_fluid.addComponent(\"ethane\", 8.0)\n", + " test_fluid.addComponent(\"propane\", 5.0)\n", + " test_fluid.addComponent(\"i-butane\", 1.5)\n", + " test_fluid.addComponent(\"n-butane\", 2.5)\n", + " test_fluid.addComponent(\"i-pentane\", 1.0)\n", + " test_fluid.addComponent(\"n-pentane\", 1.0)\n", + " test_fluid.addComponent(\"n-hexane\", 2.0)\n", + " test_fluid.addComponent(\"n-heptane\", 3.0)\n", + " test_fluid.addComponent(\"n-octane\", 2.0)\n", + " test_fluid.setMixingRule(2)\n", + " test_fluid.setMultiPhaseCheck(True)\n", + " test_fluid.setTemperature(55.0, \"C\")\n", + " test_fluid.setPressure(60.0, \"bara\")\n", + " test_fluid.setTotalFlowRate(10.0, \"MSm3/day\")\n", + " \n", + " # Build process with this LP pressure\n", + " clearProcess()\n", + " b = ProcessBuilder(\"Study\")\n", + " b.add_stream(\"well\", test_fluid)\n", + " b.add_valve(\"v1\", inlet=\"well\", pressure=50.0)\n", + " b.add_separator(\"sep1\", inlet=\"v1\", three_phase=True)\n", + " b.add_valve(\"v2\", inlet=\"sep1.oil\", pressure=15.0)\n", + " b.add_separator(\"sep2\", inlet=\"v2\")\n", + " b.add_valve(\"v3\", inlet=\"sep2.liquid\", pressure=p3)\n", + " b.add_separator(\"sep3\", inlet=\"v3\")\n", + " \n", + " # Recompression\n", + " b.add_cooler(\"c1\", inlet=\"sep3.gas\", temperature=273.15+35)\n", + " b.add_separator(\"s1\", inlet=\"c1\")\n", + " b.add_compressor(\"comp1\", inlet=\"s1.gas\", pressure=15.0)\n", + " b.add_mixer(\"m1\", inlets=[\"comp1\", \"sep2.gas\"])\n", + " b.add_cooler(\"c2\", inlet=\"m1\", temperature=273.15+35)\n", + " b.add_separator(\"s2\", inlet=\"c2\")\n", + " b.add_compressor(\"comp2\", inlet=\"s2.gas\", pressure=50.0)\n", + " b.add_mixer(\"m2\", inlets=[\"comp2\", \"sep1.gas\"])\n", + " b.add_cooler(\"c3\", inlet=\"m2\", temperature=273.15+35)\n", + " b.add_separator(\"s3\", inlet=\"c3\")\n", + " b.add_compressor(\"comp3\", inlet=\"s3.gas\", pressure=80.0)\n", + " b.add_cooler(\"c4\", inlet=\"comp3\", temperature=273.15+35)\n", + " b.add_compressor(\"comp4\", inlet=\"c4\", pressure=120.0)\n", + " b.add_cooler(\"c5\", inlet=\"comp4\", temperature=273.15+35)\n", + " \n", + " b.run()\n", + " \n", + " power = (b.get(\"comp1\").getPower('kW') + b.get(\"comp2\").getPower('kW') + \n", + " b.get(\"comp3\").getPower('kW') + b.get(\"comp4\").getPower('kW'))\n", + " gas_rate = b.get(\"c5\").getOutStream().getFlowRate('MSm3/day')\n", + " \n", + " total_powers.append(power / 1000) # MW\n", + " gas_rates.append(gas_rate)\n", + "\n", + "print(f\"Completed {len(sep3_pressures)} simulations\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f2dc0d5b", + "metadata": {}, + "outputs": [], + "source": [ + "# Plot results\n", + "fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))\n", + "\n", + "# Power vs pressure\n", + "ax1.plot(sep3_pressures, total_powers, 'b-o', linewidth=2, markersize=8)\n", + "ax1.set_xlabel('3rd Stage Separator Pressure (bara)', fontsize=12)\n", + "ax1.set_ylabel('Total Compression Power (MW)', fontsize=12)\n", + "ax1.set_title('Total Compression Power vs LP Separator Pressure', fontsize=14)\n", + "ax1.grid(True, alpha=0.3)\n", + "\n", + "# Gas rate vs pressure\n", + "ax2.plot(sep3_pressures, gas_rates, 'g-s', linewidth=2, markersize=8)\n", + "ax2.set_xlabel('3rd Stage Separator Pressure (bara)', fontsize=12)\n", + "ax2.set_ylabel('Export Gas Rate (MSm3/day)', fontsize=12)\n", + "ax2.set_title('Export Gas Flow vs LP Separator Pressure', fontsize=14)\n", + "ax2.grid(True, alpha=0.3)\n", + "\n", + "plt.tight_layout()\n", + "plt.show()" + ] + }, + { + "cell_type": "markdown", + "id": "2f19d129", + "metadata": {}, + "source": [ + "## 11. Alternative: Using ProcessContext (Context Manager Pattern)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7b06430b", + "metadata": {}, + "outputs": [], + "source": [ + "from neqsim.process import ProcessContext\n", + "\n", + "# Create a simple fluid\n", + "simple_fluid = fluid('srk')\n", + "simple_fluid.addComponent(\"methane\", 90.0)\n", + "simple_fluid.addComponent(\"ethane\", 7.0)\n", + "simple_fluid.addComponent(\"propane\", 3.0)\n", + "simple_fluid.setMixingRule(2)\n", + "simple_fluid.setTemperature(30.0, \"C\")\n", + "simple_fluid.setPressure(40.0, \"bara\")\n", + "simple_fluid.setTotalFlowRate(5.0, \"MSm3/day\")\n", + "\n", + "# Use context manager pattern\n", + "with ProcessContext() as ctx:\n", + " # Create process equipment\n", + " feed = ctx.stream(\"feed\", simple_fluid)\n", + " sep = ctx.separator(\"inlet separator\", feed)\n", + " comp = ctx.compressor(\"compressor\", sep.getGasOutStream(), pres=100.0)\n", + " cool = ctx.cooler(\"cooler\", comp.getOutStream(), T=308.15) # 35C\n", + " \n", + " # Run the process\n", + " ctx.run()\n", + " \n", + " # Print results\n", + " print(f\"Compressor power: {comp.getPower('kW'):.2f} kW\")\n", + " print(f\"Compressor outlet temp: {comp.getOutStream().getTemperature('C'):.2f} C\")\n", + " print(f\"Cooler duty: {cool.getDuty()/1000:.2f} kW\")" + ] + }, + { + "cell_type": "markdown", + "id": "60ce7192", + "metadata": {}, + "source": [ + "## 12. Building a Complex Process from YAML Configuration\n", + "\n", + "One of the powerful features of ProcessBuilder is the ability to construct processes from configuration data. This is particularly useful for:\n", + "- Storing process configurations in version control\n", + "- Building processes dynamically from external sources\n", + "- Separating process design from code\n", + "\n", + "In this example, we'll build a complete **gas compression and dehydration process** from a YAML file." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "13784aca", + "metadata": {}, + "outputs": [], + "source": [ + "import yaml\n", + "\n", + "# Define a complex process configuration in YAML format\n", + "yaml_config = \"\"\"\n", + "# Gas Compression and Dehydration Process Configuration\n", + "# =====================================================\n", + "\n", + "process:\n", + " name: \"Offshore Gas Processing\"\n", + " description: \"Multi-stage compression with TEG dehydration\"\n", + "\n", + "# Fluid compositions (mole fractions will be normalized)\n", + "fluids:\n", + " wellstream:\n", + " model: \"pr\"\n", + " mixing_rule: 2\n", + " components:\n", + " nitrogen: 1.0\n", + " CO2: 3.5\n", + " methane: 78.0\n", + " ethane: 8.5\n", + " propane: 4.0\n", + " i-butane: 1.2\n", + " n-butane: 1.8\n", + " i-pentane: 0.6\n", + " n-pentane: 0.5\n", + " n-hexane: 0.5\n", + " n-heptane: 0.25\n", + " water: 0.15\n", + " conditions:\n", + " temperature: 45.0 # C\n", + " pressure: 35.0 # bara\n", + " flow_rate: 8.0 # MSm3/day\n", + " flow_unit: \"MSm3/day\"\n", + "\n", + "# Equipment sequence\n", + "equipment:\n", + " # Inlet separation\n", + " - type: stream\n", + " name: inlet_stream\n", + " fluid: wellstream\n", + " \n", + " - type: separator\n", + " name: inlet_separator\n", + " inlet: inlet_stream\n", + " description: \"Inlet slug catcher / separator\"\n", + " \n", + " # First stage compression \n", + " - type: gas_scrubber\n", + " name: scrubber_1\n", + " inlet: inlet_separator.gas\n", + " description: \"1st stage suction scrubber\"\n", + " \n", + " - type: compressor\n", + " name: compressor_1\n", + " inlet: scrubber_1.gas\n", + " outlet_pressure: 65.0\n", + " isentropic_efficiency: 0.75\n", + " description: \"1st stage compressor\"\n", + " \n", + " - type: cooler\n", + " name: aftercooler_1\n", + " inlet: compressor_1\n", + " outlet_temperature: 40.0\n", + " description: \"1st stage aftercooler\"\n", + " \n", + " - type: gas_scrubber\n", + " name: ko_drum_1\n", + " inlet: aftercooler_1\n", + " description: \"1st stage KO drum\"\n", + "\n", + " # Second stage compression\n", + " - type: compressor\n", + " name: compressor_2\n", + " inlet: ko_drum_1.gas\n", + " outlet_pressure: 120.0\n", + " isentropic_efficiency: 0.75\n", + " description: \"2nd stage compressor\"\n", + " \n", + " - type: cooler\n", + " name: aftercooler_2\n", + " inlet: compressor_2\n", + " outlet_temperature: 40.0\n", + " description: \"2nd stage aftercooler\"\n", + " \n", + " - type: gas_scrubber\n", + " name: ko_drum_2\n", + " inlet: aftercooler_2\n", + " description: \"2nd stage KO drum\"\n", + "\n", + " # Third stage compression (export pressure)\n", + " - type: compressor\n", + " name: compressor_3\n", + " inlet: ko_drum_2.gas\n", + " outlet_pressure: 200.0\n", + " isentropic_efficiency: 0.75\n", + " description: \"Export compressor\"\n", + " \n", + " - type: cooler\n", + " name: export_cooler\n", + " inlet: compressor_3\n", + " outlet_temperature: 45.0\n", + " description: \"Export gas cooler\"\n", + "\n", + " # Condensate handling - collect from all scrubbers\n", + " - type: mixer\n", + " name: condensate_mixer\n", + " inlets:\n", + " - inlet_separator.liquid\n", + " - scrubber_1.liquid\n", + " - ko_drum_1.liquid\n", + " - ko_drum_2.liquid\n", + " description: \"Condensate collection header\"\n", + " \n", + " - type: pump\n", + " name: condensate_pump\n", + " inlet: condensate_mixer\n", + " outlet_pressure: 40.0\n", + " description: \"Condensate export pump\"\n", + "\n", + "# Results to extract after simulation\n", + "outputs:\n", + " - equipment: compressor_1\n", + " properties: [power_kW, outlet_temperature_C, compression_ratio]\n", + " - equipment: compressor_2\n", + " properties: [power_kW, outlet_temperature_C, compression_ratio]\n", + " - equipment: compressor_3\n", + " properties: [power_kW, outlet_temperature_C, compression_ratio]\n", + " - equipment: export_cooler\n", + " properties: [duty_kW, outlet_flow_MSm3day]\n", + " - equipment: condensate_pump\n", + " properties: [power_kW, outlet_flow_m3hr]\n", + "\"\"\"\n", + "\n", + "# Parse the YAML\n", + "config = yaml.safe_load(yaml_config)\n", + "print(\"Loaded configuration for:\", config['process']['name'])\n", + "print(\"Description:\", config['process']['description'])\n", + "print(f\"\\nFluids defined: {list(config['fluids'].keys())}\")\n", + "print(f\"Equipment count: {len(config['equipment'])}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a98d78bb", + "metadata": {}, + "outputs": [], + "source": [ + "def create_fluid_from_config(fluid_config: dict):\n", + " \"\"\"Create a NeqSim fluid from YAML configuration.\"\"\"\n", + " # Create fluid with equation of state\n", + " f = fluid(fluid_config.get('model', 'srk'))\n", + " \n", + " # Add components\n", + " for component, amount in fluid_config['components'].items():\n", + " f.addComponent(component, float(amount))\n", + " \n", + " # Set mixing rule\n", + " if 'mixing_rule' in fluid_config:\n", + " f.setMixingRule(fluid_config['mixing_rule'])\n", + " \n", + " # Enable multiphase if water present\n", + " if 'water' in fluid_config['components']:\n", + " f.setMultiPhaseCheck(True)\n", + " \n", + " # Set conditions\n", + " conditions = fluid_config.get('conditions', {})\n", + " if 'temperature' in conditions:\n", + " f.setTemperature(conditions['temperature'], 'C')\n", + " if 'pressure' in conditions:\n", + " f.setPressure(conditions['pressure'], 'bara')\n", + " if 'flow_rate' in conditions:\n", + " flow_unit = conditions.get('flow_unit', 'kg/hr')\n", + " f.setTotalFlowRate(conditions['flow_rate'], flow_unit)\n", + " \n", + " return f\n", + "\n", + "def build_process_from_config(config: dict):\n", + " \"\"\"Build a process from YAML configuration dictionary.\"\"\"\n", + " from neqsim.process import ProcessBuilder, clearProcess\n", + " \n", + " # Clear any previous process\n", + " clearProcess()\n", + " \n", + " # Create fluids\n", + " fluids = {}\n", + " for name, fluid_config in config.get('fluids', {}).items():\n", + " fluids[name] = create_fluid_from_config(fluid_config)\n", + " print(f\" Created fluid: {name}\")\n", + " \n", + " # Create process builder\n", + " process_name = config.get('process', {}).get('name', 'Process')\n", + " builder = ProcessBuilder(process_name)\n", + " \n", + " # Add equipment from configuration\n", + " for equip in config.get('equipment', []):\n", + " equip_type = equip['type']\n", + " name = equip['name']\n", + " description = equip.get('description', '')\n", + " \n", + " # Handle different equipment types\n", + " if equip_type == 'stream':\n", + " fluid_name = equip['fluid']\n", + " builder.add_stream(name, fluids[fluid_name])\n", + " \n", + " elif equip_type == 'separator':\n", + " three_phase = equip.get('three_phase', False)\n", + " if three_phase:\n", + " builder.add_three_phase_separator(name, equip['inlet'])\n", + " else:\n", + " builder.add_separator(name, equip['inlet'])\n", + " \n", + " elif equip_type == 'gas_scrubber':\n", + " builder.add_gas_scrubber(name, equip['inlet'])\n", + " \n", + " elif equip_type == 'compressor':\n", + " builder.add_compressor(\n", + " name, \n", + " equip['inlet'],\n", + " outlet_pressure=equip.get('outlet_pressure'),\n", + " isentropic_efficiency=equip.get('isentropic_efficiency', 0.75)\n", + " )\n", + " \n", + " elif equip_type == 'cooler':\n", + " builder.add_cooler(name, equip['inlet'], \n", + " outlet_temperature=equip.get('outlet_temperature'))\n", + " \n", + " elif equip_type == 'heater':\n", + " builder.add_heater(name, equip['inlet'],\n", + " outlet_temperature=equip.get('outlet_temperature'))\n", + " \n", + " elif equip_type == 'valve':\n", + " builder.add_valve(name, equip['inlet'],\n", + " outlet_pressure=equip.get('outlet_pressure'))\n", + " \n", + " elif equip_type == 'pump':\n", + " builder.add_pump(name, equip['inlet'],\n", + " outlet_pressure=equip.get('outlet_pressure'))\n", + " \n", + " elif equip_type == 'mixer':\n", + " builder.add_mixer(name, inlets=equip.get('inlets', []))\n", + " \n", + " elif equip_type == 'splitter':\n", + " builder.add_splitter(name, equip['inlet'],\n", + " split_fractions=equip.get('split_fractions'))\n", + " \n", + " elif equip_type == 'pipe':\n", + " builder.add_pipe(name, equip['inlet'],\n", + " length=equip.get('length'),\n", + " diameter=equip.get('diameter'))\n", + " \n", + " elif equip_type == 'heat_exchanger':\n", + " builder.add_heat_exchanger(name,\n", + " hot_inlet=equip.get('hot_inlet'),\n", + " cold_inlet=equip.get('cold_inlet'))\n", + " else:\n", + " # Try generic add method\n", + " kwargs = {k: v for k, v in equip.items() \n", + " if k not in ['type', 'name', 'description']}\n", + " builder.add(equip_type, name, **kwargs)\n", + " \n", + " if description:\n", + " print(f\" Added {equip_type}: {name} - {description}\")\n", + " else:\n", + " print(f\" Added {equip_type}: {name}\")\n", + " \n", + " return builder, fluids\n", + "\n", + "# Build the process\n", + "print(\"Building process from YAML configuration...\")\n", + "print(\"-\" * 50)\n", + "print(\"Creating fluids:\")\n", + "builder, fluids = build_process_from_config(config)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3fe03239", + "metadata": {}, + "outputs": [], + "source": [ + "# Run the simulation\n", + "print(\"\\nRunning process simulation...\")\n", + "builder.run()\n", + "print(\"Simulation complete!\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "e9740b2b", + "metadata": {}, + "outputs": [], + "source": [ + "def extract_equipment_results(builder, outputs_config: list):\n", + " \"\"\"Extract results from equipment based on configuration.\"\"\"\n", + " results = {}\n", + " \n", + " for output in outputs_config:\n", + " equip_name = output['equipment']\n", + " equip = builder.get(equip_name)\n", + " \n", + " if equip is None:\n", + " print(f\"Warning: Equipment '{equip_name}' not found\")\n", + " continue\n", + " \n", + " equip_results = {}\n", + " \n", + " for prop in output.get('properties', []):\n", + " try:\n", + " if prop == 'power_kW':\n", + " equip_results[prop] = equip.getPower('kW')\n", + " elif prop == 'outlet_temperature_C':\n", + " equip_results[prop] = equip.getOutletStream().getTemperature('C')\n", + " elif prop == 'inlet_temperature_C':\n", + " equip_results[prop] = equip.getInletStream().getTemperature('C')\n", + " elif prop == 'outlet_pressure_bara':\n", + " equip_results[prop] = equip.getOutletStream().getPressure('bara')\n", + " elif prop == 'inlet_pressure_bara':\n", + " equip_results[prop] = equip.getInletStream().getPressure('bara')\n", + " elif prop == 'compression_ratio':\n", + " p_out = equip.getOutletStream().getPressure('bara')\n", + " p_in = equip.getInletStream().getPressure('bara')\n", + " equip_results[prop] = p_out / p_in\n", + " elif prop == 'duty_kW':\n", + " equip_results[prop] = abs(equip.getDuty()) / 1000\n", + " elif prop == 'outlet_flow_MSm3day':\n", + " equip_results[prop] = equip.getOutletStream().getFlowRate('MSm3/day')\n", + " elif prop == 'outlet_flow_m3hr':\n", + " equip_results[prop] = equip.getOutletStream().getFlowRate('m3/hr')\n", + " elif prop == 'outlet_flow_kghr':\n", + " equip_results[prop] = equip.getOutletStream().getFlowRate('kg/hr')\n", + " else:\n", + " equip_results[prop] = f\"Unknown property: {prop}\"\n", + " except Exception as e:\n", + " equip_results[prop] = f\"Error: {e}\"\n", + " \n", + " results[equip_name] = equip_results\n", + " \n", + " return results\n", + "\n", + "# Extract results based on YAML configuration\n", + "results = extract_equipment_results(builder, config.get('outputs', []))\n", + "\n", + "# Display results\n", + "print(\"\\n\" + \"=\" * 60)\n", + "print(\"PROCESS SIMULATION RESULTS\")\n", + "print(\"=\" * 60)\n", + "\n", + "for equip_name, props in results.items():\n", + " print(f\"\\n{equip_name}:\")\n", + " for prop, value in props.items():\n", + " if isinstance(value, float):\n", + " print(f\" {prop}: {value:.2f}\")\n", + " else:\n", + " print(f\" {prop}: {value}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "0dd00ca7", + "metadata": {}, + "outputs": [], + "source": [ + "# Create a summary table\n", + "import pandas as pd\n", + "\n", + "# Compressor summary\n", + "compressor_data = []\n", + "for i in [1, 2, 3]:\n", + " comp = builder.get(f'compressor_{i}')\n", + " compressor_data.append({\n", + " 'Stage': i,\n", + " 'Inlet P (bara)': comp.getInletStream().getPressure('bara'),\n", + " 'Outlet P (bara)': comp.getOutletStream().getPressure('bara'),\n", + " 'Ratio': comp.getOutletStream().getPressure('bara') / comp.getInletStream().getPressure('bara'),\n", + " 'Inlet T (°C)': comp.getInletStream().getTemperature('C'),\n", + " 'Outlet T (°C)': comp.getOutletStream().getTemperature('C'),\n", + " 'Power (kW)': comp.getPower('kW'),\n", + " 'Efficiency': comp.getIsentropicEfficiency()\n", + " })\n", + "\n", + "comp_df = pd.DataFrame(compressor_data)\n", + "print(\"\\nCOMPRESSOR PERFORMANCE SUMMARY\")\n", + "print(\"-\" * 80)\n", + "print(comp_df.to_string(index=False))\n", + "\n", + "# Calculate totals\n", + "total_power = comp_df['Power (kW)'].sum()\n", + "print(f\"\\nTotal Compression Power: {total_power:.1f} kW ({total_power/1000:.2f} MW)\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d3575bd1", + "metadata": {}, + "outputs": [], + "source": [ + "# Visualize the compression stages\n", + "import matplotlib.pyplot as plt\n", + "\n", + "fig, axes = plt.subplots(1, 3, figsize=(15, 5))\n", + "\n", + "# Pressure profile\n", + "stages = ['Inlet', 'Stage 1', 'IC 1', 'Stage 2', 'IC 2', 'Stage 3', 'Export']\n", + "pressures = [\n", + " builder.get('inlet_stream').getOutletStream().getPressure('bara'),\n", + " builder.get('compressor_1').getOutletStream().getPressure('bara'),\n", + " builder.get('aftercooler_1').getOutletStream().getPressure('bara'),\n", + " builder.get('compressor_2').getOutletStream().getPressure('bara'),\n", + " builder.get('aftercooler_2').getOutletStream().getPressure('bara'),\n", + " builder.get('compressor_3').getOutletStream().getPressure('bara'),\n", + " builder.get('export_cooler').getOutletStream().getPressure('bara'),\n", + "]\n", + "axes[0].plot(stages, pressures, 'b-o', linewidth=2, markersize=10)\n", + "axes[0].set_ylabel('Pressure (bara)', fontsize=12)\n", + "axes[0].set_title('Pressure Profile', fontsize=14)\n", + "axes[0].tick_params(axis='x', rotation=45)\n", + "axes[0].grid(True, alpha=0.3)\n", + "\n", + "# Temperature profile\n", + "temperatures = [\n", + " builder.get('inlet_stream').getOutletStream().getTemperature('C'),\n", + " builder.get('compressor_1').getOutletStream().getTemperature('C'),\n", + " builder.get('aftercooler_1').getOutletStream().getTemperature('C'),\n", + " builder.get('compressor_2').getOutletStream().getTemperature('C'),\n", + " builder.get('aftercooler_2').getOutletStream().getTemperature('C'),\n", + " builder.get('compressor_3').getOutletStream().getTemperature('C'),\n", + " builder.get('export_cooler').getOutletStream().getTemperature('C'),\n", + "]\n", + "axes[1].plot(stages, temperatures, 'r-s', linewidth=2, markersize=10)\n", + "axes[1].set_ylabel('Temperature (°C)', fontsize=12)\n", + "axes[1].set_title('Temperature Profile', fontsize=14)\n", + "axes[1].tick_params(axis='x', rotation=45)\n", + "axes[1].grid(True, alpha=0.3)\n", + "\n", + "# Power distribution\n", + "comp_names = ['Stage 1', 'Stage 2', 'Stage 3']\n", + "powers = [builder.get(f'compressor_{i}').getPower('kW') for i in [1, 2, 3]]\n", + "colors = ['#2ecc71', '#3498db', '#9b59b6']\n", + "bars = axes[2].bar(comp_names, powers, color=colors, edgecolor='black')\n", + "axes[2].set_ylabel('Power (kW)', fontsize=12)\n", + "axes[2].set_title('Compressor Power Distribution', fontsize=14)\n", + "axes[2].grid(True, alpha=0.3, axis='y')\n", + "\n", + "# Add value labels on bars\n", + "for bar, power in zip(bars, powers):\n", + " axes[2].text(bar.get_x() + bar.get_width()/2, bar.get_height() + 50,\n", + " f'{power:.0f}', ha='center', fontsize=11)\n", + "\n", + "plt.tight_layout()\n", + "plt.show()" + ] + }, + { + "cell_type": "markdown", + "id": "982ac9c4", + "metadata": {}, + "source": [ + "### Saving Configuration to a YAML File\n", + "\n", + "In practice, you would save your process configuration to a file for reuse:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "dfee4cd4", + "metadata": {}, + "outputs": [], + "source": [ + "# Save the configuration to a file\n", + "config_filename = 'gas_compression_process.yaml'\n", + "\n", + "with open(config_filename, 'w') as f:\n", + " f.write(yaml_config)\n", + "\n", + "print(f\"Configuration saved to: {config_filename}\")\n", + "print(f\"\\nTo load and run this process later:\")\n", + "print(f\"\"\"\n", + " import yaml\n", + " from neqsim.process import ProcessBuilder\n", + " \n", + " with open('{config_filename}', 'r') as f:\n", + " config = yaml.safe_load(f)\n", + " \n", + " builder, fluids = build_process_from_config(config)\n", + " builder.run()\n", + "\"\"\")" + ] + }, + { + "cell_type": "markdown", + "id": "db3d4208", + "metadata": {}, + "source": [ + "### Benefits of YAML Configuration Approach\n", + "\n", + "| Benefit | Description |\n", + "|---------|-------------|\n", + "| **Version Control** | Track process changes in Git alongside code |\n", + "| **Separation of Concerns** | Process design separate from simulation logic |\n", + "| **Reusability** | Same config can be used by different scripts |\n", + "| **Validation** | Easy to add JSON Schema validation for configs |\n", + "| **Documentation** | YAML comments serve as inline documentation |\n", + "| **Parametric Studies** | Easy to modify parameters programmatically |\n", + "| **Collaboration** | Process engineers can edit YAML without Python knowledge |" + ] + }, + { + "cell_type": "markdown", + "id": "0b2722c8", + "metadata": {}, + "source": [ + "## Summary\n", + "\n", + "The **ProcessBuilder fluent API** provides:\n", + "\n", + "1. **Chainable methods** - Build complex processes with readable chains\n", + "2. **Named equipment** - Access any equipment by name using `builder.get(\"name\")`\n", + "3. **Dot notation for outlets** - Use `'separator.gas'`, `'separator.oil'`, `'separator.liquid'` to connect specific outlets\n", + "4. **Three-phase separators** - Use `three_phase=True` in `add_separator()` or `add_three_phase_separator()`\n", + "5. **Easy parametric studies** - Clean syntax for sensitivity analysis\n", + "\n", + "### Available Methods\n", + "\n", + "| Method | Description |\n", + "|--------|-------------|\n", + "| `add_stream(name, fluid)` | Add inlet stream |\n", + "| `add_separator(name, inlet, three_phase=False)` | Two or three-phase separator |\n", + "| `add_valve(name, inlet, pressure)` | Throttling valve |\n", + "| `add_compressor(name, inlet, pressure, efficiency=0.75)` | Gas compressor |\n", + "| `add_cooler(name, inlet, temperature)` | Cooler (temp in Kelvin) |\n", + "| `add_heater(name, inlet, temperature)` | Heater (temp in Kelvin) |\n", + "| `add_pump(name, inlet, pressure)` | Liquid pump |\n", + "| `add_mixer(name, inlets=[...])` | Stream mixer |\n", + "| `add_splitter(name, inlet, split_factors=[...])` | Stream splitter |\n", + "| `add_gas_scrubber(name, inlet)` | Gas scrubber |\n", + "| `run()` | Run the process simulation |\n", + "| `get(name)` | Get equipment by name |\n", + "\n", + "### Dot Notation for Outlets\n", + "\n", + "| Notation | Outlet Stream |\n", + "|----------|---------------|\n", + "| `'separator.gas'` or `'separator.vapor'` | Gas outlet |\n", + "| `'separator.liquid'` | Liquid outlet (2-phase) |\n", + "| `'separator.oil'` | Oil outlet (3-phase) |\n", + "| `'separator.water'` | Water outlet (3-phase) |" + ] + } + ], + "metadata": { + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/examples/process_config_example.yaml b/examples/process_config_example.yaml new file mode 100644 index 00000000..55a05496 --- /dev/null +++ b/examples/process_config_example.yaml @@ -0,0 +1,215 @@ +# NeqSim Process Configuration Example +# ===================================== +# This YAML file demonstrates the format for create_process_from_config() +# +# Usage: +# from neqsim.process import create_process_from_config +# process = create_process_from_config('process_config_example.yaml') +# print(process.results_dataframe()) + +# Process name (optional) +name: "Gas Compression and Separation Example" + +# ============================================================================= +# FLUIDS SECTION +# ============================================================================= +# Define fluids inline - these are automatically created. +# Each fluid can specify: +# - model: Equation of state (srk, pr, cpa, etc.) +# - temperature: Initial temperature in Kelvin +# - pressure: Initial pressure in bara +# - components: List of components with amounts +# - mixing_rule: Optional mixing rule +# - multiphase: Enable multiphase calculations (true/false) + +fluids: + # Natural gas feed + feed_gas: + model: srk + temperature: 303.15 # 30°C + pressure: 50.0 + components: + - name: methane + moles: 0.85 + - name: ethane + moles: 0.07 + - name: propane + moles: 0.04 + - name: n-butane + moles: 0.02 + - name: CO2 + moles: 0.01 + - name: nitrogen + moles: 0.01 + + # Alternative: Using flow rates instead of mole fractions + glycol_stream: + model: cpa + temperature: 298.15 + pressure: 1.0 + components: + - name: MEG + rate: 100.0 + unit: kg/hr + - name: water + rate: 5.0 + unit: kg/hr + + # Predefined fluid types + dry_gas: + type: predefined + name: dry gas + +# ============================================================================= +# EQUIPMENT SECTION +# ============================================================================= +# Equipment is processed in order. Each item must have: +# - type: Equipment type (stream, compressor, separator, etc.) +# - name: Unique name for the equipment +# - Additional parameters specific to each equipment type +# +# Outlet Notation: +# - separator: use .gas or .liquid +# - three_phase_separator: use .gas, .oil, or .water +# - splitter: use .split_0, .split_1, etc. +# - manifold: use .mixed or .split_0, .split_1, etc. +# - virtual_stream: use .out + +equipment: + # ----- INLET STREAM ----- + - type: stream + name: inlet + fluid: feed_gas # Reference to fluid defined in 'fluids' section + temperature: 303.15 # Override fluid temperature + pressure: 50.0 # Override fluid pressure + flow_rate: 10.0 # Standard volume flow + flow_unit: MSm3/day # Flow unit + + # ----- HEATING ----- + - type: heater + name: inlet_heater + inlet: inlet + temperature: 320.0 # Target outlet temperature in K + + # ----- TWO-PHASE SEPARATION ----- + - type: separator + name: inlet_separator + inlet: inlet_heater + # Outputs: inlet_separator.gas, inlet_separator.liquid + + # ----- GAS COMPRESSION ----- + - type: compressor + name: stage1_compressor + inlet: inlet_separator.gas # Use .gas outlet from separator + pressure: 100.0 # Discharge pressure in bara + # Optional: polytropic_efficiency: 0.75 + + # ----- COOLING ----- + - type: cooler + name: aftercooler1 + inlet: stage1_compressor + temperature: 303.15 + + # ----- SECOND STAGE COMPRESSION ----- + - type: compressor + name: stage2_compressor + inlet: aftercooler1 + pressure: 150.0 + + # ----- FINAL COOLING ----- + - type: cooler + name: aftercooler2 + inlet: stage2_compressor + temperature: 303.15 + + # ----- FLOW SPLITTING ----- + # Split stream into multiple outlets based on fractions + - type: splitter + name: product_splitter + inlet: aftercooler2 + split_factors: + - 0.7 # 70% to split_0 + - 0.3 # 30% to split_1 + # Alternative: Use absolute flow rates + # flow_rates: + # - 7.0 + # - 3.0 + # flow_unit: MSm3/day + # Outputs: product_splitter.split_0, product_splitter.split_1 + + # ----- VALVE ----- + - type: valve + name: export_valve + inlet: product_splitter.split_0 # Main product + pressure: 70.0 # Downstream pressure + + # ----- THREE-PHASE SEPARATION ----- + - type: three_phase_separator + name: final_separator + inlet: product_splitter.split_1 + # Outputs: final_separator.gas, final_separator.oil, final_separator.water + + # ----- STREAM MIXING ----- + # Combine multiple streams + - type: mixer + name: gas_mixer + inlets: + - inlet_separator.liquid # Liquid from first separator + - final_separator.oil # Oil from three-phase separator + +# ============================================================================= +# RECYCLE LOOPS (Example) +# ============================================================================= +# For recycle loops, use virtual_stream and recycle equipment: +# +# - type: virtual_stream +# name: recycle_estimate +# fluid: feed_gas +# temperature: 320.0 +# pressure: 50.0 +# flow_rate: 0.5 +# +# - type: mixer +# name: recycle_mixer +# inlets: +# - inlet +# - recycle_estimate.out +# +# # ... process equipment ... +# +# - type: splitter +# name: recycle_splitter +# inlet: some_equipment +# split_factors: [0.9, 0.1] +# +# - type: recycle +# name: recycle_loop +# inlet: recycle_splitter.split_1 +# target_stream: recycle_estimate +# tolerance: 1e-4 +# max_iterations: 50 + +# ============================================================================= +# MANIFOLD (Multiple Inlets/Outlets) +# ============================================================================= +# Manifolds combine multiple inlet streams and optionally split the output: +# +# - type: manifold +# name: well_manifold +# inlets: +# - well_1 +# - well_2 +# - well_3 +# split_factors: [0.5, 0.3, 0.2] # Optional: split output +# # Outputs: well_manifold.mixed, well_manifold.split_0, etc. + +# ============================================================================= +# HEAT EXCHANGER +# ============================================================================= +# Two-stream heat exchangers: +# +# - type: heat_exchanger +# name: gas_gas_exchanger +# hot_inlet: hot_stream +# cold_inlet: cold_stream +# approach_temperature: 10.0 # Minimum temperature approach diff --git a/src/neqsim/process/processTools.py b/src/neqsim/process/processTools.py index 226df098..353e6d11 100644 --- a/src/neqsim/process/processTools.py +++ b/src/neqsim/process/processTools.py @@ -380,6 +380,106 @@ def recycle(self, name: str, inlet_stream: Any = None) -> Any: r.addStream(inlet_stream) return self.add(r) + def gas_scrubber(self, name: str, inlet_stream: Any) -> Any: + """Create a gas scrubber and add to this process.""" + scrubber = jneqsim.process.equipment.separator.GasScrubber(name, inlet_stream) + return self.add(scrubber) + + def distillation_column( + self, name: str, trays: int = 5, reboiler: bool = True, condenser: bool = True + ) -> Any: + """Create a distillation column and add to this process.""" + column = jneqsim.process.equipment.distillation.DistillationColumn( + trays, reboiler, condenser + ) + column.setName(name) + return self.add(column) + + def teg_absorber(self, name: str) -> Any: + """Create a simple TEG absorber and add to this process.""" + absorber = jneqsim.process.equipment.absorber.SimpleTEGAbsorber(name) + return self.add(absorber) + + def water_stripper_column(self, name: str) -> Any: + """Create a water stripper column and add to this process.""" + column = jneqsim.process.equipment.absorber.WaterStripperColumn(name) + return self.add(column) + + def component_splitter( + self, name: str, inlet_stream: Any, split_factors: List[float] = None + ) -> Any: + """Create a component splitter and add to this process.""" + splitter = jneqsim.process.equipment.splitter.ComponentSplitter( + name, inlet_stream + ) + if split_factors: + splitter.setSplitFactors(split_factors) + return self.add(splitter) + + def saturator(self, name: str, inlet_stream: Any) -> Any: + """Create a stream saturator and add to this process.""" + sat = jneqsim.process.equipment.util.StreamSaturatorUtil(name, inlet_stream) + return self.add(sat) + + def filters(self, name: str, inlet_stream: Any) -> Any: + """Create a filter and add to this process.""" + f = jneqsim.process.equipment.filter.Filter(name, inlet_stream) + return self.add(f) + + def calculator(self, name: str) -> Any: + """Create a calculator and add to this process.""" + calc = jneqsim.process.equipment.util.Calculator(name) + return self.add(calc) + + def setpoint( + self, + name: str, + target_equipment: Any, + target_variable: str, + source_equipment: Any, + ) -> Any: + """Create a setpoint controller and add to this process.""" + sp = jneqsim.process.equipment.util.SetPoint( + name, target_equipment, target_variable, source_equipment + ) + return self.add(sp) + + def adjuster( + self, + name: str, + target_equipment: Any = None, + target_variable: str = None, + target_value: float = None, + ) -> Any: + """Create an adjuster and add to this process.""" + adj = jneqsim.process.equipment.util.Adjuster(name) + if target_equipment is not None and target_variable is not None: + adj.setAdjustedVariable(target_equipment, target_variable) + if target_value is not None: + adj.setTargetValue(target_value) + return self.add(adj) + + def ejector(self, name: str, motive_stream: Any, suction_stream: Any) -> Any: + """Create an ejector and add to this process.""" + ej = jneqsim.process.equipment.ejector.Ejector( + name, motive_stream, suction_stream + ) + return self.add(ej) + + def flare(self, name: str, inlet_stream: Any = None) -> Any: + """Create a flare and add to this process.""" + f = jneqsim.process.equipment.flare.Flare(name) + if inlet_stream is not None: + f.addStream(inlet_stream) + return self.add(f) + + def tank(self, name: str, inlet_stream: Any = None) -> Any: + """Create a tank and add to this process.""" + t = jneqsim.process.equipment.tank.Tank(name) + if inlet_stream is not None: + t.addStream(inlet_stream) + return self.add(t) + class ProcessBuilder: """ @@ -389,25 +489,319 @@ class ProcessBuilder: step by step. Equipment is referenced by name, making it easy to construct processes from configuration data. - Example: - >>> from neqsim.thermo import fluid - >>> from neqsim.process import ProcessBuilder - >>> - >>> feed = fluid('srk') - >>> feed.addComponent('methane', 0.9) - >>> feed.addComponent('ethane', 0.1) - >>> feed.setPressure(30.0, 'bara') - >>> feed.setTemperature(30.0, 'C') - >>> - >>> process = (ProcessBuilder("Compression Train") - ... .add_stream('inlet', feed) - ... .add_compressor('comp1', 'inlet', pressure=60.0) - ... .add_cooler('cooler1', 'comp1', temperature=303.15) - ... .add_compressor('comp2', 'cooler1', pressure=120.0) - ... .run()) - >>> - >>> print(f"Stage 1 power: {process.get('comp1').getPower()/1e6:.2f} MW") - >>> print(f"Stage 2 power: {process.get('comp2').getPower()/1e6:.2f} MW") + QUICK START + =========== + + Basic usage follows a simple pattern: + 1. Create a fluid (thermodynamic system) + 2. Create a ProcessBuilder + 3. Chain equipment additions using the fluent API + 4. Call .run() to execute the simulation + 5. Access results via .get() or .results() + + Example:: + + from neqsim.thermo import fluid + from neqsim.process import ProcessBuilder + + # Create feed fluid + feed = fluid('srk') + feed.addComponent('methane', 0.9) + feed.addComponent('ethane', 0.1) + feed.setMolarFlowRate(100.0, 'mol/sec') + feed.setTemperature(25.0, 'C') + feed.setPressure(10.0, 'bara') + + # Build and run process + process = ( + ProcessBuilder("My Process") + .add_stream('feed', feed) + .add_compressor('comp', 'feed', outlet_pressure=50.0) + .add_cooler('cooler', 'comp', outlet_temperature=30.0) + .add_separator('sep', 'cooler') + .run() + ) + + # Access results + print(f"Compressor power: {process.get('comp').getPower()/1e3:.1f} kW") + print(f"Gas out temp: {process.get('sep').getGasOutStream().getTemperature('C'):.1f} C") + + FLUENT API PATTERN + ================== + + All add_* methods return ``self``, enabling method chaining:: + + builder = ( + ProcessBuilder() + .add_stream(...) + .add_compressor(...) + .add_cooler(...) + .run() + ) + + EQUIPMENT CONNECTIONS + ===================== + + Equipment is connected by referencing upstream equipment by name. + The builder automatically gets the appropriate outlet stream. + + **Basic connection** - just use the equipment name:: + + .add_compressor('comp', inlet='feed') # Gets feed's outlet + .add_cooler('cooler', inlet='comp') # Gets comp's outlet + + **Separator outlets** - use dot notation for specific outlets:: + + .add_separator('sep', inlet='cooler') + .add_compressor('gas_comp', inlet='sep.gas') # Gas outlet + .add_pump('oil_pump', inlet='sep.oil') # Oil outlet + .add_pump('water_pump', inlet='sep.water') # Water outlet (3-phase) + + **Available outlet types**: + - ``.gas`` or ``.vapor`` - gas/vapor phase outlet + - ``.liquid`` - liquid outlet (2-phase separator) + - ``.oil`` - oil outlet (3-phase separator) + - ``.water`` or ``.aqueous`` - water outlet (3-phase separator) + - ``.out`` - generic outlet (VirtualStream, etc.) + + EQUIPMENT CATEGORIES + ==================== + + **Streams**:: + + .add_stream(name, fluid, temperature=None, pressure=None, + flow_rate=None, flow_unit='kg/sec') + .add_virtual_stream(name, source=None, flow_rate=None, ...) + .add_neq_stream(name, fluid) # Non-equilibrium stream + .add_energy_stream(name) # Energy/duty stream + .add_well_stream(name, fluid) # Well stream + + **Separation**:: + + .add_separator(name, inlet) # 2-phase separator + .add_three_phase_separator(name, inlet) # 3-phase separator + .add_separator_with_dimensions(name, inlet, inner_diameter, length) + .add_gas_scrubber(name, inlet) + .add_gas_scrubber_with_options(name, inlet, ...) + + **Compression**:: + + .add_compressor(name, inlet, outlet_pressure=None, isentropic_efficiency=None) + .add_compressor_with_chart(name, inlet) # With performance curves + .add_polytopic_compressor(name, inlet, outlet_pressure, efficiency) + + **Pumping**:: + + .add_pump(name, inlet, outlet_pressure=None, efficiency=None) + + **Heat Transfer**:: + + .add_heater(name, inlet, outlet_temperature=None) + .add_cooler(name, inlet, outlet_temperature=None) + .add_heat_exchanger(name, hot_inlet=None, cold_inlet=None) + + **Pressure Control**:: + + .add_valve(name, inlet, outlet_pressure=None) + .add_valve_with_options(name, inlet, outlet_pressure, cv=None, ...) + + **Mixing/Splitting**:: + + .add_mixer(name, inlets=None) # List of inlet names + .add_static_mixer(name, inlets=None) + .add_splitter(name, inlet, split_fractions=None) + .add_splitter_with_flowrates(name, inlet, flowrates, flow_unit) + + **Pipelines**:: + + .add_pipe(name, inlet, length=None, diameter=None) + .add_beggs_brill_pipe(name, inlet, length, elevation, diameter, ...) + .add_two_phase_pipe(name, inlet, length, elevation, diameter, ...) + + **Distillation/Absorption**:: + + .add_distillation_column(name, trays=5, reboiler=True, condenser=True) + .add_teg_absorber(name) + .add_simple_absorber(name, inlet_gas=None, inlet_liquid=None) + .add_water_stripper(name) + + **Process Control**:: + + .add_setpoint(name, source_equipment, source_variable, target_equipment, + target_variable, target_value) + .add_adjuster(name, target_equipment, target_variable, target_value, + adjust_equipment, adjust_variable) + .add_calculator(name) + .add_pid_controller(name, transmitter=None, valve=None, setpoint=None, ...) + + **Measurement**:: + + .add_pressure_transmitter(name, equipment, measurement_point='outlet') + .add_temperature_transmitter(name, equipment, measurement_point='outlet') + .add_flow_transmitter(name, equipment, measurement_point='outlet') + .add_level_transmitter(name, separator) + + **Other Equipment**:: + + .add_ejector(name, motive_inlet=None, suction_inlet=None) + .add_flare(name, inlet=None) + .add_tank(name, inlet=None) + .add_saturator(name, inlet=None) + .add_filter(name, inlet=None) + .add_reactor(name, inlet) # Gibbs reactor + .add_component_splitter(name, inlet, split_factors=None) + + RECYCLE LOOPS + ============= + + Recycles handle "streams that go back" - a common challenge because you + need to reference equipment that doesn't exist yet. + + **The Pattern**: Create a virtual stream (initial guess) FIRST, build + forward through the process, then connect the actual output back. + + Example - Anti-surge recycle:: + + process = ( + ProcessBuilder() + .add_stream('feed', fluid, flow_rate=100, flow_unit='kg/hr') + + # Step 1: Create virtual stream as initial guess + .add_virtual_stream('recycle_guess', source='feed', + flow_rate=5.0, flow_unit='kg/hr') + + # Step 2: Build forward using the guess + .add_mixer('mixer', inlets=['feed', 'recycle_guess.out']) + .add_compressor('compressor', inlet='mixer', outlet_pressure=50) + .add_cooler('cooler', inlet='compressor', outlet_temperature=30) + .add_separator('separator', inlet='cooler') + + # Step 3: Connect actual output back to virtual stream + .add_recycle('antisurge', + inlet='separator.liquid', # actual stream + outlet='recycle_guess.out', # virtual stream + tolerance=1e-6) + .run() + ) + + **Alternative helper methods**:: + + # Using setup/close pattern + .setup_recycle_loop('my_recycle', 'recycle_guess', 'feed', + initial_flow=5.0, initial_flow_unit='kg/hr') + # ... build process ... + .close_recycle_loop('my_recycle', 'separator.liquid') + + CONFIGURATION & ACCESS + ====================== + + **Configure equipment after creation**:: + + .configure('equipment_name', lambda eq: eq.setSomeProperty(value)) + + **Access equipment**:: + + process.get('compressor') # Get Java equipment object + process.get('separator').run() # Run single equipment + process['compressor'] # Dict-style access + + **Get results**:: + + process.results() # Returns underlying ProcessSystem + process.run() # Run simulation, returns self + + BUILDING FROM CONFIG + ==================== + + For dynamic process construction, use ``add()`` or ``add_from_config()``:: + + # Single equipment with type string + builder.add('compressor', 'comp1', inlet='feed', outlet_pressure=50) + + # From configuration list + config = [ + {'type': 'stream', 'name': 'feed', 'fluid': my_fluid}, + {'type': 'compressor', 'name': 'comp', 'inlet': 'feed', + 'outlet_pressure': 50}, + {'type': 'cooler', 'name': 'cooler', 'inlet': 'comp', + 'outlet_temperature': 30}, + ] + builder.add_from_config(config, fluids={'feed_fluid': my_fluid}) + + COMPLETE EXAMPLES + ================= + + **Gas Compression Train**:: + + feed = fluid('srk') + feed.addComponent('methane', 0.85) + feed.addComponent('ethane', 0.10) + feed.addComponent('propane', 0.05) + feed.setMolarFlowRate(1000, 'mol/sec') + feed.setTemperature(25, 'C') + feed.setPressure(5, 'bara') + + process = ( + ProcessBuilder("Compression Train") + .add_stream('inlet', feed) + .add_gas_scrubber('scrubber', 'inlet') + .add_compressor('stage1', 'scrubber.gas', outlet_pressure=15) + .add_cooler('ic1', 'stage1', outlet_temperature=35) + .add_gas_scrubber('kd1', 'ic1') + .add_compressor('stage2', 'kd1.gas', outlet_pressure=45) + .add_cooler('ic2', 'stage2', outlet_temperature=35) + .add_gas_scrubber('kd2', 'ic2') + .add_compressor('stage3', 'kd2.gas', outlet_pressure=120) + .add_cooler('aftercooler', 'stage3', outlet_temperature=40) + .run() + ) + + total_power = sum( + process.get(f'stage{i}').getPower() for i in [1, 2, 3] + ) / 1e6 + print(f"Total compression power: {total_power:.2f} MW") + + **Oil/Gas Separation**:: + + wellstream = fluid('srk') + # ... configure wellstream ... + + process = ( + ProcessBuilder("Separation") + .add_stream('well', wellstream) + .add_heater('heater', 'well', outlet_temperature=80) + .add_three_phase_separator('hp_sep', 'heater') + .add_valve('gas_valve', 'hp_sep.gas', outlet_pressure=20) + .add_valve('oil_valve', 'hp_sep.oil', outlet_pressure=5) + .add_three_phase_separator('lp_sep', 'oil_valve') + .add_pump('export_pump', 'lp_sep.oil', outlet_pressure=30) + .run() + ) + + **With Recycle (TEG Dehydration)**:: + + process = ( + ProcessBuilder("TEG Dehydration") + .add_stream('wet_gas', wet_gas_fluid) + .add_virtual_stream('lean_teg_recycle', source='teg_stream', + flow_rate=1000, flow_unit='kg/hr') + .add_teg_absorber('absorber') + .configure('absorber', lambda a: ( + a.addGasInStream(process.get('wet_gas').getOutletStream()), + a.addSolventInStream(process.get('lean_teg_recycle').getOutStream()) + )) + # ... regeneration equipment ... + .add_recycle('teg_recycle', + inlet='teg_cooler', + outlet='lean_teg_recycle.out', + tolerance=1e-4) + .run() + ) + + See Also + -------- + neqsim.thermo.fluid : Create thermodynamic systems + neqsim.process.newProcess : Alternative process creation function """ def __init__(self, name: str = ""): @@ -420,15 +814,28 @@ def _get_outlet(self, ref: Union[str, Any]) -> Any: """ Get outlet stream from equipment reference (name or object). - Supports dot notation for selecting specific outlets from separators: + Supports dot notation for selecting specific outlets: + + Separators: - 'separator.gas' or 'separator.vapor' - gas/vapor outlet - 'separator.liquid' - liquid outlet (2-phase separator) - 'separator.oil' - oil outlet (3-phase separator) - - 'separator.water' or 'separator.aqueous' - water outlet (3-phase separator) + - 'separator.water' or 'separator.aqueous' - water outlet + + Splitters and Manifolds: + - 'splitter.split_0', 'splitter.split_1', etc. - numbered outlets + - 'manifold.split_0', 'manifold.split_1', etc. - numbered outlets + - 'manifold.mixed' - mixed stream (before splitting) + + Other: + - 'virtual_stream.out' - output from VirtualStream/Recycle Examples: >>> builder.add_compressor('comp', 'sep.gas', pressure=100) >>> builder.add_pump('pump', 'sep.oil', pressure=50) + >>> builder.add_mixer('mixer', inlets=['feed', 'recycle_guess.out']) + >>> builder.add_valve('valve1', 'splitter.split_0', pressure=10) + >>> builder.add_compressor('comp1', 'manifold.split_0', pressure=100) """ if isinstance(ref, str): # Check for dot notation (e.g., 'separator.gas') @@ -441,6 +848,45 @@ def _get_outlet(self, ref: Union[str, Any]) -> Any: if equip is None: raise ValueError(f"Equipment '{equip_name}' not found") + # Handle VirtualStream .out notation + if outlet_type == "out": + if hasattr(equip, "getOutStream"): + return equip.getOutStream() + elif hasattr(equip, "getOutletStream"): + return equip.getOutletStream() + raise ValueError( + f"Equipment '{equip_name}' does not have an output stream method" + ) + + # Handle manifold mixed stream outlet (.mixed) + if outlet_type == "mixed": + if hasattr(equip, "getMixedStream"): + return equip.getMixedStream() + raise ValueError( + f"Equipment '{equip_name}' does not have a mixed stream outlet" + ) + + # Handle splitter/manifold outlets (.split_0, .split_1, etc.) + if outlet_type.startswith("split_"): + if hasattr(equip, "getSplitStream"): + try: + index = int(outlet_type.replace("split_", "")) + split_stream = equip.getSplitStream(index) + if split_stream is not None: + return split_stream + raise ValueError( + f"Equipment '{equip_name}' does not have outlet at index {index}" + ) + except ValueError as e: + if "does not have outlet" in str(e): + raise + raise ValueError( + f"Invalid splitter outlet format: '{outlet_type}'. Use 'split_0', 'split_1', etc." + ) + raise ValueError( + f"Equipment '{equip_name}' is not a splitter or manifold" + ) + # Map outlet type to method outlet_methods = { "gas": ["getGasOutStream", "getOutletStream"], @@ -454,7 +900,7 @@ def _get_outlet(self, ref: Union[str, Any]) -> Any: if outlet_type not in outlet_methods: raise ValueError( f"Unknown outlet type '{outlet_type}'. " - f"Valid types: {list(outlet_methods.keys())}" + f"Valid types: {list(outlet_methods.keys()) + ['out', 'mixed', 'split_N']}" ) for method_name in outlet_methods[outlet_type]: @@ -613,7 +1059,25 @@ def add_cooler( return self def add_mixer(self, name: str, inlets: List[str]) -> "ProcessBuilder": - """Add a mixer to the process.""" + """ + Add a mixer to the process. + + Args: + name: Name of the mixer. + inlets: List of inlet stream/equipment names. Supports dot notation: + - 'equipment_name' - default outlet + - 'separator.gas', 'separator.liquid' - separator outlets + - 'splitter.split_0', 'splitter.split_1' - splitter outlets + - 'virtual_stream.out' - virtual stream outlet + + Returns: + Self for method chaining. + + Examples: + >>> builder.add_mixer('mixer', inlets=['stream1', 'stream2']) + >>> builder.add_mixer('mix', inlets=['splitter.split_0', 'splitter.split_1']) + >>> builder.add_mixer('recycle_mix', inlets=['feed', 'recycle_guess.out']) + """ m = jneqsim.process.equipment.mixer.Mixer(name) for inlet in inlets: inlet_stream = self._get_outlet(inlet) @@ -622,49 +1086,1834 @@ def add_mixer(self, name: str, inlets: List[str]) -> "ProcessBuilder": self.process.add(m) return self - def add_splitter( - self, name: str, inlet: str, split_factors: List[float] = None + def add_splitter( + self, + name: str, + inlet: str, + split_factors: List[float] = None, + flow_rates: List[float] = None, + flow_unit: str = "MSm3/day", + ) -> "ProcessBuilder": + """ + Add a splitter to the process. + + Args: + name: Name of the splitter. + inlet: Name of inlet stream/equipment. + split_factors: List of split fractions (should sum to 1.0). + Length determines number of outlets. Use this OR flow_rates. + flow_rates: List of absolute flow rates for each outlet. + Use -1 for one outlet to auto-calculate remainder. + flow_unit: Unit for flow_rates (default 'MSm3/day'). + Options: 'MSm3/day', 'Sm3/day', 'kg/hr', 'kg/sec', 'mole/sec' + + Returns: + Self for method chaining. + + Notes: + Access splitter outlets using dot notation: + - 'splitter.split_0' for first outlet + - 'splitter.split_1' for second outlet + - etc. + + Examples: + # Using split factors (fractions) + >>> builder.add_splitter('split', 'feed', split_factors=[0.5, 0.5]) + + # Using absolute flow rates + >>> builder.add_splitter('split', 'feed', + ... flow_rates=[5.0, 2.0], flow_unit='MSm3/day') + + # Auto-calculate one outlet (use -1) + >>> builder.add_splitter('split', 'feed', + ... flow_rates=[-1, 1.0], flow_unit='MSm3/day') # First gets remainder + """ + inlet_stream = self._get_outlet(inlet) + s = jneqsim.process.equipment.splitter.Splitter(name, inlet_stream) + if flow_rates is not None: + s.setFlowRates(flow_rates, flow_unit) + elif split_factors is not None: + s.setSplitFactors(split_factors) + self.equipment[name] = s + self.process.add(s) + return self + + def add_heat_exchanger( + self, name: str, hot_inlet: str, cold_inlet: str, approach_temp: float = 10.0 + ) -> "ProcessBuilder": + """Add a heat exchanger to the process.""" + hot_stream = self._get_outlet(hot_inlet) + cold_stream = self._get_outlet(cold_inlet) + hx = jneqsim.process.equipment.heatexchanger.HeatExchanger( + name, hot_stream, cold_stream + ) + hx.setApproachTemperature(approach_temp) + self.equipment[name] = hx + self.process.add(hx) + return self + + def add_pipe( + self, + name: str, + inlet: str, + length: float = 100.0, + diameter: float = 0.1, + elevation: float = 0.0, + ) -> "ProcessBuilder": + """Add a pipe to the process.""" + inlet_stream = self._get_outlet(inlet) + p = jneqsim.process.equipment.pipeline.AdiabaticPipe(name, inlet_stream) + p.setLength(length) + p.setDiameter(diameter) + if elevation != 0: + p.setElevation(elevation) + self.equipment[name] = p + self.process.add(p) + return self + + def add_gas_scrubber(self, name: str, inlet: str) -> "ProcessBuilder": + """ + Add a gas scrubber to the process. + + Args: + name: Name of the gas scrubber. + inlet: Name of inlet equipment or stream. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + scrubber = jneqsim.process.equipment.separator.GasScrubber(name, inlet_stream) + self.equipment[name] = scrubber + self.process.add(scrubber) + return self + + def add_recycle( + self, + name: str, + inlet: str = None, + outlet: str = None, + tolerance: float = 1e-4, + priority: int = None, + max_iterations: int = None, + downstream_property: str = None, + ) -> "ProcessBuilder": + """ + Add a recycle stream to the process. + + RECYCLE PATTERN EXPLANATION: + ============================ + Recycles handle "streams that go back" in a process. The pattern is: + + 1. Create a virtual stream FIRST as an initial guess for the recycle + 2. Build forward through the process (using virtual stream as input) + 3. Add the recycle to connect actual output back to virtual stream + + Example - Anti-surge recycle loop: + ---------------------------------- + ```python + builder = ( + ProcessBuilder() + .add_stream("feed", fluid, flow_rate=100, flow_unit="kg/hr") + # Step 1: Create virtual stream as initial guess for recycle + .add_virtual_stream("recycle_guess", source="feed") + # Configure it with estimated recycle flow + .configure("recycle_guess", lambda vs: vs.setFlowRate(10.0, "kg/hr")) + # Step 2: Mix feed with recycle guess + .add_mixer("suction_mixer", inlets=["feed", "recycle_guess.out"]) + # Continue forward through process + .add_compressor("compressor", inlet="suction_mixer", outlet_pressure=50.0) + .add_cooler("aftercooler", inlet="compressor", outlet_temperature=30.0) + .add_separator("separator", inlet="aftercooler") + # Step 3: Add recycle connecting separator liquid back to virtual stream + .add_recycle("antisurge_recycle", + inlet="separator.liquid", + outlet="recycle_guess.out", + tolerance=1e-6) + .run() + ) + ``` + + Args: + name: Name of the recycle. + inlet: Name of inlet equipment (the actual stream to recycle). + outlet: Name of outlet stream (typically a virtual stream's output). + tolerance: Convergence tolerance for recycle iteration (default 1e-4). + priority: Recycle priority (higher = later in solve order). + max_iterations: Maximum iterations for this recycle. + downstream_property: Property to pass downstream (e.g., "flow rate"). + + Returns: + Self for method chaining. + """ + r = jneqsim.process.equipment.util.Recycle(name) + if inlet is not None: + inlet_stream = self._get_outlet(inlet) + r.addStream(inlet_stream) + if outlet is not None: + outlet_stream = self._get_outlet(outlet) + r.setOutletStream(outlet_stream) + r.setTolerance(tolerance) + if priority is not None: + r.setPriority(priority) + if max_iterations is not None: + r.setMaximumIterations(max_iterations) + if downstream_property is not None: + r.setDownstreamProperty(downstream_property) + self.equipment[name] = r + self.process.add(r) + return self + + def setup_recycle_loop( + self, + recycle_name: str, + virtual_stream_name: str, + source_stream: str, + initial_flow: float = None, + initial_flow_unit: str = "kg/hr", + ) -> "ProcessBuilder": + """ + Set up a recycle loop by creating a virtual stream for the initial guess. + + This is a convenience method that creates a virtual stream to be used + as the recycle guess. You then build your process forward using this + virtual stream, and finally connect the actual recycle back to it. + + Example workflow: + ----------------- + ```python + builder = ( + ProcessBuilder() + .add_stream("feed", fluid, flow_rate=100, flow_unit="kg/hr") + # Set up the recycle initial guess + .setup_recycle_loop("my_recycle", "recycle_guess", "feed", + initial_flow=10.0, initial_flow_unit="kg/hr") + # Mix feed with recycle + .add_mixer("mixer", inlets=["feed", "recycle_guess.out"]) + # Process forward... + .add_separator("sep", inlet="mixer") + # Close the loop + .close_recycle_loop("my_recycle", "sep.liquid", "recycle_guess.out") + .run() + ) + ``` + + Args: + recycle_name: Name for the recycle (used later in close_recycle_loop). + virtual_stream_name: Name for the virtual stream. + source_stream: Stream to use as template for composition. + initial_flow: Initial guess for recycle flow rate. + initial_flow_unit: Unit for the flow rate. + + Returns: + Self for method chaining. + """ + # Create the virtual stream + source = self._get_outlet(source_stream) + vs = jneqsim.process.equipment.stream.VirtualStream(virtual_stream_name, source) + if initial_flow is not None: + vs.setFlowRate(initial_flow, initial_flow_unit) + self.equipment[virtual_stream_name] = vs + self.process.add(vs) + # Store recycle info for later + if not hasattr(self, "_pending_recycles"): + self._pending_recycles = {} + self._pending_recycles[recycle_name] = virtual_stream_name + return self + + def close_recycle_loop( + self, + recycle_name: str, + inlet: str, + outlet: str = None, + tolerance: float = 1e-4, + priority: int = None, + ) -> "ProcessBuilder": + """ + Close a recycle loop that was set up with setup_recycle_loop. + + This creates the actual Recycle object that connects the process + output back to the virtual stream. + + Args: + recycle_name: Name for the recycle (should match setup_recycle_loop). + inlet: The actual stream to recycle (e.g., "separator.liquid"). + outlet: The virtual stream output (if None, uses the one from setup). + tolerance: Convergence tolerance. + priority: Recycle priority. + + Returns: + Self for method chaining. + """ + # Get the virtual stream from setup if outlet not specified + if outlet is None: + if ( + hasattr(self, "_pending_recycles") + and recycle_name in self._pending_recycles + ): + vs_name = self._pending_recycles[recycle_name] + outlet = f"{vs_name}.out" + else: + raise ValueError( + f"No outlet specified and no setup_recycle_loop found for '{recycle_name}'" + ) + return self.add_recycle( + recycle_name, + inlet=inlet, + outlet=outlet, + tolerance=tolerance, + priority=priority, + ) + + def add_distillation_column( + self, + name: str, + trays: int = 5, + reboiler: bool = True, + condenser: bool = True, + ) -> "ProcessBuilder": + """ + Add a distillation column to the process. + + Args: + name: Name of the column. + trays: Number of theoretical trays. + reboiler: Whether to include a reboiler. + condenser: Whether to include a condenser. + + Returns: + Self for method chaining. + + Note: + Use get() to access the column and add feed streams, set temperatures, etc. + """ + column = jneqsim.process.equipment.distillation.DistillationColumn( + trays, reboiler, condenser + ) + column.setName(name) + self.equipment[name] = column + self.process.add(column) + return self + + def add_teg_absorber(self, name: str) -> "ProcessBuilder": + """ + Add a simple TEG absorber to the process. + + Args: + name: Name of the absorber. + + Returns: + Self for method chaining. + + Note: + Use get() to access the absorber and configure it (add streams, set parameters). + """ + absorber = jneqsim.process.equipment.absorber.SimpleTEGAbsorber(name) + self.equipment[name] = absorber + self.process.add(absorber) + return self + + def add_water_stripper(self, name: str) -> "ProcessBuilder": + """ + Add a water stripper column to the process. + + Args: + name: Name of the stripper. + + Returns: + Self for method chaining. + """ + column = jneqsim.process.equipment.absorber.WaterStripperColumn(name) + self.equipment[name] = column + self.process.add(column) + return self + + def add_component_splitter( + self, name: str, inlet: str, split_factors: List[float] = None + ) -> "ProcessBuilder": + """ + Add a component splitter to the process. + + A component splitter separates a stream by component, useful for + creating pseudo-streams for specific components. + + Args: + name: Name of the splitter. + inlet: Name of inlet equipment. + split_factors: List of split factors for each component (0 to 1). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + splitter = jneqsim.process.equipment.splitter.ComponentSplitter( + name, inlet_stream + ) + if split_factors: + splitter.setSplitFactors(split_factors) + self.equipment[name] = splitter + self.process.add(splitter) + return self + + def add_saturator(self, name: str, inlet: str) -> "ProcessBuilder": + """ + Add a stream saturator to the process. + + A saturator ensures the stream is at its saturation point. + + Args: + name: Name of the saturator. + inlet: Name of inlet equipment. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + sat = jneqsim.process.equipment.util.StreamSaturatorUtil(name, inlet_stream) + self.equipment[name] = sat + self.process.add(sat) + return self + + def add_filter(self, name: str, inlet: str) -> "ProcessBuilder": + """ + Add a filter to the process. + + Args: + name: Name of the filter. + inlet: Name of inlet equipment. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + f = jneqsim.process.equipment.filter.Filter(name, inlet_stream) + self.equipment[name] = f + self.process.add(f) + return self + + def add_calculator(self, name: str) -> "ProcessBuilder": + """ + Add a calculator to the process. + + Calculators allow custom calculations based on process variables. + + Args: + name: Name of the calculator. + + Returns: + Self for method chaining. + + Note: + Use get() to access the calculator and configure it. + """ + calc = jneqsim.process.equipment.util.Calculator(name) + self.equipment[name] = calc + self.process.add(calc) + return self + + def add_setpoint( + self, + name: str, + target: str, + target_variable: str, + source: str, + ) -> "ProcessBuilder": + """ + Add a setpoint controller to the process. + + A setpoint sets a variable in one unit based on another unit's property. + + Args: + name: Name of the setpoint. + target: Name of target equipment to control. + target_variable: Variable name to control (e.g., 'pressure', 'temperature'). + source: Name of source equipment to read from. + + Returns: + Self for method chaining. + """ + target_eq = self.equipment.get(target) + source_eq = self.equipment.get(source) + if target_eq is None: + raise ValueError(f"Target equipment '{target}' not found") + if source_eq is None: + raise ValueError(f"Source equipment '{source}' not found") + sp = jneqsim.process.equipment.util.SetPoint( + name, target_eq, target_variable, source_eq + ) + self.equipment[name] = sp + self.process.add(sp) + return self + + def add_adjuster( + self, + name: str, + adjust_equipment: str = None, + adjust_variable: str = None, + target_equipment: str = None, + target_variable: str = None, + target_value: float = None, + ) -> "ProcessBuilder": + """ + Add an adjuster to the process. + + An adjuster iteratively adjusts a variable to achieve a target value. + + Args: + name: Name of the adjuster. + adjust_equipment: Name of equipment with variable to adjust. + adjust_variable: Variable name to adjust. + target_equipment: Name of equipment with target variable. + target_variable: Target variable name. + target_value: Target value to achieve. + + Returns: + Self for method chaining. + """ + adj = jneqsim.process.equipment.util.Adjuster(name) + if adjust_equipment is not None and adjust_variable is not None: + adj_eq = self.equipment.get(adjust_equipment) + if adj_eq is None: + raise ValueError(f"Adjust equipment '{adjust_equipment}' not found") + adj.setAdjustedVariable(adj_eq, adjust_variable) + if target_equipment is not None and target_variable is not None: + tgt_eq = self.equipment.get(target_equipment) + if tgt_eq is None: + raise ValueError(f"Target equipment '{target_equipment}' not found") + adj.setTargetVariable(tgt_eq, target_variable) + if target_value is not None: + adj.setTargetValue(target_value) + self.equipment[name] = adj + self.process.add(adj) + return self + + def add_ejector( + self, name: str, motive_inlet: str, suction_inlet: str + ) -> "ProcessBuilder": + """ + Add an ejector to the process. + + Args: + name: Name of the ejector. + motive_inlet: Name of motive (high-pressure) stream equipment. + suction_inlet: Name of suction (low-pressure) stream equipment. + + Returns: + Self for method chaining. + """ + motive_stream = self._get_outlet(motive_inlet) + suction_stream = self._get_outlet(suction_inlet) + ej = jneqsim.process.equipment.ejector.Ejector( + name, motive_stream, suction_stream + ) + self.equipment[name] = ej + self.process.add(ej) + return self + + def add_flare(self, name: str, inlet: str = None) -> "ProcessBuilder": + """ + Add a flare to the process. + + Args: + name: Name of the flare. + inlet: Optional name of inlet equipment. + + Returns: + Self for method chaining. + """ + f = jneqsim.process.equipment.flare.Flare(name) + if inlet is not None: + inlet_stream = self._get_outlet(inlet) + f.addStream(inlet_stream) + self.equipment[name] = f + self.process.add(f) + return self + + def add_tank(self, name: str, inlet: str = None) -> "ProcessBuilder": + """ + Add a tank to the process. + + Args: + name: Name of the tank. + inlet: Optional name of inlet equipment. + + Returns: + Self for method chaining. + """ + t = jneqsim.process.equipment.tank.Tank(name) + if inlet is not None: + inlet_stream = self._get_outlet(inlet) + t.addStream(inlet_stream) + self.equipment[name] = t + self.process.add(t) + return self + + def add_virtual_stream( + self, + name: str, + source: str = None, + flow_rate: float = None, + flow_unit: str = "kg/hr", + temperature: float = None, + temperature_unit: str = "C", + pressure: float = None, + pressure_unit: str = "bara", + ) -> "ProcessBuilder": + """ + Add a virtual stream to the process. + + Virtual streams are used with recycling to create initial guesses + for recycle convergence. They copy composition from a source stream + but can have independent flow/T/P settings. + + USAGE WITH RECYCLES: + ==================== + Virtual streams are essential for recycle loops. The pattern is: + + 1. Create virtual stream as initial guess (copies composition from source) + 2. Use virtual_stream.out as input to equipment in the loop + 3. Create a Recycle that connects actual output back to virtual_stream.out + + Example: + -------- + ```python + builder = ( + ProcessBuilder() + .add_stream("feed", fluid, flow_rate=100, flow_unit="kg/hr") + # Virtual stream as recycle guess (small flow to start) + .add_virtual_stream("recycle_guess", source="feed", + flow_rate=5.0, flow_unit="kg/hr") + # Mix feed with recycle + .add_mixer("mixer", inlets=["feed", "recycle_guess.out"]) + .add_separator("sep", inlet="mixer") + # Connect separator liquid back to virtual stream + .add_recycle("recycle", inlet="sep.liquid", outlet="recycle_guess.out") + .run() + ) + ``` + + Args: + name: Name of the virtual stream. + source: Name of source stream to copy composition from. + flow_rate: Initial flow rate guess. + flow_unit: Unit for flow rate (default "kg/hr"). + temperature: Temperature (if different from source). + temperature_unit: Unit for temperature (default "C"). + pressure: Pressure (if different from source). + pressure_unit: Unit for pressure (default "bara"). + + Returns: + Self for method chaining. + + Note: + Access the virtual stream's output with '.out' suffix in inlet parameters, + e.g., 'virtual_stream_name.out' or just 'virtual_stream_name' (auto-detected). + """ + if source is not None: + source_stream = self._get_outlet(source) + vs = jneqsim.process.equipment.stream.VirtualStream(name, source_stream) + else: + vs = jneqsim.process.equipment.stream.VirtualStream(name) + if flow_rate is not None: + vs.setFlowRate(flow_rate, flow_unit) + if temperature is not None: + vs.setTemperature(temperature, temperature_unit) + if pressure is not None: + vs.setPressure(pressure, pressure_unit) + self.equipment[name] = vs + self.process.add(vs) + return self + + def add_water_stream( + self, + name: str, + temperature: float = None, + temperature_unit: str = "C", + pressure: float = None, + pressure_unit: str = "bara", + flow_rate: float = None, + flow_unit: str = "kg/hr", + ) -> "ProcessBuilder": + """ + Add a water stream to the process. + + Creates a stream containing pure water at specified conditions. + + Args: + name: Name of the stream. + temperature: Temperature value. + temperature_unit: Unit for temperature (default 'C'). + pressure: Pressure value. + pressure_unit: Unit for pressure (default 'bara'). + flow_rate: Flow rate value. + flow_unit: Unit for flow rate (default 'kg/hr'). + + Returns: + Self for method chaining. + + Example: + >>> builder.add_water_stream('water_feed', temperature=80.0, pressure=10.0, flow_rate=1000.0) + """ + water_fluid = jneqsim.thermo.system.SystemSrkEos(273.15 + 25.0, 1.0) + water_fluid.addComponent("water", 1.0) + water_fluid.setMixingRule(2) + water_fluid.init(0) + + s = jneqsim.process.equipment.stream.Stream(name, water_fluid) + if temperature is not None: + s.setTemperature(temperature, temperature_unit) + if pressure is not None: + s.setPressure(pressure, pressure_unit) + if flow_rate is not None: + s.setFlowRate(flow_rate, flow_unit) + self.equipment[name] = s + self.process.add(s) + return self + + def add_three_phase_separator( + self, + name: str, + inlet: str = None, + inlets: List[str] = None, + entrainment: Dict[str, Any] = None, + internal_diameter: float = None, + ) -> "ProcessBuilder": + """ + Add a three-phase separator to the process. + + Three-phase separators separate gas, oil, and water phases. + + Args: + name: Name of the separator. + inlet: Name of primary inlet equipment (for single inlet). + inlets: List of inlet equipment names (for multiple inlets). + entrainment: Optional entrainment settings dict with keys: + - 'value': entrainment fraction + - 'from_phase': source phase ('feed') + - 'unit': unit type ('volume') + - 'from': phase to entrain from ('aqueous', 'oil', 'gas') + - 'to': phase to entrain to ('oil', 'gas', 'aqueous') + internal_diameter: Optional internal diameter in meters. + + Returns: + Self for method chaining. + + Example: + >>> builder.add_three_phase_separator( + ... '1st_stage_sep', + ... inlet='feed_heater', + ... entrainment={'value': 0.001, 'from_phase': 'feed', 'unit': 'volume', + ... 'from': 'aqueous', 'to': 'oil'} + ... ) + """ + if inlet is not None: + inlet_stream = self._get_outlet(inlet) + sep = jneqsim.process.equipment.separator.ThreePhaseSeparator( + name, inlet_stream + ) + else: + sep = jneqsim.process.equipment.separator.ThreePhaseSeparator(name) + + # Add additional inlet streams + if inlets is not None: + for inlet_name in inlets: + inlet_stream = self._get_outlet(inlet_name) + sep.addStream(inlet_stream) + + if entrainment is not None: + sep.setEntrainment( + entrainment.get("value", 0.0), + entrainment.get("from_phase", "feed"), + entrainment.get("unit", "volume"), + entrainment.get("from", "aqueous"), + entrainment.get("to", "oil"), + ) + + if internal_diameter is not None: + sep.setInternalDiameter(internal_diameter) + + self.equipment[name] = sep + self.process.add(sep) + return self + + def add_manifold( + self, + name: str, + inlets: List[str] = None, + split_factors: List[float] = None, + flow_rates: List[float] = None, + flow_unit: str = "MSm3/day", + ) -> "ProcessBuilder": + """ + Add a manifold to the process. + + A manifold combines multiple inlet streams (like a mixer) and then + distributes to multiple outlets (like a splitter). + + Args: + name: Name of the manifold. + inlets: List of inlet equipment/stream names. Supports dot notation. + split_factors: List of split fractions for each outlet (should sum to 1). + Use this OR flow_rates. + flow_rates: List of absolute flow rates for each outlet. + Use -1 for one outlet to auto-calculate remainder. + flow_unit: Unit for flow_rates (default 'MSm3/day'). + Options: 'MSm3/day', 'Sm3/day', 'kg/hr', 'kg/sec', 'mole/sec' + + Returns: + Self for method chaining. + + Notes: + Access manifold outlets using dot notation: + - 'manifold.split_0' for first outlet + - 'manifold.split_1' for second outlet + - 'manifold.mixed' for the mixed stream (before splitting) + + Examples: + # Using split factors + >>> builder.add_manifold('prod_manifold', + ... inlets=['well1', 'well2', 'well3'], + ... split_factors=[0.5, 0.3, 0.2]) + + # Using absolute flow rates + >>> builder.add_manifold('prod_manifold', + ... inlets=['well1', 'well2'], + ... flow_rates=[10.0, 5.0], flow_unit='MSm3/day') + + Note: + For manifolds, flow_rates are set on the internal splitter after + the streams are mixed. Use -1 for one outlet to auto-calculate. + """ + m = jneqsim.process.equipment.manifold.Manifold(name) + if inlets is not None: + for inlet in inlets: + inlet_stream = self._get_outlet(inlet) + m.addStream(inlet_stream) + if flow_rates is not None: + # Manifold uses internal splitter - access via localsplitter field + # First set split factors to initialize the right number of outlets + m.setSplitFactors([1.0 / len(flow_rates)] * len(flow_rates)) + # Then set flow rates on internal splitter + if hasattr(m, "localsplitter"): + m.localsplitter.setFlowRates(flow_rates, flow_unit) + elif split_factors is not None: + m.setSplitFactors(split_factors) + self.equipment[name] = m + self.process.add(m) + return self + + def add_compressor_with_chart( + self, + name: str, + inlet: str, + outlet_pressure: float = None, + pressure_unit: str = "bara", + polytopic_efficiency: float = 0.75, + use_polytropic: bool = True, + speed: float = None, + chart_conditions: Any = None, + chart_speeds: List[float] = None, + chart_flow: List[List[float]] = None, + chart_head: List[List[float]] = None, + chart_eff_flow: List[List[float]] = None, + chart_efficiency: List[List[float]] = None, + surge_flow: List[float] = None, + surge_head: List[float] = None, + use_chart: bool = False, + use_energy_efficiency_chart: bool = True, + ) -> "ProcessBuilder": + """ + Add a compressor with performance charts to the process. + + This method provides full control over compressor performance curves + including surge curves. + + Args: + name: Name of the compressor. + inlet: Name of inlet equipment. + outlet_pressure: Outlet pressure. + pressure_unit: Unit for pressure (default 'bara'). + polytopic_efficiency: Polytropic efficiency (0-1). + use_polytropic: Use polytropic calculation (default True). + speed: Rotational speed in RPM. + chart_conditions: Reference conditions for chart. + chart_speeds: List of speeds for chart curves. + chart_flow: Nested list of flow values for each speed. + chart_head: Nested list of head values for each speed. + chart_eff_flow: Nested list of flow values for efficiency. + chart_efficiency: Nested list of efficiency values. + surge_flow: Flow values for surge curve. + surge_head: Head values for surge curve. + use_chart: Whether to use compressor chart (default False). + use_energy_efficiency_chart: Use energy efficiency chart (default True). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + comp = jneqsim.process.equipment.compressor.Compressor(name, inlet_stream) + comp.setCompressorChartType("interpolate and extrapolate") + + if use_polytropic: + comp.setUsePolytropicCalc(True) + comp.setPolytropicEfficiency(polytopic_efficiency) + else: + comp.setIsentropicEfficiency(polytopic_efficiency) + + if outlet_pressure is not None: + comp.setOutletPressure(outlet_pressure, pressure_unit) + + if speed is not None: + comp.setSpeed(speed) + + # Set performance chart if provided + if chart_conditions is not None and chart_speeds is not None: + comp.getCompressorChart().setCurves( + chart_conditions, + chart_speeds, + chart_flow, + chart_head, + chart_eff_flow, + chart_efficiency, + ) + + comp.getCompressorChart().setUseCompressorChart(use_chart) + comp.setUseEnergyEfficiencyChart(use_energy_efficiency_chart) + + # Set surge curve if provided + if ( + surge_flow is not None + and surge_head is not None + and chart_conditions is not None + ): + comp.getCompressorChart().getSurgeCurve().setCurve( + chart_conditions, surge_flow, surge_head + ) + + self.equipment[name] = comp + self.process.add(comp) + return self + + def add_valve_with_options( + self, + name: str, + inlet: str, + outlet_pressure: float = None, + delta_pressure: float = None, + pressure_unit: str = "bara", + percent_opening: float = None, + tag_name: str = None, + ) -> "ProcessBuilder": + """ + Add a throttling valve with extended options. + + Args: + name: Name of the valve. + inlet: Name of inlet equipment. + outlet_pressure: Outlet pressure value. + delta_pressure: Pressure drop across valve (alternative to outlet_pressure). + pressure_unit: Unit for pressure (default 'bara'). + percent_opening: Valve opening percentage (0-100). + tag_name: Equipment tag for identification. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + v = jneqsim.process.equipment.valve.ThrottlingValve(name, inlet_stream) + + if outlet_pressure is not None: + v.setOutletPressure(outlet_pressure, pressure_unit) + elif delta_pressure is not None: + v.setDeltaPressure(delta_pressure, pressure_unit) + + if percent_opening is not None: + v.setPercentValveOpening(percent_opening) + + if tag_name is not None: + v.setTagName(tag_name) + + self.equipment[name] = v + self.process.add(v) + return self + + def add_splitter_with_flowrates( + self, + name: str, + inlet: str, + flow_rates: List[float] = None, + flow_unit: str = "kg/hr", + split_factors: List[float] = None, + ) -> "ProcessBuilder": + """ + Add a splitter with flow rate specifications. + + Use -1 in flow_rates to indicate "remainder" (balance stream). + + Args: + name: Name of the splitter. + inlet: Name of inlet equipment. + flow_rates: List of flow rates for each split (-1 for remainder). + flow_unit: Unit for flow rates (default 'kg/hr'). + split_factors: Alternative: split by fractions (0 to 1 each). + + Returns: + Self for method chaining. + + Example: + >>> # Split with specific flow rate and remainder + >>> builder.add_splitter_with_flowrates('splitter', 'stream', + ... flow_rates=[-1, 100.0]) + """ + inlet_stream = self._get_outlet(inlet) + spl = jneqsim.process.equipment.splitter.Splitter(name, inlet_stream) + + if flow_rates is not None: + spl.setFlowRates(flow_rates, flow_unit) + elif split_factors is not None: + spl.setSplitFactors(split_factors) + + self.equipment[name] = spl + self.process.add(spl) + return self + + def add_gas_scrubber_with_options( + self, + name: str, + inlet: str, + internal_diameter: float = None, + orientation: str = None, + ) -> "ProcessBuilder": + """ + Add a gas scrubber with configuration options. + + Args: + name: Name of the scrubber. + inlet: Name of inlet equipment. + internal_diameter: Internal diameter in meters. + orientation: 'vertical' or 'horizontal'. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + scrubber = jneqsim.process.equipment.separator.GasScrubber(name, inlet_stream) + + if internal_diameter is not None: + scrubber.setInternalDiameter(internal_diameter) + + if orientation is not None: + scrubber.setOrientation(orientation) + + self.equipment[name] = scrubber + self.process.add(scrubber) + return self + + def add_stream_from_outlet( + self, + name: str, + source: str, + ) -> "ProcessBuilder": + """ + Add a named stream from another equipment's outlet. + + Useful for creating named reference streams from separator outlets + or other equipment. + + Args: + name: Name of the new stream. + source: Source equipment with optional outlet spec (e.g., 'sep.water'). + + Returns: + Self for method chaining. + + Example: + >>> builder.add_stream_from_outlet('water_product', 'separator.water') + """ + source_stream = self._get_outlet(source) + s = jneqsim.process.equipment.stream.Stream(name, source_stream) + self.equipment[name] = s + self.process.add(s) + return self + + def add_process( + self, + sub_process: "ProcessBuilder", + prefix: str = "", + ) -> "ProcessBuilder": + """ + Add all equipment from another ProcessBuilder as a sub-process. + + This allows building modular process sections that can be combined. + + Args: + sub_process: Another ProcessBuilder instance to incorporate. + prefix: Optional prefix to add to all equipment names. + + Returns: + Self for method chaining. + + Example: + >>> compression_train = (ProcessBuilder("Compression") + ... .add_compressor('comp1', 'inlet', pressure=50) + ... .add_cooler('cooler1', 'comp1', temperature=303)) + >>> main_process = (ProcessBuilder("Main") + ... .add_stream('feed', fluid) + ... .add_process(compression_train, prefix='train1_')) + """ + for eq_name, eq in sub_process.equipment.items(): + new_name = f"{prefix}{eq_name}" if prefix else eq_name + self.equipment[new_name] = eq + self.process.add(eq) + return self + + def configure(self, equipment_name: str, **kwargs) -> "ProcessBuilder": + """ + Configure an existing equipment item with additional settings. + + This method allows setting arbitrary properties on equipment + after it has been added to the process. + + Args: + equipment_name: Name of the equipment to configure. + **kwargs: Property-value pairs to set. Property names should + match Java setter methods (without 'set' prefix). + + Returns: + Self for method chaining. + + Example: + >>> builder.add_stream('feed', fluid) + >>> builder.configure('feed', Temperature=(80, 'C'), Pressure=(10, 'bara')) + """ + eq = self.equipment.get(equipment_name) + if eq is None: + raise ValueError(f"Equipment '{equipment_name}' not found") + + for prop, value in kwargs.items(): + setter_name = f"set{prop}" + if hasattr(eq, setter_name): + setter = getattr(eq, setter_name) + if isinstance(value, (tuple, list)): + setter(*value) + else: + setter(value) + else: + raise ValueError( + f"Equipment '{equipment_name}' has no setter '{setter_name}'" + ) + + return self + + # ============ PIPELINE EQUIPMENT ============ + + def add_beggs_brill_pipe( + self, + name: str, + inlet: str, + length: float = 100.0, + diameter: float = 0.1, + elevation: float = 0.0, + wall_roughness: float = 1e-5, + increments: int = 10, + ) -> "ProcessBuilder": + """ + Add a Beggs and Brill pipeline to the process. + + Beggs-Brill is a multiphase flow correlation commonly used for + oil and gas pipelines. + + Args: + name: Name of the pipe. + inlet: Name of inlet equipment. + length: Pipe length in meters (default 100.0). + diameter: Pipe internal diameter in meters (default 0.1). + elevation: Elevation change in meters (default 0.0). + wall_roughness: Pipe wall roughness in meters (default 1e-5). + increments: Number of calculation increments (default 10). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + p = jneqsim.process.equipment.pipeline.PipeBeggsAndBrills(name, inlet_stream) + p.setLength(length) + p.setDiameter(diameter) + p.setElevation(elevation) + p.setPipeWallRoughness(wall_roughness) + p.setNumberOfIncrements(increments) + self.equipment[name] = p + self.process.add(p) + return self + + def add_two_phase_pipe( + self, + name: str, + inlet: str, + length: float = 100.0, + diameter: float = 0.1, + elevation: float = 0.0, + ) -> "ProcessBuilder": + """ + Add a two-phase pipeline to the process. + + Args: + name: Name of the pipe. + inlet: Name of inlet equipment. + length: Pipe length in meters (default 100.0). + diameter: Pipe internal diameter in meters (default 0.1). + elevation: Elevation change in meters (default 0.0). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + p = jneqsim.process.equipment.pipeline.TwoPhasePipeLine(name, inlet_stream) + p.setLength(length) + p.setDiameter(diameter) + if elevation != 0: + p.setElevation(elevation) + self.equipment[name] = p + self.process.add(p) + return self + + # ============ MEASUREMENT DEVICES ============ + + def add_pressure_transmitter( + self, + name: str, + inlet: str, + unit: str = "bara", + min_value: float = 0.0, + max_value: float = 100.0, + ) -> "ProcessBuilder": + """ + Add a pressure transmitter to the process. + + Args: + name: Name of the transmitter. + inlet: Name of stream to measure. + unit: Pressure unit (default 'bara'). + min_value: Minimum value for scaling. + max_value: Maximum value for scaling. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + pt = jneqsim.process.measurementdevice.PressureTransmitter(inlet_stream) + pt.setName(name) + pt.setUnit(unit) + pt.setMinimumValue(min_value) + pt.setMaximumValue(max_value) + self.equipment[name] = pt + self.process.add(pt) + return self + + def add_level_transmitter( + self, + name: str, + separator: str, + min_value: float = 0.0, + max_value: float = 1.0, + ) -> "ProcessBuilder": + """ + Add a level transmitter to the process. + + Args: + name: Name of the transmitter. + separator: Name of separator to measure level. + min_value: Minimum value for scaling (default 0.0). + max_value: Maximum value for scaling (default 1.0). + + Returns: + Self for method chaining. + """ + sep = self.equipment.get(separator) + if sep is None: + raise ValueError(f"Separator '{separator}' not found") + lt = jneqsim.process.measurementdevice.LevelTransmitter(sep) + lt.setName(name) + lt.setMinimumValue(min_value) + lt.setMaximumValue(max_value) + self.equipment[name] = lt + self.process.add(lt) + return self + + def add_flow_transmitter( + self, + name: str, + inlet: str, + unit: str = "kg/hr", + min_value: float = 0.0, + max_value: float = 1000.0, + ) -> "ProcessBuilder": + """ + Add a volume/mass flow transmitter to the process. + + Args: + name: Name of the transmitter. + inlet: Name of stream to measure. + unit: Flow unit (default 'kg/hr'). + min_value: Minimum value for scaling. + max_value: Maximum value for scaling. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + ft = jneqsim.process.measurementdevice.VolumeFlowTransmitter(inlet_stream) + ft.setName(name) + ft.setUnit(unit) + ft.setMinimumValue(min_value) + ft.setMaximumValue(max_value) + self.equipment[name] = ft + self.process.add(ft) + return self + + def add_temperature_transmitter( + self, + name: str, + inlet: str, + unit: str = "C", + min_value: float = -50.0, + max_value: float = 200.0, + ) -> "ProcessBuilder": + """ + Add a temperature transmitter to the process. + + Args: + name: Name of the transmitter. + inlet: Name of stream to measure. + unit: Temperature unit (default 'C'). + min_value: Minimum value for scaling. + max_value: Maximum value for scaling. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + tt = jneqsim.process.measurementdevice.TemperatureTransmitter(inlet_stream) + tt.setName(name) + tt.setUnit(unit) + tt.setMinimumValue(min_value) + tt.setMaximumValue(max_value) + self.equipment[name] = tt + self.process.add(tt) + return self + + # ============ CONTROLLERS ============ + + def add_pid_controller( + self, + name: str, + transmitter: str, + setpoint: float, + kp: float = 1.0, + ti: float = 100.0, + td: float = 0.0, + reverse_acting: bool = False, + ) -> "ProcessBuilder": + """ + Add a PID controller to the process. + + Args: + name: Name of the controller. + transmitter: Name of transmitter to use as input. + setpoint: Controller setpoint value. + kp: Proportional gain (default 1.0). + ti: Integral time constant (default 100.0). + td: Derivative time constant (default 0.0). + reverse_acting: If True, controller output decreases when + process variable increases (default False). + + Returns: + Self for method chaining. + + Note: + After adding, use builder.get(name) to access the controller and + attach it to equipment with equipment.setController(controller). + """ + trans = self.equipment.get(transmitter) + if trans is None: + raise ValueError(f"Transmitter '{transmitter}' not found") + ctrl = jneqsim.process.controllerdevice.ControllerDeviceBaseClass(name) + ctrl.setTransmitter(trans) + ctrl.setControllerSetPoint(setpoint) + ctrl.setControllerParameters(kp, ti, td) + ctrl.setReverseActing(reverse_acting) + self.equipment[name] = ctrl + # Note: Controllers are not added to process.add() in NeqSim + # They are attached to equipment using equipment.setController() + return self + + # ============ FLOW UTILITIES ============ + + def add_flow_setter( + self, + name: str, + inlet: str, + gas_flow: float = None, + oil_flow: float = None, + water_flow: float = None, + gas_unit: str = "Sm3/day", + oil_unit: str = "m3/hr", + water_unit: str = "m3/hr", + ) -> "ProcessBuilder": + """ + Add a flow setter to the process. + + Flow setters adjust the composition and flow rates of a stream + to match specified gas, oil, and water rates. + + Args: + name: Name of the flow setter. + inlet: Name of inlet equipment. + gas_flow: Desired gas flow rate. + oil_flow: Desired oil flow rate. + water_flow: Desired water flow rate. + gas_unit: Gas flow unit (default 'Sm3/day'). + oil_unit: Oil flow unit (default 'm3/hr'). + water_unit: Water flow unit (default 'm3/hr'). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + fs = jneqsim.process.equipment.util.FlowSetter(name, inlet_stream) + if gas_flow is not None: + fs.setGasFlowRate(gas_flow, gas_unit) + if oil_flow is not None: + fs.setOilFlowRate(oil_flow, oil_unit) + if water_flow is not None: + fs.setWaterFlowRate(water_flow, water_unit) + self.equipment[name] = fs + self.process.add(fs) + return self + + def add_flow_rate_adjuster( + self, + name: str, + inlet: str, + gas_flow: float = None, + oil_flow: float = None, + water_flow: float = None, + unit: str = "Sm3/hr", + ) -> "ProcessBuilder": + """ + Add a flow rate adjuster to the process. + + Flow rate adjusters modify stream composition to achieve + desired phase flow rates. + + Args: + name: Name of the flow rate adjuster. + inlet: Name of inlet equipment. + gas_flow: Desired gas flow rate. + oil_flow: Desired oil flow rate. + water_flow: Desired water flow rate (optional). + unit: Flow rate unit for all phases (default 'Sm3/hr'). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + fra = jneqsim.process.equipment.util.FlowRateAdjuster(name, inlet_stream) + if water_flow is not None: + fra.setAdjustedFlowRates(gas_flow or 0.0, oil_flow or 0.0, water_flow, unit) + elif gas_flow is not None or oil_flow is not None: + fra.setAdjustedFlowRates(gas_flow or 0.0, oil_flow or 0.0, unit) + self.equipment[name] = fra + self.process.add(fra) + return self + + # ============ STREAM VARIANTS ============ + + def add_neq_stream( + self, + name: str, + thermo_system: Any, + temperature: float = None, + pressure: float = None, + flow_rate: float = None, + flow_unit: str = "kg/sec", + ) -> "ProcessBuilder": + """ + Add a non-equilibrium stream to the process. + + NeqStream uses non-equilibrium thermodynamics for better accuracy + in some conditions. + + Args: + name: Name of the stream. + thermo_system: Fluid/thermodynamic system. + temperature: Optional temperature in Kelvin. + pressure: Optional pressure in bara. + flow_rate: Optional flow rate. + flow_unit: Unit for flow rate (default 'kg/sec'). + + Returns: + Self for method chaining. + """ + if temperature is not None: + thermo_system.setTemperature(temperature) + if pressure is not None: + thermo_system.setPressure(pressure) + if flow_rate is not None: + thermo_system.setTotalFlowRate(flow_rate, flow_unit) + s = jneqsim.process.equipment.stream.NeqStream(name, thermo_system) + self.equipment[name] = s + self.process.add(s) + return self + + def add_energy_stream(self, name: str, duty: float = 0.0) -> "ProcessBuilder": + """ + Add an energy stream to the process. + + Energy streams are used to connect heat/work flows between equipment. + + Args: + name: Name of the energy stream. + duty: Initial energy/duty in kW (default 0.0). + + Returns: + Self for method chaining. + """ + es = jneqsim.process.equipment.stream.EnergyStream(name) + if duty != 0.0: + es.setDuty(duty) + self.equipment[name] = es + self.process.add(es) + return self + + # ============ ENHANCED SEPARATOR ============ + + def add_separator_with_dimensions( + self, + name: str, + inlet: str, + three_phase: bool = False, + length: float = None, + diameter: float = None, + liquid_level: float = None, + orientation: str = None, + ) -> "ProcessBuilder": + """ + Add a separator with physical dimensions to the process. + + Used for dynamic simulations and mechanical design calculations. + + Args: + name: Name of the separator. + inlet: Name of inlet equipment. + three_phase: If True, creates a ThreePhaseSeparator. + length: Separator length in meters. + diameter: Internal diameter in meters. + liquid_level: Initial liquid level (0-1 fraction). + orientation: 'horizontal' or 'vertical'. + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + if three_phase: + sep = jneqsim.process.equipment.separator.ThreePhaseSeparator( + name, inlet_stream + ) + else: + sep = jneqsim.process.equipment.separator.Separator(name, inlet_stream) + + if length is not None: + sep.setSeparatorLength(length) + if diameter is not None: + sep.setInternalDiameter(diameter) + if liquid_level is not None: + sep.setLiquidLevel(liquid_level) + if orientation is not None: + sep.setOrientation(orientation) + self.equipment[name] = sep + self.process.add(sep) + return self + + # ============ ENHANCED COMPRESSOR ============ + + def add_polytopic_compressor( + self, + name: str, + inlet: str, + pressure: float = None, + polytopic_efficiency: float = 0.75, + use_polytopic_calc: bool = True, + ) -> "ProcessBuilder": + """ + Add a compressor using polytropic calculation method. + + Args: + name: Name of the compressor. + inlet: Name of inlet equipment. + pressure: Outlet pressure in bara. + polytopic_efficiency: Polytropic efficiency (0-1, default 0.75). + use_polytopic_calc: If True, use polytropic calculation (default True). + + Returns: + Self for method chaining. + """ + inlet_stream = self._get_outlet(inlet) + comp = jneqsim.process.equipment.compressor.Compressor(name, inlet_stream) + if pressure is not None: + comp.setOutletPressure(pressure) + comp.setPolytropicEfficiency(polytopic_efficiency) + comp.setUsePolytropicCalc(use_polytopic_calc) + self.equipment[name] = comp + self.process.add(comp) + return self + + # ============ MIXERS ============ + + def add_static_mixer(self, name: str, inlets: List[str] = None) -> "ProcessBuilder": + """ + Add a static mixer to the process. + + Static mixers instantaneously mix streams without pressure drop. + + Args: + name: Name of the mixer. + inlets: List of inlet equipment names (optional, can add later). + + Returns: + Self for method chaining. + """ + m = jneqsim.process.equipment.mixer.StaticMixer(name) + if inlets: + for inlet in inlets: + inlet_stream = self._get_outlet(inlet) + m.addStream(inlet_stream) + self.equipment[name] = m + self.process.add(m) + return self + + def add_static_phase_mixer( + self, name: str, inlets: List[str] = None + ) -> "ProcessBuilder": + """ + Add a static phase mixer to the process. + + Static phase mixers mix phases without equilibrium calculation. + + Args: + name: Name of the mixer. + inlets: List of inlet equipment names (optional). + + Returns: + Self for method chaining. + """ + m = jneqsim.process.equipment.mixer.StaticPhaseMixer(name) + if inlets: + for inlet in inlets: + inlet_stream = self._get_outlet(inlet) + m.addStream(inlet_stream) + self.equipment[name] = m + self.process.add(m) + return self + + # ============ TRANSIENT SIMULATION HELPERS ============ + + def set_transient_mode( + self, equipment_name: str, steady_state: bool = False + ) -> "ProcessBuilder": + """ + Set equipment to transient calculation mode. + + Args: + equipment_name: Name of equipment to configure. + steady_state: If False, use transient calculation (default False). + + Returns: + Self for method chaining. + """ + eq = self.equipment.get(equipment_name) + if eq is None: + raise ValueError(f"Equipment '{equipment_name}' not found") + if hasattr(eq, "setCalculateSteadyState"): + eq.setCalculateSteadyState(steady_state) + else: + raise ValueError( + f"Equipment '{equipment_name}' does not support transient mode" + ) + return self + + def set_valve_opening( + self, valve_name: str, opening: float, min_opening: float = None + ) -> "ProcessBuilder": + """ + Set valve opening percentage. + + Args: + valve_name: Name of the valve. + opening: Valve opening percentage (0-100). + min_opening: Minimum valve opening percentage (optional). + + Returns: + Self for method chaining. + """ + valve = self.equipment.get(valve_name) + if valve is None: + raise ValueError(f"Valve '{valve_name}' not found") + valve.setPercentValveOpening(opening) + if min_opening is not None: + valve.setMinimumValveOpening(min_opening) + return self + + def attach_controller( + self, equipment_name: str, controller_name: str + ) -> "ProcessBuilder": + """ + Attach a controller to equipment. + + Args: + equipment_name: Name of equipment to attach controller to. + controller_name: Name of the controller. + + Returns: + Self for method chaining. + """ + eq = self.equipment.get(equipment_name) + if eq is None: + raise ValueError(f"Equipment '{equipment_name}' not found") + ctrl = self.equipment.get(controller_name) + if ctrl is None: + raise ValueError(f"Controller '{controller_name}' not found") + if hasattr(eq, "setController"): + eq.setController(ctrl) + else: + raise ValueError( + f"Equipment '{equipment_name}' does not support controllers" + ) + return self + + # ============ REACTOR ============ + + def add_reactor( + self, + name: str, + inlet: str = None, + reactor_type: str = "gibbs", ) -> "ProcessBuilder": - """Add a splitter to the process.""" - inlet_stream = self._get_outlet(inlet) - s = jneqsim.process.equipment.splitter.Splitter(name, inlet_stream) - if split_factors: - s.setSplitFactors(split_factors) - self.equipment[name] = s - self.process.add(s) + """ + Add a reactor to the process. + + Args: + name: Name of the reactor. + inlet: Name of inlet equipment (optional). + reactor_type: Type of reactor - 'gibbs' (default), 'plug_flow', etc. + + Returns: + Self for method chaining. + """ + if reactor_type.lower() == "gibbs": + r = jneqsim.process.equipment.reactor.GibbsReactor(name) + else: + r = jneqsim.process.equipment.reactor.GibbsReactor(name) + + if inlet is not None: + inlet_stream = self._get_outlet(inlet) + r.addStream(inlet_stream) + self.equipment[name] = r + self.process.add(r) return self - def add_heat_exchanger( - self, name: str, hot_inlet: str, cold_inlet: str, approach_temp: float = 10.0 + # ============ ABSORPTION/STRIPPING ============ + + def add_simple_absorber( + self, + name: str, + gas_inlet: str = None, + liquid_inlet: str = None, + stages: int = 5, ) -> "ProcessBuilder": - """Add a heat exchanger to the process.""" - hot_stream = self._get_outlet(hot_inlet) - cold_stream = self._get_outlet(cold_inlet) - hx = jneqsim.process.equipment.heatexchanger.HeatExchanger( - name, hot_stream, cold_stream - ) - hx.setApproachTemperature(approach_temp) - self.equipment[name] = hx - self.process.add(hx) + """ + Add a simple absorber to the process. + + Args: + name: Name of the absorber. + gas_inlet: Name of gas inlet stream. + liquid_inlet: Name of liquid inlet stream (e.g., lean TEG). + stages: Number of equilibrium stages (default 5). + + Returns: + Self for method chaining. + """ + absorber = jneqsim.process.equipment.absorber.SimpleTEGAbsorber(name) + absorber.setNumberOfStages(stages) + if gas_inlet is not None: + gas_stream = self._get_outlet(gas_inlet) + absorber.addGasInStream(gas_stream) + if liquid_inlet is not None: + liquid_stream = self._get_outlet(liquid_inlet) + absorber.addSolventInStream(liquid_stream) + self.equipment[name] = absorber + self.process.add(absorber) return self - def add_pipe( + # ============ WELL/RESERVOIR ============ + + def add_well_stream( self, name: str, - inlet: str, - length: float = 100.0, - diameter: float = 0.1, - elevation: float = 0.0, + thermo_system: Any, + gor: float = None, + wc: float = None, + flow_rate: float = None, + flow_unit: str = "Sm3/day", ) -> "ProcessBuilder": - """Add a pipe to the process.""" - inlet_stream = self._get_outlet(inlet) - p = jneqsim.process.equipment.pipeline.AdiabaticPipe(name, inlet_stream) - p.setLength(length) - p.setDiameter(diameter) - if elevation != 0: - p.setElevation(elevation) - self.equipment[name] = p - self.process.add(p) + """ + Add a well stream with GOR and water cut specifications. + + Args: + name: Name of the stream. + thermo_system: Fluid/thermodynamic system. + gor: Gas-oil ratio (Sm3/Sm3). + wc: Water cut (fraction 0-1). + flow_rate: Oil flow rate. + flow_unit: Flow rate unit (default 'Sm3/day'). + + Returns: + Self for method chaining. + """ + s = jneqsim.process.equipment.stream.Stream(name, thermo_system) + if flow_rate is not None: + s.setFlowRate(flow_rate, flow_unit) + self.equipment[name] = s + self.process.add(s) + return self + + def add_stream_to(self, target: str, source: str) -> "ProcessBuilder": + """ + Add a stream from one equipment to another (e.g., adding feed to separator). + + Some equipment types like ThreePhaseSeparator allow multiple inlets. + + Args: + target: Name of target equipment to add stream to. + source: Name of source equipment (the stream to add). + + Returns: + Self for method chaining. + + Example: + >>> builder.add_stream_to('separator', 'recycle_stream') + """ + target_eq = self.equipment.get(target) + if target_eq is None: + raise ValueError(f"Target equipment '{target}' not found") + + source_stream = self._get_outlet(source) + + if hasattr(target_eq, "addStream"): + target_eq.addStream(source_stream) + else: + raise ValueError(f"Equipment '{target}' does not support addStream()") + return self def add_equipment( @@ -697,6 +2946,63 @@ def add_equipment( "splitter": self.add_splitter, "heat_exchanger": self.add_heat_exchanger, "pipe": self.add_pipe, + "gas_scrubber": self.add_gas_scrubber, + "recycle": self.add_recycle, + "distillation_column": self.add_distillation_column, + "teg_absorber": self.add_teg_absorber, + "water_stripper": self.add_water_stripper, + "component_splitter": self.add_component_splitter, + "saturator": self.add_saturator, + "filter": self.add_filter, + "calculator": self.add_calculator, + "setpoint": self.add_setpoint, + "adjuster": self.add_adjuster, + "ejector": self.add_ejector, + "flare": self.add_flare, + "tank": self.add_tank, + "virtual_stream": self.add_virtual_stream, + # Extended equipment types + "water_stream": self.add_water_stream, + "three_phase_separator": self.add_three_phase_separator, + "manifold": self.add_manifold, + "compressor_with_chart": self.add_compressor_with_chart, + "valve_with_options": self.add_valve_with_options, + "splitter_with_flowrates": self.add_splitter_with_flowrates, + "gas_scrubber_with_options": self.add_gas_scrubber_with_options, + "stream_from_outlet": self.add_stream_from_outlet, + # Pipeline equipment + "beggs_brill_pipe": self.add_beggs_brill_pipe, + "two_phase_pipe": self.add_two_phase_pipe, + # Measurement devices + "pressure_transmitter": self.add_pressure_transmitter, + "level_transmitter": self.add_level_transmitter, + "flow_transmitter": self.add_flow_transmitter, + "temperature_transmitter": self.add_temperature_transmitter, + # Controllers + "pid_controller": self.add_pid_controller, + # Flow utilities + "flow_setter": self.add_flow_setter, + "flow_rate_adjuster": self.add_flow_rate_adjuster, + # Stream variants + "neq_stream": self.add_neq_stream, + "energy_stream": self.add_energy_stream, + # Enhanced equipment + "separator_with_dimensions": self.add_separator_with_dimensions, + "polytopic_compressor": self.add_polytopic_compressor, + "polytropic_compressor": self.add_polytopic_compressor, # Alias + # Mixers + "static_mixer": self.add_static_mixer, + "static_phase_mixer": self.add_static_phase_mixer, + # Reactor + "reactor": self.add_reactor, + "gibbs_reactor": self.add_reactor, + # Absorption + "simple_absorber": self.add_simple_absorber, + # Well + "well_stream": self.add_well_stream, + # Recycle helpers + "recycle_loop": self.setup_recycle_loop, + "close_recycle": self.close_recycle_loop, } method = method_map.get(equipment_type.lower()) if method is None: @@ -1004,6 +3310,608 @@ def print_results(self) -> "ProcessBuilder": return self + # ========================================================================= + # GUI HELPER METHODS + # ========================================================================= + + @classmethod + def get_equipment_types(cls) -> List[str]: + """ + Get a list of all available equipment types. + + This is useful for GUI development to populate dropdown menus + or equipment palettes. + + Returns: + List of equipment type strings that can be used with add(). + + Example: + >>> types = ProcessBuilder.get_equipment_types() + >>> print(types) + ['stream', 'compressor', 'separator', 'valve', ...] + """ + # This mirrors the method_map keys + return [ + # Core equipment + "stream", + "separator", + "three_phase_separator", + "compressor", + "pump", + "valve", + "heater", + "cooler", + "mixer", + "splitter", + "heat_exchanger", + "pipe", + "gas_scrubber", + "recycle", + "virtual_stream", + # Extended equipment + "distillation_column", + "teg_absorber", + "water_stripper", + "component_splitter", + "saturator", + "filter", + "calculator", + "setpoint", + "adjuster", + "ejector", + "flare", + "tank", + # Pipeline + "beggs_brill_pipe", + "two_phase_pipe", + # Measurement + "pressure_transmitter", + "level_transmitter", + "flow_transmitter", + "temperature_transmitter", + # Control + "pid_controller", + # Flow utilities + "flow_setter", + "flow_rate_adjuster", + # Streams + "neq_stream", + "energy_stream", + "water_stream", + "well_stream", + # Enhanced + "separator_with_dimensions", + "polytopic_compressor", + "compressor_with_chart", + # Mixers + "static_mixer", + "static_phase_mixer", + # Reactor + "reactor", + "simple_absorber", + ] + + @classmethod + def get_equipment_parameters(cls, equipment_type: str) -> Dict[str, Any]: + """ + Get parameter schema for an equipment type. + + Returns a dictionary describing the parameters for the specified + equipment type, useful for generating dynamic forms in a GUI. + + Args: + equipment_type: Type of equipment (e.g., 'compressor', 'separator') + + Returns: + Dictionary with parameter names, types, defaults, and descriptions. + + Example: + >>> schema = ProcessBuilder.get_equipment_parameters('compressor') + >>> print(schema) + { + 'inlet': {'type': 'str', 'required': True, 'description': 'Inlet equipment name'}, + 'outlet_pressure': {'type': 'float', 'required': False, 'unit': 'bara'}, + ... + } + """ + # Parameter schemas for common equipment types + schemas = { + "stream": { + "fluid": { + "type": "fluid", + "required": True, + "description": "Thermodynamic system", + }, + "temperature": { + "type": "float", + "required": False, + "unit": "K", + "description": "Temperature", + }, + "pressure": { + "type": "float", + "required": False, + "unit": "bara", + "description": "Pressure", + }, + "flow_rate": { + "type": "float", + "required": False, + "description": "Flow rate", + }, + "flow_unit": { + "type": "str", + "required": False, + "default": "kg/sec", + "options": ["kg/sec", "kg/hr", "MSm3/day", "Sm3/day", "mole/sec"], + }, + }, + "separator": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + }, + "three_phase_separator": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + }, + "compressor": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "outlet_pressure": { + "type": "float", + "required": False, + "unit": "bara", + "description": "Outlet pressure", + }, + "isentropic_efficiency": { + "type": "float", + "required": False, + "default": 0.75, + "min": 0.0, + "max": 1.0, + "description": "Isentropic efficiency", + }, + "use_polytropic": { + "type": "bool", + "required": False, + "default": False, + "description": "Use polytropic calculation", + }, + }, + "pump": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "outlet_pressure": { + "type": "float", + "required": False, + "unit": "bara", + "description": "Outlet pressure", + }, + "efficiency": { + "type": "float", + "required": False, + "default": 0.75, + "min": 0.0, + "max": 1.0, + "description": "Pump efficiency", + }, + }, + "valve": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "outlet_pressure": { + "type": "float", + "required": False, + "unit": "bara", + "description": "Outlet pressure", + }, + }, + "heater": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "outlet_temperature": { + "type": "float", + "required": False, + "unit": "C", + "description": "Outlet temperature", + }, + }, + "cooler": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "outlet_temperature": { + "type": "float", + "required": False, + "unit": "C", + "description": "Outlet temperature", + }, + }, + "mixer": { + "inlets": { + "type": "list[str]", + "required": False, + "description": "List of inlet equipment/stream names", + }, + }, + "splitter": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "split_fractions": { + "type": "list[float]", + "required": False, + "description": "Split fractions for each outlet", + }, + }, + "pipe": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + "length": { + "type": "float", + "required": False, + "unit": "m", + "description": "Pipe length", + }, + "diameter": { + "type": "float", + "required": False, + "unit": "m", + "description": "Pipe inner diameter", + }, + }, + "gas_scrubber": { + "inlet": { + "type": "str", + "required": True, + "description": "Inlet equipment/stream name", + }, + }, + "virtual_stream": { + "source": { + "type": "str", + "required": False, + "description": "Source stream to copy from", + }, + "flow_rate": { + "type": "float", + "required": False, + "description": "Initial flow rate guess", + }, + "flow_unit": { + "type": "str", + "required": False, + "default": "kg/hr", + "options": ["kg/hr", "kg/sec", "MSm3/day"], + }, + }, + "recycle": { + "inlet": { + "type": "str", + "required": False, + "description": "Actual stream to recycle", + }, + "outlet": { + "type": "str", + "required": False, + "description": "Virtual stream to connect to", + }, + "tolerance": { + "type": "float", + "required": False, + "default": 1e-4, + "description": "Convergence tolerance", + }, + "priority": { + "type": "int", + "required": False, + "description": "Solve priority (higher = later)", + }, + }, + } + + return schemas.get(equipment_type.lower(), {}) + + def get_outlets(self, equipment_name: str) -> List[str]: + """ + Get available outlet types for an equipment. + + Returns the outlet connection points available for the specified + equipment, useful for drawing connections in a visual editor. + + Args: + equipment_name: Name of equipment in the process. + + Returns: + List of outlet names (e.g., ['gas', 'liquid', 'oil'] for 3-phase sep) + + Example: + >>> builder.add_three_phase_separator('sep', 'feed') + >>> outlets = builder.get_outlets('sep') + >>> print(outlets) # ['gas', 'liquid', 'oil', 'water'] + """ + equip = self.equipment.get(equipment_name) + if equip is None: + return [] + + outlets = [] + + # Check for various outlet methods + if hasattr(equip, "getGasOutStream"): + outlets.append("gas") + if hasattr(equip, "getLiquidOutStream"): + outlets.append("liquid") + if hasattr(equip, "getOilOutStream"): + outlets.append("oil") + if hasattr(equip, "getWaterOutStream"): + outlets.append("water") + if hasattr(equip, "getOutletStream"): + if not outlets: # Only add 'out' if no specific outlets + outlets.append("out") + if hasattr(equip, "getOutStream"): + if not outlets: + outlets.append("out") + + # For manifolds, add mixed stream outlet + if hasattr(equip, "getMixedStream"): + outlets.append("mixed") + + # For splitters and manifolds, check for multiple split outlets + if hasattr(equip, "getSplitStream"): + # Splitters/Manifolds have numbered outlets + try: + for i in range(10): # Check up to 10 outlets + if equip.getSplitStream(i) is not None: + outlets.append(f"split_{i}") + except: + pass + + return outlets if outlets else ["out"] + + def get_inlets(self, equipment_name: str) -> List[str]: + """ + Get inlet connection points for an equipment. + + Args: + equipment_name: Name of equipment in the process. + + Returns: + List of inlet names (usually ['inlet'] or ['hot_inlet', 'cold_inlet']) + """ + equip = self.equipment.get(equipment_name) + if equip is None: + return [] + + inlets = [] + + # Check for various inlet patterns + if hasattr(equip, "addStream"): + inlets.append("inlet") # Mixers accept multiple + if hasattr(equip, "setInletStream"): + inlets.append("inlet") + if hasattr(equip, "setFeedStream"): + inlets.append("feed") + + # Heat exchangers have two inlets + equip_class = type(equip).__name__ + if "HeatExchanger" in equip_class: + inlets = ["hot_inlet", "cold_inlet"] + + return inlets if inlets else ["inlet"] + + def validate_connection(self, source: str, target_equipment: str) -> Dict[str, Any]: + """ + Validate if a connection between equipment is valid. + + Checks if the source outlet exists and if the target can accept + the connection. Useful for GUI validation before allowing connections. + + Args: + source: Source in format 'equipment.outlet' or just 'equipment' + target_equipment: Target equipment name + + Returns: + Dictionary with 'valid' (bool), 'message' (str), and 'warnings' (list) + + Example: + >>> result = builder.validate_connection('separator.gas', 'compressor') + >>> if result['valid']: + ... builder.add_compressor('comp', 'separator.gas', ...) + """ + result = {"valid": True, "message": "Connection is valid", "warnings": []} + + # Parse source + if "." in source: + equip_name, outlet_type = source.split(".", 1) + else: + equip_name = source + outlet_type = None + + # Check source equipment exists + source_equip = self.equipment.get(equip_name) + if source_equip is None: + result["valid"] = False + result["message"] = f"Source equipment '{equip_name}' not found" + return result + + # Check outlet type exists + if outlet_type: + available_outlets = self.get_outlets(equip_name) + if outlet_type.lower() not in [o.lower() for o in available_outlets]: + result["valid"] = False + result["message"] = ( + f"Outlet '{outlet_type}' not available on '{equip_name}'. " + f"Available: {available_outlets}" + ) + return result + + # Check target equipment exists (if already added) + if target_equipment in self.equipment: + result["warnings"].append( + f"Target '{target_equipment}' already exists - will reconnect" + ) + + # Try to get the actual stream to validate it works + try: + stream = self._get_outlet(source) + if stream is None: + result["valid"] = False + result["message"] = f"Could not get stream from '{source}'" + return result + except Exception as e: + result["valid"] = False + result["message"] = f"Error accessing '{source}': {str(e)}" + return result + + return result + + def to_dict(self) -> Dict[str, Any]: + """ + Export current process configuration as a dictionary. + + This can be used to save process designs, or to recreate + the process later using from_dict(). + + Returns: + Dictionary representation of the process. + + Example: + >>> config = builder.to_dict() + >>> # Save to file + >>> import json + >>> with open('process.json', 'w') as f: + ... json.dump(config, f, indent=2) + """ + config = { + "process": {"name": self._name}, + "equipment": [], + "connections": [], + } + + # Note: This captures equipment names but not full reconstruction info + # For full reconstruction, use YAML config from the start + for name, equip in self.equipment.items(): + equip_info = { + "name": name, + "type": type(equip).__name__, + } + + # Try to get outlet info + outlets = self.get_outlets(name) + if outlets: + equip_info["outlets"] = outlets + + # Try to get some properties + try: + if hasattr(equip, "getOutletStream"): + stream = equip.getOutletStream() + if stream: + equip_info["outlet_temperature_C"] = round( + stream.getTemperature("C"), 2 + ) + equip_info["outlet_pressure_bara"] = round( + stream.getPressure("bara"), 2 + ) + except: + pass + + config["equipment"].append(equip_info) + + return config + + def to_yaml(self) -> str: + """ + Export current process configuration as a YAML string. + + Returns: + YAML string representation of the process. + + Example: + >>> yaml_str = builder.to_yaml() + >>> print(yaml_str) + >>> # Or save to file + >>> with open('process.yaml', 'w') as f: + ... f.write(yaml_str) + """ + import yaml + + config = self.to_dict() + return yaml.dump(config, default_flow_style=False, sort_keys=False) + + def get_equipment_list(self) -> List[Dict[str, Any]]: + """ + Get a list of all equipment with their properties. + + Useful for populating equipment lists in GUI. + + Returns: + List of dictionaries with equipment info. + + Example: + >>> equipment = builder.get_equipment_list() + >>> for eq in equipment: + ... print(f"{eq['name']}: {eq['type']}") + """ + equipment_list = [] + for name, equip in self.equipment.items(): + info = { + "name": name, + "type": type(equip).__name__, + "outlets": self.get_outlets(name), + "inlets": self.get_inlets(name), + } + equipment_list.append(info) + return equipment_list + + def get_connection_graph(self) -> Dict[str, List[str]]: + """ + Get the connection graph of the process. + + Returns a dictionary where keys are equipment names and values + are lists of downstream equipment names. Useful for drawing + process flow diagrams. + + Returns: + Dictionary mapping equipment to their downstream connections. + + Example: + >>> graph = builder.get_connection_graph() + >>> print(graph) + {'feed': ['separator'], 'separator': ['compressor', 'pump'], ...} + """ + # This is a simplified version - full implementation would track + # actual connections made during build + graph = {name: [] for name in self.equipment} + + # For now, return empty graph - connections aren't tracked + # A full implementation would require tracking during add_* calls + return graph + def save_results(self, filename: str, format: str = "json") -> "ProcessBuilder": """ Save simulation results to a file. @@ -2530,3 +5438,350 @@ def energystream(name: str, power: float = 0.0) -> Any: if not _loop_mode: processoperations.add(es) return es + + +# ============================================================================= +# YAML/Config-Based Process Creation Functions +# ============================================================================= + + +def create_fluid_from_config(config: Dict[str, Any]) -> Any: + """ + Create a fluid (thermodynamic system) from a configuration dictionary. + + Supports various fluid creation methods: + - Predefined types: 'dry gas', 'black oil', etc. + - Custom compositions with equation of state + - DataFrame-based reservoir fluids + + Args: + config: Dictionary with fluid configuration. Supported keys: + - type: 'predefined', 'custom', or 'dataframe' (default 'custom') + - model: Equation of state ('srk', 'pr', 'cpa', etc.) + - name: Predefined fluid name (for type='predefined') + - temperature: Initial temperature in Kelvin (default 298.15) + - pressure: Initial pressure in bara (default 1.01325) + - components: List of component definitions for custom fluids + - dataframe: Dict with 'data' for DataFrame-based fluids + + Returns: + Fluid object (thermodynamic system). + + Example: + >>> # Predefined fluid + >>> config = {'type': 'predefined', 'name': 'dry gas'} + >>> fluid = create_fluid_from_config(config) + + >>> # Custom fluid with composition + >>> config = { + ... 'model': 'srk', + ... 'temperature': 303.15, + ... 'pressure': 50.0, + ... 'components': [ + ... {'name': 'methane', 'moles': 0.9}, + ... {'name': 'ethane', 'moles': 0.05}, + ... {'name': 'propane', 'moles': 0.03}, + ... {'name': 'n-butane', 'moles': 0.02} + ... ] + ... } + >>> fluid = create_fluid_from_config(config) + + >>> # Custom with flow rates + >>> config = { + ... 'model': 'cpa', + ... 'temperature': 288.15, + ... 'pressure': 100.0, + ... 'components': [ + ... {'name': 'water', 'rate': 100.0, 'unit': 'kg/hr'}, + ... {'name': 'MEG', 'rate': 500.0, 'unit': 'kg/hr'} + ... ] + ... } + >>> fluid = create_fluid_from_config(config) + """ + from neqsim.thermo import thermoTools + + fluid_type = config.get("type", "custom") + + # Predefined fluid types + if fluid_type == "predefined": + name = config.get("name", "dry gas") + return thermoTools.createfluid(name) + + # Custom fluid with components + model = config.get("model", "srk") + temperature = config.get("temperature", 298.15) + pressure = config.get("pressure", 1.01325) + + # Create base fluid + thermo_system = thermoTools.fluid(model, temperature, pressure) + + # Add components + components = config.get("components", []) + for comp in components: + comp_name = comp.get("name") + if comp_name is None: + continue + + # Support different ways to specify amount + if "moles" in comp: + thermoTools.addComponent(thermo_system, comp_name, comp["moles"]) + elif "rate" in comp: + rate = comp["rate"] + unit = comp.get("unit", "mol/sec") + thermoTools.addComponent(thermo_system, comp_name, rate, unit) + elif "mole_fraction" in comp: + # Will normalize later + thermoTools.addComponent(thermo_system, comp_name, comp["mole_fraction"]) + else: + # Default to 1 mole + thermoTools.addComponent(thermo_system, comp_name, 1.0) + + # Apply mixing rule if specified + mixing_rule = config.get("mixing_rule") + if mixing_rule: + ge_model = config.get("ge_model", "") + thermoTools.mixingRule(thermo_system, mixing_rule, ge_model) + + # Enable multiphase if specified + if config.get("multiphase", False): + thermoTools.multiphase(thermo_system, 1) + + # Enable solid check if specified + if config.get("solid_check", False): + thermoTools.solidcheck(thermo_system, 1) + + return thermo_system + + +def create_process_from_config( + config: Union[str, Dict[str, Any]], + fluids: Dict[str, Any] = None, + run: bool = True, +) -> "ProcessBuilder": + """ + Create a complete process from a YAML file or configuration dictionary. + + This is a general-purpose function for building NeqSim processes from + configuration files. It can automatically create fluids from configuration + or use pre-created fluid objects. + + Args: + config: Either a path to a YAML file or a configuration dictionary. + fluids: Optional dictionary mapping fluid names to fluid objects. + If the config includes a 'fluids' section, fluids are created + automatically and merged with this dictionary. + run: If True (default), run the process after building. + + Returns: + ProcessBuilder instance with the built process. + + YAML Format: + The YAML file should have this structure: + + ```yaml + # Process configuration + name: "My Process" + + # Optional: Define fluids inline (auto-created) + fluids: + feed: + model: srk + temperature: 303.15 + pressure: 50.0 + components: + - name: methane + moles: 0.85 + - name: ethane + moles: 0.10 + - name: propane + moles: 0.05 + + glycol: + model: cpa + temperature: 298.15 + pressure: 1.0 + components: + - name: MEG + rate: 100.0 + unit: kg/hr + + # Equipment list (processed in order) + equipment: + - type: stream + name: inlet + fluid: feed # References fluid defined above or in fluids dict + temperature: 303.15 + pressure: 50.0 + flow_rate: 10.0 + flow_unit: MSm3/day + + - type: heater + name: heater1 + inlet: inlet + temperature: 320.0 + + - type: separator + name: sep1 + inlet: heater1 + + - type: compressor + name: comp1 + inlet: sep1.gas + pressure: 100.0 + + - type: cooler + name: cooler1 + inlet: comp1 + temperature: 303.15 + + - type: splitter + name: split1 + inlet: cooler1 + split_factors: [0.7, 0.3] + + - type: mixer + name: mixer1 + inlets: + - split1.split_0 + - some_other_stream + ``` + + Equipment Types: + The following equipment types are supported (use lowercase): + - stream: Process stream (requires 'fluid' reference) + - heater, cooler: Temperature changers + - separator, three_phase_separator: Phase separators + - compressor, expander: Pressure changers + - pump, valve: Liquid/gas handling + - heat_exchanger: Two-stream heat exchange + - mixer, splitter: Stream combining/splitting + - manifold: Multiple inlet/outlet handling + - recycle, virtual_stream: Recycle loops + - distillation_column, absorber: Columns + - And many more... + + Outlet Notation: + For equipment with multiple outlets, use dot notation: + - separator: .gas, .liquid + - three_phase_separator: .gas, .oil, .water + - splitter: .split_0, .split_1, .split_2, ... + - manifold: .mixed (combined), .split_0, .split_1, ... + - virtual_stream: .out + + Example: + >>> # From YAML file + >>> process = create_process_from_config('process.yaml') + >>> print(process.results_dataframe()) + + >>> # From dictionary with pre-created fluids + >>> from neqsim.thermo import fluid, addComponent + >>> feed = fluid('srk') + >>> addComponent(feed, 'methane', 0.9) + >>> addComponent(feed, 'ethane', 0.1) + >>> + >>> config = { + ... 'name': 'Simple Compression', + ... 'equipment': [ + ... {'type': 'stream', 'name': 'inlet', 'fluid': 'feed', + ... 'temperature': 300, 'pressure': 10, 'flow_rate': 5}, + ... {'type': 'compressor', 'name': 'comp1', 'inlet': 'inlet', + ... 'pressure': 50} + ... ] + ... } + >>> process = create_process_from_config(config, fluids={'feed': feed}) + + >>> # Don't run immediately + >>> process = create_process_from_config('process.yaml', run=False) + >>> # ... modify process ... + >>> process.run() + + See Also: + - ProcessBuilder.from_yaml(): Class method for YAML loading + - ProcessBuilder.from_dict(): Class method for dict loading + - create_fluid_from_config(): Create fluids from config + """ + # Load config from file if string path provided + if isinstance(config, str): + try: + import yaml + except ImportError: + raise ImportError( + "PyYAML is required for YAML support. Install with: pip install pyyaml" + ) + + with open(config, "r") as f: + config = yaml.safe_load(f) + + # Initialize fluids dictionary + all_fluids = fluids.copy() if fluids else {} + + # Create fluids from config if 'fluids' section exists + if "fluids" in config: + for fluid_name, fluid_config in config["fluids"].items(): + if fluid_name not in all_fluids: # Don't override provided fluids + all_fluids[fluid_name] = create_fluid_from_config(fluid_config) + + # Build process using ProcessBuilder + builder = ProcessBuilder.from_dict(config, all_fluids) + + # Run if requested + if run: + builder.run() + + return builder + + +def load_process_config(yaml_path: str) -> Dict[str, Any]: + """ + Load a process configuration from a YAML file without creating the process. + + Useful for inspecting or modifying configurations before building. + + Args: + yaml_path: Path to YAML configuration file. + + Returns: + Dictionary with the configuration. + + Example: + >>> config = load_process_config('process.yaml') + >>> # Modify configuration + >>> config['equipment'][0]['pressure'] = 100.0 + >>> # Now create process + >>> process = create_process_from_config(config) + """ + try: + import yaml + except ImportError: + raise ImportError( + "PyYAML is required for YAML support. Install with: pip install pyyaml" + ) + + with open(yaml_path, "r") as f: + return yaml.safe_load(f) + + +def save_process_config(config: Dict[str, Any], yaml_path: str) -> None: + """ + Save a process configuration to a YAML file. + + Args: + config: Configuration dictionary. + yaml_path: Path to save the YAML file. + + Example: + >>> config = { + ... 'name': 'My Process', + ... 'equipment': [...] + ... } + >>> save_process_config(config, 'process.yaml') + """ + try: + import yaml + except ImportError: + raise ImportError( + "PyYAML is required for YAML support. Install with: pip install pyyaml" + ) + + with open(yaml_path, "w") as f: + yaml.dump(config, f, default_flow_style=False, sort_keys=False)