Flowsheet Runner#

Structured flowsheet runner API#

The structured flowsheet runner is an API in the structfs subpackage, and in particular that package’s runner and fsrunner modules.

Overview#

The core idea of the FlowsheetRunner class is that flowsheets should follow a standard set of “steps”. By standardizing the naming and ordering of these steps, it becomes easier to build tools that run and inspect flowsheets. The Python mechanics of this are to put each step in a function and wrap that function with decorator. The decorator uses a string to indicate which standard step the function implements.

Once these functions are defined, the API can be used to execute and inspect a wrapped flowsheet.

The framework can perform arbitrary actions before and after each run, and before and after a given set of steps. This is implemented with the Actions class and methods add_action, get_action, and remove_action on the base Runner class. More details are given below in the Actions section.

Step 1: Define flowsheet#

It is assumed here that you have Python code to build, configure, and run an IDAES flowsheet. You will first arrange this code to follow the standard “steps” of a flowsheet workflow, which are listed in the BaseFlowsheetRunner class’ STEPS attribute. Not all the steps need to be defined: the API will skip over steps with no definition when executing a range of steps. To make the code more structured you can also define internal sub-steps, as described later.

The set of defined steps is:

build - create the flowsheet
set_operating_conditions - set initial variable values
set_scaling - set scaling
initialize - initialize the flowsheet
set_solver - choose the solver
solve_initial - perform an initial (square problem) solve
add_costing - add costing information (if any)
check_model_structure - check the model for structural issues
initialize_costing - initialize costing variables
solve_optimization - setup and solve the optimization problem
check_model_numerics - check the model for numerical issues

Example: Flash flowsheet#

This is illustrated below with a before/after of an extremely simple flowsheet with a single Flash unit model.

Before#

For now, let’s assume this flowsheet uses only four of the standard steps: “build”, “set_operating_conditions”, “initialize”, and “solve_optimization”. Let’s also assume you have four functions defined that correspond to these steps. Below is a sample flowsheet (for a single Flash unit) that we will use as an example:

from pyomo.environ import ConcreteModel, SolverFactory, SolverStatus
from idaes.core import FlowsheetBlock
from idaes.models.properties.activity_coeff_models.BTX_activity_coeff_VLE import (
    BTXParameterBlock,
)
from idaes.models.unit_models import Flash

def build_model():
    m = ConcreteModel()
    m.fs = FlowsheetBlock(dynamic=False)
    m.fs.properties = BTXParameterBlock(
        valid_phase=("Liq", "Vap"), activity_coeff_model="Ideal", state_vars="FTPz"
    )
    m.fs.flash = Flash(property_package=m.fs.properties)
    return m


def set_operating_conditions(m):
    m.fs.flash.inlet.flow_mol.fix(1)
    m.fs.flash.inlet.temperature.fix(368)
    m.fs.flash.inlet.pressure.fix(101325)
    m.fs.flash.inlet.mole_frac_comp[0, "benzene"].fix(0.5)
    m.fs.flash.inlet.mole_frac_comp[0, "toluene"].fix(0.5)
    m.fs.flash.heat_duty.fix(0)
    m.fs.flash.deltaP.fix(0)


def init_model(m):
    m.fs.flash.initialize()


def solve(m):
    solver = SolverFactory("ipopt")
    return solver.solve(m, tee=True)

After#

In order to make this into a FlowsheetRunner-wrapped flowsheet, we need to do make a few changes. The modified file is shown below, with changed lines highlighted and descriptions below.

from pyomo.environ import ConcreteModel, SolverFactory
from idaes.core import FlowsheetBlock
import idaes.logger as idaeslog
from idaes.models.properties.activity_coeff_models.BTX_activity_coeff_VLE import ( BTXParameterBlock, )
from idaes.models.unit_models import Flash

from idaes.core.util.structfs.fsrunner import FlowsheetRunner

FS = FlowsheetRunner()

@FS.step("build") 
def build_model(ctx):
    """Build the model."""
    m = ConcreteModel()
    m.fs = FlowsheetBlock(dynamic=False)
    m.fs.properties = BTXParameterBlock(
        valid_phase=("Liq", "Vap"),
        activity_coeff_model="Ideal",
        state_vars="FTPz"
    )
    m.fs.flash = Flash(property_package=m.fs.properties) 
    # assert degrees_of_freedom(m) == 7
    ctx.model = m

@FS.step("set_operating_conditions")
def set_operating_conditions(ctx):
    """Set operating conditions."""
    m = ctx.model
    m.fs.flash.inlet.flow_mol.fix(1)
    m.fs.flash.inlet.temperature.fix(368)
    m.fs.flash.inlet.pressure.fix(101325)
    m.fs.flash.inlet.mole_frac_comp[0, "benzene"].fix(0.5)
    m.fs.flash.inlet.mole_frac_comp[0, "toluene"].fix(0.5)
    m.fs.flash.heat_duty.fix(0)
    m.fs.flash.deltaP.fix(0)

@FS.step("initialize") 
def init_model(ctx): 
    """ "Initialize the model."""
    m = ctx.model
    m.fs.flash.initialize()

@FS.step("set_solver") 
def set_solver(ctx):
    """Set the solver."""
    ctx.solver = SolverFactory("ipopt")

@FS.step("solve_optimization")
def solve_opt(ctx):
    ctx["results"] = ctx.solver.solve(ctx.model, tee=ctx["tee"])

Details on the changes:

7: Import the FlowsheetRunner class.
9: Create a global FlowsheetRunner object, here called FS.
11, 25, 37, 43, 48: Add a @FS.step() decorator in front of each function with the name of the associated step.
12, 26, 38, 44, 49: Make each function take a single argument which is a fsrunner.Context instance used to pass state information between functions (here, that argument is named ctx).
23: Assign the model created in the “build” step to ctx.model, a standard attribute of the context object.
28, 40: Replace the direct passing of the model object (in this case, called m) with a context object that has a .model attribute.
44-46: Add a function for the set_solver step, to select the solver (here, IPOPT).
46: In the “solve_optimization” step, assign the solver result to ctx["results"].

Step 2: Execute and inspect#

Once the flowsheet has been ‘wrapped’ in the flowsheet runner interface, it can be run and manipulated via the wrapper object. The basic steps to do this are: import the flowsheet-runner object, build and execute the flowsheet, and inspect the flowsheet.

For example to run all the steps and get the status of the solve, you could do this:

FS.run_steps()
assert FS.results.solver.status == SolverStatus.ok

Some more examples of using the FlowsheetRunner are shown in the example notebooks found under the docs/examples/structfs directory (docs link).

Actions#

The Action class implements a simple framework to run arbitrary functions before and/or after each step and/or run performed by the Runner class.

To create and use your own Action, inherit from this class and then define one or more of the methods:

before_step - Called before a given step is executed
after_step - Called after a given step is executed
before/after_substep - Called before/after a named substep is executed (these can have arbitrary names)
before_run - Called before the first step is executed
after_run - Called after the last step is executed

Then add the action to the Runner class (e.g., FlowsheetRunner) instance with add_action(). Note that you pass the action class, not instance. Additional settings can be passed to the created action instance with arguments to add_action. Also note that the name argument is used to retrieve the action instance later, as needed.

All actions must also implement the report() method, which returns the results of the action to the caller as either a Pydantic BaseModel subclass or a Python dict.

Example

Below is a simple example that prints a message before/after every step and prints the total number of steps run at the end of the run.

Runner HelloGoodbye class#

from idaes.core.util.structfs.runner import Action
class HelloGoodbye(Action):
    "Example action, for tutorial purposes."

    def __init__(self, runner, hello="hi", goodbye="bye", **kwargs):
        super().__init__(runner, **kwargs)
        self._hello, self._goodbye = hello, goodbye
        self.step_counter = -1

    def before_run(self):
        self.step_counter = 0

    def before_step(self, name):
        print(f">> {self._hello} from step {name}")

    def before_substep(self, name, subname):
        print(f"  >> {self._hello} from sub-step {subname}")

    def after_step(self, name):
        print(f"<< {self._goodbye} from step {name}")
        self.step_counter += 1

    def after_substep(self, name, subname):
        print(f"  << {self._goodbye} from sub-step {subname}")

    def after_run(self):
        print(f"Ran {self.step_counter} steps")

    def report(self):
        return {"steps": self.step_counter}

You could add the above example to a Runner subclass, here called my_runner, like this:

my_runner.add_action(
    "hg",
    HelloGoodbye,
    hello="Greetings and salutations",
    goodbye="Smell you later",
)

Then, after running steps, you could print the value of the step_counter attribute with:

print(my_runner.get_action("hg").step_counter)

See the pre-defined actions in the runner_actions module, and their usage in the FlowsheetRunner class, for more examples.

Annotation#

You can also ‘annotate’ variables for special treatment in display, etc. with the annotate_var function in the FlowsheetRunner class.

annotate_var(variable: object, key: str = None, title: str = None, desc: str = None, units: str = None, rounding: int = 3, is_input: bool = True, is_output: bool = True, input_category: str = 'main', output_category: str = 'main') → object

Annotate a Pyomo variable.

Args:

variable: Pyomo variable being annotated
key: Key for this block in dict. Defaults to object name.
title: Name / title of the block. Defaults to object name.
desc: Description of the block. Defaults to object name.
units: Units. Defaults to string value of native units.
rounding: Significant digits
is_input: Is this variable an input
is_output: Is this variable an output
input_category: Name of input grouping to display under
output_category: Name of output grouping to display under

Returns:

Input block (for chaining)

Raises:

ValueError: if is_input and is_output are both False

Example#

Here is an example of annotating a single Pyomo variable. You can apply this same technique to any Pyomo, and thus IDAES, object.

from idaes.core.util.structfs.fsrunner import FlowsheetRunner
from pyomo.environ import *

def example(f: FlowsheetRunner):
    v = Var()
    v.construct()
    f.annotate_var(v, key="example", title="Example variable").fix(1)

To retrieve the annotated variables, use the annotated_vars property:

example(fr := FlowsheetRunner())
print(fr.annotated_vars)
# prints something like this:
# {'example': {'var': <pyomo.core.base.var.ScalarVar object at 0x762ffb124b40>,
# 'fullname': 'ScalarVar', 'title': 'Example variable',
# 'description': 'ScalarVar', 'units': 'dimensionless',
# 'rounding': 3, 'is_input': True, 'is_output': True, 'input_category': 'main',
# 'output_category': 'main'}}

Flowsheet Runner

Contents