Merge pull request #1088 from oemof/feature/multi_input_sink

Allow sinks/sources to have multiple inputs/outputs

docs/whatsnew/v0-5-4.rst

@@ -4,6 +4,8 @@ v0.5.4
 API changes
 ###########
 
+* Explicitly allow to have multiple inputs or outputs for Sinks and Sources,
+  respectively.
 
 New features
 ############
@@ -29,3 +31,4 @@ Known issues
 Contributors
 ############
 
+* Patrik Schönfeldt

src/oemof/solph/components/_sink.py

@@ -13,10 +13,7 @@
 SPDX-License-Identifier: MIT
 
 """
-from warnings import warn
-
 from oemof.network import Node
-from oemof.tools import debugging
 
 
 class Sink(Node):
@@ -27,6 +24,9 @@ class Sink(Node):
     label : str
         String holding the label of the Sink object.
         The label of each object must be unique.
+    inputs: dict
+        A dictionary mapping input nodes to corresponding inflows
+        (i.e. input values).
 
     Examples
     --------
@@ -39,30 +39,14 @@ class Sink(Node):
     ...    label='el_export',
     ...    inputs={bel: solph.flows.Flow()})
 
-
-    Notes
-    -----
-    It is theoretically possible to use the Sink object with multiple inputs.
-    However, we strongly recommend using multiple Sink objects instead.
     """
 
-    def __init__(self, label=None, inputs=None, custom_attributes=None):
+    def __init__(self, label=None, *, inputs, custom_attributes=None):
         if inputs is None:
             inputs = {}
         if custom_attributes is None:
             custom_attributes = {}
 
-        if len(inputs) != 1:
-            msg = (
-                "A Sink is designed to have one input but you provided {0}."
-                " If this is intended and you know what you are doing you can "
-                "disable the SuspiciousUsageWarning globally."
-            )
-            warn(
-                msg.format(len(inputs)),
-                debugging.SuspiciousUsageWarning,
-            )
-
         super().__init__(
             label=label, inputs=inputs, custom_properties=custom_attributes
         )

src/oemof/solph/components/_source.py

@@ -13,11 +13,7 @@
 SPDX-License-Identifier: MIT
 
 """
-
-from warnings import warn
-
 from oemof.network import Node
-from oemof.tools import debugging
 
 
 class Source(Node):
@@ -28,6 +24,9 @@ class Source(Node):
     label : str
         String holding the label of the Source object.
         The label of each object must be unique.
+    outputs: dict
+        A dictionary mapping input nodes to corresponding outflows
+        (i.e. output values).
 
     Examples
     --------
@@ -48,31 +47,14 @@ class Source(Node):
 
     >>> str(pv_plant.outputs[bel].output)
     'electricity'
-
-    Notes
-    -----
-    It is theoretically possible to use the Source object with multiple
-    outputs. However, we strongly recommend using multiple Source objects
-    instead.
     """
 
-    def __init__(self, label=None, outputs=None, custom_attributes=None):
+    def __init__(self, label=None, *, outputs, custom_attributes=None):
         if outputs is None:
             outputs = {}
         if custom_attributes is None:
             custom_attributes = {}
 
-        if len(outputs) != 1:
-            msg = (
-                "A Source is designed to have one output but you provided {0}."
-                " If this is intended and you know what you are doing you can "
-                "disable the SuspiciousUsageWarning globally."
-            )
-            warn(
-                msg.format(len(outputs)),
-                debugging.SuspiciousUsageWarning,
-            )
-
         super().__init__(
             label=label, outputs=outputs, custom_properties=custom_attributes
         )

tests/test_components/test_sink.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from oemof import solph
+
+
+def test_multi_input_sink():
+    num_in = 3
+    steps = 10
+    costs = -0.1
+
+    es = solph.EnergySystem(
+        timeindex=solph.create_time_index(year=2023, number=steps),
+        infer_last_interval=False,
+    )
+
+    for i in range(num_in):
+        bus_label = f"bus input {i}"
+        b = solph.Bus(bus_label)
+        es.add(b)
+        es.add(
+            solph.components.Source(f"source {i}", outputs={b: solph.Flow()})
+        )
+
+    es.add(
+        solph.components.Sink(
+            inputs={
+                es.node[f"bus input {i}"]: solph.Flow(
+                    nominal_value=1,
+                    variable_costs=costs,
+                )
+                for i in range(num_in)
+            }
+        )
+    )
+
+    model = solph.Model(es)
+    model.solve("cbc")
+
+    assert (
+        model.solver_results["Solver"][0]["Termination condition"]
+        != "infeasible"
+    )
+    meta_results = solph.processing.meta_results(model)
+
+    assert meta_results["objective"] == pytest.approx(num_in * steps * costs)

tests/test_components/test_source.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from oemof import solph
+
+
+def test_multi_output_source():
+    num_out = 3
+    steps = 10
+    costs = -0.1
+
+    es = solph.EnergySystem(
+        timeindex=solph.create_time_index(year=2023, number=steps),
+        infer_last_interval=False,
+    )
+
+    for i in range(num_out):
+        bus_label = f"bus input {i}"
+        b = solph.Bus(bus_label)
+        es.add(b)
+        es.add(solph.components.Sink(f"source {i}", inputs={b: solph.Flow()}))
+
+    es.add(
+        solph.components.Source(
+            outputs={
+                es.node[f"bus input {i}"]: solph.Flow(
+                    nominal_value=1,
+                    variable_costs=costs,
+                )
+                for i in range(num_out)
+            }
+        )
+    )
+
+    model = solph.Model(es)
+    model.solve("cbc")
+
+    assert (
+        model.solver_results["Solver"][0]["Termination condition"]
+        != "infeasible"
+    )
+    meta_results = solph.processing.meta_results(model)
+
+    assert meta_results["objective"] == pytest.approx(num_out * steps * costs)

tests/test_warnings.py

@@ -28,50 +28,6 @@ def warning_fixture():
     warnings.simplefilter(action="ignore", category=FutureWarning)
 
 
-def test_that_the_sink_errors_actually_get_raised(warning_fixture):
-    """Sink doesn't warn about potentially erroneous usage."""
-    look_out = solph.Bus()
-    with pytest.raises(
-        TypeError, match="got an unexpected keyword argument 'outputs'"
-    ):
-        solph.components.Sink(label="test_sink", outputs={look_out: "A typo!"})
-
-    msg = (
-        "A Sink is designed to have one input but you provided 0."
-        " If this is intended and you know what you are doing you can "
-        "disable the SuspiciousUsageWarning globally."
-    )
-    with warnings.catch_warnings(record=True) as w:
-        solph.components.Sink(
-            label="no input",
-        )
-        assert len(w) == 1
-        assert msg in str(w[-1].message)
-
-
-def test_that_the_source_warnings_actually_get_raised(warning_fixture):
-    """Source doesn't warn about potentially erroneous usage."""
-    look_out = solph.Bus()
-    with pytest.raises(
-        TypeError, match="got an unexpected keyword argument 'inputs'"
-    ):
-        solph.components.Source(
-            label="test_source", inputs={look_out: "A typo!"}
-        )
-
-    msg = (
-        "A Source is designed to have one output but you provided 0."
-        " If this is intended and you know what you are doing you can "
-        "disable the SuspiciousUsageWarning globally."
-    )
-    with warnings.catch_warnings(record=True) as w:
-        solph.components.Source(
-            label="no output",
-        )
-        assert len(w) == 1
-        assert msg in str(w[-1].message)
-
-
 def test_that_the_converter_warnings_actually_get_raised(warning_fixture):
     """Converter doesn't warn about potentially erroneous usage."""
     look_out = solph.Bus()