multiple choice costing (watertap-org#1183)

docs/technical_reference/costing/index.rst

@@ -7,4 +7,5 @@ Costing Components
    costing_base
    watertap_costing
    zero_order_costing
+   multiple_choice_costing_block
    util

docs/technical_reference/costing/multiple_choice_costing_block.rst

@@ -0,0 +1,145 @@
+.. _multiple_choice_costing_block:
+
+Multiple Choice Unit Model Costing Block
+========================================
+
+.. currentmodule:: watertap.costing.multiple_choice_costing_block
+
+The MultiUnitModelCostingBlock class is a wrapper around the IDAES
+`UnitModelCostingBlock` which allows the modeller to easily explore the implications of
+different unit model costing choices on overall flowsheet costs without rebuilding
+a new flowsheet or replacing Pyomo components.
+
+MultiUnitModelCostingBlock objects instead construct every single costing relationship
+specified for the unit model, and controls which one is active through the indexed
+mutable parameter `costing_block_selector`, which takes the value `1` when the
+associated costing block is active and the value `0` otherwise. The helper method
+`select_costing_block` handles activating a single costing block whilst deactivating
+all other costing blocks.
+
+Usage
+-----
+
+The MultiUnitModelCostingBlock allows the user to pre-specify all costing methods
+on-the-fly and change between them without rebuilding the flowsheet or even the base
+costing relationships on the costing package.
+
+The code below demonstrates its use on a reverse osmosis unit model.
+
+.. testcode::
+
+    import pyomo.environ as pyo
+    from idaes.core import FlowsheetBlock
+
+    import watertap.property_models.NaCl_prop_pack as props
+    from watertap.unit_models.reverse_osmosis_0D import (
+        ReverseOsmosis0D,
+        ConcentrationPolarizationType,
+        MassTransferCoefficient,
+        PressureChangeType,
+    )
+    from watertap.costing import WaterTAPCosting, MultiUnitModelCostingBlock
+    from watertap.costing.unit_models.reverse_osmosis import (
+        cost_reverse_osmosis,
+    )
+
+    m = pyo.ConcreteModel()
+    m.fs = FlowsheetBlock(dynamic=False)
+
+    m.fs.properties = props.NaClParameterBlock()
+    m.fs.costing = WaterTAPCosting()
+
+    m.fs.RO = ReverseOsmosis0D(
+        property_package=m.fs.properties,
+        has_pressure_change=True,
+        pressure_change_type=PressureChangeType.calculated,
+        mass_transfer_coefficient=MassTransferCoefficient.calculated,
+        concentration_polarization_type=ConcentrationPolarizationType.calculated,
+    )
+
+    def my_reverse_osmosis_costing(blk):
+        blk.variable_operating_cost = pyo.Var(
+            initialize=42,
+            units=blk.costing_package.base_currency / blk.costing_package.base_period,
+            doc="Unit variable operating cost",
+        )
+        blk.variable_operating_cost_constraint = pyo.Constraint(
+            expr=blk.variable_operating_cost
+            == 42 * blk.costing_package.base_currency / blk.costing_package.base_period
+        )
+
+    m.fs.RO.costing = MultiUnitModelCostingBlock(
+        # always needed, just like for UnitModelCostingBlock
+        flowsheet_costing_block=m.fs.costing,
+
+        # The keys to the costing-block are a user-defined name.
+        # The values are either the costing method itself or another
+        # dictionary with the keys "costing_method", specifying the
+        # costing method, and optionally "costing_method_arguments",
+        # which defines the keyword arguments into the costing method
+        costing_blocks={
+            "normal_pressure": cost_reverse_osmosis,
+            "high_pressure": {
+                "costing_method": cost_reverse_osmosis,
+                "costing_method_arguments": {"ro_type": "high_pressure"},
+            },
+            "custom_costing": my_reverse_osmosis_costing,
+        },
+
+        # This argument is optional, if it is not specified the block
+        # utilizes the first method, in this case "normal_pressure"
+        initial_costing_block="normal_pressure",
+    )
+
+    # set an area
+    m.fs.RO.area.set_value(100)
+
+    # create the system-level aggregates
+    m.fs.costing.cost_process()
+
+    # initialize the unit level costing blocks and the flowsheet costing block
+    m.fs.costing.initialize()
+
+    # Since the `initial_costing_block` was active, its aggregates are used:
+    assert ( pyo.value(m.fs.RO.costing.capital_cost) == 
+        pyo.value(m.fs.RO.costing.costing_blocks["normal_pressure"].capital_cost)
+    )
+    assert ( pyo.value(m.fs.costing.aggregate_capital_cost) == 
+        pyo.value(m.fs.RO.costing.costing_blocks["normal_pressure"].capital_cost)
+    )
+    assert pyo.value(m.fs.RO.costing.variable_operating_cost) == 0
+    assert pyo.value(m.fs.costing.aggregate_variable_operating_cost) == 0
+
+    # We can activate the "high_pressure" costing block:
+    m.fs.RO.costing.select_costing_block("high_pressure")
+
+    # Need re-initialize to have the new values in the aggregates
+    m.fs.costing.initialize()
+
+    assert ( pyo.value(m.fs.RO.costing.capital_cost) == 
+        pyo.value(m.fs.RO.costing.costing_blocks["high_pressure"].capital_cost)
+    )
+    assert ( pyo.value(m.fs.costing.aggregate_capital_cost) == 
+        pyo.value(m.fs.RO.costing.costing_blocks["high_pressure"].capital_cost)
+    )
+    assert pyo.value(m.fs.RO.costing.variable_operating_cost) == 0
+    assert pyo.value(m.fs.costing.aggregate_variable_operating_cost) == 0
+
+    # We can activate the "custom_costing" costing block:
+    m.fs.RO.costing.select_costing_block("custom_costing")
+
+    # Need re-initialize to have the new values in the aggregates
+    m.fs.costing.initialize()
+
+    # No capital cost for block "custom_costing", but it does have
+    # a "variable" operating cost 
+    assert pyo.value(m.fs.RO.costing.capital_cost) == 0
+    assert pyo.value(m.fs.costing.aggregate_capital_cost) == 0
+    assert pyo.value(m.fs.RO.costing.variable_operating_cost) == 42
+    assert pyo.value(m.fs.costing.aggregate_variable_operating_cost) == 42
+
+
+Class Documentation
+-------------------
+
+* :class:`MultiUnitModelCostingBlock`

watertap/costing/__init__.py

@@ -12,6 +12,7 @@
 
 from .watertap_costing_package import WaterTAPCosting
 from .zero_order_costing import ZeroOrderCosting
+from .multiple_choice_costing_block import MultiUnitModelCostingBlock
 
 from .util import (
     register_costing_parameter_block,

watertap/costing/multiple_choice_costing_block.py

@@ -0,0 +1,210 @@
+#################################################################################
+# WaterTAP Copyright (c) 2020-2023, The Regents of the University of California,
+# through Lawrence Berkeley National Laboratory, Oak Ridge National Laboratory,
+# National Renewable Energy Laboratory, and National Energy Technology
+# Laboratory (subject to receipt of any required approvals from the U.S. Dept.
+# of Energy). All rights reserved.
+#
+# Please see the files COPYRIGHT.md and LICENSE.md for full copyright and license
+# information, respectively. These files are also available online at the URL
+# "https://github.com/watertap-org/watertap/"
+#################################################################################
+
+import pyomo.environ as pyo
+from pyomo.common.config import ConfigBlock, ConfigValue
+from pyomo.util.calc_var_value import calculate_variable_from_constraint
+from idaes.core import declare_process_block_class, ProcessBlockData, UnitModelBlockData
+from idaes.core.util.misc import add_object_reference
+from idaes.core.base.costing_base import (
+    UnitModelCostingBlockData,
+    UnitModelCostingBlock,
+    assert_flowsheet_costing_block,
+    DefaultCostingComponents,
+)
+import idaes.logger as idaeslog
+
+_log = idaeslog.getLogger(__name__)
+
+
+@declare_process_block_class("MultiUnitModelCostingBlock")
+class MultiUnitModelCostingBlockData(UnitModelCostingBlockData, UnitModelCostingBlock):
+    """
+    Class for constructing several costing blocks on the same
+    unit model and then allowing for choice between them
+    """
+
+    CONFIG = ConfigBlock()
+
+    CONFIG.declare(
+        "flowsheet_costing_block",
+        ConfigValue(
+            domain=assert_flowsheet_costing_block,
+            doc="Reference to associated FlowsheetCostingBlock to use.",
+        ),
+    )
+    CONFIG.declare(
+        "costing_blocks",
+        ConfigValue(
+            domain=dict,
+            doc="Costing blocks to use for unit. Should be a dictionary whose keys "
+            "are the names of the block and whose values are either the costing "
+            "method used to construct the block, or another dictionary whose keys "
+            "are `costing_method`, which has the value of the costing method, and "
+            "optionally `costing_method_arguments`, which has the keyword arguments "
+            "for the costing method specified.",
+        ),
+    )
+    CONFIG.declare(
+        "initial_costing_block",
+        ConfigValue(
+            doc="Costing block to be initially active",
+            default=None,
+        ),
+    )
+
+    def build(self):
+        ProcessBlockData.build(self)
+
+        # Alias flowsheet costing block reference
+        fcb = self.config.flowsheet_costing_block
+
+        # Get reference to unit model
+        unit_model = self.parent_block()
+
+        # Check that parent is an instance of a UnitModelBlockData
+        if UnitModelBlockData not in unit_model.__class__.__mro__:
+            raise TypeError(
+                f"{self.name} - parent object ({unit_model.name}) is not an "
+                f"instance of a UnitModelBlockData object. "
+                "UnitModelCostingBlocks can only be added to UnitModelBlocks."
+            )
+
+        # Check to see if unit model already has costing
+        for b in unit_model.component_objects(pyo.Block, descend_into=False):
+            if b is not self and isinstance(
+                b, (UnitModelCostingBlock, MultiUnitModelCostingBlock)
+            ):
+                # Block already has costing, clean up and raise exception
+                raise RuntimeError(
+                    f"Unit model {unit_model.name} already has a costing block"
+                    f" registered: {b.name}. Each unit may only have a single "
+                    "UnitModelCostingBlock associated with it."
+                )
+        # Add block to unit model initialization order
+        unit_model._initialization_order.append(self)
+
+        # Register unit model with this costing package
+        fcb._registered_unit_costing.append(self)
+
+        # Assign object references for costing package and unit model
+        add_object_reference(self, "costing_package", fcb)
+        add_object_reference(self, "unit_model", unit_model)
+
+        self.costing_blocks = pyo.Block(self.config.costing_blocks)
+        self.costing_block_selector = pyo.Param(
+            self.config.costing_blocks, domain=pyo.Boolean, default=0, mutable=True
+        )
+
+        if self.config.initial_costing_block is None:
+            for k in self.config.costing_blocks:
+                self.costing_block_selector[k].set_value(1)
+                break
+        else:
+            self.costing_block_selector[self.config.initial_costing_block].set_value(1)
+
+        # Get costing method if not provided
+        for k, val in self.config.costing_blocks.items():
+
+            # if we get a callable, just use it
+            if callable(val):
+                method = val
+                kwds = {}
+            # else we'll assume it's a dictionary with keys
+            # `costing_method` and potentially `costing_method_arguments`.
+            # We raise an error if we find something else.
+            else:
+                for l in val:
+                    if l not in ("costing_method", "costing_method_arguments"):
+                        raise RuntimeError(
+                            f"Unrecognized key {l} for costing block {k}."
+                        )
+                try:
+                    method = val["costing_method"]
+                except KeyError:
+                    raise KeyError(
+                        f"Must specify a `costing_method` key for costing "
+                        f"block {k}."
+                    )
+                kwds = val.get("costing_method_arguments", {})
+
+            blk = self.costing_blocks[k]
+
+            # Assign object references for costing package and unit model
+            add_object_reference(blk, "costing_package", fcb)
+            add_object_reference(blk, "unit_model", unit_model)
+
+            # Call unit costing method
+            method(blk, **kwds)
+
+            # Check that costs are Vars and have lower bound of 0
+            cost_vars = DefaultCostingComponents
+            for v in cost_vars:
+                try:
+                    cvar = getattr(blk, v)
+                    if not isinstance(cvar, pyo.Var):
+                        raise TypeError(
+                            f"{unit_model.name} {v} component must be a Var. "
+                            "Please check the costing package you are using to "
+                            "ensure that all costing components are declared as "
+                            "variables."
+                        )
+                    elif cvar.lb is None or cvar.lb < 0:
+                        _log.warning(
+                            f"{unit_model.name} {v} component has a lower bound "
+                            "less than zero. Be aware that this may result in "
+                            "negative costs during optimization."
+                        )
+                except AttributeError:
+                    pass
+
+        # Now we need to tie them all together
+        cost_vars = list(DefaultCostingComponents) + ["direct_capital_cost"]
+        for vname in cost_vars:
+            for blk in self.costing_blocks.values():
+                if hasattr(blk, vname):
+                    break
+            else:  # no break
+                continue
+
+            expr = 0.0
+            for name, blk in self.costing_blocks.items():
+                cvar = blk.component(vname)
+                if cvar is None:
+                    continue
+                expr += self.costing_block_selector[name] * cvar
+
+            self.add_component(vname, pyo.Expression(expr=expr))
+
+    def select_costing_block(self, costing_block_name):
+        """
+        Set the active costing block
+        """
+        # zero out everything else
+        self.costing_block_selector[:].set_value(0)
+        self.costing_block_selector[costing_block_name].set_value(1)
+
+    def initialize(self, *args, **kwargs):
+        """
+        Initialize all costing blocks for easy switching between
+        """
+        # TODO: Implement an initialization method
+        # TODO: Need to have a general purpose method (block triangularisation?)
+        # TODO: Should also allow registering custom methods
+
+        # Vars and Constraints
+        for blk in self.costing_blocks.values():
+            for c in DefaultCostingComponents:
+                if hasattr(blk, c):
+                    var = getattr(blk, c)
+                    cons = getattr(blk, f"{c}_constraint")
+                    calculate_variable_from_constraint(var, cons)