neqsim-agentic-process-optimization

# Agentic Process-Model Optimization This skill is the recipe for an **agent that optimizes a large, already-built multi-area plant** (e.g. an offshore separation + recompression + export train) by turning the newest NeqSim automation and introspection APIs into a robust optimization loop. It assumes the flowsheet is a `ProcessModel` assembled from several named `ProcessSystem` areas (see `neqsim-platform-modeling`). Use `neqsim-optimization-and-doe` for the *algorithm* (SQP, PSO, BatchStudy, ProcessSimulationEvaluator → SciPy/Pyomo). Use *this* skill for the *plumbing*: how to read the decision space, evaluate one trial robustly, gate feasibility, and score the objective from real equipment results. --- ## 1. The four pillars (all verified in NeqSim ≥ 3.13.0) | Need | API | Returns | |------|-----|---------| | Decision space (bounded knobs) | `ProcessAutomation.getAdjustableParameters()` / `getAdjustableParametersJson()` | `List<AdjustableParameter>` with `name/address/unit/lowerBound/upperBound/source` | | Robust convergence of a coupled plant | `ProcessModel.runUntilConverged(int maxIterations, double tolerance)` | `boolean` converged; pair with `getConvergenceReportJson()` | | Per-trial feasibility / failure gating | `ProcessModel.getRunStatus()` / `getRunStatusJson()`, `ProcessSystem.getRunStatus()` | `RunStatus` (completed/success/failedUnitName/failedUnitError) | | Objective + constraints from equipment | `Compressor.getOperatingPoint()`, `Standard_ASTM_D6377.RvpResult` | power, surge/stonewall margins; certified RVP | > **Why these matter for an agent:** they replace the fragile "call `.run()` twice > and hope" pattern with explicit *did-it-converge* and *did-any-unit-fail* > signals, and they expose objective/constraint numbers (compression power, surge > distance, RVP spec) as structured JSON the agent can parse without walking Java > object trees. --- ## 2. Discover the decision space ```python from neqsim import jneqsim # or devtools `ns` (see §8) import json auto = plant.getAutomation() # plant = ProcessModel # NOTE: jpype returns java.lang.String, not Python str — json.loads needs str(...). params = json.loads(str(auto.getAdjustableParametersJson())) for p in params["parameters"]: print(p["name"], p["address"], p["unit"], p["lowerBound"], p["upperBound"], p["source"]) ``` - `source = "INPUT_VARIABLE"` → a settable equipment input (compressor outlet pressure, heater outlet temperature, valve outlet pressure, …). - `source = "ADJUSTER"` → a knob already wired to an `Adjuster`; let the model solve it, do **not** also optimize it (double control = divergence). - `UNBOUNDED_THRESHOLD = 1.0e9` → bounds at/above this are "no real bound"; the agent MUST supply a physically meaningful `[lo, hi]` before optimizing. > `getAdjustableParameters()` surfaces **model-side** inputs with bounds. If your > decision variables live in a Python driver (e.g. a `ProcessInput` dataclass or > a year selector), map each one explicitly to a NeqSim address or a rebuild > argument — the registry will not invent them for you. > **Prerequisite — the model must already carry the optimization data basis.** > `getAdjustableParameters()` only surfaces knobs the *model* has, and capacity > constraints only fire if the equipment has design limits. Before optimizing, > confirm the model was built with: line sizes (ID, length, roughness, elevation) > and manifold/header sizes for hydraulics; valve/choke **Cv** (and choke > Cv-vs-opening); compressor/pump **performance maps + speeds**; separator > dimensions + design **K**; equipment **design limits** (rated power, surge > margin, NPSH, erosional velocity, design P/T, MAWP, valve max Cv); manipulable > setpoints with **physical bounds**; and the **objective + economic/spec basis**. > Use `enterprise-process-model-build-verify` with > `target_fidelity="optimization_ready"` (its `OPTIMIZATION_DATA_BASIS` / > `optimization_data_basis` output) to gather and gate this basis, and > `neqsim-process-modeling` for the community build checklist. --- ## 3. Robust single-trial evaluation (the core helper) Always gate on convergence **and** run status. A trial that diverges or throws in one unit must return a large penalty, never a misleading objective. ```python def converge(plant, max_iter=30, tol=5e-3, settle_passes=2, soft_maxerr=0.05): """Run the coupled ProcessModel until boundary streams stop moving. Recycle-heavy plants have near-zero-flow anti-surge recycles that inflate the *relative* boundary-error metric (a tiny stream gives a large % error even when physically settled), so a strict `tol=1e-4` rarely passes. Accept a trial when NO unit threw AND the model either strictly converged or reached a relaxed boundary band; keep genuine unit failures infeasible. """ converged = bool(plant.runUntilConverged(int(max_iter), float(tol))) for _ in range(settle_passes): # settle slow recompression loops try: plant.run() except Exception: break report = json.loads(str(plant.getConvergenceReportJson())) # str() — jpype String status = json.loads(str(plant.getRunStatusJson())) # str() — jpype String failed = status.get("failedUnitName") not in (None, "", "null") max_err = report.get("maxError", float("inf")) soft_ok = (max_err == max_err) and (max_err < soft_maxerr) # not NaN and small ok = (not failed) and (converged or soft_ok) return ok, report, status def evaluate(plant, setpoints, *, max_iter=30, tol=5e-3): """Apply setpoints, converge, score. Returns (objective, record).""" try: apply_setpoints(plant, setpoints) # see §4 ok, report, status = converge(plant, max_iter, tol) if not ok: failed = status.get("failedUnitName") or report.get("maxError") return 1e9, {"feasible": False, "reason": f"non-converged/{failed}", **setpoints} obj, parts = objective_and_constraints(plant) # see §5 return obj, {"feasible": True, **parts, **setpoints} except Exception as exc: # JVM exception from a unit return 1e9, {"feasible": False, "reason": f"exception:{exc}", **setpoints} ``` Key rules: - **Penalty, not crash.** Optimizers must keep exploring after a bad trial. - **`maxError` is a *relative* boundary error.** Near-zero anti-surge recycles inflate it to ~0.8 even when settled — don't demand `tol=1e-4`. Gate on *unit failure* (`failedUnitName`) plus a relaxed `maxError` band, and add a couple of settling `run()` passes. - **Twice-run is obsolete as a *gate*** — `runUntilConverged` drives convergence; the extra passes only damp slow recompression loops. Raise `max_iter`, not hacks. - **Bound every trial with the solver watchdog (opt-in, NeqSim releases after 3.26).** A trial that never converges must not hang the optimizer: call `plant.setMaxRunTimeSeconds(120.0)` (spans all areas and all passes of `runUntilConverged`) and `plant.setStallDetection(8, 1e-3)` (no residual improvement for 8 iterations) once, and call `plant.requestStop()` from a watchdog thread to cancel from outside. A tripped run returns normally; read `status["terminationReason"]` (`TIMED_OUT` / `STALLED` / `STOPPED`), `getRunStatus().getBudgetUnitName()` (unit that was running) and `plant.getLastRunDiagnostics().toJson()`, and score the trial as infeasible. Cooperative only: one unit that never returns cannot be interrupted. Default is off. See `docs/process/processmodel/run_status.md`. - **Record everything.** Keep a list of `record` dicts → tornado/trace plots and `results.json` later. Report the **best feasible trial** (min objective among `feasible=True`), never the last `evaluate()` result — the final call can land on a non-converged retry and write `null` power/RVP/surge into `results.json`. - **One lever may be infeasible.** If a single setpoint (e.g. oil-heater T) can't satisfy two coupled constraints (RVP *and* compressor surge), say so and add the relieving lever (recompression suction pressure) to `DECISION_VARS`; switch the optimizer from `minimize_scalar` to `scipy.minimize(Nelder-Mead, bounds=…)`. --- ## 4. Apply setpoints (string-addressable, dirty-tracked) Prefer `ProcessAutomation` string addresses over walking the object graph: ```python def apply_setpoints(plant, sp): auto = plant.getAutomation() updates = {} # address -> value (one unit family) if "export_P_bara" in sp: updates["Compression::export compressor.outletPressure"] = sp["export_P_bara"] auto.setValues(updates, "bara", False) # batch, no run yet # temperatures use a different unit -> separate batch if "oil_heater_T_C" in sp: auto.setVariableValue("Sep train A::oil heater second stage.outletTemperature", sp["oil_heater_T_C"], "C") auto.setVariableValue("Sep train B::oil heater second stage.outletTemperature", sp["oil_heater_T_C"], "C") # convergence happens in converge() via runUntilConverged ``` If addresses are uncertain, self-heal: `auto.setVariableValueSafe(addr, val, unit)` returns JSON with an auto-corrected address instead of throwing. Use `auto.validateAddress(addr)` (returns `None` when valid) as a pre-flight check. When the plant is built imperatively (handles in Python scope), it is equally valid to set levers directly on the unit and let `converge()` settle them: `train_A.getUnit("oil heater second stage").setOutTemperature(T, "C")`. --- ## 5. Objective and constraints from real equipment ### Compression power + surge/stonewall margin ```python def compressor_metrics(plant): total_power_MW = 0.0 min_surge_margin = float("inf") within_chart_all = True for area in plant.getAllProcesses(): # each ProcessSystem for u in area.getUnitOperations(): if u.getClass().getSimpleName() != "Compressor": continue op = json.loads(str(u.getOperatingPointJson())) # str() — jpype String p = op.get("power_MW") if p == p: # not NaN total_power_MW += p d = op.get("distanceToSurge") # fraction; NaN if no chart if d == d: min_surge_margin = min(min_surge_margin, d) if op.get("withinChart") is False: within_chart_all = False return total_power_MW, min_surge_margin, within_chart_all ``` `getOperatingPoint()` / `getOperatingPointJson()` fields: `power_MW`, `polytropicEfficiency`, `head_kJkg`, `flow_m3hr`, `speed_rpm`, `distanceToSurge`, `distanceToStoneWall`, `surgeFlowRateMargin_m3hr`, `withinChart`, `limitingConstraint` (`none`/`surge`/`stonewall`/`no_chart`). Uncomputable margins are `NaN` — always test `x == x` before using them. ### Compressor control mode: solve-speed vs predictive (choose per objective) A charted compressor can run two ways, and **the mode decides what is an input vs an output** in your optimization — pick it deliberately: | Mode | Setup | Fixed (input) | Computed (output) | Use for | |------|-------|---------------|-------------------|---------| | **Solve-speed** (default for most plants) | `setOutletPressure(P)` + `setUseCompressorChart(True)` + `setSolveSpeed(True)` | discharge pressure | **speed, power, surge margin** | spec/capacity/max-throughput studies where the network has fixed pressure boundaries | | **Predictive** | `setUseCompressorChart(True)` + `setSpeed(rpm)` + `setSolveSpeed(False)` (do **not** pin outlet P) | **shaft speed** | **discharge pressure, head, power, surge margin** | speed-as-decision-variable optimization, surge/turndown and dynamic studies | Rules of thumb: - If the discharge feeds a **fixed pressure boundary** (export pipeline pressure, a process level held by a downstream scrubber/level controller, an import header), **solve-speed is the physically correct steady-state control mode** — the machine trims speed to hold that pressure, and *speed / power / surge margin are the outputs* you optimize against. This is the right mode for "max production while meeting RVP / dew-point / cricondenbar", because product specs are set by separation temperatures/pressures, not by throughput, so the binding limit is compressor power/surge or separator gas load — all of which fall out of a solve-speed run. - If you want **speed itself as a decision variable** ("what speed gives the lowest power while still delivering P_export?"), use predictive mode: fix the speed, let the chart compute discharge pressure and power. Sweeping speed then traces the head/power curve directly (e.g. 90 %→95 %→100 %→105 % speed maps to a monotonically rising discharge pressure and head). - **Making a *whole* recycle flowsheet predictive** requires letting the internal pressure nodes float (e.g. a recompressor whose discharge mixes back into an upstream stage): fix all speeds, drop the interior `setOutletPressure` calls, and let a pressure-node/recycle pass settle the mixer pressures. Do this behind a `PREDICTIVE_PRESSURE` toggle so the default solve-speed model (used for spec and capacity studies) stays intact. Mixing the two modes on the same recycle loop without a pressure-node pass will diverge. - Either way keep anti-surge active (`getAntiSurge().setActive(True)`, `setSurgeControlFactor(...)`) so a turned-down point recycles onto the control line instead of falling outside the map and returning `NaN` margins. ### Export-oil RVP (certified spec) — use RvpResult ```python D6377 = jneqsim.standards.oilquality.Standard_ASTM_D6377 def export_oil_rvp_bara(stream, ref_T_C=37.8): std = D6377(stream.getFluid()) std.setReferenceTemperature(ref_T_C, "C") # if available on the build res = std.getRvpResult(D6377.RvpMethod.RVP_ASTM_D6377) r = json.loads(str(res.toJson())) # str() — jpype String; {value, unit, method, referenceTemperatureC, valid} return r["value"], r["valid"] ``` The convenience `stream.getRVP(37.8, "C", "bara", "RVP_ASTM_D6377")` still works and is fine inside a hot loop; switch to `RvpResult` when you need the method label, reference temperature, and `valid` flag for the report. ### Assemble objective with penalties ```python def objective_and_constraints(plant, rvp_target=0.79, surge_floor=0.10): power_MW, surge_margin, within = compressor_metrics(plant) rvp, rvp_valid = export_oil_rvp_bara(export_oil_stream()) penalty = 0.0