diff --git a/docs/technical_reference/costing/index.rst b/docs/technical_reference/costing/index.rst index c71c8d7b8d..df2b84ec5f 100644 --- a/docs/technical_reference/costing/index.rst +++ b/docs/technical_reference/costing/index.rst @@ -7,4 +7,5 @@ Costing Components costing_base watertap_costing zero_order_costing + multiple_choice_costing_block util diff --git a/docs/technical_reference/costing/multiple_choice_costing_block.rst b/docs/technical_reference/costing/multiple_choice_costing_block.rst new file mode 100644 index 0000000000..3126531d75 --- /dev/null +++ b/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` diff --git a/watertap/costing/__init__.py b/watertap/costing/__init__.py index b1e3db0fdf..a236eec3ad 100644 --- a/watertap/costing/__init__.py +++ b/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, diff --git a/watertap/costing/multiple_choice_costing_block.py b/watertap/costing/multiple_choice_costing_block.py new file mode 100644 index 0000000000..84ee832ab2 --- /dev/null +++ b/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) diff --git a/watertap/costing/tests/test_multiple_choice_costing_block.py b/watertap/costing/tests/test_multiple_choice_costing_block.py new file mode 100644 index 0000000000..b8dacc28b8 --- /dev/null +++ b/watertap/costing/tests/test_multiple_choice_costing_block.py @@ -0,0 +1,272 @@ +################################################################################# +# 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 pytest + +import os +import re + +import pyomo.environ as pyo +from idaes.core import FlowsheetBlock, UnitModelCostingBlock +from watertap.costing import MultiUnitModelCostingBlock + +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, ROType +from watertap.costing.unit_models.reverse_osmosis import ( + cost_reverse_osmosis, + cost_high_pressure_reverse_osmosis, +) + + +def setup_flowsheet(): + + 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, + ) + m.fs.RO.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + }, + ) + + with pytest.raises( + RuntimeError, + match="Unit model fs.RO already has a costing block " + "registered: fs.RO.costing. Each unit may only have a single " + "UnitModelCostingBlock associated with it.", + ): + m.fs.RO.costing_2 = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + }, + ) + m.fs.RO.del_component(m.fs.RO.costing_2) + + with pytest.raises( + RuntimeError, + match="Unit model fs.RO already has a costing block " + "registered: fs.RO.costing. Each unit may only have a single " + "UnitModelCostingBlock associated with it.", + ): + m.fs.RO.costing_3 = UnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_method=cost_reverse_osmosis, + costing_method_arguments={"ro_type": "high_pressure"}, + ) + m.fs.RO.del_component(m.fs.RO.costing_3) + + m.fs.RO.area.set_value(100) + + m.fs.RO2 = 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, + ) + with pytest.raises( + RuntimeError, + match="Unrecognized key costing_mehtod for costing block foo.", + ): + m.fs.RO2.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + "foo": {"costing_mehtod": cost_reverse_osmosis}, + }, + initial_costing_block="high_pressure", + ) + m.fs.RO2.del_component(m.fs.RO2.costing) + + with pytest.raises( + KeyError, + match="Must specify a `costing_method` key for costing block foo.", + ): + m.fs.RO2.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + "foo": {"costing_method_arguments": {"ro_type": "high_pressure"}}, + }, + initial_costing_block="high_pressure", + ) + m.fs.RO2.del_component(m.fs.RO2.costing) + + def dummy_method(blk): + blk.capital_cost = pyo.Expression() + + with pytest.raises( + TypeError, + match="fs.RO2 capital_cost component must be a " + "Var. Please check the costing package you are " + "using to ensure that all costing components are " + "declared as variables.", + ): + m.fs.RO2.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={"bar": dummy_method}, + ) + m.fs.RO2.del_component(m.fs.RO.costing) + + m.fs.RO2.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + }, + initial_costing_block="high_pressure", + ) + m.fs.RO2.area.set_value(50) + + m.fs.RO3 = 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_own_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.RO3.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + "my_own": my_own_reverse_osmosis_costing, + }, + ) + m.fs.RO3.area.set_value(25) + m.fs.RO3.costing.select_costing_block("my_own") + + m.fs.costing.cost_process() + + m.fs.foo = pyo.Block() + with pytest.raises( + TypeError, + match=re.escape( + "fs.foo.costing - parent object (fs.foo) is not an instance " + "of a UnitModelBlockData object. UnitModelCostingBlocks can only be " + "added to UnitModelBlocks." + ), + ): + m.fs.foo.costing = MultiUnitModelCostingBlock( + flowsheet_costing_block=m.fs.costing, + costing_blocks={ + "normal_pressure": cost_reverse_osmosis, + "high_pressure": { + "costing_method": cost_reverse_osmosis, + "costing_method_arguments": {"ro_type": "high_pressure"}, + }, + "my_own": my_own_reverse_osmosis_costing, + }, + ) + + return m + + +def test_multiple_choice_costing_block(): + + m = setup_flowsheet() + + m.fs.costing.initialize() + + # first method is the default + assert pyo.value(m.fs.RO.costing.capital_cost) == pyo.value( + m.fs.RO.costing.costing_blocks["normal_pressure"].capital_cost + ) + # manual default + assert pyo.value(m.fs.RO2.costing.capital_cost) == pyo.value( + m.fs.RO2.costing.costing_blocks["high_pressure"].capital_cost + ) + + assert m.fs.costing.total_capital_cost.value == 2 * ( + m.fs.RO.costing.costing_blocks["normal_pressure"].capital_cost.value + + m.fs.RO2.costing.costing_blocks["high_pressure"].capital_cost.value + ) + + m.fs.RO.costing.select_costing_block("high_pressure") + + assert pyo.value(m.fs.RO.costing.capital_cost) == pyo.value( + m.fs.RO.costing.costing_blocks["high_pressure"].capital_cost + ) + + # need to re-initialize + assert m.fs.costing.total_capital_cost.value == 2 * ( + m.fs.RO.costing.costing_blocks["normal_pressure"].capital_cost.value + + m.fs.RO2.costing.costing_blocks["high_pressure"].capital_cost.value + ) + + m.fs.costing.initialize() + assert m.fs.costing.total_capital_cost.value == 2 * ( + m.fs.RO.costing.costing_blocks["high_pressure"].capital_cost.value + + m.fs.RO2.costing.costing_blocks["high_pressure"].capital_cost.value + ) + assert m.fs.costing.aggregate_variable_operating_cost.value == 42 + + m.fs.RO3.costing.select_costing_block("high_pressure") + m.fs.costing.initialize() + assert m.fs.costing.total_capital_cost.value == 2 * ( + m.fs.RO.costing.costing_blocks["high_pressure"].capital_cost.value + + m.fs.RO2.costing.costing_blocks["high_pressure"].capital_cost.value + + m.fs.RO3.costing.costing_blocks["high_pressure"].capital_cost.value + ) + assert m.fs.costing.aggregate_variable_operating_cost.value == 0