Feature/prevent simultanious flow rates #431
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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: 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[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.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)}' | ||
| ) | ||
|
|
||
|
Comment on lines
+195
to
+206
Contributor
There was a problem hiding this comment. 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 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 |
||
|
|
||
| @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 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: | ||
| # 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
Contributor
There was a problem hiding this comment. 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 |
||
|
|
||
| def results_structure(self): | ||
| return { | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Bug: iteration on None when outputs=None and flag=True
Guard None to avoid TypeError.
🤖 Prompt for AI Agents