Enable derived costing package to define a flow cost

idaes/core/base/costing_base.py

@@ -330,20 +330,37 @@ def register_flow_type(self, flow_type, cost):
         """
         This method allows users to register new material and utility flows
         with the FlowsheetCostingBlock for use when costing flows.
+        This method creates a new `Var` on the FlowsheetCostingBlock named
+        f`{flow_type}_cost` whose value is fixed to `cost`.
+
+        If a component named f`{flow_type}_cost` already exists on the
+        FlowsheetCostingBlock, then an error is raised unless f`{flow_type}_cost`
+        is `cost`. If f`{flow_type}_cost` is `cost`, no error is raised and
+        the existing component f`{flow_type}_cost` is used to cost the flow.
 
         Args:
             flow_type: string name to represent flow type
             cost: a Pyomo expression with units representing the flow cost
         """
         self.flow_types.add(flow_type)
 
-        # Create a Var to hold the cost
-        # Units will be different between flows, so have to use scalar Vars
-        fvar = pyo.Var(
-            units=pyo.units.get_units(cost), doc=f"Cost parameter for {flow_type} flow"
-        )
-        self.add_component(f"{flow_type}_cost", fvar)
-        fvar.fix(cost)
+        name = f"{flow_type}_cost"
+        current_component = self.component(name)
+        if current_component is not None:
+            if current_component is not cost:
+                raise RuntimeError(
+                    f"Component {name} already exists on {self} but is not {cost}."
+                )
+            # now self.{flow_type}_cost is cost, so just use it
+        else:
+            # Create a Var to hold the cost
+            # Units will be different between flows, so have to use scalar Vars
+            fvar = pyo.Var(
+                units=pyo.units.get_units(cost),
+                doc=f"Cost parameter for {flow_type} flow",
+            )
+            self.add_component(name, fvar)
+            fvar.fix(cost)
 
         self._registered_flows[flow_type] = []
 

idaes/core/base/tests/test_costing_base.py

@@ -15,7 +15,7 @@
 """
 import pytest
 
-from pyomo.environ import ConcreteModel, Constraint, Set, units as pyunits, Var
+from pyomo.environ import ConcreteModel, Constraint, Set, units as pyunits, Var, Param
 from pyomo.util.check_units import assert_units_consistent, assert_units_equivalent
 
 from idaes.core import declare_process_block_class, UnitModelBlockData
@@ -165,7 +165,12 @@ def build_global_params(self):
         self.base_currency = pyunits.USD_test
         self.base_period = pyunits.year
 
-        self.defined_flows = {"test_flow_1": 0.2 * pyunits.J}
+        self.test_flow_2_cost = Param(initialize=0.07, units=pyunits.kW)
+
+        self.defined_flows = {
+            "test_flow_1": 0.2 * pyunits.J,
+            "test_flow_2": self.test_flow_2_cost,
+        }
 
         self._bgp = True
 
@@ -219,9 +224,12 @@ def costing(self):
     def test_basic_attributes(self, costing):
         assert costing.costing._registered_unit_costing == []
         assert isinstance(costing.costing.flow_types, Set)
-        assert len(costing.costing.flow_types) == 1
+        assert len(costing.costing.flow_types) == 2
         assert "test_flow_1" in costing.costing.flow_types
-        assert costing.costing._registered_flows == {"test_flow_1": []}
+        assert costing.costing._registered_flows == {
+            "test_flow_1": [],
+            "test_flow_2": [],
+        }
 
         assert costing.costing._costing_methods_map == {
             TypeAData: TestCostingPackageData.method_1,
@@ -234,6 +242,8 @@ def test_basic_attributes(self, costing):
         assert costing.costing.test_flow_1_cost.value == 0.2
         assert_units_equivalent(costing.costing.test_flow_1_cost.get_units(), pyunits.J)
 
+        assert isinstance(costing.costing.test_flow_2_cost, Param)
+
         # Test that build_global_parameters was called successfully
         assert costing.costing._bgp
 
@@ -250,7 +260,23 @@ def test_register_flow_type(self, costing):
         )
         assert "test_flow" in costing.costing.flow_types
 
-        assert costing.costing._registered_flows == {"test_flow_1": [], "test_flow": []}
+        assert costing.costing._registered_flows == {
+            "test_flow_1": [],
+            "test_flow_2": [],
+            "test_flow": [],
+        }
+
+    @pytest.mark.unit
+    def test_register_flow_component_exists(self, costing):
+        costing.costing.test_flow_3_cost = Var()
+        with pytest.raises(
+            RuntimeError,
+            match="Component test_flow_3_cost already exists on costing but is not 42.",
+        ):
+            costing.costing.register_flow_type("test_flow_3", 42)
+        # cleanup for next test
+        costing.costing.flow_types.remove("test_flow_3")
+        costing.costing.del_component(costing.costing.test_flow_3_cost)
 
     @pytest.mark.unit
     def test_cost_flow_invalid_type(self, costing):