Feature/prevent simultanious flow rates

CHANGELOG.md

@@ -53,12 +53,15 @@ If upgrading from v2.x, see the [v3.0.0 release notes](https://github.com/flixOp
 If upgrading from v2.x, see the [v3.0.0 release notes](https://github.com/flixOpt/flixOpt/releases/tag/v3.0.0) and [Migration Guide](https://flixopt.github.io/flixopt/latest/user-guide/migration-guide-v3/).
 
 ### ✨ Added
+- **Multiple constraint groups for `prevent_simultaneous_flows`**: Components can now define multiple independent mutual exclusivity constraints. Use `[['fuel1', 'fuel2'], ['cooling1', 'cooling2']]` to enforce "at most 1 fuel AND at most 1 cooling method" while allowing combinations like (fuel1+cooling2). Previously only single constraint groups were supported.
+- Added `prevent_simultaneous_flows` to `LinearConverter` class
 
 ### 💥 Breaking Changes
 
 ### ♻️ Changed
 
 ### 🗑️ Deprecated
+- **Flow objects in `prevent_simultaneous_flows`**: Use flow label strings instead of Flow objects. Example: `prevent_simultaneous_flows=['flow1', 'flow2']` (preferred) instead of `prevent_simultaneous_flows=[flow1_obj, flow2_obj]` (deprecated). Flow objects still work but trigger a DeprecationWarning.
 
 ### 🔥 Removed
 
@@ -72,6 +75,9 @@ If upgrading from v2.x, see the [v3.0.0 release notes](https://github.com/flixOp
 
 ### 👷 Development
 
+- Added comprehensive tests for multiple constraint groups in `prevent_simultaneous_flows`
+- Fixed type hints for `submodel` attributes to include `| None` for consistency with runtime behavior
+
 ### 🚧 Known Issues
 
 ---

flixopt/components.py

@@ -62,6 +62,10 @@ class LinearConverter(Component):
             of different flows. Enables modeling of non-linear conversion behavior through
             linear approximation. Either 'conversion_factors' or 'piecewise_conversion'
             can be used, but not both.
+        prevent_simultaneous_flows: Flow labels (strings) that cannot be active simultaneously.
+            Can be a single list or list of lists for multiple independent constraint groups.
+            See Component class documentation for details and examples.
+            Note: Passing Flow objects is deprecated.
         meta_data: Used to store additional information about the Element. Not used
             internally, but saved in results. Only use Python native types.
 
@@ -168,9 +172,10 @@ def __init__(
         on_off_parameters: OnOffParameters | None = None,
         conversion_factors: list[dict[str, TemporalDataUser]] | None = None,
         piecewise_conversion: PiecewiseConversion | None = None,
+        prevent_simultaneous_flows: list[str | Flow] | list[list[str | Flow]] | None = None,
         meta_data: dict | None = None,
     ):
-        super().__init__(label, inputs, outputs, on_off_parameters, meta_data=meta_data)
+        super().__init__(label, inputs, outputs, on_off_parameters, prevent_simultaneous_flows, meta_data=meta_data)
         self.conversion_factors = conversion_factors or []
         self.piecewise_conversion = piecewise_conversion
 
@@ -401,7 +406,9 @@ def __init__(
             label,
             inputs=[charging],
             outputs=[discharging],
-            prevent_simultaneous_flows=[charging, discharging] if prevent_simultaneous_charge_and_discharge else None,
+            prevent_simultaneous_flows=[charging.label, discharging.label]
+            if prevent_simultaneous_charge_and_discharge
+            else None,
             meta_data=meta_data,
         )
 
@@ -661,7 +668,7 @@ def __init__(
             on_off_parameters=on_off_parameters,
             prevent_simultaneous_flows=None
             if in2 is None or prevent_simultaneous_flows_in_both_directions is False
-            else [in1, in2],
+            else [in1.label, in2.label],
             meta_data=meta_data,
         )
         self.in1 = in1
@@ -1078,7 +1085,9 @@ def __init__(
             label,
             inputs=inputs,
             outputs=outputs,
-            prevent_simultaneous_flows=(inputs or []) + (outputs or []) if prevent_simultaneous_flow_rates else None,
+            prevent_simultaneous_flows=[flow.label for flow in (inputs or []) + (outputs or [])]
+            if prevent_simultaneous_flow_rates
+            else None,
             meta_data=meta_data,
         )
         self.prevent_simultaneous_flow_rates = prevent_simultaneous_flow_rates
@@ -1206,7 +1215,7 @@ def __init__(
             label,
             outputs=outputs,
             meta_data=meta_data,
-            prevent_simultaneous_flows=outputs if prevent_simultaneous_flow_rates else None,
+            prevent_simultaneous_flows=[flow.label for flow in outputs] if prevent_simultaneous_flow_rates else None,
         )
 
     @property
@@ -1331,7 +1340,7 @@ def __init__(
             label,
             inputs=inputs,
             meta_data=meta_data,
-            prevent_simultaneous_flows=inputs if prevent_simultaneous_flow_rates else None,
+            prevent_simultaneous_flows=[flow.label for flow in inputs] if prevent_simultaneous_flow_rates else None,
         )
 
     @property

flixopt/elements.py

@@ -50,9 +50,30 @@ class Component(Element):
             component has discrete on/off states. Creates binary variables for all
             connected Flows. For better performance, prefer defining OnOffParameters
             on individual Flows when possible.
-        prevent_simultaneous_flows: list of Flows that cannot be active simultaneously.
-            Creates binary variables to enforce mutual exclusivity. Use sparingly as
-            it increases computational complexity.
+        prevent_simultaneous_flows: Defines mutual exclusivity constraints between flows.
+            Each constraint group allows at most 1 flow to be active simultaneously.
+
+            **Single constraint group** (one list):
+                `['flow1', 'flow2', 'flow3']` - At most 1 of these 3 flows can be active.
+
+                Use case: A boiler that can burn different fuels (coal, gas, or biomass)
+                but only one at a time.
+
+            **Multiple constraint groups** (list of lists):
+                `[['coal', 'gas', 'biomass'], ['water_cooling', 'air_cooling']]`
+                Creates two independent constraints:
+                - At most 1 fuel source active (coal OR gas OR biomass)
+                - At most 1 cooling method active (water OR air)
+
+                Use case: A power plant that must choose one fuel AND one cooling method,
+                allowing combinations like (coal+water), (gas+air), etc.
+
+            **Performance note**: Creates binary variables for mutual exclusivity constraints.
+            Use sparingly as it increases computational complexity. The implementation uses
+            string-based flow labels internally for efficient serialization and deserialization.
+
+            Note:
+                Passing Flow objects directly is deprecated. Always use flow label strings.
         meta_data: Used to store additional information. Not used internally but saved
             in results. Only use Python native types.
 
@@ -80,18 +101,74 @@ def __init__(
         inputs: list[Flow] | None = None,
         outputs: list[Flow] | None = None,
         on_off_parameters: OnOffParameters | None = None,
-        prevent_simultaneous_flows: list[Flow] | None = None,
+        prevent_simultaneous_flows: list[str | Flow] | list[list[str | Flow]] | None = None,
         meta_data: dict | None = None,
     ):
         super().__init__(label, meta_data=meta_data)
         self.inputs: list[Flow] = inputs or []
         self.outputs: list[Flow] = outputs or []
         self._check_unique_flow_labels()
         self.on_off_parameters = on_off_parameters
-        self.prevent_simultaneous_flows: list[Flow] = prevent_simultaneous_flows or []
 
         self.flows: dict[str, Flow] = {flow.label: flow for flow in self.inputs + self.outputs}
 
+        # Normalize prevent_simultaneous_flows to always be list of lists of flow labels (strings)
+        self.prevent_simultaneous_flows: list[list[str]] | None = self._normalize_simultaneous_flows(
+            prevent_simultaneous_flows
+        )
+
+    @staticmethod
+    def _normalize_simultaneous_flows(
+        prevent_simultaneous_flows: (
+            list[str | Flow] | tuple[str | Flow, ...] | list[list[str | Flow] | tuple[str | Flow, ...]] | None
+        ),
+    ) -> list[list[str]] | None:
+        """Normalize prevent_simultaneous_flows to always be a list of constraint groups with flow labels.
+
+        Args:
+            prevent_simultaneous_flows: Either None, a single list/tuple of flow labels/objects,
+                or a list of lists/tuples of flow labels/objects.
+                Passing Flow objects is deprecated - use flow label strings instead.
+
+        Returns:
+            List of constraint groups as flow label strings (list of lists of strings), or None if input is None.
+
+        Examples:
+            None -> None
+            ['flow1', 'flow2'] -> [['flow1', 'flow2']]  (preferred)
+            ['flow1', 'flow2', 'flow1'] -> [['flow1', 'flow2']]  (duplicates removed)
+            ['flow1'] -> None  (< 2 unique labels, skipped with warning)
+            [['flow1', 'flow2'], ['flow3']] -> [['flow1', 'flow2']]  (second group skipped)
+        """
+        if prevent_simultaneous_flows is None:
+            return None
+        elif not isinstance(prevent_simultaneous_flows, (list, tuple)):
+            raise TypeError('prevent_simultaneous_flows must be a list or tuple')
+
+        def extract_label(item) -> str:
+            """Extract label from item, warn if it's a Flow object."""
+            if isinstance(item, str):
+                return item
+            elif isinstance(item, Flow):
+                warnings.warn(
+                    'Passing Flow objects to prevent_simultaneous_flows is deprecated. '
+                    'Please use flow label strings instead. '
+                    f"Example: prevent_simultaneous_flows=['{item.label}', ...] instead of [flow_object, ...]",
+                    DeprecationWarning,
+                    stacklevel=4,
+                )
+                return item.label
+            else:
+                raise TypeError(f'Expected str or Flow object, got {type(item).__name__}')
+
+        # Check if it's a list of lists/tuples (multiple groups) or a single list
+        if len(prevent_simultaneous_flows) > 0 and isinstance(prevent_simultaneous_flows[0], (list, tuple)):
+            # Multiple groups: [['flow1', 'flow2'], ['flow3', 'flow4']]
+            return [[extract_label(item) for item in group] for group in prevent_simultaneous_flows]
+        else:
+            # Single group: ['flow1', 'flow2', 'flow3']
+            return [[extract_label(item) for item in prevent_simultaneous_flows]]
+
     def create_model(self, model: FlowSystemModel) -> ComponentModel:
         self._plausibility_checks()
         self.submodel = ComponentModel(model, self)
@@ -115,6 +192,18 @@ def _check_unique_flow_labels(self):
     def _plausibility_checks(self) -> None:
         self._check_unique_flow_labels()
 
+        if self.prevent_simultaneous_flows is not None:
+            for group in self.prevent_simultaneous_flows:
+                if len(set(group)) != len(group):
+                    raise ValueError(
+                        f'Flow names must not occure multiple times in "prevent_simultaneous_flows"! Got {group}'
+                    )
+                for flow_name in group:
+                    if flow_name not in self.flows:
+                        raise ValueError(
+                            f'Flow name "{flow_name}" is not present in the component "{self.label_full}". You cant use it in "prevent_simultaneous_flows". Availlable flows: {list(self.flows)}'
+                        )
+
 
 @register_class_for_io
 class Bus(Element):
@@ -787,10 +876,13 @@ def _do_modeling(self):
                 if flow.on_off_parameters is None:
                     flow.on_off_parameters = OnOffParameters()
 
-        if self.element.prevent_simultaneous_flows:
-            for flow in self.element.prevent_simultaneous_flows:
-                if flow.on_off_parameters is None:
-                    flow.on_off_parameters = OnOffParameters()
+        if self.element.prevent_simultaneous_flows is not None:
+            # Iterate over all flows in all constraint groups (flow_label is a string)
+            for group in self.element.prevent_simultaneous_flows:
+                for flow_label in group:
+                    flow = self.element.flows[flow_label]
+                    if flow.on_off_parameters is None:
+                        flow.on_off_parameters = OnOffParameters()
 
         for flow in all_flows:
             self.add_submodels(flow.create_model(self._model), short_name=flow.label)
@@ -820,12 +912,22 @@ def _do_modeling(self):
             )
 
         if self.element.prevent_simultaneous_flows:
-            # Simultanious Useage --> Only One FLow is On at a time, but needs a Binary for every flow
-            ModelingPrimitives.mutual_exclusivity_constraint(
-                self,
-                binary_variables=[flow.submodel.on_off.on for flow in self.element.prevent_simultaneous_flows],
-                short_name='prevent_simultaneous_use',
-            )
+            # Create mutual exclusivity constraint for each group
+            # Each group enforces "at most 1 flow active at a time"
+            # flow_labels_group is a list of flow label strings
+            for group_idx, flow_labels_group in enumerate(self.element.prevent_simultaneous_flows):
+                constraint_name = (
+                    'prevent_simultaneous_use'
+                    if len(self.element.prevent_simultaneous_flows) == 1
+                    else f'prevent_simultaneous_use|group{group_idx}'
+                )
+                # Look up flows by their labels and get their binary variables
+                flows_in_group = [self.element.flows[label] for label in flow_labels_group]
+                ModelingPrimitives.mutual_exclusivity_constraint(
+                    self,
+                    binary_variables=[flow.submodel.on_off.on for flow in flows_in_group],
+                    short_name=constraint_name,
+                )
 
     def results_structure(self):
         return {