From b4a92361ca5777e126c638b1d6c17e752e1a3889 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Wed, 23 Jul 2025 16:48:13 +0200 Subject: [PATCH 01/10] Update Effect name in tests to be 'costs' instead of 'Costs' Everywhere Simplify testing by creating a Element Library --- tests/conftest.py | 742 +++++++++++++++---------- tests/test_effect.py | 16 +- tests/test_effects_shares_summation.py | 16 +- tests/test_flow.py | 46 +- tests/test_integration.py | 6 +- tests/test_linear_converter.py | 16 +- tests/test_scenarios.py | 8 +- 7 files changed, 487 insertions(+), 363 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index d12ca9eca..3d6623eb7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,7 +5,7 @@ """ import os -from typing import Iterable +from typing import Dict, Iterable import linopy.testing import numpy as np @@ -16,6 +16,10 @@ import flixopt as fx from flixopt.structure import FlowSystemModel +# ============================================================================ +# SOLVER FIXTURES +# ============================================================================ + @pytest.fixture() def highs_solver(): @@ -32,20 +36,320 @@ def solver_fixture(request): return request.getfixturevalue(request.param.__name__) -# Custom assertion function -def assert_almost_equal_numeric( - actual, desired, err_msg, relative_error_range_in_percent=0.011, absolute_tolerance=1e-7 -): - """ - Custom assertion function for comparing numeric values with relative and absolute tolerances - """ - relative_tol = relative_error_range_in_percent / 100 +# ============================================================================ +# HIERARCHICAL ELEMENT LIBRARY +# ============================================================================ - if isinstance(desired, (int, float)): - delta = abs(relative_tol * desired) if desired != 0 else absolute_tolerance - assert np.isclose(actual, desired, atol=delta), err_msg - else: - np.testing.assert_allclose(actual, desired, rtol=relative_tol, atol=absolute_tolerance, err_msg=err_msg) + +class Buses: + """Standard buses used across flow systems""" + + @staticmethod + def electricity(): + return fx.Bus('Strom') + + @staticmethod + def heat(): + return fx.Bus('Fernwärme') + + @staticmethod + def gas(): + return fx.Bus('Gas') + + @staticmethod + def coal(): + return fx.Bus('Kohle') + + @staticmethod + def defaults(): + """Get all standard buses at once""" + return [Buses.electricity(), Buses.heat(), Buses.gas()] + + +class Effects: + """Standard effects used across flow systems""" + + @staticmethod + def costs(): + return fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True) + + @staticmethod + def co2(): + return fx.Effect('CO2', 'kg', 'CO2_e-Emissionen') + + @staticmethod + def co2_with_costs_share(): + return fx.Effect( + 'CO2', + 'kg', + 'CO2_e-Emissionen', + specific_share_to_other_effects_operation={'costs': 0.2}, + ) + + @staticmethod + def primary_energy(): + return fx.Effect('PE', 'kWh_PE', 'Primärenergie') + + +class Converters: + """Energy conversion components""" + + class Boilers: + @staticmethod + def simple(): + """Simple boiler from simple_flow_system""" + return fx.linear_converters.Boiler( + 'Boiler', + eta=0.5, + Q_th=fx.Flow( + 'Q_th', + bus='Fernwärme', + size=50, + relative_minimum=5 / 50, + relative_maximum=1, + on_off_parameters=fx.OnOffParameters(), + ), + Q_fu=fx.Flow('Q_fu', bus='Gas'), + ) + + @staticmethod + def complex(): + """Complex boiler with investment parameters from flow_system_complex""" + return fx.linear_converters.Boiler( + 'Kessel', + eta=0.5, + on_off_parameters=fx.OnOffParameters(effects_per_running_hour={'costs': 0, 'CO2': 1000}), + Q_th=fx.Flow( + 'Q_th', + bus='Fernwärme', + load_factor_max=1.0, + load_factor_min=0.1, + relative_minimum=5 / 50, + relative_maximum=1, + previous_flow_rate=50, + size=fx.InvestParameters( + fix_effects=1000, fixed_size=50, optional=False, specific_effects={'costs': 10, 'PE': 2} + ), + on_off_parameters=fx.OnOffParameters( + on_hours_total_min=0, + on_hours_total_max=1000, + consecutive_on_hours_max=10, + consecutive_on_hours_min=1, + consecutive_off_hours_max=10, + effects_per_switch_on=0.01, + switch_on_total_max=1000, + ), + flow_hours_total_max=1e6, + ), + Q_fu=fx.Flow('Q_fu', bus='Gas', size=200, relative_minimum=0, relative_maximum=1), + ) + + class CHPs: + @staticmethod + def simple(): + """Simple CHP from simple_flow_system""" + return fx.linear_converters.CHP( + 'CHP_unit', + eta_th=0.5, + eta_el=0.4, + P_el=fx.Flow( + 'P_el', bus='Strom', size=60, relative_minimum=5 / 60, on_off_parameters=fx.OnOffParameters() + ), + Q_th=fx.Flow('Q_th', bus='Fernwärme'), + Q_fu=fx.Flow('Q_fu', bus='Gas'), + ) + + @staticmethod + def base(): + """CHP from flow_system_base""" + return fx.linear_converters.CHP( + 'KWK', + eta_th=0.5, + eta_el=0.4, + on_off_parameters=fx.OnOffParameters(effects_per_switch_on=0.01), + P_el=fx.Flow('P_el', bus='Strom', size=60, relative_minimum=5 / 60, previous_flow_rate=10), + Q_th=fx.Flow('Q_th', bus='Fernwärme', size=1e3), + Q_fu=fx.Flow('Q_fu', bus='Gas', size=1e3), + ) + + class LinearConverters: + @staticmethod + def piecewise(): + """Piecewise converter from flow_system_piecewise_conversion""" + return fx.LinearConverter( + 'KWK', + inputs=[fx.Flow('Q_fu', bus='Gas')], + outputs=[ + fx.Flow('P_el', bus='Strom', size=60, relative_maximum=55, previous_flow_rate=10), + fx.Flow('Q_th', bus='Fernwärme'), + ], + piecewise_conversion=fx.PiecewiseConversion( + { + 'P_el': fx.Piecewise([fx.Piece(5, 30), fx.Piece(40, 60)]), + 'Q_th': fx.Piecewise([fx.Piece(6, 35), fx.Piece(45, 100)]), + 'Q_fu': fx.Piecewise([fx.Piece(12, 70), fx.Piece(90, 200)]), + } + ), + on_off_parameters=fx.OnOffParameters(effects_per_switch_on=0.01), + ) + + @staticmethod + def segments(timesteps_length): + """Segments converter with time-varying piecewise conversion""" + return fx.LinearConverter( + 'KWK', + inputs=[fx.Flow('Q_fu', bus='Gas')], + outputs=[ + fx.Flow('P_el', bus='Strom', size=60, relative_maximum=55, previous_flow_rate=10), + fx.Flow('Q_th', bus='Fernwärme'), + ], + piecewise_conversion=fx.PiecewiseConversion( + { + 'P_el': fx.Piecewise( + [ + fx.Piece(np.linspace(5, 6, timesteps_length), 30), + fx.Piece(40, np.linspace(60, 70, timesteps_length)), + ] + ), + 'Q_th': fx.Piecewise([fx.Piece(6, 35), fx.Piece(45, 100)]), + 'Q_fu': fx.Piecewise([fx.Piece(12, 70), fx.Piece(90, 200)]), + } + ), + on_off_parameters=fx.OnOffParameters(effects_per_switch_on=0.01), + ) + + +class Storage: + """Energy storage components""" + + @staticmethod + def simple(): + """Simple storage from simple_flow_system""" + return fx.Storage( + 'Speicher', + charging=fx.Flow('Q_th_load', bus='Fernwärme', size=1e4), + discharging=fx.Flow('Q_th_unload', bus='Fernwärme', size=1e4), + capacity_in_flow_hours=fx.InvestParameters(fix_effects=20, fixed_size=30, optional=False), + initial_charge_state=0, + relative_maximum_charge_state=1 / 100 * np.array([80.0, 70.0, 80.0, 80, 80, 80, 80, 80, 80]), + relative_maximum_final_charge_state=0.8, + eta_charge=0.9, + eta_discharge=1, + relative_loss_per_hour=0.08, + prevent_simultaneous_charge_and_discharge=True, + ) + + @staticmethod + def complex(): + """Complex storage with piecewise investment from flow_system_complex""" + invest_speicher = fx.InvestParameters( + fix_effects=0, + piecewise_effects=fx.PiecewiseEffects( + piecewise_origin=fx.Piecewise([fx.Piece(5, 25), fx.Piece(25, 100)]), + piecewise_shares={ + 'costs': fx.Piecewise([fx.Piece(50, 250), fx.Piece(250, 800)]), + 'PE': fx.Piecewise([fx.Piece(5, 25), fx.Piece(25, 100)]), + }, + ), + optional=False, + specific_effects={'costs': 0.01, 'CO2': 0.01}, + minimum_size=0, + maximum_size=1000, + ) + return fx.Storage( + 'Speicher', + charging=fx.Flow('Q_th_load', bus='Fernwärme', size=1e4), + discharging=fx.Flow('Q_th_unload', bus='Fernwärme', size=1e4), + capacity_in_flow_hours=invest_speicher, + initial_charge_state=0, + maximal_final_charge_state=10, + eta_charge=0.9, + eta_discharge=1, + relative_loss_per_hour=0.08, + prevent_simultaneous_charge_and_discharge=True, + ) + + +class LoadProfiles: + """Standard load and price profiles""" + + @staticmethod + def thermal_simple(): + return np.array([30.0, 0.0, 90.0, 110, 110, 20, 20, 20, 20]) + + @staticmethod + def thermal_complex(): + return np.array([30, 0, 90, 110, 110, 20, 20, 20, 20]) + + @staticmethod + def electrical_simple(): + return 1 / 1000 * np.array([80.0, 80.0, 80.0, 80, 80, 80, 80, 80, 80]) + + @staticmethod + def electrical_scenario(): + return np.array([0.08, 0.1, 0.15]) + + @staticmethod + def electrical_complex(): + return np.array([40, 40, 40, 40, 40, 40, 40, 40, 40]) + + @staticmethod + def random_thermal(length=10, seed=42): + np.random.seed(seed) + return np.array([np.random.random() for _ in range(length)]) * 180 + + @staticmethod + def random_electrical(length=10, seed=42): + np.random.seed(seed) + return (np.array([np.random.random() for _ in range(length)]) + 0.5) / 1.5 * 50 + + +class Sinks: + """Energy sinks (loads)""" + + @staticmethod + def heat_load(thermal_profile): + """Create thermal heat load sink""" + return fx.Sink( + 'Wärmelast', sink=fx.Flow('Q_th_Last', bus='Fernwärme', size=1, fixed_relative_profile=thermal_profile) + ) + + @staticmethod + def electricity_feed_in(electrical_price_profile): + """Create electricity feed-in sink""" + return fx.Sink( + 'Einspeisung', sink=fx.Flow('P_el', bus='Strom', effects_per_flow_hour=-1 * electrical_price_profile) + ) + + @staticmethod + def electricity_load(electrical_profile): + """Create electrical load sink (for flow_system_long)""" + return fx.Sink( + 'Stromlast', sink=fx.Flow('P_el_Last', bus='Strom', size=1, fixed_relative_profile=electrical_profile) + ) + + +class Sources: + """Energy sources""" + + @staticmethod + def gas_with_costs_and_co2(): + """Standard gas tariff with CO2 emissions""" + source = Sources.gas_with_costs() + source.source.effects_per_flow_hour = {'costs': 0.04, 'CO2': 0.3} + return source + + @staticmethod + def gas_with_costs(): + """Simple gas tariff without CO2""" + return fx.Source( + 'Gastarif', source=fx.Flow(label='Q_Gas', bus='Gas', size=1000, effects_per_flow_hour={'costs': 0.04}) + ) + + +# ============================================================================ +# RECREATED FIXTURES USING HIERARCHICAL LIBRARY +# ============================================================================ @pytest.fixture @@ -53,72 +357,26 @@ def simple_flow_system() -> fx.FlowSystem: """ Create a simple energy system for testing """ - base_thermal_load = np.array([30.0, 0.0, 90.0, 110, 110, 20, 20, 20, 20]) - base_electrical_price = 1 / 1000 * np.array([80.0, 80.0, 80.0, 80, 80, 80, 80, 80, 80]) + base_thermal_load = LoadProfiles.thermal_simple() + base_electrical_price = LoadProfiles.electrical_simple() base_timesteps = pd.date_range('2020-01-01', periods=9, freq='h', name='time') + # Define effects - costs = fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True) - co2 = fx.Effect( - 'CO2', - 'kg', - 'CO2_e-Emissionen', - specific_share_to_other_effects_operation={costs.label: 0.2}, - maximum_operation_per_hour=1000, - ) + costs = Effects.costs() + co2 = Effects.co2_with_costs_share() + co2.maximum_operation_per_hour = 1000 # Create components - boiler = fx.linear_converters.Boiler( - 'Boiler', - eta=0.5, - Q_th=fx.Flow( - 'Q_th', - bus='Fernwärme', - size=50, - relative_minimum=5 / 50, - relative_maximum=1, - on_off_parameters=fx.OnOffParameters(), - ), - Q_fu=fx.Flow('Q_fu', bus='Gas'), - ) - - chp = fx.linear_converters.CHP( - 'CHP_unit', - eta_th=0.5, - eta_el=0.4, - P_el=fx.Flow('P_el', bus='Strom', size=60, relative_minimum=5 / 60, on_off_parameters=fx.OnOffParameters()), - Q_th=fx.Flow('Q_th', bus='Fernwärme'), - Q_fu=fx.Flow('Q_fu', bus='Gas'), - ) - - storage = fx.Storage( - 'Speicher', - charging=fx.Flow('Q_th_load', bus='Fernwärme', size=1e4), - discharging=fx.Flow('Q_th_unload', bus='Fernwärme', size=1e4), - capacity_in_flow_hours=fx.InvestParameters(fix_effects=20, fixed_size=30, optional=False), - initial_charge_state=0, - relative_maximum_charge_state=1 / 100 * np.array([80.0, 70.0, 80.0, 80, 80, 80, 80, 80, 80]), - relative_maximum_final_charge_state=0.8, - eta_charge=0.9, - eta_discharge=1, - relative_loss_per_hour=0.08, - prevent_simultaneous_charge_and_discharge=True, - ) - - heat_load = fx.Sink( - 'Wärmelast', sink=fx.Flow('Q_th_Last', bus='Fernwärme', size=1, fixed_relative_profile=base_thermal_load) - ) - - gas_tariff = fx.Source( - 'Gastarif', source=fx.Flow('Q_Gas', bus='Gas', size=1000, effects_per_flow_hour={'costs': 0.04, 'CO2': 0.3}) - ) - - electricity_feed_in = fx.Sink( - 'Einspeisung', sink=fx.Flow('P_el', bus='Strom', effects_per_flow_hour=-1 * base_electrical_price) - ) + boiler = Converters.Boilers.simple() + chp = Converters.CHPs.simple() + storage = Storage.simple() + heat_load = Sinks.heat_load(base_thermal_load) + gas_tariff = Sources.gas_with_costs_and_co2() + electricity_feed_in = Sinks.electricity_feed_in(base_electrical_price) # Create flow system flow_system = fx.FlowSystem(base_timesteps) - flow_system.add_elements(fx.Bus('Strom'), fx.Bus('Fernwärme'), fx.Bus('Gas')) + flow_system.add_elements(*Buses.defaults()) flow_system.add_elements(storage, costs, co2, boiler, heat_load, gas_tariff, electricity_feed_in, chp) return flow_system @@ -129,74 +387,28 @@ def simple_flow_system_scenarios() -> fx.FlowSystem: """ Create a simple energy system for testing """ - base_thermal_load = np.array([30.0, 0.0, 90.0, 110, 110, 20, 20, 20, 20]) - base_electrical_price = np.array([0.08, 0.1, 0.15]) + base_thermal_load = LoadProfiles.thermal_simple() + base_electrical_price = LoadProfiles.electrical_scenario() base_timesteps = pd.date_range('2020-01-01', periods=9, freq='h', name='time') + # Define effects - costs = fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True) - co2 = fx.Effect( - 'CO2', - 'kg', - 'CO2_e-Emissionen', - specific_share_to_other_effects_operation={costs.label: 0.2}, - maximum_operation_per_hour=1000, - ) + costs = Effects.costs() + co2 = Effects.co2_with_costs_share() + co2.maximum_operation_per_hour = 1000 # Create components - boiler = fx.linear_converters.Boiler( - 'Boiler', - eta=0.5, - Q_th=fx.Flow( - 'Q_th', - bus='Fernwärme', - size=50, - relative_minimum=5 / 50, - relative_maximum=1, - on_off_parameters=fx.OnOffParameters(), - ), - Q_fu=fx.Flow('Q_fu', bus='Gas'), - ) - - chp = fx.linear_converters.CHP( - 'CHP_unit', - eta_th=0.5, - eta_el=0.4, - P_el=fx.Flow('P_el', bus='Strom', size=60, relative_minimum=5 / 60, on_off_parameters=fx.OnOffParameters()), - Q_th=fx.Flow('Q_th', bus='Fernwärme'), - Q_fu=fx.Flow('Q_fu', bus='Gas'), - ) - - storage = fx.Storage( - 'Speicher', - charging=fx.Flow('Q_th_load', bus='Fernwärme', size=1e4), - discharging=fx.Flow('Q_th_unload', bus='Fernwärme', size=1e4), - capacity_in_flow_hours=fx.InvestParameters(fix_effects=20, fixed_size=30, optional=False), - initial_charge_state=0, - relative_maximum_charge_state=1 / 100 * np.array([80.0, 70.0, 80.0, 80, 80, 80, 80, 80, 80]), - relative_maximum_final_charge_state=0.8, - eta_charge=0.9, - eta_discharge=1, - relative_loss_per_hour=0.08, - prevent_simultaneous_charge_and_discharge=True, - ) - - heat_load = fx.Sink( - 'Wärmelast', sink=fx.Flow('Q_th_Last', bus='Fernwärme', size=1, fixed_relative_profile=base_thermal_load) - ) - - gas_tariff = fx.Source( - 'Gastarif', source=fx.Flow('Q_Gas', bus='Gas', size=1000, effects_per_flow_hour={'costs': 0.04, 'CO2': 0.3}) - ) - - electricity_feed_in = fx.Sink( - 'Einspeisung', sink=fx.Flow('P_el', bus='Strom', effects_per_flow_hour=-1 * base_electrical_price) - ) + boiler = Converters.Boilers.simple() + chp = Converters.CHPs.simple() + storage = Storage.simple() + heat_load = Sinks.heat_load(base_thermal_load) + gas_tariff = Sources.gas_with_costs_and_co2() + electricity_feed_in = Sinks.electricity_feed_in(base_electrical_price) # Create flow system flow_system = fx.FlowSystem( base_timesteps, scenarios=pd.Index(['A', 'B', 'C']), weights=np.array([0.5, 0.25, 0.25]) ) - flow_system.add_elements(fx.Bus('Strom'), fx.Bus('Fernwärme'), fx.Bus('Gas')) + flow_system.add_elements(*Buses.defaults()) flow_system.add_elements(storage, costs, co2, boiler, heat_load, gas_tariff, electricity_feed_in, chp) return flow_system @@ -206,18 +418,17 @@ def simple_flow_system_scenarios() -> fx.FlowSystem: def basic_flow_system() -> fx.FlowSystem: """Create basic elements for component testing""" flow_system = fx.FlowSystem(pd.date_range('2020-01-01', periods=10, freq='h', name='time')) - thermal_load = np.array([np.random.random() for _ in range(10)]) * 180 - p_el = (np.array([np.random.random() for _ in range(10)]) + 0.5) / 1.5 * 50 - flow_system.add_elements( - fx.Bus('Strom'), - fx.Bus('Fernwärme'), - fx.Bus('Gas'), - fx.Effect('Costs', '€', 'Kosten', is_standard=True, is_objective=True), - fx.Sink('Wärmelast', sink=fx.Flow('Q_th_Last', 'Fernwärme', size=1, fixed_relative_profile=thermal_load)), - fx.Source('Gastarif', source=fx.Flow('Q_Gas', 'Gas', size=1000, effects_per_flow_hour=0.04)), - fx.Sink('Einspeisung', sink=fx.Flow('P_el', 'Strom', effects_per_flow_hour=-1 * p_el)), - ) + thermal_load = LoadProfiles.random_thermal(10) + p_el = LoadProfiles.random_electrical(10) + + costs = Effects.costs() + heat_load = Sinks.heat_load(thermal_load) + gas_source = Sources.gas_with_costs() + electricity_sink = Sinks.electricity_feed_in(p_el) + + flow_system.add_elements(*Buses.defaults()) + flow_system.add_elements(costs, heat_load, gas_source, electricity_sink) return flow_system @@ -227,79 +438,26 @@ def flow_system_complex() -> fx.FlowSystem: """ Helper method to create a base model with configurable parameters """ - thermal_load = np.array([30, 0, 90, 110, 110, 20, 20, 20, 20]) - electrical_load = np.array([40, 40, 40, 40, 40, 40, 40, 40, 40]) + thermal_load = LoadProfiles.thermal_complex() + electrical_load = LoadProfiles.electrical_complex() flow_system = fx.FlowSystem(pd.date_range('2020-01-01', periods=9, freq='h', name='time')) + # Define the components and flow_system - flow_system.add_elements( - fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True), - fx.Effect('CO2', 'kg', 'CO2_e-Emissionen', specific_share_to_other_effects_operation={'costs': 0.2}), - fx.Effect('PE', 'kWh_PE', 'Primärenergie', maximum_total=3.5e3), - fx.Bus('Strom'), - fx.Bus('Fernwärme'), - fx.Bus('Gas'), - fx.Sink('Wärmelast', sink=fx.Flow('Q_th_Last', 'Fernwärme', size=1, fixed_relative_profile=thermal_load)), - fx.Source( - 'Gastarif', source=fx.Flow('Q_Gas', 'Gas', size=1000, effects_per_flow_hour={'costs': 0.04, 'CO2': 0.3}) - ), - fx.Sink('Einspeisung', sink=fx.Flow('P_el', 'Strom', effects_per_flow_hour=-1 * electrical_load)), - ) + costs = Effects.costs() + co2 = Effects.co2() + co2.specific_share_to_other_effects_operation = {'costs': 0.2} + pe = Effects.primary_energy() + pe.maximum_total = 3.5e3 - boiler = fx.linear_converters.Boiler( - 'Kessel', - eta=0.5, - on_off_parameters=fx.OnOffParameters(effects_per_running_hour={'costs': 0, 'CO2': 1000}), - Q_th=fx.Flow( - 'Q_th', - bus='Fernwärme', - load_factor_max=1.0, - load_factor_min=0.1, - relative_minimum=5 / 50, - relative_maximum=1, - previous_flow_rate=50, - size=fx.InvestParameters( - fix_effects=1000, fixed_size=50, optional=False, specific_effects={'costs': 10, 'PE': 2} - ), - on_off_parameters=fx.OnOffParameters( - on_hours_total_min=0, - on_hours_total_max=1000, - consecutive_on_hours_max=10, - consecutive_on_hours_min=1, - consecutive_off_hours_max=10, - effects_per_switch_on=0.01, - switch_on_total_max=1000, - ), - flow_hours_total_max=1e6, - ), - Q_fu=fx.Flow('Q_fu', bus='Gas', size=200, relative_minimum=0, relative_maximum=1), - ) + heat_load = Sinks.heat_load(thermal_load) + gas_tariff = Sources.gas_with_costs_and_co2() + electricity_feed_in = Sinks.electricity_feed_in(electrical_load) - invest_speicher = fx.InvestParameters( - fix_effects=0, - piecewise_effects=fx.PiecewiseEffects( - piecewise_origin=fx.Piecewise([fx.Piece(5, 25), fx.Piece(25, 100)]), - piecewise_shares={ - 'costs': fx.Piecewise([fx.Piece(50, 250), fx.Piece(250, 800)]), - 'PE': fx.Piecewise([fx.Piece(5, 25), fx.Piece(25, 100)]), - }, - ), - optional=False, - specific_effects={'costs': 0.01, 'CO2': 0.01}, - minimum_size=0, - maximum_size=1000, - ) - speicher = fx.Storage( - 'Speicher', - charging=fx.Flow('Q_th_load', bus='Fernwärme', size=1e4), - discharging=fx.Flow('Q_th_unload', bus='Fernwärme', size=1e4), - capacity_in_flow_hours=invest_speicher, - initial_charge_state=0, - maximal_final_charge_state=10, - eta_charge=0.9, - eta_discharge=1, - relative_loss_per_hour=0.08, - prevent_simultaneous_charge_and_discharge=True, - ) + flow_system.add_elements(*Buses.defaults()) + flow_system.add_elements(costs, co2, pe, heat_load, gas_tariff, electricity_feed_in) + + boiler = Converters.Boilers.complex() + speicher = Storage.complex() flow_system.add_elements(boiler, speicher) @@ -312,45 +470,16 @@ def flow_system_base(flow_system_complex) -> fx.FlowSystem: Helper method to create a base model with configurable parameters """ flow_system = flow_system_complex - - flow_system.add_elements( - fx.linear_converters.CHP( - 'KWK', - eta_th=0.5, - eta_el=0.4, - on_off_parameters=fx.OnOffParameters(effects_per_switch_on=0.01), - P_el=fx.Flow('P_el', bus='Strom', size=60, relative_minimum=5 / 60, previous_flow_rate=10), - Q_th=fx.Flow('Q_th', bus='Fernwärme', size=1e3), - Q_fu=fx.Flow('Q_fu', bus='Gas', size=1e3), - ) - ) - + chp = Converters.CHPs.base() + flow_system.add_elements(chp) return flow_system @pytest.fixture def flow_system_piecewise_conversion(flow_system_complex) -> fx.FlowSystem: flow_system = flow_system_complex - - flow_system.add_elements( - fx.LinearConverter( - 'KWK', - inputs=[fx.Flow('Q_fu', bus='Gas')], - outputs=[ - fx.Flow('P_el', bus='Strom', size=60, relative_maximum=55, previous_flow_rate=10), - fx.Flow('Q_th', bus='Fernwärme'), - ], - piecewise_conversion=fx.PiecewiseConversion( - { - 'P_el': fx.Piecewise([fx.Piece(5, 30), fx.Piece(40, 60)]), - 'Q_th': fx.Piecewise([fx.Piece(6, 35), fx.Piece(45, 100)]), - 'Q_fu': fx.Piecewise([fx.Piece(12, 70), fx.Piece(90, 200)]), - } - ), - on_off_parameters=fx.OnOffParameters(effects_per_switch_on=0.01), - ) - ) - + converter = Converters.LinearConverters.piecewise() + flow_system.add_elements(converter) return flow_system @@ -360,38 +489,16 @@ def flow_system_segments_of_flows_2(flow_system_complex) -> fx.FlowSystem: Use segments/Piecewise with numeric data """ flow_system = flow_system_complex - - flow_system.add_elements( - fx.LinearConverter( - 'KWK', - inputs=[fx.Flow('Q_fu', bus='Gas')], - outputs=[ - fx.Flow('P_el', bus='Strom', size=60, relative_maximum=55, previous_flow_rate=10), - fx.Flow('Q_th', bus='Fernwärme'), - ], - piecewise_conversion=fx.PiecewiseConversion( - { - 'P_el': fx.Piecewise( - [ - fx.Piece(np.linspace(5, 6, len(flow_system.timesteps)), 30), - fx.Piece(40, np.linspace(60, 70, len(flow_system.timesteps))), - ] - ), - 'Q_th': fx.Piecewise([fx.Piece(6, 35), fx.Piece(45, 100)]), - 'Q_fu': fx.Piecewise([fx.Piece(12, 70), fx.Piece(90, 200)]), - } - ), - on_off_parameters=fx.OnOffParameters(effects_per_switch_on=0.01), - ) - ) - + converter = Converters.LinearConverters.segments(len(flow_system.timesteps)) + flow_system.add_elements(converter) return flow_system @pytest.fixture def flow_system_long(): """ - Fixture to create and return the flow system with loaded data + Special fixture with CSV data loading - kept separate for backward compatibility + Uses library components where possible, but has special elements inline """ # Load data filename = os.path.join(os.path.dirname(__file__), 'ressources', 'Zeitreihen2020.csv') @@ -404,25 +511,22 @@ def flow_system_long(): p_el = data['Strompr.€/MWh'].values gas_price = data['Gaspr.€/MWh'].values - flow_system = fx.FlowSystem(pd.DatetimeIndex(data.index)) - thermal_load_ts, electrical_load_ts = ( - fx.TimeSeriesData(thermal_load, coords={'time': flow_system.timesteps}), - fx.TimeSeriesData(electrical_load, aggregation_weight=0.7, coords={'time': flow_system.timesteps}), + fx.TimeSeriesData(thermal_load), + fx.TimeSeriesData(electrical_load, aggregation_weight=0.7), ) p_feed_in, p_sell = ( - fx.TimeSeriesData(-(p_el - 0.5), aggregation_group='p_el', coords={'time': flow_system.timesteps}), - fx.TimeSeriesData(p_el + 0.5, aggregation_group='p_el', coords={'time': flow_system.timesteps}), + fx.TimeSeriesData(-(p_el - 0.5), aggregation_group='p_el'), + fx.TimeSeriesData(p_el + 0.5, aggregation_group='p_el'), ) + flow_system = fx.FlowSystem(pd.DatetimeIndex(data.index)) flow_system.add_elements( - fx.Bus('Strom'), - fx.Bus('Fernwärme'), - fx.Bus('Gas'), - fx.Bus('Kohle'), - fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True), - fx.Effect('CO2', 'kg', 'CO2_e-Emissionen'), - fx.Effect('PE', 'kWh_PE', 'Primärenergie'), + *Buses.defaults(), + Buses.coal(), + Effects.costs(), + Effects.co2(), + Effects.primary_energy(), fx.Sink( 'Wärmelast', sink=fx.Flow('Q_th_Last', bus='Fernwärme', size=1, fixed_relative_profile=thermal_load_ts) ), @@ -487,6 +591,51 @@ def flow_system_long(): } +@pytest.fixture(params=['h', '3h']) +def timesteps_linopy(request): + return pd.date_range('2020-01-01', periods=10, freq=request.param, name='time') + + +@pytest.fixture +def basic_flow_system_linopy(timesteps_linopy) -> fx.FlowSystem: + """Create basic elements for component testing""" + flow_system = fx.FlowSystem(timesteps_linopy) + + thermal_load = LoadProfiles.random_thermal(10) + p_el = LoadProfiles.random_electrical(10) + + costs = Effects.costs() + heat_load = Sinks.heat_load(thermal_load) + gas_source = Sources.gas_with_costs() + electricity_sink = Sinks.electricity_feed_in(p_el) + + flow_system.add_elements(*Buses.defaults()) + flow_system.add_elements(costs, heat_load, gas_source, electricity_sink) + + return flow_system + + +# ============================================================================ +# UTILITY FUNCTIONS (kept for backward compatibility) +# ============================================================================ + + +# Custom assertion function +def assert_almost_equal_numeric( + actual, desired, err_msg, relative_error_range_in_percent=0.011, absolute_tolerance=1e-7 +): + """ + Custom assertion function for comparing numeric values with relative and absolute tolerances + """ + relative_tol = relative_error_range_in_percent / 100 + + if isinstance(desired, (int, float)): + delta = abs(relative_tol * desired) if desired != 0 else absolute_tolerance + assert np.isclose(actual, desired, atol=delta), err_msg + else: + np.testing.assert_allclose(actual, desired, rtol=relative_tol, atol=absolute_tolerance, err_msg=err_msg) + + def create_calculation_and_solve( flow_system: fx.FlowSystem, solver, name: str, allow_infeasible: bool = False ) -> fx.FullCalculation: @@ -508,31 +657,6 @@ def create_linopy_model(flow_system: fx.FlowSystem) -> FlowSystemModel: return calculation.model -@pytest.fixture(params=['h', '3h']) -def timesteps_linopy(request): - return pd.date_range('2020-01-01', periods=10, freq=request.param, name='time') - - -@pytest.fixture -def basic_flow_system_linopy(timesteps_linopy) -> fx.FlowSystem: - """Create basic elements for component testing""" - flow_system = fx.FlowSystem(timesteps_linopy) - thermal_load = np.array([np.random.random() for _ in range(10)]) * 180 - p_el = (np.array([np.random.random() for _ in range(10)]) + 0.5) / 1.5 * 50 - - flow_system.add_elements( - fx.Bus('Strom'), - fx.Bus('Fernwärme'), - fx.Bus('Gas'), - fx.Effect('Costs', '€', 'Kosten', is_standard=True, is_objective=True), - fx.Sink('Wärmelast', sink=fx.Flow('Q_th_Last', 'Fernwärme', size=1, fixed_relative_profile=thermal_load)), - fx.Source('Gastarif', source=fx.Flow('Q_Gas', 'Gas', size=1000, effects_per_flow_hour=0.04)), - fx.Sink('Einspeisung', sink=fx.Flow('P_el', 'Strom', effects_per_flow_hour=-1 * p_el)), - ) - - return flow_system - - def assert_conequal(actual: linopy.Constraint, desired: linopy.Constraint): """Assert that two constraints are equal with detailed error messages.""" name = actual.name diff --git a/tests/test_effect.py b/tests/test_effect.py index e2807cc89..13e878041 100644 --- a/tests/test_effect.py +++ b/tests/test_effect.py @@ -204,7 +204,7 @@ def test_shares(self, basic_flow_system_linopy): class TestEffectResults: def test_shares(self, basic_flow_system_linopy): flow_system = basic_flow_system_linopy - flow_system.effects['Costs'].specific_share_to_other_effects_operation['Effect1'] = 0.5 + flow_system.effects['costs'].specific_share_to_other_effects_operation['Effect1'] = 0.5 flow_system.add_elements( fx.Effect( 'Effect1', @@ -230,9 +230,9 @@ def test_shares(self, basic_flow_system_linopy): effect_share_factors = { 'operation': { - ('Costs', 'Effect1'): 0.5, - ('Costs', 'Effect2'): 0.5 * 1.1, - ('Costs', 'Effect3'): 0.5 * 1.1 * 5 + 0.5 * 1.2, # This is where the issue lies + ('costs', 'Effect1'): 0.5, + ('costs', 'Effect2'): 0.5 * 1.1, + ('costs', 'Effect3'): 0.5 * 1.1 * 5 + 0.5 * 1.2, # This is where the issue lies ('Effect1', 'Effect2'): 1.1, ('Effect1', 'Effect3'): 1.2 + 1.1 * 5, ('Effect2', 'Effect3'): 5, @@ -249,8 +249,8 @@ def test_shares(self, basic_flow_system_linopy): np.testing.assert_allclose(results.effect_share_factors['invest'][key].values, value) xr.testing.assert_allclose( - results.effects_per_component('operation').sum('component')['Costs'], - results.solution['Costs(operation)|total_per_timestep'].fillna(0), + results.effects_per_component('operation').sum('component')['costs'], + results.solution['costs(operation)|total_per_timestep'].fillna(0), ) xr.testing.assert_allclose( @@ -270,7 +270,7 @@ def test_shares(self, basic_flow_system_linopy): # Invest mode checks xr.testing.assert_allclose( - results.effects_per_component('invest').sum('component')['Costs'], results.solution['Costs(invest)|total'] + results.effects_per_component('invest').sum('component')['costs'], results.solution['costs(invest)|total'] ) xr.testing.assert_allclose( @@ -290,7 +290,7 @@ def test_shares(self, basic_flow_system_linopy): # Total mode checks xr.testing.assert_allclose( - results.effects_per_component('total').sum('component')['Costs'], results.solution['Costs|total'] + results.effects_per_component('total').sum('component')['costs'], results.solution['costs|total'] ) xr.testing.assert_allclose( diff --git a/tests/test_effects_shares_summation.py b/tests/test_effects_shares_summation.py index d4d22d6df..15de93481 100644 --- a/tests/test_effects_shares_summation.py +++ b/tests/test_effects_shares_summation.py @@ -99,7 +99,7 @@ def test_effect_shares_example(): """Test the specific example from the effects share factors test.""" # Create the conversion dictionary based on test example conversion_dict = { - 'Costs': {'Effect1': xr.DataArray(0.5)}, + 'costs': {'Effect1': xr.DataArray(0.5)}, 'Effect1': {'Effect2': xr.DataArray(1.1), 'Effect3': xr.DataArray(1.2)}, 'Effect2': {'Effect3': xr.DataArray(5.0)}, } @@ -107,19 +107,19 @@ def test_effect_shares_example(): result = calculate_all_conversion_paths(conversion_dict) # Test direct paths - assert result[('Costs', 'Effect1')].item() == 0.5 + assert result[('costs', 'Effect1')].item() == 0.5 assert result[('Effect1', 'Effect2')].item() == 1.1 assert result[('Effect2', 'Effect3')].item() == 5.0 # Test indirect paths - # Costs -> Effect2 = Costs -> Effect1 -> Effect2 = 0.5 * 1.1 - assert result[('Costs', 'Effect2')].item() == 0.5 * 1.1 + # costs -> Effect2 = costs -> Effect1 -> Effect2 = 0.5 * 1.1 + assert result[('costs', 'Effect2')].item() == 0.5 * 1.1 - # Costs -> Effect3 has two paths: - # 1. Costs -> Effect1 -> Effect3 = 0.5 * 1.2 = 0.6 - # 2. Costs -> Effect1 -> Effect2 -> Effect3 = 0.5 * 1.1 * 5 = 2.75 + # costs -> Effect3 has two paths: + # 1. costs -> Effect1 -> Effect3 = 0.5 * 1.2 = 0.6 + # 2. costs -> Effect1 -> Effect2 -> Effect3 = 0.5 * 1.1 * 5 = 2.75 # Total = 0.6 + 2.75 = 3.35 - assert result[('Costs', 'Effect3')].item() == 0.5 * 1.2 + 0.5 * 1.1 * 5 + assert result[('costs', 'Effect3')].item() == 0.5 * 1.2 + 0.5 * 1.1 * 5 # Effect1 -> Effect3 has two paths: # 1. Effect1 -> Effect2 -> Effect3 = 1.1 * 5.0 = 5.5 diff --git a/tests/test_flow.py b/tests/test_flow.py index 93658bf2e..2ee609f68 100644 --- a/tests/test_flow.py +++ b/tests/test_flow.py @@ -101,11 +101,11 @@ def test_effects_per_flow_hour(self, basic_flow_system_linopy): co2_per_flow_hour = xr.DataArray(np.linspace(4, 5, timesteps.size), coords=(timesteps,)) flow = fx.Flow( - 'Wärme', bus='Fernwärme', effects_per_flow_hour={'Costs': costs_per_flow_hour, 'CO2': co2_per_flow_hour} + 'Wärme', bus='Fernwärme', effects_per_flow_hour={'costs': costs_per_flow_hour, 'CO2': co2_per_flow_hour} ) flow_system.add_elements(fx.Sink('Sink', sink=flow), fx.Effect('CO2', 't', '')) model = create_linopy_model(flow_system) - costs, co2 = flow_system.effects['Costs'], flow_system.effects['CO2'] + costs, co2 = flow_system.effects['costs'], flow_system.effects['CO2'] assert_sets_equal( set(flow.submodel.variables), @@ -114,12 +114,12 @@ def test_effects_per_flow_hour(self, basic_flow_system_linopy): ) assert_sets_equal(set(flow.submodel.constraints), {'Sink(Wärme)|total_flow_hours'}, msg='Incorrect constraints') - assert 'Sink(Wärme)->Costs(operation)' in set(costs.submodel.constraints) + assert 'Sink(Wärme)->costs(operation)' in set(costs.submodel.constraints) assert 'Sink(Wärme)->CO2(operation)' in set(co2.submodel.constraints) assert_conequal( - model.constraints['Sink(Wärme)->Costs(operation)'], - model.variables['Sink(Wärme)->Costs(operation)'] + model.constraints['Sink(Wärme)->costs(operation)'], + model.variables['Sink(Wärme)->costs(operation)'] == flow.submodel.variables['Sink(Wärme)|flow_rate'] * model.hours_per_step * costs_per_flow_hour, ) @@ -426,8 +426,8 @@ def test_flow_invest_with_effects(self, basic_flow_system_linopy): minimum_size=20, maximum_size=100, optional=True, - fix_effects={'Costs': 1000, 'CO2': 5}, # Fixed investment effects - specific_effects={'Costs': 500, 'CO2': 0.1}, # Specific investment effects + fix_effects={'costs': 1000, 'CO2': 5}, # Fixed investment effects + specific_effects={'costs': 500, 'CO2': 0.1}, # Specific investment effects ), ) @@ -435,13 +435,13 @@ def test_flow_invest_with_effects(self, basic_flow_system_linopy): model = create_linopy_model(flow_system) # Check investment effects - assert 'Sink(Wärme)->Costs(invest)' in model.variables + assert 'Sink(Wärme)->costs(invest)' in model.variables assert 'Sink(Wärme)->CO2(invest)' in model.variables # Check fix effects (applied only when is_invested=1) assert_conequal( - model.constraints['Sink(Wärme)->Costs(invest)'], - model.variables['Sink(Wärme)->Costs(invest)'] + model.constraints['Sink(Wärme)->costs(invest)'], + model.variables['Sink(Wärme)->costs(invest)'] == flow.submodel.variables['Sink(Wärme)|is_invested'] * 1000 + flow.submodel.variables['Sink(Wärme)|size'] * 500, ) @@ -464,7 +464,7 @@ def test_flow_invest_divest_effects(self, basic_flow_system_linopy): minimum_size=20, maximum_size=100, optional=True, - divest_effects={'Costs': 500}, # Cost incurred when NOT investing + divest_effects={'costs': 500}, # Cost incurred when NOT investing ), ) @@ -472,11 +472,11 @@ def test_flow_invest_divest_effects(self, basic_flow_system_linopy): model = create_linopy_model(flow_system) # Check divestment effects - assert 'Sink(Wärme)->Costs(invest)' in model.constraints + assert 'Sink(Wärme)->costs(invest)' in model.constraints assert_conequal( - model.constraints['Sink(Wärme)->Costs(invest)'], - model.variables['Sink(Wärme)->Costs(invest)'] + (model.variables['Sink(Wärme)|is_invested'] - 1) * 500 == 0, + model.constraints['Sink(Wärme)->costs(invest)'], + model.variables['Sink(Wärme)->costs(invest)'] + (model.variables['Sink(Wärme)|is_invested'] - 1) * 500 == 0, ) @@ -558,12 +558,12 @@ def test_effects_per_running_hour(self, basic_flow_system_linopy): 'Wärme', bus='Fernwärme', on_off_parameters=fx.OnOffParameters( - effects_per_running_hour={'Costs': costs_per_running_hour, 'CO2': co2_per_running_hour} + effects_per_running_hour={'costs': costs_per_running_hour, 'CO2': co2_per_running_hour} ), ) flow_system.add_elements(fx.Sink('Sink', sink=flow), fx.Effect('CO2', 't', '')) model = create_linopy_model(flow_system) - costs, co2 = flow_system.effects['Costs'], flow_system.effects['CO2'] + costs, co2 = flow_system.effects['costs'], flow_system.effects['CO2'] assert_sets_equal( set(flow.submodel.variables), @@ -586,12 +586,12 @@ def test_effects_per_running_hour(self, basic_flow_system_linopy): msg='Incorrect constraints', ) - assert 'Sink(Wärme)->Costs(operation)' in set(costs.submodel.constraints) + assert 'Sink(Wärme)->costs(operation)' in set(costs.submodel.constraints) assert 'Sink(Wärme)->CO2(operation)' in set(co2.submodel.constraints) assert_conequal( - model.constraints['Sink(Wärme)->Costs(operation)'], - model.variables['Sink(Wärme)->Costs(operation)'] + model.constraints['Sink(Wärme)->costs(operation)'], + model.variables['Sink(Wärme)->costs(operation)'] == flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step * costs_per_running_hour, ) @@ -943,7 +943,7 @@ def test_switch_on_constraints(self, basic_flow_system_linopy): size=100, on_off_parameters=fx.OnOffParameters( switch_on_total_max=5, # Maximum 5 startups - effects_per_switch_on={'Costs': 100}, # 100 EUR startup cost + effects_per_switch_on={'costs': 100}, # 100 EUR startup cost ), ) @@ -984,12 +984,12 @@ def test_switch_on_constraints(self, basic_flow_system_linopy): ) # Check that startup cost effect constraint exists - assert 'Sink(Wärme)->Costs(operation)' in model.constraints + assert 'Sink(Wärme)->costs(operation)' in model.constraints # Verify the startup cost effect constraint assert_conequal( - model.constraints['Sink(Wärme)->Costs(operation)'], - model.variables['Sink(Wärme)->Costs(operation)'] == flow.submodel.variables['Sink(Wärme)|switch|on'] * 100, + model.constraints['Sink(Wärme)->costs(operation)'], + model.variables['Sink(Wärme)->costs(operation)'] == flow.submodel.variables['Sink(Wärme)|switch|on'] * 100, ) def test_on_hours_limits(self, basic_flow_system_linopy): diff --git a/tests/test_integration.py b/tests/test_integration.py index babc7b131..97876c251 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -248,7 +248,7 @@ def test_piecewise_conversion(self, flow_system_piecewise_conversion, highs_solv assert_almost_equal_numeric( comps['Speicher'].submodel.variables['Speicher|PiecewiseEffects|costs'].solution.values, 454.74666666666667, - 'Speicher investCosts_segmented_costs doesnt match expected value', + 'Speicher investcosts_segmented_costs doesnt match expected value', ) @@ -309,13 +309,13 @@ def test_modeling_types_costs(self, modeling_calculation): assert_almost_equal_numeric( calc.results.model['costs|total'].solution.item(), expected_costs[modeling_type], - f'Costs do not match for {modeling_type} modeling type', + f'costs do not match for {modeling_type} modeling type', ) else: assert_almost_equal_numeric( calc.results.solution_without_overlap('costs(operation)|total_per_timestep').sum(), expected_costs[modeling_type], - f'Costs do not match for {modeling_type} modeling type', + f'costs do not match for {modeling_type} modeling type', ) def test_segmented_io(self, modeling_calculation): diff --git a/tests/test_linear_converter.py b/tests/test_linear_converter.py index 42dc80077..322a5f6f0 100644 --- a/tests/test_linear_converter.py +++ b/tests/test_linear_converter.py @@ -146,7 +146,7 @@ def test_linear_converter_with_on_off(self, basic_flow_system_linopy): # Create OnOffParameters on_off_params = fx.OnOffParameters( - on_hours_total_min=10, on_hours_total_max=40, effects_per_running_hour={'Costs': 5} + on_hours_total_min=10, on_hours_total_max=40, effects_per_running_hour={'costs': 5} ) # Create a linear converter with OnOffParameters @@ -186,10 +186,10 @@ def test_linear_converter_with_on_off(self, basic_flow_system_linopy): ) # Check on_off effects - assert 'Converter->Costs(operation)' in model.constraints + assert 'Converter->costs(operation)' in model.constraints assert_conequal( - model.constraints['Converter->Costs(operation)'], - model.variables['Converter->Costs(operation)'] + model.constraints['Converter->costs(operation)'], + model.variables['Converter->costs(operation)'] == model.variables['Converter|on'] * model.hours_per_step * 5, ) @@ -398,7 +398,7 @@ def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy): # Create OnOffParameters on_off_params = fx.OnOffParameters( - on_hours_total_min=10, on_hours_total_max=40, effects_per_running_hour={'Costs': 5} + on_hours_total_min=10, on_hours_total_max=40, effects_per_running_hour={'costs': 5} ) # Create a linear converter with piecewise conversion and on/off parameters @@ -489,10 +489,10 @@ def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy): ) # Verify that the costs effect is applied - assert 'Converter->Costs(operation)' in model.constraints + assert 'Converter->costs(operation)' in model.constraints assert_conequal( - model.constraints['Converter->Costs(operation)'], - model.variables['Converter->Costs(operation)'] + model.constraints['Converter->costs(operation)'], + model.variables['Converter->costs(operation)'] == model.variables['Converter|on'] * model.hours_per_step * 5, ) diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 0ccc1a5dd..897122242 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -67,9 +67,9 @@ def test_system(): size=InvestParameters( minimum_size=0, maximum_size=20, - specific_effects={'Costs': 100}, # €/kW + specific_effects={'costs': 100}, # €/kW ), - effects_per_flow_hour={'Costs': 20}, # €/MWh + effects_per_flow_hour={'costs': 20}, # €/MWh ) generator = Source('Generator', source=power_gen) @@ -83,7 +83,7 @@ def test_system(): capacity_in_flow_hours=InvestParameters( minimum_size=0, maximum_size=50, - specific_effects={'Costs': 50}, # €/kWh + specific_effects={'costs': 50}, # €/kWh ), eta_charge=0.95, eta_discharge=0.95, @@ -91,7 +91,7 @@ def test_system(): ) # Create effects and objective - cost_effect = Effect(label='Costs', unit='€', description='Total costs', is_standard=True, is_objective=True) + cost_effect = Effect(label='costs', unit='€', description='Total costs', is_standard=True, is_objective=True) # Add all elements to the flow system flow_system.add_elements(electricity_bus, generator, demand_sink, storage, cost_effect) From b7734f8f2241b661d73d6b12cc1b42e106460788 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Wed, 23 Jul 2025 18:08:58 +0200 Subject: [PATCH 02/10] Improve some of the modeling and coord handling --- flixopt/features.py | 9 ++- flixopt/modeling.py | 131 ++++++++++++++++---------------------------- 2 files changed, 53 insertions(+), 87 deletions(-) diff --git a/flixopt/features.py b/flixopt/features.py index fc80f0eb3..d07844c8d 100644 --- a/flixopt/features.py +++ b/flixopt/features.py @@ -171,7 +171,7 @@ def _do_modeling(self): self.parameters.on_hours_total_max if self.parameters.on_hours_total_max is not None else np.inf, ), # TODO: self._model.hours_per_step.sum('time').item() + self._get_previous_on_duration()) short_name='on_hours_total', - coords=self.get_coords(['year', 'scenario']), + coords=['year', 'scenario'], ) # 4. Switch tracking using existing pattern @@ -179,13 +179,14 @@ def _do_modeling(self): self.add_variables(binary=True, short_name='switch|on', coords=self.get_coords()) self.add_variables(binary=True, short_name='switch|off', coords=self.get_coords()) - ModelingPrimitives.state_transition_variables( + BoundingPatterns.state_transition_bounds( self, state_variable=self.on, switch_on=self.switch_on, switch_off=self.switch_off, name=f'{self.label_of_model}|switch', previous_state=self._previous_states.isel(time=-1) if self._previous_states is not None else 0, + coord='time', ) if self.parameters.switch_on_total_max is not None: @@ -408,7 +409,9 @@ def _do_modeling(self): rhs = self.zero_point elif self._zero_point is True: self.zero_point = self.add_variables( - coords=self._model.get_coords(), binary=True, short_name='zero_point' + coords=self._model.get_coords(('year', 'scenario') if self._as_time_series else None), + binary=True, + short_name='zero_point', ) rhs = self.zero_point else: diff --git a/flixopt/modeling.py b/flixopt/modeling.py index fa13aeea8..14f8c45f3 100644 --- a/flixopt/modeling.py +++ b/flixopt/modeling.py @@ -189,7 +189,7 @@ def expression_tracking_variable( name: str = None, short_name: str = None, bounds: Tuple[TemporalData, TemporalData] = None, - coords: List[str] = None, + coords: Optional[Union[str, List[str]]] = None, ) -> Tuple[linopy.Variable, linopy.Constraint]: """ Creates variable that equals a given expression. @@ -205,8 +205,6 @@ def expression_tracking_variable( if not isinstance(model, Submodel): raise ValueError('ModelingPrimitives.expression_tracking_variable() can only be used with a Submodel') - coords = coords or ['year', 'scenario'] - if not bounds: tracker = model.add_variables(name=name, coords=model.get_coords(coords), short_name=short_name) else: @@ -223,86 +221,6 @@ def expression_tracking_variable( return tracker, tracking - @staticmethod - def state_transition_variables( - model: Submodel, - state_variable: linopy.Variable, - switch_on: linopy.Variable, - switch_off: linopy.Variable, - name: str, - previous_state=0, - ) -> Tuple[linopy.Constraint, linopy.Constraint, linopy.Constraint]: - """ - Creates switch-on/off variables with state transition logic. - - Mathematical formulation: - switch_on[t] - switch_off[t] = state[t] - state[t-1] ∀t > 0 - switch_on[0] - switch_off[0] = state[0] - previous_state - switch_on[t] + switch_off[t] ≤ 1 ∀t - switch_on[t], switch_off[t] ∈ {0, 1} - - Returns: - variables: {'switch_on': binary_var, 'switch_off': binary_var} - constraints: {'transition': constraint, 'initial': constraint, 'mutex': constraint} - """ - if not isinstance(model, Submodel): - raise ValueError('ModelingPrimitives.state_transition_variables() can only be used with a Submodel') - - # State transition constraints for t > 0 - transition = model.add_constraints( - switch_on.isel(time=slice(1, None)) - switch_off.isel(time=slice(1, None)) - == state_variable.isel(time=slice(1, None)) - state_variable.isel(time=slice(None, -1)), - name=f'{name}|transition', - ) - - # Initial state transition for t = 0 - initial = model.add_constraints( - switch_on.isel(time=0) - switch_off.isel(time=0) == state_variable.isel(time=0) - previous_state, - name=f'{name}|initial', - ) - - # At most one switch per timestep - mutex = model.add_constraints(switch_on + switch_off <= 1, name=f'{name}|mutex') - - return transition, initial, mutex - - @staticmethod - def sum_up_variable( - model: Submodel, - variable_to_count: linopy.Variable, - name: str = None, - bounds: Tuple[NonTemporalData, NonTemporalData] = None, - factor: TemporalData = 1, - ) -> Tuple[linopy.Variable, linopy.Constraint]: - """ - SUms up a variable over time, applying a factor to the variable. - - Args: - model: The optimization model instance - variable_to_count: The variable to be summed up - name: The name of the constraint - bounds: The bounds of the constraint - factor: The factor to be applied to the variable - """ - if not isinstance(model, Submodel): - raise ValueError('ModelingPrimitives.sum_up_variable() can only be used with a Submodel') - - if bounds is None: - bounds = (0, np.inf) - else: - bounds = (bounds[0] if bounds[0] is not None else 0, bounds[1] if bounds[1] is not None else np.inf) - - count = model.add_variables( - lower=bounds[0], - upper=bounds[1], - coords=model.get_coords(['year', 'scenario']), - name=name, - ) - - count_constraint = model.add_constraints(count == (variable_to_count * factor).sum('time'), name=name) - - return count, count_constraint - @staticmethod def consecutive_duration_tracking( model: Submodel, @@ -346,7 +264,7 @@ def consecutive_duration_tracking( duration = model.add_variables( lower=0, upper=maximum_duration if maximum_duration is not None else mega, - coords=model.get_coords(['time']), + coords=model.get_coords(), name=name, short_name=short_name, ) @@ -625,3 +543,48 @@ def scaled_bounds_with_state( binary_lower = model.add_constraints(variable_state * big_m_lower <= variable, name=f'{name}|lb1') return [scaling_lower, scaling_upper, binary_lower, binary_upper] + + @staticmethod + def state_transition_bounds( + model: Submodel, + state_variable: linopy.Variable, + switch_on: linopy.Variable, + switch_off: linopy.Variable, + name: str, + previous_state=0, + coord: str = 'time', + ) -> Tuple[linopy.Constraint, linopy.Constraint, linopy.Constraint]: + """ + Creates switch-on/off variables with state transition logic. + + Mathematical formulation: + switch_on[t] - switch_off[t] = state[t] - state[t-1] ∀t > 0 + switch_on[0] - switch_off[0] = state[0] - previous_state + switch_on[t] + switch_off[t] ≤ 1 ∀t + switch_on[t], switch_off[t] ∈ {0, 1} + + Returns: + variables: {'switch_on': binary_var, 'switch_off': binary_var} + constraints: {'transition': constraint, 'initial': constraint, 'mutex': constraint} + """ + if not isinstance(model, Submodel): + raise ValueError('ModelingPrimitives.state_transition_variables() can only be used with a Submodel') + + # State transition constraints for t > 0 + transition = model.add_constraints( + switch_on.isel({coord: slice(1, None)}) - switch_off.isel({coord: slice(1, None)}) + == state_variable.isel({coord: slice(1, None)}) - state_variable.isel({coord: slice(None, -1)}), + name=f'{name}|transition', + ) + + # Initial state transition for t = 0 + initial = model.add_constraints( + switch_on.isel({coord: 0}) - switch_off.isel({coord: 0}) + == state_variable.isel({coord: 0}) - previous_state, + name=f'{name}|initial', + ) + + # At most one switch per timestep + mutex = model.add_constraints(switch_on + switch_off <= 1, name=f'{name}|mutex') + + return transition, initial, mutex From e06692b473f1c2cfd298efa2e4e335a7780e1c07 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Wed, 23 Jul 2025 21:28:23 +0200 Subject: [PATCH 03/10] Add tests with years and scenarios --- tests/conftest.py | 45 +++++++++++++++++++++++++++++++++++++++++++++ tests/test_bus.py | 43 ++++++++++++++++++++++++++++++++++++------- 2 files changed, 81 insertions(+), 7 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 3d6623eb7..b581233fb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -36,6 +36,32 @@ def solver_fixture(request): return request.getfixturevalue(request.param.__name__) +# ================================= +# COORDINATE CONFIGURATION FIXTURES +# ================================= + + +@pytest.fixture( + params=[ + {'timesteps': pd.date_range('2020-01-01', periods=10, freq='h', name='time'), 'years': None, 'scenarios': None}, + { + 'timesteps': pd.date_range('2020-01-01', periods=10, freq='h', name='time'), + 'years': pd.Index([2020, 2030, 2040], name='year'), + 'scenarios': None, + }, + { + 'timesteps': pd.date_range('2020-01-01', periods=10, freq='h', name='time'), + 'years': pd.Index([2020, 2030, 2040], name='year'), + 'scenarios': pd.Index(['A', 'B'], name='scenario'), + }, + ], + ids=['time_only', 'time+years', 'time+years+scenarios'], +) +def coords_config(request): + """Coordinate configurations for parametrized testing.""" + return request.param + + # ============================================================================ # HIERARCHICAL ELEMENT LIBRARY # ============================================================================ @@ -615,6 +641,25 @@ def basic_flow_system_linopy(timesteps_linopy) -> fx.FlowSystem: return flow_system +@pytest.fixture +def basic_flow_system_linopy_coords(coords_config) -> fx.FlowSystem: + """Create basic elements for component testing with coordinate parametrization.""" + flow_system = fx.FlowSystem(**coords_config) + + thermal_load = LoadProfiles.random_thermal(10) + p_el = LoadProfiles.random_electrical(10) + + costs = Effects.costs() + heat_load = Sinks.heat_load(thermal_load) + gas_source = Sources.gas_with_costs() + electricity_sink = Sinks.electricity_feed_in(p_el) + + flow_system.add_elements(*Buses.defaults()) + flow_system.add_elements(costs, heat_load, gas_source, electricity_sink) + + return flow_system + + # ============================================================================ # UTILITY FUNCTIONS (kept for backward compatibility) # ============================================================================ diff --git a/tests/test_bus.py b/tests/test_bus.py index c9bf3956c..e4e0de6fd 100644 --- a/tests/test_bus.py +++ b/tests/test_bus.py @@ -11,9 +11,9 @@ class TestBusModel: """Test the FlowModel class.""" - def test_bus(self, basic_flow_system_linopy): + def test_bus(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system = basic_flow_system_linopy_coords bus = fx.Bus('TestBus', excess_penalty_per_flow_hour=None) flow_system.add_elements( bus, @@ -30,10 +30,9 @@ def test_bus(self, basic_flow_system_linopy): model.variables['GastarifTest(Q_Gas)|flow_rate'] == model.variables['WärmelastTest(Q_th_Last)|flow_rate'], ) - def test_bus_penalty(self, basic_flow_system_linopy): + def test_bus_penalty(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy - timesteps = flow_system.timesteps + flow_system = basic_flow_system_linopy_coords bus = fx.Bus('TestBus') flow_system.add_elements( bus, @@ -50,8 +49,12 @@ def test_bus_penalty(self, basic_flow_system_linopy): } assert set(bus.submodel.constraints) == {'TestBus|balance'} - assert_var_equal(model.variables['TestBus|excess_input'], model.add_variables(lower=0, coords=(timesteps,))) - assert_var_equal(model.variables['TestBus|excess_output'], model.add_variables(lower=0, coords=(timesteps,))) + assert_var_equal( + model.variables['TestBus|excess_input'], model.add_variables(lower=0, coords=model.get_coords()) + ) + assert_var_equal( + model.variables['TestBus|excess_output'], model.add_variables(lower=0, coords=model.get_coords()) + ) assert_conequal( model.constraints['TestBus|balance'], @@ -68,3 +71,29 @@ def test_bus_penalty(self, basic_flow_system_linopy): == (model.variables['TestBus|excess_input'] * 1e5 * model.hours_per_step).sum() + (model.variables['TestBus|excess_output'] * 1e5 * model.hours_per_step).sum(), ) + + def test_bus_with_coords(self, basic_flow_system_linopy_coords, coords_config): + """Test bus behavior across different coordinate configurations.""" + flow_system = basic_flow_system_linopy_coords + bus = fx.Bus('TestBus', excess_penalty_per_flow_hour=None) + flow_system.add_elements( + bus, + fx.Sink('WärmelastTest', sink=fx.Flow('Q_th_Last', 'TestBus')), + fx.Source('GastarifTest', source=fx.Flow('Q_Gas', 'TestBus')), + ) + model = create_linopy_model(flow_system) + + # Same core assertions as your existing test + assert set(bus.submodel.variables) == {'WärmelastTest(Q_th_Last)|flow_rate', 'GastarifTest(Q_Gas)|flow_rate'} + assert set(bus.submodel.constraints) == {'TestBus|balance'} + + assert_conequal( + model.constraints['TestBus|balance'], + model.variables['GastarifTest(Q_Gas)|flow_rate'] == model.variables['WärmelastTest(Q_th_Last)|flow_rate'], + ) + + # Just verify coordinate dimensions are correct + gas_var = model.variables['GastarifTest(Q_Gas)|flow_rate'] + if flow_system.scenarios is not None: + assert 'scenario' in gas_var.dims + assert 'time' in gas_var.dims From cf0186cec45d98de1629338bbadd55154f3df2e6 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Wed, 23 Jul 2025 22:13:52 +0200 Subject: [PATCH 04/10] Update tests to run with multiple coords --- tests/test_bus.py | 6 +-- tests/test_component.py | 19 ++++---- tests/test_effect.py | 57 ++++++++++++++-------- tests/test_flow.py | 88 +++++++++++++++++----------------- tests/test_linear_converter.py | 32 ++++++------- tests/test_storage.py | 38 ++++++++------- 6 files changed, 131 insertions(+), 109 deletions(-) diff --git a/tests/test_bus.py b/tests/test_bus.py index e4e0de6fd..58e00a1dc 100644 --- a/tests/test_bus.py +++ b/tests/test_bus.py @@ -13,7 +13,7 @@ class TestBusModel: def test_bus(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy_coords + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config bus = fx.Bus('TestBus', excess_penalty_per_flow_hour=None) flow_system.add_elements( bus, @@ -32,7 +32,7 @@ def test_bus(self, basic_flow_system_linopy_coords, coords_config): def test_bus_penalty(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy_coords + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config bus = fx.Bus('TestBus') flow_system.add_elements( bus, @@ -74,7 +74,7 @@ def test_bus_penalty(self, basic_flow_system_linopy_coords, coords_config): def test_bus_with_coords(self, basic_flow_system_linopy_coords, coords_config): """Test bus behavior across different coordinate configurations.""" - flow_system = basic_flow_system_linopy_coords + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config bus = fx.Bus('TestBus', excess_penalty_per_flow_hour=None) flow_system.add_elements( bus, diff --git a/tests/test_component.py b/tests/test_component.py index 90388ef26..2ed9bea3c 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -17,9 +17,8 @@ class TestComponentModel: - def test_flow_label_check(self, basic_flow_system_linopy): + def test_flow_label_check(self): """Test that flow model constraints are correctly generated.""" - _ = basic_flow_system_linopy inputs = [ fx.Flow('Q_th_Last', 'Fernwärme', relative_minimum=np.ones(10) * 0.1), fx.Flow('Q_Gas', 'Fernwärme', relative_minimum=np.ones(10) * 0.1), @@ -31,9 +30,9 @@ def test_flow_label_check(self, basic_flow_system_linopy): with pytest.raises(ValueError, match='Flow names must be unique!'): _ = flixopt.elements.Component('TestComponent', inputs=inputs, outputs=outputs) - def test_component(self, basic_flow_system_linopy): + def test_component(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config inputs = [ fx.Flow('In1', 'Fernwärme', relative_minimum=np.ones(10) * 0.1), fx.Flow('In2', 'Fernwärme', relative_minimum=np.ones(10) * 0.1), @@ -72,9 +71,9 @@ def test_component(self, basic_flow_system_linopy): msg='Incorrect constraints', ) - def test_on_with_multiple_flows(self, basic_flow_system_linopy): + def test_on_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps ub_out2 = np.linspace(1, 1.5, 10).round(2) inputs = [ @@ -171,9 +170,9 @@ def test_on_with_multiple_flows(self, basic_flow_system_linopy): + 1e-5, ) - def test_on_with_single_flow(self, basic_flow_system_linopy): + def test_on_with_single_flow(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps inputs = [ fx.Flow('In1', 'Fernwärme', relative_minimum=np.ones(10) * 0.1, size=100), @@ -231,9 +230,9 @@ def test_on_with_single_flow(self, basic_flow_system_linopy): model.variables['TestComponent|on'] == model.variables['TestComponent(In1)|on'], ) - def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy): + def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps ub_out2 = np.linspace(1, 1.5, 10).round(2) inputs = [ diff --git a/tests/test_effect.py b/tests/test_effect.py index 13e878041..a0a6fecfe 100644 --- a/tests/test_effect.py +++ b/tests/test_effect.py @@ -17,9 +17,8 @@ class TestEffectModel: """Test the FlowModel class.""" - def test_minimal(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy - timesteps = flow_system.timesteps + def test_minimal(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config effect = fx.Effect('Effect1', '€', 'Testing Effect') flow_system.add_elements(effect) @@ -47,11 +46,18 @@ def test_minimal(self, basic_flow_system_linopy): msg='Incorrect constraints', ) - assert_var_equal(model.variables['Effect1|total'], model.add_variables()) - assert_var_equal(model.variables['Effect1(invest)|total'], model.add_variables()) - assert_var_equal(model.variables['Effect1(operation)|total'], model.add_variables()) assert_var_equal( - model.variables['Effect1(operation)|total_per_timestep'], model.add_variables(coords=(timesteps,)) + model.variables['Effect1|total'], model.add_variables(coords=model.get_coords(['year', 'scenario'])) + ) + assert_var_equal( + model.variables['Effect1(invest)|total'], model.add_variables(coords=model.get_coords(['year', 'scenario'])) + ) + assert_var_equal( + model.variables['Effect1(operation)|total'], + model.add_variables(coords=model.get_coords(['year', 'scenario'])), + ) + assert_var_equal( + model.variables['Effect1(operation)|total_per_timestep'], model.add_variables(coords=model.get_coords()) ) assert_conequal( @@ -63,16 +69,15 @@ def test_minimal(self, basic_flow_system_linopy): assert_conequal( model.constraints['Effect1(operation)|total'], model.variables['Effect1(operation)|total'] - == model.variables['Effect1(operation)|total_per_timestep'].sum(), + == model.variables['Effect1(operation)|total_per_timestep'].sum('time'), ) assert_conequal( model.constraints['Effect1(operation)|total_per_timestep'], model.variables['Effect1(operation)|total_per_timestep'] == 0, ) - def test_bounds(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy - timesteps = flow_system.timesteps + def test_bounds(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config effect = fx.Effect( 'Effect1', '€', @@ -112,13 +117,24 @@ def test_bounds(self, basic_flow_system_linopy): msg='Incorrect constraints', ) - assert_var_equal(model.variables['Effect1|total'], model.add_variables(lower=3.0, upper=3.1)) - assert_var_equal(model.variables['Effect1(invest)|total'], model.add_variables(lower=2.0, upper=2.1)) - assert_var_equal(model.variables['Effect1(operation)|total'], model.add_variables(lower=1.0, upper=1.1)) + assert_var_equal( + model.variables['Effect1|total'], + model.add_variables(lower=3.0, upper=3.1, coords=model.get_coords(['year', 'scenario'])), + ) + assert_var_equal( + model.variables['Effect1(invest)|total'], + model.add_variables(lower=2.0, upper=2.1, coords=model.get_coords(['year', 'scenario'])), + ) + assert_var_equal( + model.variables['Effect1(operation)|total'], + model.add_variables(lower=1.0, upper=1.1, coords=model.get_coords(['year', 'scenario'])), + ) assert_var_equal( model.variables['Effect1(operation)|total_per_timestep'], model.add_variables( - lower=4.0 * model.hours_per_step, upper=4.1 * model.hours_per_step, coords=(timesteps,) + lower=4.0 * model.hours_per_step, + upper=4.1 * model.hours_per_step, + coords=model.get_coords(['time', 'year', 'scenario']), ), ) @@ -131,15 +147,15 @@ def test_bounds(self, basic_flow_system_linopy): assert_conequal( model.constraints['Effect1(operation)|total'], model.variables['Effect1(operation)|total'] - == model.variables['Effect1(operation)|total_per_timestep'].sum(), + == model.variables['Effect1(operation)|total_per_timestep'].sum('time'), ) assert_conequal( model.constraints['Effect1(operation)|total_per_timestep'], model.variables['Effect1(operation)|total_per_timestep'] == 0, ) - def test_shares(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_shares(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config effect1 = fx.Effect( 'Effect1', '€', @@ -202,8 +218,8 @@ def test_shares(self, basic_flow_system_linopy): class TestEffectResults: - def test_shares(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_shares(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config flow_system.effects['costs'].specific_share_to_other_effects_operation['Effect1'] = 0.5 flow_system.add_elements( fx.Effect( @@ -221,6 +237,7 @@ def test_shares(self, basic_flow_system_linopy): Q_th=fx.Flow( 'Q_th', bus='Fernwärme', + size=fx.InvestParameters(specific_effects=10, minimum_size=20, optional=False), ), Q_fu=fx.Flow('Q_fu', bus='Gas'), ), diff --git a/tests/test_flow.py b/tests/test_flow.py index 2ee609f68..357b0a352 100644 --- a/tests/test_flow.py +++ b/tests/test_flow.py @@ -11,9 +11,9 @@ class TestFlowModel: """Test the FlowModel class.""" - def test_flow_minimal(self, basic_flow_system_linopy): + def test_flow_minimal(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow('Wärme', bus='Fernwärme', size=100) @@ -36,8 +36,8 @@ def test_flow_minimal(self, basic_flow_system_linopy): ) assert_sets_equal(set(flow.submodel.constraints), {'Sink(Wärme)|total_flow_hours'}, msg='Incorrect constraints') - def test_flow(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -93,8 +93,8 @@ def test_flow(self, basic_flow_system_linopy): msg='Incorrect constraints', ) - def test_effects_per_flow_hour(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_effects_per_flow_hour(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps costs_per_flow_hour = xr.DataArray(np.linspace(1, 2, timesteps.size), coords=(timesteps,)) @@ -133,8 +133,8 @@ def test_effects_per_flow_hour(self, basic_flow_system_linopy): class TestFlowInvestModel: """Test the FlowModel class.""" - def test_flow_invest(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_invest(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -192,8 +192,8 @@ def test_flow_invest(self, basic_flow_system_linopy): * xr.DataArray(np.linspace(0.5, 1, timesteps.size), coords=(timesteps,)), ) - def test_flow_invest_optional(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_invest_optional(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -260,8 +260,8 @@ def test_flow_invest_optional(self, basic_flow_system_linopy): flow.submodel.variables['Sink(Wärme)|size'] >= flow.submodel.variables['Sink(Wärme)|is_invested'] * 20, ) - def test_flow_invest_optional_wo_min_size(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_invest_optional_wo_min_size(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -328,8 +328,8 @@ def test_flow_invest_optional_wo_min_size(self, basic_flow_system_linopy): flow.submodel.variables['Sink(Wärme)|size'] >= flow.submodel.variables['Sink(Wärme)|is_invested'] * 1e-5, ) - def test_flow_invest_wo_min_size_non_optional(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_invest_wo_min_size_non_optional(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -382,9 +382,9 @@ def test_flow_invest_wo_min_size_non_optional(self, basic_flow_system_linopy): * xr.DataArray(np.linspace(0.5, 1, timesteps.size), coords=(timesteps,)), ) - def test_flow_invest_fixed_size(self, basic_flow_system_linopy): + def test_flow_invest_fixed_size(self, basic_flow_system_linopy_coords, coords_config): """Test flow with fixed size investment.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -412,9 +412,9 @@ def test_flow_invest_fixed_size(self, basic_flow_system_linopy): flow.submodel.flow_rate, model.add_variables(lower=0.2 * 75, upper=0.9 * 75, coords=(timesteps,)) ) - def test_flow_invest_with_effects(self, basic_flow_system_linopy): + def test_flow_invest_with_effects(self, basic_flow_system_linopy_coords, coords_config): """Test flow with investment effects.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create effects co2 = fx.Effect(label='CO2', unit='ton', description='CO2 emissions') @@ -453,9 +453,9 @@ def test_flow_invest_with_effects(self, basic_flow_system_linopy): + flow.submodel.variables['Sink(Wärme)|size'] * 0.1, ) - def test_flow_invest_divest_effects(self, basic_flow_system_linopy): + def test_flow_invest_divest_effects(self, basic_flow_system_linopy_coords, coords_config): """Test flow with divestment effects.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config flow = fx.Flow( 'Wärme', @@ -483,8 +483,8 @@ def test_flow_invest_divest_effects(self, basic_flow_system_linopy): class TestFlowOnModel: """Test the FlowModel class.""" - def test_flow_on(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_on(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -547,8 +547,8 @@ def test_flow_on(self, basic_flow_system_linopy): == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum(), ) - def test_effects_per_running_hour(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_effects_per_running_hour(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps costs_per_running_hour = xr.DataArray(np.linspace(1, 2, timesteps.size), coords=(timesteps,)) @@ -601,9 +601,9 @@ def test_effects_per_running_hour(self, basic_flow_system_linopy): == flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step * co2_per_running_hour, ) - def test_consecutive_on_hours(self, basic_flow_system_linopy): + def test_consecutive_on_hours(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive on hours.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -684,9 +684,9 @@ def test_consecutive_on_hours(self, basic_flow_system_linopy): * 2, ) - def test_consecutive_on_hours_previous(self, basic_flow_system_linopy): + def test_consecutive_on_hours_previous(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive on hours.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -766,9 +766,9 @@ def test_consecutive_on_hours_previous(self, basic_flow_system_linopy): * 2, ) - def test_consecutive_off_hours(self, basic_flow_system_linopy): + def test_consecutive_off_hours(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive off hours.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -849,9 +849,9 @@ def test_consecutive_off_hours(self, basic_flow_system_linopy): * 4, ) - def test_consecutive_off_hours_previous(self, basic_flow_system_linopy): + def test_consecutive_off_hours_previous(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive off hours.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( @@ -933,9 +933,9 @@ def test_consecutive_off_hours_previous(self, basic_flow_system_linopy): * 4, ) - def test_switch_on_constraints(self, basic_flow_system_linopy): + def test_switch_on_constraints(self, basic_flow_system_linopy_coords, coords_config): """Test flow with constraints on the number of startups.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config flow = fx.Flow( 'Wärme', @@ -992,9 +992,9 @@ def test_switch_on_constraints(self, basic_flow_system_linopy): model.variables['Sink(Wärme)->costs(operation)'] == flow.submodel.variables['Sink(Wärme)|switch|on'] * 100, ) - def test_on_hours_limits(self, basic_flow_system_linopy): + def test_on_hours_limits(self, basic_flow_system_linopy_coords, coords_config): """Test flow with limits on total on hours.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config flow = fx.Flow( 'Wärme', @@ -1031,8 +1031,8 @@ def test_on_hours_limits(self, basic_flow_system_linopy): class TestFlowOnInvestModel: """Test the FlowModel class.""" - def test_flow_on_invest_optional(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_on_invest_optional(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -1130,8 +1130,8 @@ def test_flow_on_invest_optional(self, basic_flow_system_linopy): flow.submodel.variables['Sink(Wärme)|flow_rate'] <= flow.submodel.variables['Sink(Wärme)|size'] * 0.8, ) - def test_flow_on_invest_non_optional(self, basic_flow_system_linopy): - flow_system = basic_flow_system_linopy + def test_flow_on_invest_non_optional(self, basic_flow_system_linopy_coords, coords_config): + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -1222,9 +1222,9 @@ def test_flow_on_invest_non_optional(self, basic_flow_system_linopy): class TestFlowWithFixedProfile: """Test Flow with fixed relative profile.""" - def test_fixed_relative_profile(self, basic_flow_system_linopy): + def test_fixed_relative_profile(self, basic_flow_system_linopy_coords, coords_config): """Test flow with a fixed relative profile.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps # Create a time-varying profile (e.g., for a load or renewable generation) @@ -1242,9 +1242,9 @@ def test_fixed_relative_profile(self, basic_flow_system_linopy): model.add_variables(lower=profile * 100, upper=profile * 100, coords=(timesteps,)), ) - def test_fixed_profile_with_investment(self, basic_flow_system_linopy): + def test_fixed_profile_with_investment(self, basic_flow_system_linopy_coords, coords_config): """Test flow with fixed profile and investment.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps # Create a fixed profile diff --git a/tests/test_linear_converter.py b/tests/test_linear_converter.py index 322a5f6f0..e90d52f40 100644 --- a/tests/test_linear_converter.py +++ b/tests/test_linear_converter.py @@ -12,9 +12,9 @@ class TestLinearConverterModel: """Test the LinearConverterModel class.""" - def test_basic_linear_converter(self, basic_flow_system_linopy): + def test_basic_linear_converter(self, basic_flow_system_linopy_coords, coords_config): """Test basic initialization and modeling of a LinearConverter.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create input and output flows input_flow = fx.Flow('input', bus='input_bus', size=100) @@ -45,9 +45,9 @@ def test_basic_linear_converter(self, basic_flow_system_linopy): input_flow.submodel.flow_rate * 0.8 == output_flow.submodel.flow_rate * 1.0, ) - def test_linear_converter_time_varying(self, basic_flow_system_linopy): + def test_linear_converter_time_varying(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with time-varying conversion factors.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps # Create time-varying efficiency (e.g., temperature-dependent) @@ -83,9 +83,9 @@ def test_linear_converter_time_varying(self, basic_flow_system_linopy): input_flow.submodel.flow_rate * efficiency_series == output_flow.submodel.flow_rate * 1.0, ) - def test_linear_converter_multiple_factors(self, basic_flow_system_linopy): + def test_linear_converter_multiple_factors(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with multiple conversion factors.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create flows input_flow1 = fx.Flow('input1', bus='input_bus1', size=100) @@ -136,9 +136,9 @@ def test_linear_converter_multiple_factors(self, basic_flow_system_linopy): input_flow1.submodel.flow_rate * 0.2 == output_flow2.submodel.flow_rate * 0.3, ) - def test_linear_converter_with_on_off(self, basic_flow_system_linopy): + def test_linear_converter_with_on_off(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with OnOffParameters.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create input and output flows input_flow = fx.Flow('input', bus='input_bus', size=100) @@ -193,9 +193,9 @@ def test_linear_converter_with_on_off(self, basic_flow_system_linopy): == model.variables['Converter|on'] * model.hours_per_step * 5, ) - def test_linear_converter_multidimensional(self, basic_flow_system_linopy): + def test_linear_converter_multidimensional(self, basic_flow_system_linopy_coords, coords_config): """Test LinearConverter with multiple inputs, outputs, and connections between them.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create a more complex setup with multiple flows input_flow1 = fx.Flow('fuel', bus='fuel_bus', size=100) @@ -247,9 +247,9 @@ def test_linear_converter_multidimensional(self, basic_flow_system_linopy): input_flow1.submodel.flow_rate * 0.1 == output_flow2.submodel.flow_rate * 0.5, ) - def test_edge_case_time_varying_conversion(self, basic_flow_system_linopy): + def test_edge_case_time_varying_conversion(self, basic_flow_system_linopy_coords, coords_config): """Test edge case with extreme time-varying conversion factors.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps # Create fluctuating conversion efficiency (e.g., for a heat pump) @@ -288,9 +288,9 @@ def test_edge_case_time_varying_conversion(self, basic_flow_system_linopy): input_flow.submodel.flow_rate * fluctuating_cop == output_flow.submodel.flow_rate * 1.0, ) - def test_piecewise_conversion(self, basic_flow_system_linopy): + def test_piecewise_conversion(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with PiecewiseConversion.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps # Create input and output flows @@ -377,9 +377,9 @@ def test_piecewise_conversion(self, basic_flow_system_linopy): <= 1, ) - def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy): + def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with PiecewiseConversion and OnOffParameters.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps # Create input and output flows diff --git a/tests/test_storage.py b/tests/test_storage.py index f6b6f2079..479f66a87 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -11,9 +11,9 @@ class TestStorageModel: """Test that storage model variables and constraints are correctly generated.""" - def test_basic_storage(self, basic_flow_system_linopy): + def test_basic_storage(self, basic_flow_system_linopy_coords, coords_config): """Test that basic storage model variables and constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps timesteps_extra = flow_system.timesteps_extra @@ -85,9 +85,9 @@ def test_basic_storage(self, basic_flow_system_linopy): model.variables['TestStorage|charge_state'].isel(time=0) == 0, ) - def test_lossy_storage(self, basic_flow_system_linopy): + def test_lossy_storage(self, basic_flow_system_linopy_coords, coords_config): """Test that basic storage model variables and constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps timesteps_extra = flow_system.timesteps_extra @@ -170,9 +170,9 @@ def test_lossy_storage(self, basic_flow_system_linopy): model.variables['TestStorage|charge_state'].isel(time=0) == 0, ) - def test_charge_state_bounds(self, basic_flow_system_linopy): + def test_charge_state_bounds(self, basic_flow_system_linopy_coords, coords_config): """Test that basic storage model variables and constraints are correctly generated.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps timesteps_extra = flow_system.timesteps_extra @@ -251,9 +251,9 @@ def test_charge_state_bounds(self, basic_flow_system_linopy): model.variables['TestStorage|charge_state'].isel(time=0) == 3, ) - def test_storage_with_investment(self, basic_flow_system_linopy): + def test_storage_with_investment(self, basic_flow_system_linopy_coords, coords_config): """Test storage with investment parameters.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create storage with investment parameters storage = fx.Storage( @@ -297,9 +297,9 @@ def test_storage_with_investment(self, basic_flow_system_linopy): model.variables['InvestStorage|size'] >= model.variables['InvestStorage|is_invested'] * 20, ) - def test_storage_with_final_state_constraints(self, basic_flow_system_linopy): + def test_storage_with_final_state_constraints(self, basic_flow_system_linopy_coords, coords_config): """Test storage with final state constraints.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create storage with final state constraints storage = fx.Storage( @@ -342,9 +342,9 @@ def test_storage_with_final_state_constraints(self, basic_flow_system_linopy): model.variables['FinalStateStorage|charge_state'].isel(time=-1) <= 25, ) - def test_storage_cyclic_initialization(self, basic_flow_system_linopy): + def test_storage_cyclic_initialization(self, basic_flow_system_linopy_coords, coords_config): """Test storage with cyclic initialization.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create storage with cyclic initialization storage = fx.Storage( @@ -375,9 +375,9 @@ def test_storage_cyclic_initialization(self, basic_flow_system_linopy): 'prevent_simultaneous', [True, False], ) - def test_simultaneous_charge_discharge(self, basic_flow_system_linopy, prevent_simultaneous): + def test_simultaneous_charge_discharge(self, basic_flow_system_linopy_coords, coords_config, prevent_simultaneous): """Test prevent_simultaneous_charge_and_discharge parameter.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create storage with or without simultaneous charge/discharge prevention storage = fx.Storage( @@ -424,10 +424,16 @@ def test_simultaneous_charge_discharge(self, basic_flow_system_linopy, prevent_s ], ) def test_investment_parameters( - self, basic_flow_system_linopy, optional, minimum_size, expected_vars, expected_constraints + self, + basic_flow_system_linopy_coords, + coords_config, + optional, + minimum_size, + expected_vars, + expected_constraints, ): """Test different investment parameter combinations.""" - flow_system = basic_flow_system_linopy + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config # Create investment parameters invest_params = { From 5510297e0ba5f63887f2b531384c39ce923ba1ab Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Wed, 23 Jul 2025 22:15:03 +0200 Subject: [PATCH 05/10] Fix Effects dataset computation in case of empty effects --- flixopt/interface.py | 20 ++++++------ flixopt/results.py | 76 +++++++++++++++++++++++++++++--------------- tests/test_effect.py | 29 ++++++++++------- 3 files changed, 76 insertions(+), 49 deletions(-) diff --git a/flixopt/interface.py b/flixopt/interface.py index 374c3fb44..8bcab1de0 100644 --- a/flixopt/interface.py +++ b/flixopt/interface.py @@ -339,15 +339,13 @@ def use_consecutive_off_hours(self) -> bool: @property def use_switch_on(self) -> bool: """Determines wether a Variable for SWITCH-ON is needed or not""" - return ( - any( - param not in (None, {}) - for param in [ - self.effects_per_switch_on, - self.switch_on_total_max, - self.on_hours_total_min, - self.on_hours_total_max, - ] - ) - or self.force_switch_on + if self.force_switch_on: + return True + + return any( + param is not None or param != {} + for param in [ + self.effects_per_switch_on, + self.switch_on_total_max, + ] ) diff --git a/flixopt/results.py b/flixopt/results.py index 512af4ad7..dfe6a759b 100644 --- a/flixopt/results.py +++ b/flixopt/results.py @@ -200,7 +200,7 @@ def __init__( self._flow_rates = None self._flow_hours = None self._sizes = None - self._effects_per_component = {'operation': None, 'invest': None, 'total': None} + self._effects_per_component = None def __getitem__(self, key: str) -> Union['ComponentResults', 'BusResults', 'EffectResults', 'FlowResults']: if key in self.components: @@ -312,20 +312,24 @@ def filter_solution( startswith=startswith, ) - def effects_per_component(self, mode: Literal['operation', 'invest', 'total'] = 'total') -> xr.Dataset: - """Returns a dataset containing effect totals for each components (including their flows). - - Args: - mode: Which effects to contain. (operation, invest, total) + @property + def effects_per_component(self) -> xr.Dataset: + """Returns a dataset containing effect results for each mode, aggregated by Component Returns: An xarray Dataset with an additional component dimension and effects as variables. """ - if mode not in ['operation', 'invest', 'total']: - raise ValueError(f'Invalid mode {mode}') - if self._effects_per_component[mode] is None: - self._effects_per_component[mode] = self._create_effects_dataset(mode) - return self._effects_per_component[mode] + if self._effects_per_component is None: + self._effects_per_component = xr.Dataset( + { + mode: self._create_effects_dataset(mode).to_dataarray('effect', name=mode) + for mode in ['operation', 'invest', 'total'] + } + ) + dim_order = ['time', 'year', 'scenario', 'component', 'effect'] + self._effects_per_component = self._effects_per_component.transpose(*dim_order, missing_dims='ignore') + + return self._effects_per_component def flow_rates( self, @@ -580,7 +584,7 @@ def _compute_effect_total( total = xr.DataArray(np.nan) return total.rename(f'{element}->{effect}({mode})') - def _create_effects_dataset(self, mode: Literal['operation', 'invest', 'total'] = 'total') -> xr.Dataset: + def _create_effects_dataset(self, mode: Literal['operation', 'invest', 'total']) -> xr.Dataset: """Creates a dataset containing effect totals for all components (including their flows). The dataset does contain the direct as well as the indirect effects of each component. @@ -590,24 +594,44 @@ def _create_effects_dataset(self, mode: Literal['operation', 'invest', 'total'] Returns: An xarray Dataset with components as dimension and effects as variables. """ - # Create an empty dataset ds = xr.Dataset() + all_arrays = {} + template = None # Template is needed to determine the dimensions of the arrays. This handles the case of no shares for an effect + + components_list = list(self.components) - # Add each effect as a variable to the dataset + # First pass: collect arrays and find template for effect in self.effects: - # Create a list of DataArrays, one for each component - component_arrays = [ - self._compute_effect_total(element=component, effect=effect, mode=mode, include_flows=True).expand_dims( - component=[component] - ) # Add component dimension to each array - for component in list(self.components) - ] + effect_arrays = [] + for component in components_list: + da = self._compute_effect_total(element=component, effect=effect, mode=mode, include_flows=True) + effect_arrays.append(da) + + if template is None and (da.dims or not da.isnull().all()): + template = da + + all_arrays[effect] = effect_arrays + + # Ensure we have a template + if template is None: + raise ValueError( + f"No template with proper dimensions found for mode '{mode}'. " + f'All computed arrays are scalars, which indicates a data issue.' + ) + + # Second pass: process all effects (guaranteed to include all) + for effect in self.effects: + dataarrays = all_arrays[effect] + component_arrays = [] + + for component, arr in zip(components_list, dataarrays, strict=False): + # Expand scalar NaN arrays to match template dimensions + if not arr.dims and np.isnan(arr.item()): + arr = xr.full_like(template, np.nan, dtype=float).rename(arr.name) + + component_arrays.append(arr.expand_dims(component=[component])) - # Combine all components into one DataArray for this effect - if component_arrays: - effect_array = xr.concat(component_arrays, dim='component', coords='minimal') - # Add this effect as a variable to the dataset - ds[effect] = effect_array + ds[effect] = xr.concat(component_arrays, dim='component', coords='minimal') # For now include a test to ensure correctness suffix = { diff --git a/tests/test_effect.py b/tests/test_effect.py index a0a6fecfe..cc4841900 100644 --- a/tests/test_effect.py +++ b/tests/test_effect.py @@ -266,58 +266,63 @@ def test_shares(self, basic_flow_system_linopy_coords, coords_config): np.testing.assert_allclose(results.effect_share_factors['invest'][key].values, value) xr.testing.assert_allclose( - results.effects_per_component('operation').sum('component')['costs'], + results.effects_per_component['operation'].sum('component').sel(effect='costs', drop=True), results.solution['costs(operation)|total_per_timestep'].fillna(0), ) xr.testing.assert_allclose( - results.effects_per_component('operation').sum('component')['Effect1'], + results.effects_per_component['operation'].sum('component').sel(effect='Effect1', drop=True), results.solution['Effect1(operation)|total_per_timestep'].fillna(0), ) xr.testing.assert_allclose( - results.effects_per_component('operation').sum('component')['Effect2'], + results.effects_per_component['operation'].sum('component').sel(effect='Effect2', drop=True), results.solution['Effect2(operation)|total_per_timestep'].fillna(0), ) xr.testing.assert_allclose( - results.effects_per_component('operation').sum('component')['Effect3'], + results.effects_per_component['operation'].sum('component').sel(effect='Effect3', drop=True), results.solution['Effect3(operation)|total_per_timestep'].fillna(0), ) # Invest mode checks xr.testing.assert_allclose( - results.effects_per_component('invest').sum('component')['costs'], results.solution['costs(invest)|total'] + results.effects_per_component['invest'].sum('component').sel(effect='costs', drop=True), + results.solution['costs(invest)|total'], ) xr.testing.assert_allclose( - results.effects_per_component('invest').sum('component')['Effect1'], + results.effects_per_component['invest'].sum('component').sel(effect='Effect1', drop=True), results.solution['Effect1(invest)|total'], ) xr.testing.assert_allclose( - results.effects_per_component('invest').sum('component')['Effect2'], + results.effects_per_component['invest'].sum('component').sel(effect='Effect2', drop=True), results.solution['Effect2(invest)|total'], ) xr.testing.assert_allclose( - results.effects_per_component('invest').sum('component')['Effect3'], + results.effects_per_component['invest'].sum('component').sel(effect='Effect3', drop=True), results.solution['Effect3(invest)|total'], ) # Total mode checks xr.testing.assert_allclose( - results.effects_per_component('total').sum('component')['costs'], results.solution['costs|total'] + results.effects_per_component['total'].sum('component').sel(effect='costs', drop=True), + results.solution['costs|total'], ) xr.testing.assert_allclose( - results.effects_per_component('total').sum('component')['Effect1'], results.solution['Effect1|total'] + results.effects_per_component['total'].sum('component').sel(effect='Effect1', drop=True), + results.solution['Effect1|total'], ) xr.testing.assert_allclose( - results.effects_per_component('total').sum('component')['Effect2'], results.solution['Effect2|total'] + results.effects_per_component['total'].sum('component').sel(effect='Effect2', drop=True), + results.solution['Effect2|total'], ) xr.testing.assert_allclose( - results.effects_per_component('total').sum('component')['Effect3'], results.solution['Effect3|total'] + results.effects_per_component['total'].sum('component').sel(effect='Effect3', drop=True), + results.solution['Effect3|total'], ) From b694dbe1b7605495a1e5902fdbacd8f7e3d24877 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Thu, 24 Jul 2025 13:13:41 +0200 Subject: [PATCH 06/10] Update Test for multiple dims Fix Dim order in scaled_bounds_with_state Bugfix logic in .use_switch_on --- flixopt/interface.py | 6 +- flixopt/modeling.py | 4 +- tests/test_flow.py | 220 ++++++++++++++++++++++++++----------------- 3 files changed, 140 insertions(+), 90 deletions(-) diff --git a/flixopt/interface.py b/flixopt/interface.py index 8bcab1de0..cae1757c7 100644 --- a/flixopt/interface.py +++ b/flixopt/interface.py @@ -246,8 +246,8 @@ def maximum_or_fixed_size(self) -> NonTemporalData: class OnOffParameters(Interface): def __init__( self, - effects_per_switch_on: Optional['NonTemporalEffectsUser'] = None, - effects_per_running_hour: Optional['NonTemporalEffectsUser'] = None, + effects_per_switch_on: Optional['TemporalEffectsUser'] = None, + effects_per_running_hour: Optional['TemporalEffectsUser'] = None, on_hours_total_min: Optional[int] = None, on_hours_total_max: Optional[int] = None, consecutive_on_hours_min: Optional[TemporalDataUser] = None, @@ -343,7 +343,7 @@ def use_switch_on(self) -> bool: return True return any( - param is not None or param != {} + param is not None and param != {} for param in [ self.effects_per_switch_on, self.switch_on_total_max, diff --git a/flixopt/modeling.py b/flixopt/modeling.py index 14f8c45f3..11880c5e8 100644 --- a/flixopt/modeling.py +++ b/flixopt/modeling.py @@ -536,8 +536,8 @@ def scaled_bounds_with_state( ) scaling_upper = model.add_constraints(variable <= scaling_variable * rel_upper, name=f'{name}|ub2') - big_m_upper = scaling_max * rel_upper - big_m_lower = np.maximum(CONFIG.modeling.EPSILON, scaling_min * rel_lower) + big_m_upper = rel_upper * scaling_max + big_m_lower = np.maximum(CONFIG.modeling.EPSILON, rel_lower * scaling_min) binary_upper = model.add_constraints(variable_state * big_m_upper >= variable, name=f'{name}|ub1') binary_lower = model.add_constraints(variable_state * big_m_lower <= variable, name=f'{name}|lb1') diff --git a/tests/test_flow.py b/tests/test_flow.py index 357b0a352..0d60a8bfe 100644 --- a/tests/test_flow.py +++ b/tests/test_flow.py @@ -14,7 +14,7 @@ class TestFlowModel: def test_flow_minimal(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps + flow = fx.Flow('Wärme', bus='Fernwärme', size=100) flow_system.add_elements(fx.Sink('Sink', sink=flow)) @@ -24,10 +24,12 @@ def test_flow_minimal(self, basic_flow_system_linopy_coords, coords_config): assert_conequal( model.constraints['Sink(Wärme)|total_flow_hours'], flow.submodel.variables['Sink(Wärme)|total_flow_hours'] - == (flow.submodel.variables['Sink(Wärme)|flow_rate'] * model.hours_per_step).sum(), + == (flow.submodel.variables['Sink(Wärme)|flow_rate'] * model.hours_per_step).sum('time'), + ) + assert_var_equal(flow.submodel.flow_rate, model.add_variables(lower=0, upper=100, coords=model.get_coords())) + assert_var_equal( + flow.submodel.total_flow_hours, model.add_variables(lower=0, coords=model.get_coords(['year', 'scenario'])) ) - assert_var_equal(flow.submodel.flow_rate, model.add_variables(lower=0, upper=100, coords=(timesteps,))) - assert_var_equal(flow.submodel.total_flow_hours, model.add_variables(lower=0)) assert_sets_equal( set(flow.submodel.variables), @@ -39,6 +41,7 @@ def test_flow_minimal(self, basic_flow_system_linopy_coords, coords_config): def test_flow(self, basic_flow_system_linopy_coords, coords_config): flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps + flow = fx.Flow( 'Wärme', bus='Fernwärme', @@ -58,17 +61,23 @@ def test_flow(self, basic_flow_system_linopy_coords, coords_config): assert_conequal( model.constraints['Sink(Wärme)|total_flow_hours'], flow.submodel.variables['Sink(Wärme)|total_flow_hours'] - == (flow.submodel.variables['Sink(Wärme)|flow_rate'] * model.hours_per_step).sum(), + == (flow.submodel.variables['Sink(Wärme)|flow_rate'] * model.hours_per_step).sum('time'), ) - assert_var_equal(flow.submodel.total_flow_hours, model.add_variables(lower=10, upper=1000)) + assert_var_equal( + flow.submodel.total_flow_hours, + model.add_variables(lower=10, upper=1000, coords=model.get_coords(['year', 'scenario'])), + ) + + assert flow.relative_minimum.dims == tuple(model.get_coords()) + assert flow.relative_maximum.dims == tuple(model.get_coords()) assert_var_equal( flow.submodel.flow_rate, model.add_variables( - lower=np.linspace(0, 0.5, timesteps.size) * 100, - upper=np.linspace(0.5, 1, timesteps.size) * 100, - coords=(timesteps,), + lower=flow.relative_minimum * 100, + upper=flow.relative_maximum * 100, + coords=model.get_coords(), ), ) @@ -168,28 +177,32 @@ def test_flow_invest(self, basic_flow_system_linopy_coords, coords_config): ) # size - assert_var_equal(model['Sink(Wärme)|size'], model.add_variables(lower=20, upper=100)) + assert_var_equal( + model['Sink(Wärme)|size'], + model.add_variables(lower=20, upper=100, coords=model.get_coords(['year', 'scenario'])), + ) + + assert flow.relative_minimum.dims == tuple(model.get_coords()) + assert flow.relative_maximum.dims == tuple(model.get_coords()) # flow_rate assert_var_equal( flow.submodel.flow_rate, model.add_variables( - lower=np.linspace(0.1, 0.5, timesteps.size) * 20, - upper=np.linspace(0.5, 1, timesteps.size) * 100, - coords=(timesteps,), + lower=flow.relative_minimum * 20, + upper=flow.relative_maximum * 100, + coords=model.get_coords(), ), ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|lb'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - >= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.1, 0.5, timesteps.size), coords=(timesteps,)), + >= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_minimum, ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|ub'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - <= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.5, 1, timesteps.size), coords=(timesteps,)), + <= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_maximum, ) def test_flow_invest_optional(self, basic_flow_system_linopy_coords, coords_config): @@ -224,30 +237,37 @@ def test_flow_invest_optional(self, basic_flow_system_linopy_coords, coords_conf msg='Incorrect constraints', ) - assert_var_equal(model['Sink(Wärme)|size'], model.add_variables(lower=0, upper=100)) + assert_var_equal( + model['Sink(Wärme)|size'], + model.add_variables(lower=0, upper=100, coords=model.get_coords(['year', 'scenario'])), + ) + + assert_var_equal( + model['Sink(Wärme)|is_invested'], + model.add_variables(binary=True, coords=model.get_coords(['year', 'scenario'])), + ) - assert_var_equal(model['Sink(Wärme)|is_invested'], model.add_variables(binary=True)) + assert flow.relative_minimum.dims == tuple(model.get_coords()) + assert flow.relative_maximum.dims == tuple(model.get_coords()) # flow_rate assert_var_equal( flow.submodel.flow_rate, model.add_variables( lower=0, # Optional investment - upper=np.linspace(0.5, 1, timesteps.size) * 100, - coords=(timesteps,), + upper=flow.relative_maximum * 100, + coords=model.get_coords(), ), ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|lb'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - >= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.1, 0.5, timesteps.size), coords=(timesteps,)), + >= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_minimum, ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|ub'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - <= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.5, 1, timesteps.size), coords=(timesteps,)), + <= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_maximum, ) # Is invested @@ -292,30 +312,37 @@ def test_flow_invest_optional_wo_min_size(self, basic_flow_system_linopy_coords, msg='Incorrect constraints', ) - assert_var_equal(model['Sink(Wärme)|size'], model.add_variables(lower=0, upper=100)) + assert_var_equal( + model['Sink(Wärme)|size'], + model.add_variables(lower=0, upper=100, coords=model.get_coords(['year', 'scenario'])), + ) - assert_var_equal(model['Sink(Wärme)|is_invested'], model.add_variables(binary=True)) + assert_var_equal( + model['Sink(Wärme)|is_invested'], + model.add_variables(binary=True, coords=model.get_coords(['year', 'scenario'])), + ) + + assert flow.relative_minimum.dims == tuple(model.get_coords()) + assert flow.relative_maximum.dims == tuple(model.get_coords()) # flow_rate assert_var_equal( flow.submodel.flow_rate, model.add_variables( lower=0, # Optional investment - upper=np.linspace(0.5, 1, timesteps.size) * 100, - coords=(timesteps,), + upper=flow.relative_maximum * 100, + coords=model.get_coords(), ), ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|lb'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - >= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.1, 0.5, timesteps.size), coords=(timesteps,)), + >= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_minimum, ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|ub'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - <= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.5, 1, timesteps.size), coords=(timesteps,)), + <= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_maximum, ) # Is invested @@ -358,34 +385,37 @@ def test_flow_invest_wo_min_size_non_optional(self, basic_flow_system_linopy_coo msg='Incorrect constraints', ) - assert_var_equal(model['Sink(Wärme)|size'], model.add_variables(lower=1e-5, upper=100)) + assert_var_equal( + model['Sink(Wärme)|size'], + model.add_variables(lower=1e-5, upper=100, coords=model.get_coords(['year', 'scenario'])), + ) + + assert flow.relative_minimum.dims == tuple(model.get_coords()) + assert flow.relative_maximum.dims == tuple(model.get_coords()) # flow_rate assert_var_equal( flow.submodel.flow_rate, model.add_variables( - lower=np.linspace(0.1, 0.5, timesteps.size) * 1e-5, - upper=np.linspace(0.5, 1, timesteps.size) * 100, - coords=(timesteps,), + lower=flow.relative_minimum * 1e-5, + upper=flow.relative_maximum * 100, + coords=model.get_coords(), ), ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|lb'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - >= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.1, 0.5, timesteps.size), coords=(timesteps,)), + >= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_minimum, ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|ub'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - <= flow.submodel.variables['Sink(Wärme)|size'] - * xr.DataArray(np.linspace(0.5, 1, timesteps.size), coords=(timesteps,)), + <= flow.submodel.variables['Sink(Wärme)|size'] * flow.relative_maximum, ) def test_flow_invest_fixed_size(self, basic_flow_system_linopy_coords, coords_config): """Test flow with fixed size investment.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -405,11 +435,14 @@ def test_flow_invest_fixed_size(self, basic_flow_system_linopy_coords, coords_co ) # Check that size is fixed to 75 - assert_var_equal(flow.submodel.variables['Sink(Wärme)|size'], model.add_variables(lower=75, upper=75)) + assert_var_equal( + flow.submodel.variables['Sink(Wärme)|size'], + model.add_variables(lower=75, upper=75, coords=model.get_coords(['year', 'scenario'])), + ) # Check flow rate bounds assert_var_equal( - flow.submodel.flow_rate, model.add_variables(lower=0.2 * 75, upper=0.9 * 75, coords=(timesteps,)) + flow.submodel.flow_rate, model.add_variables(lower=0.2 * 75, upper=0.9 * 75, coords=model.get_coords()) ) def test_flow_invest_with_effects(self, basic_flow_system_linopy_coords, coords_config): @@ -485,13 +518,13 @@ class TestFlowOnModel: def test_flow_on(self, basic_flow_system_linopy_coords, coords_config): flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps + flow = fx.Flow( 'Wärme', bus='Fernwärme', size=100, - relative_minimum=xr.DataArray(0.2, coords=(timesteps,)), - relative_maximum=xr.DataArray(0.8, coords=(timesteps,)), + relative_minimum=0.2, + relative_maximum=0.8, on_off_parameters=fx.OnOffParameters(), ) flow_system.add_elements(fx.Sink('Sink', sink=flow)) @@ -519,18 +552,18 @@ def test_flow_on(self, basic_flow_system_linopy_coords, coords_config): model.add_variables( lower=0, upper=0.8 * 100, - coords=(timesteps,), + coords=model.get_coords(), ), ) # OnOff assert_var_equal( flow.submodel.on_off.on, - model.add_variables(binary=True, coords=(timesteps,)), + model.add_variables(binary=True, coords=model.get_coords()), ) assert_var_equal( model.variables['Sink(Wärme)|on_hours_total'], - model.add_variables(lower=0), + model.add_variables(lower=0, coords=model.get_coords(['year', 'scenario'])), ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|lb'], @@ -544,15 +577,15 @@ def test_flow_on(self, basic_flow_system_linopy_coords, coords_config): assert_conequal( model.constraints['Sink(Wärme)|on_hours_total'], flow.submodel.variables['Sink(Wärme)|on_hours_total'] - == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum(), + == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum('time'), ) def test_effects_per_running_hour(self, basic_flow_system_linopy_coords, coords_config): flow_system, coords_config = basic_flow_system_linopy_coords, coords_config timesteps = flow_system.timesteps - costs_per_running_hour = xr.DataArray(np.linspace(1, 2, timesteps.size), coords=(timesteps,)) - co2_per_running_hour = xr.DataArray(np.linspace(4, 5, timesteps.size), coords=(timesteps,)) + costs_per_running_hour = np.linspace(1, 2, timesteps.size) + co2_per_running_hour = np.linspace(4, 5, timesteps.size) flow = fx.Flow( 'Wärme', @@ -589,6 +622,12 @@ def test_effects_per_running_hour(self, basic_flow_system_linopy_coords, coords_ assert 'Sink(Wärme)->costs(operation)' in set(costs.submodel.constraints) assert 'Sink(Wärme)->CO2(operation)' in set(co2.submodel.constraints) + costs_per_running_hour = flow.on_off_parameters.effects_per_running_hour['costs'] + co2_per_running_hour = flow.on_off_parameters.effects_per_running_hour['CO2'] + + assert costs_per_running_hour.dims == tuple(model.get_coords()) + assert co2_per_running_hour.dims == tuple(model.get_coords()) + assert_conequal( model.constraints['Sink(Wärme)->costs(operation)'], model.variables['Sink(Wärme)->costs(operation)'] @@ -604,7 +643,6 @@ def test_effects_per_running_hour(self, basic_flow_system_linopy_coords, coords_ def test_consecutive_on_hours(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive on hours.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -642,7 +680,7 @@ def test_consecutive_on_hours(self, basic_flow_system_linopy_coords, coords_conf assert_var_equal( model.variables['Sink(Wärme)|consecutive_on_hours'], - model.add_variables(lower=0, upper=8, coords=(timesteps,)), + model.add_variables(lower=0, upper=8, coords=model.get_coords()), ) mega = model.hours_per_step.sum('time') @@ -687,7 +725,6 @@ def test_consecutive_on_hours(self, basic_flow_system_linopy_coords, coords_conf def test_consecutive_on_hours_previous(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive on hours.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -724,7 +761,7 @@ def test_consecutive_on_hours_previous(self, basic_flow_system_linopy_coords, co assert_var_equal( model.variables['Sink(Wärme)|consecutive_on_hours'], - model.add_variables(lower=0, upper=8, coords=(timesteps,)), + model.add_variables(lower=0, upper=8, coords=model.get_coords()), ) mega = model.hours_per_step.sum('time') + model.hours_per_step.isel(time=0) * 3 @@ -769,7 +806,6 @@ def test_consecutive_on_hours_previous(self, basic_flow_system_linopy_coords, co def test_consecutive_off_hours(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive off hours.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -807,7 +843,7 @@ def test_consecutive_off_hours(self, basic_flow_system_linopy_coords, coords_con assert_var_equal( model.variables['Sink(Wärme)|consecutive_off_hours'], - model.add_variables(lower=0, upper=12, coords=(timesteps,)), + model.add_variables(lower=0, upper=12, coords=model.get_coords()), ) mega = model.hours_per_step.sum('time') + model.hours_per_step.isel(time=0) * 1 # previously off for 1h @@ -852,7 +888,6 @@ def test_consecutive_off_hours(self, basic_flow_system_linopy_coords, coords_con def test_consecutive_off_hours_previous(self, basic_flow_system_linopy_coords, coords_config): """Test flow with minimum and maximum consecutive off hours.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', @@ -891,7 +926,7 @@ def test_consecutive_off_hours_previous(self, basic_flow_system_linopy_coords, c assert_var_equal( model.variables['Sink(Wärme)|consecutive_off_hours'], - model.add_variables(lower=0, upper=12, coords=(timesteps,)), + model.add_variables(lower=0, upper=12, coords=model.get_coords()), ) mega = model.hours_per_step.sum('time') + model.hours_per_step.isel(time=0) * 2 @@ -974,7 +1009,10 @@ def test_switch_on_constraints(self, basic_flow_system_linopy_coords, coords_con ) # Check switch_on_nr variable bounds - assert_var_equal(flow.submodel.variables['Sink(Wärme)|switch|count'], model.add_variables(lower=0, upper=5)) + assert_var_equal( + flow.submodel.variables['Sink(Wärme)|switch|count'], + model.add_variables(lower=0, upper=5, coords=model.get_coords(['year', 'scenario'])), + ) # Verify switch_on_nr constraint (limits number of startups) assert_conequal( @@ -1017,14 +1055,15 @@ def test_on_hours_limits(self, basic_flow_system_linopy_coords, coords_config): # Check on_hours_total variable bounds assert_var_equal( - flow.submodel.variables['Sink(Wärme)|on_hours_total'], model.add_variables(lower=20, upper=100) + flow.submodel.variables['Sink(Wärme)|on_hours_total'], + model.add_variables(lower=20, upper=100, coords=model.get_coords(['year', 'scenario'])), ) # Check on_hours_total constraint assert_conequal( model.constraints['Sink(Wärme)|on_hours_total'], flow.submodel.variables['Sink(Wärme)|on_hours_total'] - == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum(), + == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum('time'), ) @@ -1033,13 +1072,12 @@ class TestFlowOnInvestModel: def test_flow_on_invest_optional(self, basic_flow_system_linopy_coords, coords_config): flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', bus='Fernwärme', size=fx.InvestParameters(minimum_size=20, maximum_size=200, optional=True), - relative_minimum=xr.DataArray(0.2, coords=(timesteps,)), - relative_maximum=xr.DataArray(0.8, coords=(timesteps,)), + relative_minimum=0.2, + relative_maximum=0.8, on_off_parameters=fx.OnOffParameters(), ) flow_system.add_elements(fx.Sink('Sink', sink=flow)) @@ -1079,18 +1117,18 @@ def test_flow_on_invest_optional(self, basic_flow_system_linopy_coords, coords_c model.add_variables( lower=0, upper=0.8 * 200, - coords=(timesteps,), + coords=model.get_coords(), ), ) # OnOff assert_var_equal( flow.submodel.on_off.on, - model.add_variables(binary=True, coords=(timesteps,)), + model.add_variables(binary=True, coords=model.get_coords()), ) assert_var_equal( model.variables['Sink(Wärme)|on_hours_total'], - model.add_variables(lower=0), + model.add_variables(lower=0, coords=model.get_coords(['year', 'scenario'])), ) assert_conequal( model.constraints['Sink(Wärme)|size|lb'], @@ -1111,11 +1149,14 @@ def test_flow_on_invest_optional(self, basic_flow_system_linopy_coords, coords_c assert_conequal( model.constraints['Sink(Wärme)|on_hours_total'], flow.submodel.variables['Sink(Wärme)|on_hours_total'] - == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum(), + == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum('time'), ) # Investment - assert_var_equal(model['Sink(Wärme)|size'], model.add_variables(lower=0, upper=200)) + assert_var_equal( + model['Sink(Wärme)|size'], + model.add_variables(lower=0, upper=200, coords=model.get_coords(['year', 'scenario'])), + ) mega = 0.2 * 200 # Relative minimum * maximum size assert_conequal( @@ -1132,13 +1173,12 @@ def test_flow_on_invest_optional(self, basic_flow_system_linopy_coords, coords_c def test_flow_on_invest_non_optional(self, basic_flow_system_linopy_coords, coords_config): flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps flow = fx.Flow( 'Wärme', bus='Fernwärme', size=fx.InvestParameters(minimum_size=20, maximum_size=200, optional=False), - relative_minimum=xr.DataArray(0.2, coords=(timesteps,)), - relative_maximum=xr.DataArray(0.8, coords=(timesteps,)), + relative_minimum=0.2, + relative_maximum=0.8, on_off_parameters=fx.OnOffParameters(), ) flow_system.add_elements(fx.Sink('Sink', sink=flow)) @@ -1175,18 +1215,18 @@ def test_flow_on_invest_non_optional(self, basic_flow_system_linopy_coords, coor model.add_variables( lower=0, upper=0.8 * 200, - coords=(timesteps,), + coords=model.get_coords(), ), ) # OnOff assert_var_equal( flow.submodel.on_off.on, - model.add_variables(binary=True, coords=(timesteps,)), + model.add_variables(binary=True, coords=model.get_coords()), ) assert_var_equal( model.variables['Sink(Wärme)|on_hours_total'], - model.add_variables(lower=0), + model.add_variables(lower=0, coords=model.get_coords(['year', 'scenario'])), ) assert_conequal( model.constraints['Sink(Wärme)|flow_rate|lb1'], @@ -1199,11 +1239,14 @@ def test_flow_on_invest_non_optional(self, basic_flow_system_linopy_coords, coor assert_conequal( model.constraints['Sink(Wärme)|on_hours_total'], flow.submodel.variables['Sink(Wärme)|on_hours_total'] - == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum(), + == (flow.submodel.variables['Sink(Wärme)|on'] * model.hours_per_step).sum('time'), ) # Investment - assert_var_equal(model['Sink(Wärme)|size'], model.add_variables(lower=20, upper=200)) + assert_var_equal( + model['Sink(Wärme)|size'], + model.add_variables(lower=20, upper=200, coords=model.get_coords(['year', 'scenario'])), + ) mega = 0.2 * 200 # Relative minimum * maximum size assert_conequal( @@ -1231,7 +1274,10 @@ def test_fixed_relative_profile(self, basic_flow_system_linopy_coords, coords_co profile = np.sin(np.linspace(0, 2 * np.pi, len(timesteps))) * 0.5 + 0.5 # Values between 0 and 1 flow = fx.Flow( - 'Wärme', bus='Fernwärme', size=100, fixed_relative_profile=xr.DataArray(profile, coords=(timesteps,)) + 'Wärme', + bus='Fernwärme', + size=100, + fixed_relative_profile=profile, ) flow_system.add_elements(fx.Sink('Sink', sink=flow)) @@ -1239,7 +1285,11 @@ def test_fixed_relative_profile(self, basic_flow_system_linopy_coords, coords_co assert_var_equal( flow.submodel.variables['Sink(Wärme)|flow_rate'], - model.add_variables(lower=profile * 100, upper=profile * 100, coords=(timesteps,)), + model.add_variables( + lower=flow.fixed_relative_profile * 100, + upper=flow.fixed_relative_profile * 100, + coords=model.get_coords(), + ), ) def test_fixed_profile_with_investment(self, basic_flow_system_linopy_coords, coords_config): @@ -1254,7 +1304,7 @@ def test_fixed_profile_with_investment(self, basic_flow_system_linopy_coords, co 'Wärme', bus='Fernwärme', size=fx.InvestParameters(minimum_size=50, maximum_size=200, optional=True), - fixed_relative_profile=xr.DataArray(profile, coords=(timesteps,)), + fixed_relative_profile=profile, ) flow_system.add_elements(fx.Sink('Sink', sink=flow)) @@ -1262,14 +1312,14 @@ def test_fixed_profile_with_investment(self, basic_flow_system_linopy_coords, co assert_var_equal( flow.submodel.variables['Sink(Wärme)|flow_rate'], - model.add_variables(lower=0, upper=profile * 200, coords=(timesteps,)), + model.add_variables(lower=0, upper=flow.fixed_relative_profile * 200, coords=model.get_coords()), ) # The constraint should link flow_rate to size * profile assert_conequal( model.constraints['Sink(Wärme)|flow_rate|fixed'], flow.submodel.variables['Sink(Wärme)|flow_rate'] - == flow.submodel.variables['Sink(Wärme)|size'] * xr.DataArray(profile, coords=(timesteps,)), + == flow.submodel.variables['Sink(Wärme)|size'] * flow.fixed_relative_profile, ) From 262e8b4f3a69125dc890a2c6815e3c1f5f9cbfe4 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 28 Jul 2025 21:33:45 +0200 Subject: [PATCH 07/10] Fix test with multiple dims --- tests/test_linear_converter.py | 24 ++++++++++-------- tests/test_storage.py | 46 ++++++++++++++++++++-------------- 2 files changed, 40 insertions(+), 30 deletions(-) diff --git a/tests/test_linear_converter.py b/tests/test_linear_converter.py index e90d52f40..62e5cbcad 100644 --- a/tests/test_linear_converter.py +++ b/tests/test_linear_converter.py @@ -176,7 +176,7 @@ def test_linear_converter_with_on_off(self, basic_flow_system_linopy_coords, coo assert_conequal( model.constraints['Converter|on_hours_total'], model.variables['Converter|on_hours_total'] - == (model.variables['Converter|on'] * model.hours_per_step).sum(), + == (model.variables['Converter|on'] * model.hours_per_step).sum('time'), ) # Check conversion constraint @@ -282,16 +282,19 @@ def test_edge_case_time_varying_conversion(self, basic_flow_system_linopy_coords # Check that the correct constraint was created assert 'VariableConverter|conversion_0' in model.constraints + factor = converter.conversion_factors[0]['electricity'] + + assert factor.dims == tuple(model.get_coords()) + # Verify the constraint has the time-varying coefficient assert_conequal( model.constraints['VariableConverter|conversion_0'], - input_flow.submodel.flow_rate * fluctuating_cop == output_flow.submodel.flow_rate * 1.0, + input_flow.submodel.flow_rate * factor == output_flow.submodel.flow_rate * 1.0, ) def test_piecewise_conversion(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with PiecewiseConversion.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps # Create input and output flows input_flow = fx.Flow('input', bus='input_bus', size=100) @@ -339,9 +342,9 @@ def test_piecewise_conversion(self, basic_flow_system_linopy_coords, coords_conf lambda1 = model.variables[f'Converter|Piece_{i}|lambda1'] inside_piece = model.variables[f'Converter|Piece_{i}|inside_piece'] - assert_var_equal(inside_piece, model.add_variables(binary=True, coords=(timesteps,))) - assert_var_equal(lambda0, model.add_variables(lower=0, upper=1, coords=(timesteps,))) - assert_var_equal(lambda1, model.add_variables(lower=0, upper=1, coords=(timesteps,))) + assert_var_equal(inside_piece, model.add_variables(binary=True, coords=model.get_coords())) + assert_var_equal(lambda0, model.add_variables(lower=0, upper=1, coords=model.get_coords())) + assert_var_equal(lambda1, model.add_variables(lower=0, upper=1, coords=model.get_coords())) # Check that the inside_piece constraint exists assert f'Converter|Piece_{i}|inside_piece' in model.constraints @@ -380,7 +383,6 @@ def test_piecewise_conversion(self, basic_flow_system_linopy_coords, coords_conf def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy_coords, coords_config): """Test a LinearConverter with PiecewiseConversion and OnOffParameters.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps # Create input and output flows input_flow = fx.Flow('input', bus='input_bus', size=100) @@ -444,9 +446,9 @@ def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy_coords, lambda1 = model.variables[f'Converter|Piece_{i}|lambda1'] inside_piece = model.variables[f'Converter|Piece_{i}|inside_piece'] - assert_var_equal(inside_piece, model.add_variables(binary=True, coords=(timesteps,))) - assert_var_equal(lambda0, model.add_variables(lower=0, upper=1, coords=(timesteps,))) - assert_var_equal(lambda1, model.add_variables(lower=0, upper=1, coords=(timesteps,))) + assert_var_equal(inside_piece, model.add_variables(binary=True, coords=model.get_coords())) + assert_var_equal(lambda0, model.add_variables(lower=0, upper=1, coords=model.get_coords())) + assert_var_equal(lambda1, model.add_variables(lower=0, upper=1, coords=model.get_coords())) # Check that the inside_piece constraint exists assert f'Converter|Piece_{i}|inside_piece' in model.constraints @@ -485,7 +487,7 @@ def test_piecewise_conversion_with_onoff(self, basic_flow_system_linopy_coords, assert 'Converter|on_hours_total' in model.constraints assert_conequal( model.constraints['Converter|on_hours_total'], - model['Converter|on_hours_total'] == (model['Converter|on'] * model.hours_per_step).sum(), + model['Converter|on_hours_total'] == (model['Converter|on'] * model.hours_per_step).sum('time'), ) # Verify that the costs effect is applied diff --git a/tests/test_storage.py b/tests/test_storage.py index 479f66a87..76226be3b 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -14,8 +14,6 @@ class TestStorageModel: def test_basic_storage(self, basic_flow_system_linopy_coords, coords_config): """Test that basic storage model variables and constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps - timesteps_extra = flow_system.timesteps_extra # Create a simple storage storage = fx.Storage( @@ -55,13 +53,14 @@ def test_basic_storage(self, basic_flow_system_linopy_coords, coords_config): # Check variable properties assert_var_equal( - model['TestStorage(Q_th_in)|flow_rate'], model.add_variables(lower=0, upper=20, coords=(timesteps,)) + model['TestStorage(Q_th_in)|flow_rate'], model.add_variables(lower=0, upper=20, coords=model.get_coords()) ) assert_var_equal( - model['TestStorage(Q_th_out)|flow_rate'], model.add_variables(lower=0, upper=20, coords=(timesteps,)) + model['TestStorage(Q_th_out)|flow_rate'], model.add_variables(lower=0, upper=20, coords=model.get_coords()) ) assert_var_equal( - model['TestStorage|charge_state'], model.add_variables(lower=0, upper=30, coords=(timesteps_extra,)) + model['TestStorage|charge_state'], + model.add_variables(lower=0, upper=30, coords=model.get_coords(extra_timestep=True)), ) # Check constraint formulations @@ -88,8 +87,6 @@ def test_basic_storage(self, basic_flow_system_linopy_coords, coords_config): def test_lossy_storage(self, basic_flow_system_linopy_coords, coords_config): """Test that basic storage model variables and constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps - timesteps_extra = flow_system.timesteps_extra # Create a simple storage storage = fx.Storage( @@ -132,13 +129,14 @@ def test_lossy_storage(self, basic_flow_system_linopy_coords, coords_config): # Check variable properties assert_var_equal( - model['TestStorage(Q_th_in)|flow_rate'], model.add_variables(lower=0, upper=20, coords=(timesteps,)) + model['TestStorage(Q_th_in)|flow_rate'], model.add_variables(lower=0, upper=20, coords=model.get_coords()) ) assert_var_equal( - model['TestStorage(Q_th_out)|flow_rate'], model.add_variables(lower=0, upper=20, coords=(timesteps,)) + model['TestStorage(Q_th_out)|flow_rate'], model.add_variables(lower=0, upper=20, coords=model.get_coords()) ) assert_var_equal( - model['TestStorage|charge_state'], model.add_variables(lower=0, upper=30, coords=(timesteps_extra,)) + model['TestStorage|charge_state'], + model.add_variables(lower=0, upper=30, coords=model.get_coords(extra_timestep=True)), ) # Check constraint formulations @@ -173,8 +171,6 @@ def test_lossy_storage(self, basic_flow_system_linopy_coords, coords_config): def test_charge_state_bounds(self, basic_flow_system_linopy_coords, coords_config): """Test that basic storage model variables and constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps - timesteps_extra = flow_system.timesteps_extra # Create a simple storage storage = fx.Storage( @@ -216,17 +212,23 @@ def test_charge_state_bounds(self, basic_flow_system_linopy_coords, coords_confi # Check variable properties assert_var_equal( - model['TestStorage(Q_th_in)|flow_rate'], model.add_variables(lower=0, upper=20, coords=(timesteps,)) + model['TestStorage(Q_th_in)|flow_rate'], model.add_variables(lower=0, upper=20, coords=model.get_coords()) ) assert_var_equal( - model['TestStorage(Q_th_out)|flow_rate'], model.add_variables(lower=0, upper=20, coords=(timesteps,)) + model['TestStorage(Q_th_out)|flow_rate'], model.add_variables(lower=0, upper=20, coords=model.get_coords()) ) assert_var_equal( model['TestStorage|charge_state'], model.add_variables( - lower=np.array([0.07, 0.11, 0.15, 0.19, 0.23, 0.27, 0.31, 0.35, 0.39, 0.43, 0.43]) * 30, - upper=np.array([0.14, 0.22, 0.3, 0.38, 0.46, 0.54, 0.62, 0.7, 0.78, 0.86, 0.86]) * 30, - coords=(timesteps_extra,), + lower=storage.relative_minimum_charge_state.reindex( + time=model.get_coords(extra_timestep=True)['time'] + ).ffill('time') + * 30, + upper=storage.relative_maximum_charge_state.reindex( + time=model.get_coords(extra_timestep=True)['time'] + ).ffill('time') + * 30, + coords=model.get_coords(extra_timestep=True), ), ) @@ -286,8 +288,14 @@ def test_storage_with_investment(self, basic_flow_system_linopy_coords, coords_c assert con_name in model.constraints, f'Missing investment constraint: {con_name}' # Check variable properties - assert_var_equal(model['InvestStorage|size'], model.add_variables(lower=0, upper=100)) - assert_var_equal(model['InvestStorage|is_invested'], model.add_variables(binary=True)) + assert_var_equal( + model['InvestStorage|size'], + model.add_variables(lower=0, upper=100, coords=model.get_coords(['year', 'scenario'])), + ) + assert_var_equal( + model['InvestStorage|is_invested'], + model.add_variables(binary=True, coords=model.get_coords(['year', 'scenario'])), + ) assert_conequal( model.constraints['InvestStorage|size|ub'], model.variables['InvestStorage|size'] <= model.variables['InvestStorage|is_invested'] * 100, From 2a469f19f2d39f355d77f34571f5987cf30f5b10 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 28 Jul 2025 21:36:33 +0200 Subject: [PATCH 08/10] Fix test with multiple dims --- tests/test_component.py | 35 +++++++++++++++++++++-------------- 1 file changed, 21 insertions(+), 14 deletions(-) diff --git a/tests/test_component.py b/tests/test_component.py index 2ed9bea3c..384a5ed28 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -74,7 +74,7 @@ def test_component(self, basic_flow_system_linopy_coords, coords_config): def test_on_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps + ub_out2 = np.linspace(1, 1.5, 10).round(2) inputs = [ fx.Flow('In1', 'Fernwärme', relative_minimum=np.ones(10) * 0.1, size=100), @@ -132,12 +132,16 @@ def test_on_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_co msg='Incorrect constraints', ) + upper_bound_flow_rate = outputs[1].relative_maximum + + assert upper_bound_flow_rate.dims == tuple(model.get_coords()) + assert_var_equal( model['TestComponent(Out2)|flow_rate'], - model.add_variables(lower=0, upper=300 * ub_out2, coords=(timesteps,)), + model.add_variables(lower=0, upper=300 * upper_bound_flow_rate, coords=model.get_coords()), ) - assert_var_equal(model['TestComponent|on'], model.add_variables(binary=True, coords=(timesteps,))) - assert_var_equal(model['TestComponent(Out2)|on'], model.add_variables(binary=True, coords=(timesteps,))) + assert_var_equal(model['TestComponent|on'], model.add_variables(binary=True, coords=model.get_coords())) + assert_var_equal(model['TestComponent(Out2)|on'], model.add_variables(binary=True, coords=model.get_coords())) assert_conequal( model.constraints['TestComponent(Out2)|flow_rate|lb'], @@ -146,7 +150,7 @@ def test_on_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_co assert_conequal( model.constraints['TestComponent(Out2)|flow_rate|ub'], model.variables['TestComponent(Out2)|flow_rate'] - <= model.variables['TestComponent(Out2)|on'] * 300 * ub_out2, + <= model.variables['TestComponent(Out2)|on'] * 300 * upper_bound_flow_rate, ) assert_conequal( @@ -173,7 +177,6 @@ def test_on_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_co def test_on_with_single_flow(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps inputs = [ fx.Flow('In1', 'Fernwärme', relative_minimum=np.ones(10) * 0.1, size=100), ] @@ -211,10 +214,10 @@ def test_on_with_single_flow(self, basic_flow_system_linopy_coords, coords_confi ) assert_var_equal( - model['TestComponent(In1)|flow_rate'], model.add_variables(lower=0, upper=100, coords=(timesteps,)) + model['TestComponent(In1)|flow_rate'], model.add_variables(lower=0, upper=100, coords=model.get_coords()) ) - assert_var_equal(model['TestComponent|on'], model.add_variables(binary=True, coords=(timesteps,))) - assert_var_equal(model['TestComponent(In1)|on'], model.add_variables(binary=True, coords=(timesteps,))) + assert_var_equal(model['TestComponent|on'], model.add_variables(binary=True, coords=model.get_coords())) + assert_var_equal(model['TestComponent(In1)|on'], model.add_variables(binary=True, coords=model.get_coords())) assert_conequal( model.constraints['TestComponent(In1)|flow_rate|lb'], @@ -233,7 +236,7 @@ def test_on_with_single_flow(self, basic_flow_system_linopy_coords, coords_confi def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy_coords, coords_config): """Test that flow model constraints are correctly generated.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config - timesteps = flow_system.timesteps + ub_out2 = np.linspace(1, 1.5, 10).round(2) inputs = [ fx.Flow( @@ -304,12 +307,16 @@ def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy_coor msg='Incorrect constraints', ) + upper_bound_flow_rate = outputs[1].relative_maximum + + assert upper_bound_flow_rate.dims == tuple(model.get_coords()) + assert_var_equal( model['TestComponent(Out2)|flow_rate'], - model.add_variables(lower=0, upper=300 * ub_out2, coords=(timesteps,)), + model.add_variables(lower=0, upper=300 * upper_bound_flow_rate, coords=model.get_coords()), ) - assert_var_equal(model['TestComponent|on'], model.add_variables(binary=True, coords=(timesteps,))) - assert_var_equal(model['TestComponent(Out2)|on'], model.add_variables(binary=True, coords=(timesteps,))) + assert_var_equal(model['TestComponent|on'], model.add_variables(binary=True, coords=model.get_coords())) + assert_var_equal(model['TestComponent(Out2)|on'], model.add_variables(binary=True, coords=model.get_coords())) assert_conequal( model.constraints['TestComponent(Out2)|flow_rate|lb'], @@ -318,7 +325,7 @@ def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy_coor assert_conequal( model.constraints['TestComponent(Out2)|flow_rate|ub'], model.variables['TestComponent(Out2)|flow_rate'] - <= model.variables['TestComponent(Out2)|on'] * 300 * ub_out2, + <= model.variables['TestComponent(Out2)|on'] * 300 * upper_bound_flow_rate, ) assert_conequal( From 159bcb3c11a6fd5165b0820a317f1d75032bd7c8 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 28 Jul 2025 21:36:53 +0200 Subject: [PATCH 09/10] New test --- tests/test_component.py | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/tests/test_component.py b/tests/test_component.py index 384a5ed28..1d1792a65 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -349,6 +349,47 @@ def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy_coor + 1e-5, ) + def test_previous_states_with_multiple_flows_2(self, basic_flow_system_linopy_coords, coords_config): + """Test that flow model constraints are correctly generated.""" + flow_system, coords_config = basic_flow_system_linopy_coords, coords_config + + ub_out2 = np.linspace(1, 1.5, 10).round(2) + inputs = [ + fx.Flow( + 'In1', + 'Fernwärme', + relative_minimum=np.ones(10) * 0.1, + size=100, + previous_flow_rate=np.array([0, 0, 1e-6, 1e-5, 1e-4, 3, 4]), + on_off_parameters=fx.OnOffParameters(consecutive_on_hours_min=3), + ), + ] + outputs = [ + fx.Flow('Out1', 'Gas', relative_minimum=np.ones(10) * 0.2, size=200, previous_flow_rate=[3, 4, 5]), + fx.Flow( + 'Out2', + 'Gas', + relative_minimum=np.ones(10) * 0.3, + relative_maximum=ub_out2, + size=300, + previous_flow_rate=20, + ), + ] + comp = flixopt.elements.Component( + 'TestComponent', + inputs=inputs, + outputs=outputs, + on_off_parameters=fx.OnOffParameters(consecutive_on_hours_min=3), + ) + flow_system.add_elements(comp) + create_linopy_model(flow_system) + + assert_conequal( + comp.submodel.constraints['TestComponent|consecutive_on_hours|initial'], + comp.submodel.variables['TestComponent|consecutive_on_hours'].isel(time=0) + == comp.submodel.variables['TestComponent|on'].isel(time=0) * 5, + ) + class TestTransmissionModel: def test_transmission_basic(self, basic_flow_system, highs_solver): From e764a1392113cce01f0726d7bc76be88c6c4fdd9 Mon Sep 17 00:00:00 2001 From: FBumann <117816358+FBumann@users.noreply.github.com> Date: Mon, 28 Jul 2025 21:45:33 +0200 Subject: [PATCH 10/10] New test for previous flow_rates --- flixopt/elements.py | 4 +++- tests/test_component.py | 32 ++++++++++++++++++++++++++------ 2 files changed, 29 insertions(+), 7 deletions(-) diff --git a/flixopt/elements.py b/flixopt/elements.py index 499ab66db..e1c0fcbc3 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -199,6 +199,7 @@ def __init__( If the load-profile is just an upper limit, use relative_maximum instead. previous_flow_rate: previous flow rate of the flow. Used to determine if and how long the flow is already on / off. If None, the flow is considered to be off for one timestep. + Currently does not support different values in different years or scenarios! meta_data: used to store more information about the Element. Is not used internally, but saved in the results. Only use python native types. """ super().__init__(label, meta_data=meta_data) @@ -305,7 +306,8 @@ def _plausibility_checks(self) -> None: ] ): raise TypeError( - f'previous_flow_rate must be None, a scalar, a list of scalars or a 1D-numpy-array. Got {type(self.previous_flow_rate)}' + f'previous_flow_rate must be None, a scalar, a list of scalars or a 1D-numpy-array. Got {type(self.previous_flow_rate)}.' + f'Different values in different years or scenarios are not yetsupported.' ) @property diff --git a/tests/test_component.py b/tests/test_component.py index 1d1792a65..ab05db5e8 100644 --- a/tests/test_component.py +++ b/tests/test_component.py @@ -349,8 +349,26 @@ def test_previous_states_with_multiple_flows(self, basic_flow_system_linopy_coor + 1e-5, ) - def test_previous_states_with_multiple_flows_2(self, basic_flow_system_linopy_coords, coords_config): - """Test that flow model constraints are correctly generated.""" + @pytest.mark.parametrize( + 'in1_previous_flow_rate, out1_previous_flow_rate, out2_previous_flow_rate, previous_on_hours', + [ + (None, None, None, 0), + (np.array([0, 1e-6, 1e-4, 5]), None, None, 2), + (np.array([0, 5, 0, 5]), None, None, 1), + (np.array([0, 5, 0, 0]), 3, 0, 1), + (np.array([0, 0, 2, 0, 4, 5]), [3, 4, 5], None, 4), + ], + ) + def test_previous_states_with_multiple_flows_parameterized( + self, + basic_flow_system_linopy_coords, + coords_config, + in1_previous_flow_rate, + out1_previous_flow_rate, + out2_previous_flow_rate, + previous_on_hours, + ): + """Test that flow model constraints are correctly generated with different previous flow rates and constraint factors.""" flow_system, coords_config = basic_flow_system_linopy_coords, coords_config ub_out2 = np.linspace(1, 1.5, 10).round(2) @@ -360,19 +378,21 @@ def test_previous_states_with_multiple_flows_2(self, basic_flow_system_linopy_co 'Fernwärme', relative_minimum=np.ones(10) * 0.1, size=100, - previous_flow_rate=np.array([0, 0, 1e-6, 1e-5, 1e-4, 3, 4]), + previous_flow_rate=in1_previous_flow_rate, on_off_parameters=fx.OnOffParameters(consecutive_on_hours_min=3), ), ] outputs = [ - fx.Flow('Out1', 'Gas', relative_minimum=np.ones(10) * 0.2, size=200, previous_flow_rate=[3, 4, 5]), + fx.Flow( + 'Out1', 'Gas', relative_minimum=np.ones(10) * 0.2, size=200, previous_flow_rate=out1_previous_flow_rate + ), fx.Flow( 'Out2', 'Gas', relative_minimum=np.ones(10) * 0.3, relative_maximum=ub_out2, size=300, - previous_flow_rate=20, + previous_flow_rate=out2_previous_flow_rate, ), ] comp = flixopt.elements.Component( @@ -387,7 +407,7 @@ def test_previous_states_with_multiple_flows_2(self, basic_flow_system_linopy_co assert_conequal( comp.submodel.constraints['TestComponent|consecutive_on_hours|initial'], comp.submodel.variables['TestComponent|consecutive_on_hours'].isel(time=0) - == comp.submodel.variables['TestComponent|on'].isel(time=0) * 5, + == comp.submodel.variables['TestComponent|on'].isel(time=0) * (previous_on_hours + 1), )