feat: Add support for defining net or gross loss in LossLink. (#1124)

pywr/nodes.py

@@ -1238,12 +1238,21 @@ class LossLink(Node):
     model : `pywr.model.Model`
     name : string
         Name of the node.
-    loss_factor : float
+    loss_factor : float, Parameter
         The proportion of flow that is lost through this node. Must be greater than or equal to zero. If zero
-        then no-losses are calculated. The percentage is calculated as a percentage of gross flow.
+        then no-losses are calculated. This value is either a proportion of gross or net flow depending on the
+        value of `loss_factor_type`.
+    loss_factor_type: str
+        Either "gross" or "net" (default) to specify whether the loss factor is applied as a proportion of gross or
+         net flow respectively.
+
+    Notes
+    -----
+    There is currently a limitation that the loss factor must be a literal constant (i.e. not a parameter) when
+    `loss_factor_type` is set to "gross".
     """
 
-    __parameter_attributes__ = ("loss_factor",)
+    __parameter_attributes__ = ("loss_factor", "max_flow", "min_flow", "cost")
 
     def __init__(self, model, name, **kwargs):
         self.allow_isolated = True
@@ -1265,36 +1274,37 @@ def __init__(self, model, name, **kwargs):
         self.gross.connect(self.net)
 
         self.agg = AggregatedNode(model, name=agg_name, nodes=[self.net, self.output])
+        self.loss_factor_type = kwargs.pop("loss_factor_type", "net")
         self.loss_factor = kwargs.pop("loss_factor", 0.0)
 
         super().__init__(model, name, **kwargs)
 
-    def loss_factor():
-        def fget(self):
-            if self.agg.factors:
-                return self.agg.factors[1]
-            elif self.output.max_flow == 0.0:
-                return 0.0
-            else:
-                return 1.0
-
-        def fset(self, value):
-            if value == 0.0:
-                # 0% loss; no flow to the output loss node.
-                self.agg.factors = None
-                self.output.max_flow = 0.0
-            elif value == 1.0:
-                # 100% loss; all flow to the output loss node
-                self.agg.factors = None
-                self.output.max_flow = float("inf")
-                self.net.max_flow = 0.0
-            else:
-                self.output.max_flow = float("inf")
+    def setup(self, model):
+        super().setup(model)
+        value = self.loss_factor
+
+        if value == 0.0:
+            # 0% loss; no flow to the output loss node.
+            self.agg.factors = None
+            self.output.max_flow = 0.0
+        elif value == 1.0 and self.loss_factor_type == "gross":
+            # 100% loss; all flow to the output loss node
+            self.agg.factors = None
+            self.output.max_flow = float("inf")
+            self.net.max_flow = 0.0
+        else:
+            self.output.max_flow = float("inf")
+            if self.loss_factor_type == "net":
                 self.agg.factors = [1.0, value]
 
-        return locals()
-
-    loss_factor = property(**loss_factor())
+            elif self.loss_factor_type == "gross":
+                # TODO this will error in the case of a `value` being a parameter
+                self.agg.factors = [1.0 - float(value), float(value)]
+            else:
+                raise ValueError(
+                    f'Unrecognised `loss_factor_type` "{self.loss_factor_type}".'
+                    f'Please use either "gross" or "net".'
+                )
 
     def min_flow():
         def fget(self):

tests/models/loss_link.json

@@ -13,7 +13,6 @@
         {
             "name": "supply1",
             "type": "Input",
-            "max_flow": 100,
             "cost": 0.1
         },
         {

tests/test_run.py

@@ -959,28 +959,48 @@ def test_breaklink_node():
     assert_allclose(transfer.storage.volume, 0)
 
 
-@pytest.mark.parametrize("loss_factor", [None, 0.2, 0.99, 1.0, 0.0])
-def test_loss_link_node(loss_factor):
+@pytest.mark.parametrize(
+    "loss_factor,loss_factor_type",
+    [
+        (None, None),
+        (0.2, "gross"),
+        (0.2, "net"),
+        (0.98, "gross"),
+        (0.98, "net"),
+        (1.0, "gross"),
+        (1.0, "net"),
+        (0.0, "gross"),
+        (0.0, "net"),
+    ],
+)
+def test_loss_link_node(loss_factor, loss_factor_type):
     """Test LossLink node"""
     model = load_model("loss_link.json")
 
     supply1 = model.nodes["supply1"]
     link1 = model.nodes["link1"]
     demand1 = model.nodes["demand1"]
 
+    if loss_factor_type is not None:
+        link1.loss_factor_type = loss_factor_type
     if loss_factor is not None:
         link1.loss_factor = loss_factor
 
     model.check()
     model.run()
 
-    if loss_factor is None:
+    if loss_factor is None and loss_factor_type is None:
+        # As per JSON file
         expected_supply = 12
         expected_demand = 10
-    elif loss_factor == 1.0:
-        # 100% loss means no flow can be provided.
-        expected_supply = 0.0
-        expected_demand = 0.0
+    elif loss_factor_type == "gross":
+        if loss_factor == 1.0:
+            # 100% loss means no flow can be provided.
+            expected_supply = 0.0
+            expected_demand = 0.0
+        else:
+            expected_supply = 10 / (1 - loss_factor)
+            expected_demand = 10
     else:
         expected_supply = 10 * (1 + loss_factor)
         expected_demand = 10