Skip to content
Open
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

---
Expand Down
21 changes: 15 additions & 6 deletions flixopt/components.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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,
)

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
)
Comment on lines +1218 to 1219
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Bug: iteration on None when outputs=None and flag=True

Guard None to avoid TypeError.

-            prevent_simultaneous_flows=[flow.label for flow in outputs] if prevent_simultaneous_flow_rates else None,
+            prevent_simultaneous_flows=[flow.label for flow in (outputs or [])]
+            if prevent_simultaneous_flow_rates
+            else None,
🤖 Prompt for AI Agents
In flixopt/components.py around lines 1218-1219, the list comprehension uses
"outputs" without guarding for None when prevent_simultaneous_flow_rates is
True, causing a TypeError if outputs is None; change the expression to only
build the list when both prevent_simultaneous_flow_rates is truthy and outputs
is not None (e.g., use a conditional like " [flow.label for flow in outputs] if
prevent_simultaneous_flow_rates and outputs else None") so the field becomes
None instead of iterating over None.


@property
Expand Down Expand Up @@ -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,
)
Comment on lines +1343 to 1344
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Bug: iteration on None when inputs=None and flag=True

Guard None to avoid TypeError.

-            prevent_simultaneous_flows=[flow.label for flow in inputs] if prevent_simultaneous_flow_rates else None,
+            prevent_simultaneous_flows=[flow.label for flow in (inputs or [])]
+            if prevent_simultaneous_flow_rates
+            else None,
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
prevent_simultaneous_flows=[flow.label for flow in inputs] if prevent_simultaneous_flow_rates else None,
)
prevent_simultaneous_flows=[flow.label for flow in (inputs or [])]
if prevent_simultaneous_flow_rates
else None,
)
🤖 Prompt for AI Agents
In flixopt/components.py around lines 1343 to 1344, the list comprehension
prevent_simultaneous_flows=[flow.label for flow in inputs] runs when inputs may
be None, causing a TypeError; fix by guarding inputs before iterating (e.g., set
prevent_simultaneous_flows to [flow.label for flow in inputs] if inputs is not
None and prevent_simultaneous_flow_rates else None, or use inputs or [] when
building the list) so no iteration occurs on None.


@property
Expand Down
132 changes: 117 additions & 15 deletions flixopt/elements.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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[Flow | str] | list[list[Flow | str] | tuple[Flow | str]] | 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 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_label', 'flow2_label']] (deprecated)
[['flow1', 'flow2'], ['flow3', 'flow4']] -> [['flow1', 'flow2'], ['flow3', 'flow4']]
[[flow1, flow2], [flow3, flow4]] -> [['flow1_label', 'flow2_label'], ['flow3_label', 'flow4_label']] (deprecated)
"""
import warnings

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]]

Comment thread
coderabbitai[bot] marked this conversation as resolved.
def create_model(self, model: FlowSystemModel) -> ComponentModel:
self._plausibility_checks()
self.submodel = ComponentModel(model, self)
Expand All @@ -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)}'
)

Comment on lines +195 to +206
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Add guard for groups with fewer than 2 flows in plausibility checks.

While the duplicate and existence checks are good, there's no validation that each group has at least 2 flows. The mutual_exclusivity_constraint (see modeling.py line ~229) asserts len(binary_variables) >= 2. A group with 0 or 1 flow will cause an assertion failure during modeling rather than a clear error message here.

Add after line 200:

                    raise ValueError(
                        f'Flow names must not occure multiple times in "prevent_simultaneous_flows"! Got {group}'
                    )
+               if len(set(group)) < 2:
+                   raise ValueError(
+                       f'Each prevent_simultaneous_flows group must contain at least 2 unique flows. '
+                       f'Got {len(set(group))} in group: {group}'
+                   )
                for flow_name in group:
🤖 Prompt for AI Agents
In flixopt/elements.py around lines 195 to 206, the plausibility checks for
prevent_simultaneous_flows lack a guard that each group contains at least two
flow names; add a check after the existing duplicate/existence checks (or before
modeling) that raises a ValueError if len(group) < 2 with a clear message naming
the component and the invalid group so users get an immediate, descriptive error
instead of an assertion failure later.


@register_class_for_io
class Bus(Element):
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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,
)
Comment on lines +915 to +930
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Validate labels and guard small groups before creating constraints

Avoid KeyError on unknown labels and assertion on groups with <2 variables; also dedup here as a final safety net.

-        for group_idx, flow_labels_group in enumerate(self.element.prevent_simultaneous_flows):
+        for group_idx, flow_labels_group in enumerate(self.element.prevent_simultaneous_flows):
+            # Dedup and validate labels early
+            unique_labels = list(dict.fromkeys(flow_labels_group))
+            missing = [lbl for lbl in unique_labels if lbl not in self.element.flows]
+            if missing:
+                raise PlausibilityError(
+                    f'{self.label_full}: prevent_simultaneous_flows references unknown flows: {missing}. '
+                    f'Available: {list(self.element.flows)}'
+                )
@@
-            flows_in_group = [self.element.flows[label] for label in flow_labels_group]
+            flows_in_group = [self.element.flows[label] for label in unique_labels]
+            if len(flows_in_group) < 2:
+                warnings.warn(
+                    f'{self.label_full}: skipping prevent_simultaneous_flows group {group_idx} '
+                    f'with fewer than 2 unique flows.',
+                    UserWarning,
+                )
+                continue
🤖 Prompt for AI Agents
In flixopt/elements.py around lines 903 to 918, the loop creating mutual
exclusivity constraints must validate flow labels and guard tiny groups: first,
for each flow_labels_group, filter and deduplicate labels while checking
membership in self.element.flows (raise a clear ValueError listing unknown
labels or optionally log and skip the group); then build flows_in_group from
only the recognized labels; if fewer than 2 unique variables remain, skip
creating the constraint (or log a debug message) to avoid assertions; finally
pass the list of binary variables as before. Ensure the constraint_name logic is
unchanged and that any mutation preserves order if needed.


def results_structure(self):
return {
Expand Down
Loading
Loading