diff --git a/CHANGELOG.md b/CHANGELOG.md index 591232bfc..a6ed04c4b 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 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 --- diff --git a/flixopt/components.py b/flixopt/components.py index 09156e1dc..4f0441989 100644 --- a/flixopt/components.py +++ b/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 diff --git a/flixopt/elements.py b/flixopt/elements.py index a0fd306c0..ca1e75b9b 100644 --- a/flixopt/elements.py +++ b/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,7 +101,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[str | Flow] | list[list[str | Flow]] | None = None, meta_data: dict | None = None, ): super().__init__(label, meta_data=meta_data) @@ -88,10 +109,66 @@ 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 [] 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 { diff --git a/tests/test_component.py b/tests/test_component.py index be1eecf3b..c4fa06c2e 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -599,3 +599,245 @@ 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 + + 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) diff --git a/tests/test_io.py b/tests/test_io.py index f5ca2174a..62b15c8c9 100644 --- a/tests/test_io.py +++ b/tests/test_io.py @@ -1,3 +1,6 @@ +import uuid + +import numpy as np import pytest import flixopt as fx @@ -31,8 +34,13 @@ 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) +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() @@ -41,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('Loaded_IO', 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() @@ -72,5 +80,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'])