From 75a26fbc867d2aedc1ca3cde7ce393f7e2d74a79 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 13:30:56 +0200 Subject: [PATCH 01/10] Add support for multiple groups to restrict simultanious usage by --- flixopt/components.py | 6 +++- flixopt/elements.py | 78 ++++++++++++++++++++++++++++++++++--------- 2 files changed, 68 insertions(+), 16 deletions(-) diff --git a/flixopt/components.py b/flixopt/components.py index 09156e1dc..8424f6afb 100644 --- a/flixopt/components.py +++ b/flixopt/components.py @@ -62,6 +62,9 @@ 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: Flows 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. meta_data: Used to store additional information about the Element. Not used internally, but saved in results. Only use Python native types. @@ -168,9 +171,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[Flow] | list[list[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 diff --git a/flixopt/elements.py b/flixopt/elements.py index a0fd306c0..bd2972853 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -50,9 +50,16 @@ 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: Flows that cannot be active simultaneously. + Can be either a single list of flows (one constraint group) or a list of lists + (multiple independent constraint groups). Each group enforces "at most 1 flow + active" within that group. Creates binary variables to enforce mutual exclusivity. + Use sparingly as it increases computational complexity. + + Examples: + - `[flow1, flow2, flow3]` - At most 1 of these 3 flows can be active + - `[[fuel1, fuel2], [cooling1, cooling2]]` - Two independent constraints: + at most 1 fuel AND at most 1 cooling method meta_data: Used to store additional information. Not used internally but saved in results. Only use Python native types. @@ -80,7 +87,7 @@ 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[Flow] | list[list[Flow]] | None = None, meta_data: dict | None = None, ): super().__init__(label, meta_data=meta_data) @@ -88,10 +95,42 @@ def __init__( 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 [] + + # Normalize prevent_simultaneous_flows to always be list of lists + self.prevent_simultaneous_flows: list[list[Flow]] = self._normalize_simultaneous_flows( + prevent_simultaneous_flows + ) self.flows: dict[str, Flow] = {flow.label: flow for flow in self.inputs + self.outputs} + @staticmethod + def _normalize_simultaneous_flows( + prevent_simultaneous_flows: list[Flow] | list[list[Flow]] | None, + ) -> list[list[Flow]] | None: + """Normalize prevent_simultaneous_flows to always be a list of constraint groups. + + Args: + prevent_simultaneous_flows: Either None, a single list of flows, or a list of lists + + Returns: + List of constraint groups (list of lists). Empty list if input is None. + + Examples: + None -> [] + [flow1, flow2] -> [[flow1, flow2]] + [[flow1, flow2], [flow3, flow4]] -> [[flow1, flow2], [flow3, flow4]] + """ + if prevent_simultaneous_flows is None: + return None + + # Check if it's a list of lists by examining the first element + if prevent_simultaneous_flows and isinstance(prevent_simultaneous_flows[0], list): + # Already a list of lists + return prevent_simultaneous_flows + else: + # Single list - wrap it in another list + return [prevent_simultaneous_flows] if prevent_simultaneous_flows else [] + def create_model(self, model: FlowSystemModel) -> ComponentModel: self._plausibility_checks() self.submodel = ComponentModel(model, self) @@ -787,10 +826,12 @@ 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 + for group in self.element.prevent_simultaneous_flows: + for flow in group: + 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 +861,19 @@ 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" + for group_idx, flow_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}' + ) + ModelingPrimitives.mutual_exclusivity_constraint( + self, + binary_variables=[flow.submodel.on_off.on for flow in flow_group], + short_name=constraint_name, + ) def results_structure(self): return { From ad0546586fd1b60de1bee449d38519ac8e4da42c Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 13:44:41 +0200 Subject: [PATCH 02/10] Add support for multiple groups to restrict simultanious usage by --- flixopt/components.py | 10 +++++----- flixopt/elements.py | 14 ++++++-------- 2 files changed, 11 insertions(+), 13 deletions(-) diff --git a/flixopt/components.py b/flixopt/components.py index 8424f6afb..a01e5828d 100644 --- a/flixopt/components.py +++ b/flixopt/components.py @@ -405,7 +405,7 @@ def __init__( label, inputs=[charging], outputs=[discharging], - prevent_simultaneous_flows=[charging, discharging] if prevent_simultaneous_charge_and_discharge else None, + prevent_simultaneous_flows=[(charging, discharging)] if prevent_simultaneous_charge_and_discharge else None, meta_data=meta_data, ) @@ -665,7 +665,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, in2)], meta_data=meta_data, ) self.in1 = in1 @@ -1082,7 +1082,7 @@ def __init__( label, inputs=inputs, outputs=outputs, - prevent_simultaneous_flows=(inputs or []) + (outputs or []) if prevent_simultaneous_flow_rates else None, + prevent_simultaneous_flows=[(inputs or []) + (outputs or [])] if prevent_simultaneous_flow_rates else None, meta_data=meta_data, ) self.prevent_simultaneous_flow_rates = prevent_simultaneous_flow_rates @@ -1210,7 +1210,7 @@ def __init__( label, outputs=outputs, meta_data=meta_data, - prevent_simultaneous_flows=outputs if prevent_simultaneous_flow_rates else None, + prevent_simultaneous_flows=[[outputs]] if prevent_simultaneous_flow_rates else None, ) @property @@ -1335,7 +1335,7 @@ def __init__( label, inputs=inputs, meta_data=meta_data, - prevent_simultaneous_flows=inputs if prevent_simultaneous_flow_rates else None, + prevent_simultaneous_flows=[inputs] if prevent_simultaneous_flow_rates else None, ) @property diff --git a/flixopt/elements.py b/flixopt/elements.py index bd2972853..d0839829c 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -97,7 +97,7 @@ def __init__( self.on_off_parameters = on_off_parameters # Normalize prevent_simultaneous_flows to always be list of lists - self.prevent_simultaneous_flows: list[list[Flow]] = self._normalize_simultaneous_flows( + self.prevent_simultaneous_flows: list[list[Flow]] | None = self._normalize_simultaneous_flows( prevent_simultaneous_flows ) @@ -105,7 +105,7 @@ def __init__( @staticmethod def _normalize_simultaneous_flows( - prevent_simultaneous_flows: list[Flow] | list[list[Flow]] | None, + prevent_simultaneous_flows: list[Flow] | list[list[Flow] | tuple[Flow]] | None, ) -> list[list[Flow]] | None: """Normalize prevent_simultaneous_flows to always be a list of constraint groups. @@ -122,14 +122,12 @@ def _normalize_simultaneous_flows( """ if prevent_simultaneous_flows is None: return None - - # Check if it's a list of lists by examining the first element - if prevent_simultaneous_flows and isinstance(prevent_simultaneous_flows[0], list): - # Already a list of lists + elif not isinstance(prevent_simultaneous_flows, (list, tuple)): + raise TypeError('Wrong type') + elif isinstance(prevent_simultaneous_flows[0], (list, tuple)): return prevent_simultaneous_flows else: - # Single list - wrap it in another list - return [prevent_simultaneous_flows] if prevent_simultaneous_flows else [] + return [prevent_simultaneous_flows] def create_model(self, model: FlowSystemModel) -> ComponentModel: self._plausibility_checks() From 7b707eb256770f83e67166bdf3444f7e75dc607a Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 14:55:55 +0200 Subject: [PATCH 03/10] Change prevent simultanious flows to use str instead of flow objects --- CHANGELOG.md | 6 ++ flixopt/components.py | 19 ++-- flixopt/elements.py | 83 ++++++++++++----- tests/test_component.py | 196 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 272 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 591232bfc..f4c3ed57c 100644 --- a/CHANGELOG.md +++ b/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 constraint groups. Each group enforces "at most 1 flow active" within that group. Example: `prevent_simultaneous_flows=[[fuel1, fuel2], [cooling1, cooling2]]` ensures at most 1 fuel AND at most 1 cooling method are active simultaneously, but allows one of each. ### 💥 Breaking Changes ### ♻️ Changed +- **Internal representation of `prevent_simultaneous_flows`**: Now stores flow labels (strings) instead of Flow object references. This improves serialization and avoids circular references. The API remains unchanged - users still pass Flow objects, which are automatically converted to labels internally. ### 🗑️ Deprecated +- **Passing Flow objects to `prevent_simultaneous_flows`**: Now deprecated in favor of passing flow label strings directly. Example: use `prevent_simultaneous_flows=['flow1', 'flow2']` instead of `prevent_simultaneous_flows=[flow1_object, flow2_object]`. Flow objects still work but will 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 --- diff --git a/flixopt/components.py b/flixopt/components.py index a01e5828d..4f0441989 100644 --- a/flixopt/components.py +++ b/flixopt/components.py @@ -62,9 +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: Flows that cannot be active simultaneously. + 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. @@ -171,7 +172,7 @@ 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[Flow] | list[list[Flow]] | 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, prevent_simultaneous_flows, meta_data=meta_data) @@ -405,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, ) @@ -665,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 @@ -1082,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 @@ -1210,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 @@ -1335,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 diff --git a/flixopt/elements.py b/flixopt/elements.py index d0839829c..715eabb17 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -50,16 +50,19 @@ 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: Flows that cannot be active simultaneously. - Can be either a single list of flows (one constraint group) or a list of lists + prevent_simultaneous_flows: Flow labels (strings) that cannot be active simultaneously. + Can be either a single list of flow labels (one constraint group) or a list of lists (multiple independent constraint groups). Each group enforces "at most 1 flow active" within that group. Creates binary variables to enforce mutual exclusivity. Use sparingly as it increases computational complexity. Examples: - - `[flow1, flow2, flow3]` - At most 1 of these 3 flows can be active - - `[[fuel1, fuel2], [cooling1, cooling2]]` - Two independent constraints: + - `['flow1', 'flow2', 'flow3']` - At most 1 of these 3 flows can be active + - `[['fuel1', 'fuel2'], ['cooling1', 'cooling2']]` - Two independent constraints: at most 1 fuel AND at most 1 cooling method + + Note: + Passing Flow objects is deprecated. Use flow label strings instead. meta_data: Used to store additional information. Not used internally but saved in results. Only use Python native types. @@ -87,7 +90,7 @@ def __init__( inputs: list[Flow] | None = None, outputs: list[Flow] | None = None, on_off_parameters: OnOffParameters | None = None, - prevent_simultaneous_flows: list[Flow] | list[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) @@ -96,38 +99,64 @@ def __init__( self._check_unique_flow_labels() self.on_off_parameters = on_off_parameters - # Normalize prevent_simultaneous_flows to always be list of lists - self.prevent_simultaneous_flows: list[list[Flow]] | None = self._normalize_simultaneous_flows( + 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 ) - self.flows: dict[str, Flow] = {flow.label: flow for flow in self.inputs + self.outputs} - @staticmethod def _normalize_simultaneous_flows( - prevent_simultaneous_flows: list[Flow] | list[list[Flow] | tuple[Flow]] | None, - ) -> list[list[Flow]] | None: - """Normalize prevent_simultaneous_flows to always be a list of constraint groups. + 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 flows, or a list of lists + 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 (list of lists). Empty list if input is None. + List of constraint groups as flow label strings (list of lists of strings), or None if input is None. Examples: - None -> [] - [flow1, flow2] -> [[flow1, flow2]] - [[flow1, flow2], [flow3, flow4]] -> [[flow1, flow2], [flow3, flow4]] + 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('Wrong type') - elif isinstance(prevent_simultaneous_flows[0], (list, tuple)): - return prevent_simultaneous_flows + 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: - return [prevent_simultaneous_flows] + # 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() @@ -825,9 +854,10 @@ def _do_modeling(self): flow.on_off_parameters = OnOffParameters() if self.element.prevent_simultaneous_flows is not None: - # Iterate over all flows in all constraint groups + # Iterate over all flows in all constraint groups (flow_label is a string) for group in self.element.prevent_simultaneous_flows: - for flow in group: + for flow_label in group: + flow = self.element.flows[flow_label] if flow.on_off_parameters is None: flow.on_off_parameters = OnOffParameters() @@ -861,15 +891,18 @@ 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" - for group_idx, flow_group in enumerate(self.element.prevent_simultaneous_flows): + # 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 flow_group], + binary_variables=[flow.submodel.on_off.on for flow in flows_in_group], short_name=constraint_name, ) diff --git a/tests/test_component.py b/tests/test_component.py index be1eecf3b..edb8e261b 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -599,3 +599,199 @@ def test_transmission_unbalanced(self, basic_flow_system, highs_solver): 10, 'Sizing does not work properly', ) + + +class TestPreventSimultaneousFlows: + """Tests for prevent_simultaneous_flows feature with multiple constraint groups.""" + + def test_single_constraint_group(self, basic_flow_system_linopy_coords, coords_config): + """Test backward compatibility: single list of flows creates one constraint.""" + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config + + # Create flows + flow1 = fx.Flow('flow1', bus='Fernwärme', size=100) + flow2 = fx.Flow('flow2', bus='Fernwärme', size=100) + flow3 = fx.Flow('flow3', bus='Fernwärme', size=100) + output_flow = fx.Flow('output', bus='Gas', size=200) + + # Create converter with single constraint group using string labels (preferred) + converter = fx.LinearConverter( + label='single_group', + inputs=[flow1, flow2, flow3], + outputs=[output_flow], + conversion_factors=[{'flow1': 1, 'output': 0.9}], + prevent_simultaneous_flows=['flow1', 'flow2', 'flow3'], # String labels (preferred) + ) + + flow_system.add_elements(converter) + model = create_linopy_model(flow_system) + + # Check that binary variables were created for each flow + assert 'single_group(flow1)|on' in model.variables + assert 'single_group(flow2)|on' in model.variables + assert 'single_group(flow3)|on' in model.variables + + # Check that the mutual exclusivity constraint exists + assert 'single_group|prevent_simultaneous_use' in model.constraints + + # Verify constraint: sum of on variables <= 1 + assert_conequal( + model.constraints['single_group|prevent_simultaneous_use'], + model.variables['single_group(flow1)|on'] + + model.variables['single_group(flow2)|on'] + + model.variables['single_group(flow3)|on'] + <= 1, + ) + + def test_multiple_constraint_groups(self, basic_flow_system_linopy_coords, coords_config): + """Test multiple independent constraint groups using string labels.""" + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config + + # Add required buses + flow_system.add_elements( + fx.Bus('fuel'), + fx.Bus('cooling'), + fx.Bus('steam'), + ) + + # Create flows for different constraint groups + coal = fx.Flow('coal', bus='fuel', size=100) + gas = fx.Flow('gas', bus='fuel', size=100) + biomass = fx.Flow('biomass', bus='fuel', size=100) + + water_cooling = fx.Flow('water_cooling', bus='cooling', size=50) + air_cooling = fx.Flow('air_cooling', bus='cooling', size=50) + + steam = fx.Flow('steam', bus='steam', size=200) + + # Create converter with two independent constraint groups using string labels (preferred) + converter = fx.LinearConverter( + label='multi_group', + inputs=[coal, gas, biomass, water_cooling, air_cooling], + outputs=[steam], + conversion_factors=[{'coal': 1, 'steam': 0.8}], + prevent_simultaneous_flows=[ + ['coal', 'gas', 'biomass'], # Group 0: at most 1 fuel (string labels) + ['water_cooling', 'air_cooling'], # Group 1: at most 1 cooling (string labels) + ], + ) + + flow_system.add_elements(converter) + model = create_linopy_model(flow_system) + + # Check that binary variables exist for all flows + for flow_name in ['coal', 'gas', 'biomass', 'water_cooling', 'air_cooling']: + assert f'multi_group({flow_name})|on' in model.variables + + # Check that two separate constraints exist + assert 'multi_group|prevent_simultaneous_use|group0' in model.constraints + assert 'multi_group|prevent_simultaneous_use|group1' in model.constraints + + # Verify first constraint: sum of fuel flow on-variables <= 1 + assert_conequal( + model.constraints['multi_group|prevent_simultaneous_use|group0'], + model.variables['multi_group(coal)|on'] + + model.variables['multi_group(gas)|on'] + + model.variables['multi_group(biomass)|on'] + <= 1, + ) + + # Verify second constraint: sum of cooling flow on-variables <= 1 + assert_conequal( + model.constraints['multi_group|prevent_simultaneous_use|group1'], + model.variables['multi_group(water_cooling)|on'] + model.variables['multi_group(air_cooling)|on'] <= 1, + ) + + def test_normalization_internal_representation(self): + """Test that single list is normalized to list of lists with flow labels internally.""" + flow1 = fx.Flow('f1', bus='test', size=10) + flow2 = fx.Flow('f2', bus='test', size=10) + flow3 = fx.Flow('f3', bus='test', size=10) + output = fx.Flow('out', bus='test', size=20) + + # Create with single list (Flow objects - deprecated but still works) + with pytest.warns(DeprecationWarning, match='Passing Flow objects to prevent_simultaneous_flows is deprecated'): + conv1 = fx.LinearConverter( + label='test1', + inputs=[flow1, flow2, flow3], + outputs=[output], + conversion_factors=[{'f1': 1, 'out': 0.9}], + prevent_simultaneous_flows=[flow1, flow2], # Single list + ) + + # Internal representation should be list of lists of flow label strings + assert isinstance(conv1.prevent_simultaneous_flows, list) + assert len(conv1.prevent_simultaneous_flows) == 1 + assert isinstance(conv1.prevent_simultaneous_flows[0], list) + assert len(conv1.prevent_simultaneous_flows[0]) == 2 + # Verify they are strings (flow labels), not Flow objects + assert all(isinstance(label, str) for label in conv1.prevent_simultaneous_flows[0]) + assert set(conv1.prevent_simultaneous_flows[0]) == {'f1', 'f2'} + + def test_string_labels_preferred(self): + """Test that using string labels (preferred way) works without deprecation warnings.""" + flow1 = fx.Flow('f1', bus='test', size=10) + flow2 = fx.Flow('f2', bus='test', size=10) + flow3 = fx.Flow('f3', bus='test', size=10) + output = fx.Flow('out', bus='test', size=20) + + # Create with string labels (preferred - no deprecation warning) + conv = fx.LinearConverter( + label='test_strings', + inputs=[flow1, flow2, flow3], + outputs=[output], + conversion_factors=[{'f1': 1, 'out': 0.9}], + prevent_simultaneous_flows=['f1', 'f2'], # String labels + ) + + # Should have same internal representation + assert conv.prevent_simultaneous_flows == [['f1', 'f2']] + + def test_multiple_groups_with_strings(self): + """Test multiple constraint groups using string labels.""" + flow1 = fx.Flow('coal', bus='test', size=10) + flow2 = fx.Flow('gas', bus='test', size=10) + flow3 = fx.Flow('water', bus='test', size=10) + flow4 = fx.Flow('air', bus='test', size=10) + output = fx.Flow('out', bus='test', size=20) + + # Multiple groups with string labels (preferred) + conv = fx.LinearConverter( + label='test_multi', + inputs=[flow1, flow2, flow3, flow4], + outputs=[output], + conversion_factors=[{'coal': 1, 'out': 0.9}], + prevent_simultaneous_flows=[ + ['coal', 'gas'], # Fuel group + ['water', 'air'], # Cooling group + ], + ) + + assert conv.prevent_simultaneous_flows == [['coal', 'gas'], ['water', 'air']] + + def test_empty_prevent_simultaneous_flows(self, basic_flow_system_linopy_coords, coords_config): + """Test that None or empty prevent_simultaneous_flows works correctly.""" + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config + + flow1 = fx.Flow('f1', bus='Fernwärme', size=10) + flow2 = fx.Flow('f2', bus='Fernwärme', size=10) + output = fx.Flow('out', bus='Gas', size=20) + + # Create with None + conv = fx.LinearConverter( + label='test_none', + inputs=[flow1, flow2], + outputs=[output], + conversion_factors=[{'f1': 1, 'out': 0.9}], + prevent_simultaneous_flows=None, + ) + + flow_system.add_elements(conv) + model = create_linopy_model(flow_system) + + # No prevent_simultaneous_use constraints should exist + prevent_constraints = [c for c in model.constraints if 'prevent_simultaneous_use' in c] + assert len(prevent_constraints) == 0 + + # Internal representation should be None + assert conv.prevent_simultaneous_flows is None From f34c40ad2aa2e26dcaf546e68bbf4ecc35eac232 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 15:51:28 +0200 Subject: [PATCH 04/10] Improve docs and add test --- CHANGELOG.md | 5 ++- flixopt/elements.py | 31 ++++++++++------ tests/test_io.py | 88 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f4c3ed57c..f8c2ff0ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,15 +53,14 @@ 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 constraint groups. Each group enforces "at most 1 flow active" within that group. Example: `prevent_simultaneous_flows=[[fuel1, fuel2], [cooling1, cooling2]]` ensures at most 1 fuel AND at most 1 cooling method are active simultaneously, but allows one of each. +- **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. ### 💥 Breaking Changes ### ♻️ Changed -- **Internal representation of `prevent_simultaneous_flows`**: Now stores flow labels (strings) instead of Flow object references. This improves serialization and avoids circular references. The API remains unchanged - users still pass Flow objects, which are automatically converted to labels internally. ### 🗑️ Deprecated -- **Passing Flow objects to `prevent_simultaneous_flows`**: Now deprecated in favor of passing flow label strings directly. Example: use `prevent_simultaneous_flows=['flow1', 'flow2']` instead of `prevent_simultaneous_flows=[flow1_object, flow2_object]`. Flow objects still work but will trigger a DeprecationWarning. +- **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 diff --git a/flixopt/elements.py b/flixopt/elements.py index 715eabb17..ac2a9e98a 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -50,19 +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: Flow labels (strings) that cannot be active simultaneously. - Can be either a single list of flow labels (one constraint group) or a list of lists - (multiple independent constraint groups). Each group enforces "at most 1 flow - active" within that group. 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. - Examples: - - `['flow1', 'flow2', 'flow3']` - At most 1 of these 3 flows can be active - - `[['fuel1', 'fuel2'], ['cooling1', 'cooling2']]` - Two independent constraints: - at most 1 fuel AND at most 1 cooling method + **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 is deprecated. Use flow label strings instead. + 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. diff --git a/tests/test_io.py b/tests/test_io.py index f5ca2174a..3e583b89e 100644 --- a/tests/test_io.py +++ b/tests/test_io.py @@ -72,5 +72,93 @@ def test_flow_system_io(flow_system): flow_system.__str__() +def test_prevent_simultaneous_flows_single_group_roundtrip(): + """Test that single constraint group serializes and deserializes correctly.""" + import pandas as pd + + timesteps = pd.date_range('2020-01-01', periods=3, freq='h') + fs = fx.FlowSystem(timesteps=timesteps) + + # Add buses + fs.add_elements( + fx.Bus('test'), + fx.Bus('output_bus'), + ) + + # Create flows and converter with single constraint group + flow1 = fx.Flow('flow1', bus='test', size=100) + flow2 = fx.Flow('flow2', bus='test', size=100) + flow3 = fx.Flow('flow3', bus='test', size=100) + output = fx.Flow('output', bus='output_bus', size=200) + + conv = fx.LinearConverter( + 'test_conv', + inputs=[flow1, flow2, flow3], + outputs=[output], + conversion_factors=[{'flow1': 1, 'output': 0.9}], + prevent_simultaneous_flows=['flow1', 'flow2', 'flow3'], # Single group (string labels) + ) + + fs.add_elements(conv) + + # Serialize and deserialize + ds = fs.to_dataset() + new_fs = fx.FlowSystem.from_dataset(ds) + + # Verify prevent_simultaneous_flows is preserved + new_conv = new_fs.components['test_conv'] + assert new_conv.prevent_simultaneous_flows == [['flow1', 'flow2', 'flow3']] + assert new_conv.prevent_simultaneous_flows == conv.prevent_simultaneous_flows + + +def test_prevent_simultaneous_flows_multiple_groups_roundtrip(): + """Test that multiple constraint groups serialize and deserialize correctly.""" + import pandas as pd + + timesteps = pd.date_range('2020-01-01', periods=3, freq='h') + fs = fx.FlowSystem(timesteps=timesteps) + + # Add buses + fs.add_elements( + fx.Bus('fuel'), + fx.Bus('cooling'), + fx.Bus('steam'), + ) + + # Create flows for different constraint groups + coal = fx.Flow('coal', bus='fuel', size=100) + gas = fx.Flow('gas', bus='fuel', size=100) + biomass = fx.Flow('biomass', bus='fuel', size=100) + + water_cooling = fx.Flow('water_cooling', bus='cooling', size=50) + air_cooling = fx.Flow('air_cooling', bus='cooling', size=50) + + steam = fx.Flow('steam', bus='steam', size=200) + + # Create converter with two independent constraint groups + conv = fx.LinearConverter( + 'power_plant', + inputs=[coal, gas, biomass, water_cooling, air_cooling], + outputs=[steam], + conversion_factors=[{'coal': 1, 'steam': 0.8}], + prevent_simultaneous_flows=[ + ['coal', 'gas', 'biomass'], # Group 0: at most 1 fuel + ['water_cooling', 'air_cooling'], # Group 1: at most 1 cooling method + ], + ) + + fs.add_elements(conv) + + # Serialize and deserialize + ds = fs.to_dataset() + new_fs = fx.FlowSystem.from_dataset(ds) + + # Verify prevent_simultaneous_flows is preserved with correct structure + new_conv = new_fs.components['power_plant'] + expected = [['coal', 'gas', 'biomass'], ['water_cooling', 'air_cooling']] + assert new_conv.prevent_simultaneous_flows == expected + assert new_conv.prevent_simultaneous_flows == conv.prevent_simultaneous_flows + + if __name__ == '__main__': pytest.main(['-v', '--disable-warnings']) From 762b717ff23b282408a80aa871d462f4a778ba86 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 15:53:23 +0200 Subject: [PATCH 05/10] Update CHANGELOG.md --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f8c2ff0ee..a6ed04c4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,7 @@ If upgrading from v2.x, see the [v3.0.0 release notes](https://github.com/flixOp ### ✨ 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 From a9694b104dc81a607e7fb7dce7234d2b0d5ea6de Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 16:45:18 +0200 Subject: [PATCH 06/10] Add plausibility checks --- flixopt/elements.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/flixopt/elements.py b/flixopt/elements.py index ac2a9e98a..5c8681410 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -192,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): From af642c97c6a0be8a967d2fe3bab3cfe4881b98ce Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 16:54:32 +0200 Subject: [PATCH 07/10] Reduce tests --- tests/test_component.py | 46 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/tests/test_component.py b/tests/test_component.py index edb8e261b..c4fa06c2e 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -795,3 +795,49 @@ def test_empty_prevent_simultaneous_flows(self, basic_flow_system_linopy_coords, # Internal representation should be None assert conv.prevent_simultaneous_flows is None + + def test_unknown_label_during_modeling(self, basic_flow_system_linopy_coords): + """Test that unknown flow label raises ValueError during full modeling.""" + flow_system = basic_flow_system_linopy_coords + + # Create flows and converter with unknown label + flow1 = fx.Flow('flow1', bus='Fernwärme', size=100) + flow2 = fx.Flow('flow2', bus='Fernwärme', size=100) + output = fx.Flow('output', bus='Gas', size=200) + + conv = fx.LinearConverter( + label='bad_conv', + inputs=[flow1, flow2], + outputs=[output], + conversion_factors=[{'flow1': 1, 'output': 0.9}], + prevent_simultaneous_flows=['flow1', 'nonexistent'], # 'nonexistent' doesn't exist + ) + + flow_system.add_elements(conv) + + # Modeling should fail with ValueError during plausibility checks + with pytest.raises(ValueError, match='Flow name "nonexistent" is not present'): + create_linopy_model(flow_system) + + def test_duplicate_labels_during_modeling(self, basic_flow_system_linopy_coords): + """Test that duplicate flow labels raise ValueError during full modeling.""" + flow_system = basic_flow_system_linopy_coords + + # Create flows and converter with duplicate labels + flow1 = fx.Flow('flow1', bus='Fernwärme', size=100) + flow2 = fx.Flow('flow2', bus='Fernwärme', size=100) + output = fx.Flow('output', bus='Gas', size=200) + + conv = fx.LinearConverter( + label='bad_conv', + inputs=[flow1, flow2], + outputs=[output], + conversion_factors=[{'flow1': 1, 'output': 0.9}], + prevent_simultaneous_flows=['flow1', 'flow2', 'flow1'], # 'flow1' appears twice + ) + + flow_system.add_elements(conv) + + # Modeling should fail with ValueError during plausibility checks + with pytest.raises(ValueError, match='Flow names must not occure multiple times'): + create_linopy_model(flow_system) From e48ca2fc8256195d939f51b10f7e01b9c15b26c0 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 17:04:57 +0200 Subject: [PATCH 08/10] Improve type hints --- flixopt/elements.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/flixopt/elements.py b/flixopt/elements.py index 5c8681410..ca1e75b9b 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -119,12 +119,14 @@ def __init__( @staticmethod def _normalize_simultaneous_flows( - prevent_simultaneous_flows: list[Flow | str] | list[list[Flow | str] | tuple[Flow | str]] | None, + 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 of flow labels/objects, + 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. @@ -134,12 +136,10 @@ def _normalize_simultaneous_flows( 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) + ['flow1', 'flow2', 'flow1'] -> [['flow1', 'flow2']] (duplicates removed) + ['flow1'] -> None (< 2 unique labels, skipped with warning) + [['flow1', 'flow2'], ['flow3']] -> [['flow1', 'flow2']] (second group skipped) """ - import warnings - if prevent_simultaneous_flows is None: return None elif not isinstance(prevent_simultaneous_flows, (list, tuple)): From 0f1a021e93aa2340a854af70f70eb92449823206 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 18:36:07 +0200 Subject: [PATCH 09/10] Fix test file io concurrency issue --- tests/test_io.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_io.py b/tests/test_io.py index 3e583b89e..7cb598736 100644 --- a/tests/test_io.py +++ b/tests/test_io.py @@ -1,3 +1,4 @@ +import numpy as np import pytest import flixopt as fx @@ -12,6 +13,8 @@ simple_flow_system_scenarios, ) +np.random.seed(42) + @pytest.fixture( params=[ @@ -32,7 +35,8 @@ def flow_system(request): @pytest.mark.slow def test_flow_system_file_io(flow_system, highs_solver): - calculation_0 = fx.FullCalculation('IO', flow_system=flow_system) + nr = np.random.randint(0, 1e12) + calculation_0 = fx.FullCalculation(f'IO-{nr}', flow_system=flow_system) calculation_0.do_modeling() calculation_0.solve(highs_solver) calculation_0.flow_system.plot_network() @@ -41,7 +45,7 @@ def test_flow_system_file_io(flow_system, highs_solver): paths = CalculationResultsPaths(calculation_0.folder, calculation_0.name) flow_system_1 = fx.FlowSystem.from_netcdf(paths.flow_system) - calculation_1 = fx.FullCalculation('Loaded_IO', flow_system=flow_system_1) + calculation_1 = fx.FullCalculation(f'Loaded_IO{nr}', flow_system=flow_system_1) calculation_1.do_modeling() calculation_1.solve(highs_solver) calculation_1.flow_system.plot_network() From dc14f0d74f9ee4c481ea0409f755f6a46546774d Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 20 Oct 2025 19:30:58 +0200 Subject: [PATCH 10/10] Fix concurrency testing issue --- tests/test_io.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/tests/test_io.py b/tests/test_io.py index 7cb598736..62b15c8c9 100644 --- a/tests/test_io.py +++ b/tests/test_io.py @@ -1,3 +1,5 @@ +import uuid + import numpy as np import pytest @@ -13,8 +15,6 @@ simple_flow_system_scenarios, ) -np.random.seed(42) - @pytest.fixture( params=[ @@ -34,9 +34,13 @@ def flow_system(request): @pytest.mark.slow -def test_flow_system_file_io(flow_system, highs_solver): - nr = np.random.randint(0, 1e12) - calculation_0 = fx.FullCalculation(f'IO-{nr}', flow_system=flow_system) +def test_flow_system_file_io(flow_system, highs_solver, request): + # Use UUID to ensure unique names across parallel test workers + unique_id = uuid.uuid4().hex[:12] + worker_id = getattr(request.config, 'workerinput', {}).get('workerid', 'main') + test_id = f'{worker_id}-{unique_id}' + + calculation_0 = fx.FullCalculation(f'IO-{test_id}', flow_system=flow_system) calculation_0.do_modeling() calculation_0.solve(highs_solver) calculation_0.flow_system.plot_network() @@ -45,7 +49,7 @@ def test_flow_system_file_io(flow_system, highs_solver): paths = CalculationResultsPaths(calculation_0.folder, calculation_0.name) flow_system_1 = fx.FlowSystem.from_netcdf(paths.flow_system) - calculation_1 = fx.FullCalculation(f'Loaded_IO{nr}', flow_system=flow_system_1) + calculation_1 = fx.FullCalculation(f'Loaded_IO-{test_id}', flow_system=flow_system_1) calculation_1.do_modeling() calculation_1.solve(highs_solver) calculation_1.flow_system.plot_network()