diff --git a/src/gsy_e/constants.py b/src/gsy_e/constants.py index 523e461ae..c1d16681b 100644 --- a/src/gsy_e/constants.py +++ b/src/gsy_e/constants.py @@ -107,3 +107,20 @@ class HeatPumpSettingsDefaultParameters: # Set the precision of the decimal numbers used in the simulation. getcontext().prec = 12 + + +class SorTesConfiguration: + """Collection of SorTes heat tank configuration parameters.""" + + MINUTES_BEFORE_SWITCH_ALLOWED = 1 * 60 + MINUTES_TIME_HORIZONT_LOW_RATES = 1 * 60 + PREFERRED_BUYING_RATE = 20 + MIN_SOC_TOLERANCE = 10 + MAX_SOC_TOLERANCE = 90 + CAPACITY_KWH = 25 + COP_HEAT_SOURCE = 1 + COP_CONDENSER = 35 # nominal value for SHSL-D1-005-1x350 + COP_EVAPORATOR = 1 + CONVERSION_CHARGE_CONDENSER_POWER = 1 / 1.2 + CONVERSION_DISCHARGE_EVAPORATOR_POWER = 1 / 1.2 + AMBIENT_TEMPERATURE_CORRECTION = 5 diff --git a/src/gsy_e/gsy_e_core/sim_results/file_export_endpoints.py b/src/gsy_e/gsy_e_core/sim_results/file_export_endpoints.py index 00120f645..718cf5f52 100644 --- a/src/gsy_e/gsy_e_core/sim_results/file_export_endpoints.py +++ b/src/gsy_e/gsy_e_core/sim_results/file_export_endpoints.py @@ -16,7 +16,7 @@ along with this program. If not, see . """ -# pylint: disable=too-many-return-statements, broad-exception-raised +# pylint: disable=too-many-return-statements, broad-exception-raised, too-many-locals from abc import ABC, abstractmethod from typing import Dict, List @@ -33,6 +33,7 @@ MultipleTankHeatPumpStrategy, HeatPumpStrategyWithoutTanks, ) +from gsy_e.models.strategy.heatpump_with_sortes_tank import HeatPumpWithSorTesTankStrategy from gsy_e.models.strategy.load_hours import LoadHoursStrategy from gsy_e.models.strategy.pv import PVStrategy from gsy_e.models.strategy.storage import StorageStrategy @@ -50,6 +51,14 @@ def is_heatpump_strategy_without_tanks(area: Area): ) +def is_heatpump_strategy_with_sortes_tank(area: Area): + """Return if area has a heat pump strategy with Sortes tanks connected.""" + return isinstance( + area.strategy, + HeatPumpWithSorTesTankStrategy, + ) + + def is_heatpump_strategy_with_tanks(area: Area): """Return if area has a heat pump strategy.""" return isinstance( @@ -407,6 +416,49 @@ def _row(self, slot, market): return rows +class HeatPumpWithSortesTankDataExporter(BaseDataExporter): + """Data exporter dedicated to heat pump areas""" + + def __init__(self, area, past_markets): + assert is_heatpump_strategy_with_sortes_tank(area) + self.area = area + self.past_markets = past_markets + + @property + def labels(self) -> List: + return [ + "slot", + "energy traded [kWh]", + "COP", + "heat demand [kJ]", + "SOC", + "auxiliary_energy_kWh", + "energy_used_for_dis_charging_kWh", + ] + + @property + def rows(self) -> List: + return [self._row(market.time_slot, market) for market in self.past_markets] + + def _traded(self, market): + return ( + market.traded_energy[self.area.name] if self.area.name in market.traded_energy else 0 + ) + + def _row(self, slot, market): + rows = [slot] + hp_stats = self.area.strategy.state.get_results_dict(slot) + rows += [ + round(self._traded(market), ROUND_TOLERANCE_EXPORT), + round(hp_stats["cop"], ROUND_TOLERANCE_EXPORT), + round(hp_stats["heat_demand_kJ"], ROUND_TOLERANCE_EXPORT), + round(hp_stats["soc"], ROUND_TOLERANCE_EXPORT), + round(hp_stats["auxiliary_energy_kWh"], 4), + round(hp_stats["energy_used_for_dis_charging_kWh"], 4), + ] + return rows + + class FileExportEndpoints: """Handle data preparation for csv-file and plot export.""" @@ -439,6 +491,8 @@ def export_data_factory( return HeatPumpDataExporter(area, area.parent.past_markets) if is_heatpump_strategy_without_tanks(area): return HeatPumpWithoutTanksDataExporter(area, area.parent.past_markets) + if is_heatpump_strategy_with_sortes_tank(area): + return HeatPumpWithSortesTankDataExporter(area, area.parent.past_markets) return LeafDataExporter(area, area.parent.past_markets) if past_market_type == AvailableMarketTypes.BALANCING: return BalancingDataExporter(area.past_balancing_markets) diff --git a/src/gsy_e/models/strategy/energy_parameters/heatpump/heat_pump.py b/src/gsy_e/models/strategy/energy_parameters/heatpump/heat_pump.py index 2ad277771..1ecd07c57 100644 --- a/src/gsy_e/models/strategy/energy_parameters/heatpump/heat_pump.py +++ b/src/gsy_e/models/strategy/energy_parameters/heatpump/heat_pump.py @@ -383,7 +383,8 @@ def combined_state(self) -> CombinedHeatpumpTanksState: """Combined heatpump and tanks state.""" return self._state - def last_time_slot(self, current_market_slot: DateTime) -> DateTime: + @staticmethod + def last_time_slot(current_market_slot: DateTime) -> DateTime: """Calculate the previous time slot from the current one.""" return current_market_slot - GlobalConfig.slot_length diff --git a/src/gsy_e/models/strategy/heat_pump.py b/src/gsy_e/models/strategy/heat_pump.py index 65295fb20..0193a55c4 100644 --- a/src/gsy_e/models/strategy/heat_pump.py +++ b/src/gsy_e/models/strategy/heat_pump.py @@ -2,8 +2,7 @@ from typing import Dict, TYPE_CHECKING, Optional, Union, List from decimal import Decimal -from gsy_framework.constants_limits import ConstSettings -from gsy_framework.constants_limits import FLOATING_POINT_TOLERANCE +from gsy_framework.constants_limits import ConstSettings, FLOATING_POINT_TOLERANCE from gsy_framework.data_classes import Trade, TraderDetails from gsy_framework.enums import AvailableMarketTypes from gsy_framework.exceptions import GSyException @@ -272,8 +271,7 @@ def __init__( ) def _init_price_params(self, order_updater_parameters, preferred_buying_rate): - self.use_default_updater_params: bool = not order_updater_parameters - if self.use_default_updater_params: + if not order_updater_parameters: order_updater_parameters = { AvailableMarketTypes.SPOT: HeatPumpOrderUpdaterParameters() } @@ -423,8 +421,7 @@ def serialize(self): } def _init_price_params(self, order_updater_parameters): - self.use_default_updater_params: bool = not order_updater_parameters - if self.use_default_updater_params: + if not order_updater_parameters: order_updater_parameters = { AvailableMarketTypes.SPOT: HeatPumpOrderUpdaterParameters() } diff --git a/src/gsy_e/models/strategy/heat_pump_soc_management.py b/src/gsy_e/models/strategy/heat_pump_soc_management.py index 4f96d7b26..0f1689bcc 100644 --- a/src/gsy_e/models/strategy/heat_pump_soc_management.py +++ b/src/gsy_e/models/strategy/heat_pump_soc_management.py @@ -73,7 +73,7 @@ def event_activate(self): else: assert False, "GlobalConfig.market_maker_rate was not initiated yet." - def calculate(self, time_slot: DateTime, _buy_rate: float = 0.0): + def calculate(self, time_slot: DateTime, buy_rate: float = 0.0): """ Calculate the bid energy depending on the current state of the heat pump, the current SOC and whether the market maker rate is cheap or expensive compared to the average market @@ -83,13 +83,10 @@ def calculate(self, time_slot: DateTime, _buy_rate: float = 0.0): if not self._is_time_for_state_change(time_slot): # If state change is not possible, but at the same time the soc is below the min or # above the max SOC value of the tank, then maintain the SOC. - if self._charger.get_average_soc(time_slot) >= self.MAX_SOC_TOLERANCE: - target_state = HeatPumpChargingState.MAINTAIN_SOC - if self._charger.get_average_soc(time_slot) <= self.MIN_SOC_TOLERANCE: - target_state = HeatPumpChargingState.MAINTAIN_SOC + target_state = self._handle_state_at_soc_limits(time_slot, target_state) else: # If the state change is possible, check the market maker rate to set the new state - target_state = self._should_charge_or_discharge(time_slot) + target_state = self._should_charge_or_discharge(time_slot, buy_rate) if self._current_state != target_state: # If the target state is the same as the current state, do nothing. Otherwise, # update the current state and the last switch timestamp. @@ -98,13 +95,32 @@ def calculate(self, time_slot: DateTime, _buy_rate: float = 0.0): return self._get_energy_from_target_state(target_state, time_slot) + def _handle_state_at_soc_limits( + self, time_slot: DateTime, current_state: HeatPumpChargingState + ) -> HeatPumpChargingState: + # If state change is not possible, but at the same time the soc is below the min or + # above the max SOC value of the tank, then maintain the SOC. + target_state = current_state + if self._get_tank_soc(time_slot) >= self.MAX_SOC_TOLERANCE: + target_state = HeatPumpChargingState.MAINTAIN_SOC + if self._get_tank_soc(time_slot) <= self.MIN_SOC_TOLERANCE: + target_state = HeatPumpChargingState.MAINTAIN_SOC + return target_state + + def _get_tank_soc(self, time_slot: DateTime) -> float: + return self._charger.get_average_soc(time_slot) + def _is_time_for_state_change(self, time_slot: DateTime) -> bool: if not self._last_switch: # If the last_switch has not been set yet, the simulation is starting and no switch has # occurred yet. Change state if needed. self._last_switch = time_slot return True - if time_slot - self._last_switch < duration(minutes=self.MINUTES_BEFORE_SWITCH_ALLOWED): + if ( + self._current_state != HeatPumpChargingState.MAINTAIN_SOC + and time_slot - self._last_switch + < duration(minutes=self.MINUTES_BEFORE_SWITCH_ALLOWED) + ): # If not enough time has passed since the last switch, do not allow state change. return False # Otherwise, allow the state change. @@ -119,20 +135,26 @@ def _get_energy_from_target_state( return self._energy_params.get_min_energy_demand_kWh(time_slot) return self._energy_params.get_energy_demand_kWh(time_slot) - def _should_charge_or_discharge(self, time_slot: DateTime) -> HeatPumpChargingState: - if GlobalConfig.market_maker_rate[time_slot] <= self._average_rate: + def _should_charge_or_discharge( + self, time_slot: DateTime, buy_rate: float + ) -> HeatPumpChargingState: + + if self._is_energy_affordable(time_slot, buy_rate): # If the market maker rate is lower than the average rate, charge except if the SOC is # too high. - if self._charger.get_average_soc(time_slot) >= self.MAX_SOC_TOLERANCE: + if self._get_tank_soc(time_slot) >= self.MAX_SOC_TOLERANCE: return HeatPumpChargingState.MAINTAIN_SOC return HeatPumpChargingState.CHARGE # If the market maker rate is higher than the average rate, discharge except if the SOC # is too low. - if self._charger.get_average_soc(time_slot) > self.MIN_SOC_TOLERANCE: + if self._get_tank_soc(time_slot) > self.MIN_SOC_TOLERANCE: return HeatPumpChargingState.DISCHARGE return HeatPumpChargingState.MAINTAIN_SOC + def _is_energy_affordable(self, time_slot: DateTime, _buy_rate: float) -> bool: + return GlobalConfig.market_maker_rate[time_slot] <= self._average_rate + def heat_pump_soc_management_factory( energy_params: "HeatPumpEnergyParameters", preferred_buying_rate: float diff --git a/src/gsy_e/models/strategy/heatpump_with_sortes_tank.py b/src/gsy_e/models/strategy/heatpump_with_sortes_tank.py new file mode 100644 index 000000000..c731079bf --- /dev/null +++ b/src/gsy_e/models/strategy/heatpump_with_sortes_tank.py @@ -0,0 +1,621 @@ +from decimal import Decimal +from typing import TYPE_CHECKING, Optional, Union +from logging import getLogger +from math import isclose + +import numpy as np +from gsy_framework.constants_limits import GlobalConfig, ConstSettings, FLOATING_POINT_TOLERANCE +from gsy_framework.enums import AvailableMarketTypes, HeatPumpSourceType +from gsy_framework.read_user_profile import InputProfileTypes +from gsy_framework.utils import ( + convert_kJ_to_kWh, + convert_pendulum_to_str_in_dict, + convert_str_to_pendulum_in_dict, + convert_kW_to_kWh, +) +from pendulum import DateTime + +from gsy_e.constants import SorTesConfiguration, RETAIN_PAST_MARKET_STRATEGIES_STATE, DEFAULT_COP +from gsy_e.models.strategy.energy_parameters.heatpump.cop_models import ( + COPModelType, + cop_model_factory, +) +from gsy_e.models.strategy.heat_pump import HeatPumpOrderUpdaterParameters, HeatPumpStrategyBase +from gsy_e.models.strategy.heat_pump_soc_management import ( + MinimiseHeatpumpSwitchStrategy, + HeatPumpChargingState, +) +from gsy_e.models.strategy.state.heatpump_state import ( + HeatPumpStateBase, +) +from gsy_e.models.strategy.strategy_profile import profile_factory, StrategyProfileBase + +if TYPE_CHECKING: + from gsy_e.models.market import MarketBase + +log = getLogger(__name__) + + +class SorTesPerformanceMaps: + """Calculate the charging and discharging performance of the SorTES tank""" + + CHARGING_POWER_MAP = { + 5: 3.8, + 10: 3.5, + 15: 3.3, + 20: 3.0, + 25: 2.6, + 30: 2.2, + 35: 1.8, + 40: 1.5, + 45: 0.5, + } # ambient_temp [K]: power [kW] + DISCHARGING_POWER_MAP = { + 5: 3.0, + 10: 3.5, + 15: 4, + 20: 4.3, + 25: 4.8, + } # ambient_temp [K]: power [kW] + + @classmethod + def get_power_charging(cls, temperature: float) -> float: + """Return the charging power.""" + return cls._get_power(temperature, cls.CHARGING_POWER_MAP) + + @classmethod + def get_power_discharging(cls, temperature: float) -> float: + """Return the discharging power.""" + return cls._get_power(temperature, cls.DISCHARGING_POWER_MAP) + + @classmethod + def _get_power(cls, temperature: float, power_table: dict) -> float: + temps = np.array(sorted(power_table.keys()), dtype=float) + powers = np.array([power_table[t] for t in sorted(power_table.keys())], dtype=float) + + return float( + np.interp(temperature, temps, powers, left=None, right=None) + if temps[0] <= temperature <= temps[-1] + else cls._extrapolate(temperature, temps, powers) + ) + + @staticmethod + def _extrapolate(temperature: float, temps: np.ndarray, powers: np.ndarray) -> float: + """Linear extrapolation using the two nearest edge points.""" + if temperature < temps[0]: + # Use the first two points for left extrapolation + slope = (powers[1] - powers[0]) / (temps[1] - temps[0]) + return powers[0] + slope * (temperature - temps[0]) + # Use the last two points for right extrapolation + slope = (powers[-1] - powers[-2]) / (temps[-1] - temps[-2]) + return powers[-1] + slope * (temperature - temps[-1]) + + +class SorTesTankMinimiseSwitchStrategy(MinimiseHeatpumpSwitchStrategy): + """Minimise number of switches between charging and discharging of the SorTES tank""" + + MINUTES_BEFORE_SWITCH_ALLOWED = SorTesConfiguration.MINUTES_BEFORE_SWITCH_ALLOWED + MIN_SOC_TOLERANCE = SorTesConfiguration.MIN_SOC_TOLERANCE + MAX_SOC_TOLERANCE = SorTesConfiguration.MAX_SOC_TOLERANCE + + # pylint: disable=super-init-not-called + def __init__( + self, + energy_params: "SorTesTankEnergyParameters", + average_trade_rate: Union[str, float, dict], + ): + self._last_switch: Optional[DateTime] = None + self._energy_params = energy_params + self._current_state = HeatPumpChargingState.MAINTAIN_SOC + self._average_trade_rate = profile_factory( + average_trade_rate, None, profile_type=InputProfileTypes.IDENTITY + ) + + @property + def current_state(self) -> HeatPumpChargingState: + """Return the current charging state of the tank.""" + return self._current_state + + def _get_tank_soc(self, time_slot: DateTime): + return self._energy_params.get_soc(time_slot) + + def _is_energy_affordable(self, time_slot: DateTime, _buy_rate: float) -> bool: + rates_in_time_horizont = [ + value + for ts, value in self._average_trade_rate.profile.items() + if time_slot + <= ts + < time_slot.add(minutes=SorTesConfiguration.MINUTES_TIME_HORIZONT_LOW_RATES) + ] + return all( + value < SorTesConfiguration.PREFERRED_BUYING_RATE for value in rates_in_time_horizont + ) + + def event_activate(self): + """Perform commands on event activate.""" + self._average_trade_rate.read_or_rotate_profiles() + + def event_market_slot(self): + """Perform commands on event market cycle.""" + self._average_trade_rate.read_or_rotate_profiles() + + def _handle_state_at_soc_limits( + self, time_slot: DateTime, current_state: HeatPumpChargingState + ) -> HeatPumpChargingState: + target_state = current_state + if ( + self._get_tank_soc(time_slot) >= self.MAX_SOC_TOLERANCE + and self._current_state == HeatPumpChargingState.CHARGE + ): + target_state = HeatPumpChargingState.MAINTAIN_SOC + if ( + self._get_tank_soc(time_slot) <= self.MIN_SOC_TOLERANCE + and self._current_state == HeatPumpChargingState.DISCHARGE + ): + target_state = HeatPumpChargingState.MAINTAIN_SOC + return target_state + + +class SorTesTankState(HeatPumpStateBase): + """State class of Sortes tank state.""" + + # pylint: disable=too-many-instance-attributes, super-init-not-called + + def __init__(self): + self._soc: dict[DateTime, float] = {} + self._cop: dict[DateTime, float] = {} # in percent + self._energy_demand_kWh: dict[DateTime, float] = {} # electricity + self._heat_demand_kJ: dict[DateTime, float] = {} + self._auxiliary_energy_kWh: dict[DateTime, float] = {} + self._energy_used_for_dis_charging_kWh: dict[DateTime, float] = {} + self._min_energy_demand_kWh: dict[DateTime, float] = {} + self._max_energy_demand_kWh: dict[DateTime, float] = {} + self._total_traded_energy_kWh: float = 0 # for KPI calculation + + def activate(self): + """Perform commands on event activate.""" + self._soc[GlobalConfig.start_date] = SorTesConfiguration.MIN_SOC_TOLERANCE + self._cop[GlobalConfig.start_date] = DEFAULT_COP + + def set_auxiliary_energy_kWh(self, time_slot: DateTime, energy_kWh): + """Set auxiliary energy.""" + self._auxiliary_energy_kWh[time_slot] = energy_kWh + + def _get_auxiliary_energy_kWh(self, time_slot: DateTime) -> float: + return self._auxiliary_energy_kWh.get(time_slot, 0) + + def update_energy_used_for_dis_charging_kWh(self, time_slot: DateTime, energy_kWh: float): + """Update the charged or discharged energy.""" + self._energy_used_for_dis_charging_kWh[time_slot] = energy_kWh + + def _get_energy_used_for_dis_charging_kWh(self, time_slot: DateTime) -> float: + """Get the charged or discharged energy.""" + return self._energy_used_for_dis_charging_kWh.get(time_slot, 0) + + def get_soc(self, time_slot: DateTime) -> float: + """Return the soc value for the given time slot.""" + return self._soc.get(time_slot, 0) + + def set_soc(self, time_slot: DateTime, soc: float): + """Set soc value for the given time slot.""" + self._soc[time_slot] = soc + + def get_min_energy_demand_kWh(self, time_slot: DateTime) -> float: + """Return the minimal energy demanded for a given time slot.""" + return self._min_energy_demand_kWh.get(time_slot, 0) + + def get_max_energy_demand_kWh(self, time_slot: DateTime) -> float: + """Return the maximal energy demanded for a given time slot.""" + return self._max_energy_demand_kWh.get(time_slot, 0) + + def set_min_energy_demand_kWh(self, time_slot: DateTime, energy_kWh: float): + """Set the minimal energy demanded for a given time slot.""" + self._min_energy_demand_kWh[time_slot] = energy_kWh + + def set_max_energy_demand_kWh(self, time_slot: DateTime, energy_kWh: float): + """Set the maximal energy demanded for a given time slot.""" + self._max_energy_demand_kWh[time_slot] = energy_kWh + + def delete_past_state_values(self, current_time_slot: Optional[DateTime] = None): + """Delete past state values.""" + if not current_time_slot or RETAIN_PAST_MARKET_STRATEGIES_STATE: + return + last_time_slot = self._last_time_slot(current_time_slot) + self._delete_time_slots(self._cop, last_time_slot) + self._delete_time_slots(self._soc, last_time_slot) + self._delete_time_slots(self._energy_demand_kWh, last_time_slot) + self._delete_time_slots(self._min_energy_demand_kWh, last_time_slot) + self._delete_time_slots(self._max_energy_demand_kWh, last_time_slot) + self._delete_time_slots(self._heat_demand_kJ, last_time_slot) + + def increase_total_traded_energy_kWh(self, energy_kWh: float): + """Add to the total traded energy of the heatpump for a given time slot.""" + self._total_traded_energy_kWh += energy_kWh + + def get_state(self) -> dict: + """Return the state.""" + return { + "soc": convert_pendulum_to_str_in_dict(self._soc), + "cop": convert_pendulum_to_str_in_dict(self._cop), + "energy_demand_kWh": convert_pendulum_to_str_in_dict(self._energy_demand_kWh), + "min_energy_demand_kWh": convert_pendulum_to_str_in_dict(self._min_energy_demand_kWh), + "max_energy_demand_kWh": convert_pendulum_to_str_in_dict(self._max_energy_demand_kWh), + "total_traded_energy_kWh": self._total_traded_energy_kWh, + } + + def restore_state(self, state_dict: dict): + """Restore the state.""" + self._soc = convert_str_to_pendulum_in_dict(state_dict["soc"]) + self._cop = convert_str_to_pendulum_in_dict(state_dict["cop"]) + self._energy_demand_kWh = convert_str_to_pendulum_in_dict(state_dict["energy_demand_kWh"]) + self._min_energy_demand_kWh = convert_str_to_pendulum_in_dict( + state_dict["min_energy_demand_kWh"] + ) + self._max_energy_demand_kWh = convert_str_to_pendulum_in_dict( + state_dict["max_energy_demand_kWh"] + ) + self._total_traded_energy_kWh = state_dict["total_traded_energy_kWh"] + + def get_results_dict(self, current_time_slot: DateTime) -> dict: + """Return the results of the given time slot.""" + return { + "cop": self.get_cop(current_time_slot), + "total_traded_energy_kWh": self._total_traded_energy_kWh, + "heat_demand_kJ": self.get_heat_demand_kJ(current_time_slot), + "soc": self.get_soc(current_time_slot), + "energy_used_for_dis_charging_kWh": self._get_energy_used_for_dis_charging_kWh( + current_time_slot + ), + "auxiliary_energy_kWh": self._get_auxiliary_energy_kWh(current_time_slot), + } + + +class SorTesTankEnergyParameters: + """Energy Parameters for the SorTes Tank heat pump""" + + # pylint: disable=too-many-instance-attributes + + def __init__( + self, + heat_demand_Q_profile: Union[str, float, dict], + ambient_temp_C_profile: Union[str, float, dict], + target_temp_C_profile: Union[str, float, dict], + average_trade_rate: Union[str, float, dict], + source_type: HeatPumpSourceType = ConstSettings.HeatPumpSettings.SOURCE_TYPE, + ): + # pylint: disable=too-many-arguments, too-many-positional-arguments + self._state = SorTesTankState() + self._source_type = source_type + self._heat_demand_Q_J: StrategyProfileBase = profile_factory( + heat_demand_Q_profile, None, profile_type=InputProfileTypes.IDENTITY + ) + self._ambient_temp_C: StrategyProfileBase = profile_factory( + ambient_temp_C_profile, None, profile_type=InputProfileTypes.IDENTITY + ) + self._target_temp_C: StrategyProfileBase = profile_factory( + target_temp_C_profile, None, profile_type=InputProfileTypes.IDENTITY + ) + self._capacity_kWh: float = SorTesConfiguration.CAPACITY_KWH + self._cop_model = cop_model_factory(COPModelType.UNIVERSAL, source_type) + self._bought_energy_kWh = 0.0 + + self._soc_management = SorTesTankMinimiseSwitchStrategy(self, average_trade_rate) + + @property + def state(self) -> SorTesTankState: + """Return the state.""" + return self._state + + def serialize(self): + """Return dict with the current energy parameter values.""" + return { + "heat_demand_Q_J": self._heat_demand_Q_J.input_profile, + "ambient_temp_C": self._ambient_temp_C.input_profile, + "target_temp_C": self._target_temp_C.input_profile, + "source_type": self._source_type, + } + + @property + def soc_management(self) -> SorTesTankMinimiseSwitchStrategy: + """Return the soc management.""" + return self._soc_management + + def event_market_cycle(self, current_time_slot: DateTime): + """Runs on market_cycle event.""" + # Order matters here + self._soc_management.event_market_slot() + self._rotate_profiles(current_time_slot) + self._populate_state(current_time_slot) + + def event_activate(self): + """Runs on activate event.""" + self._soc_management.event_activate() + self._rotate_profiles() + self._state.activate() + + def event_traded_energy(self, time_slot: DateTime, energy_kWh: float): + """React to an event_traded_energy.""" + self._bought_energy_kWh += energy_kWh + self._decrement_posted_energy(time_slot, energy_kWh) + + def get_soc(self, time_slot: DateTime): + """Return the soc of the SorTes tank""" + return self._state.get_soc(time_slot) + + def get_energy_demand_kWh(self, time_slot: DateTime): + """Return the energy_demand kWh.""" + return self._state.get_energy_demand_kWh(time_slot) + + def get_min_energy_demand_kWh(self, time_slot: DateTime): + """Return the min_energy_demand kWh.""" + return self._state.get_min_energy_demand_kWh(time_slot) + + def get_max_energy_demand_kWh(self, time_slot: DateTime): + """Return the max_energy_demand kWh.""" + return self._state.get_max_energy_demand_kWh(time_slot) + + def _populate_state(self, time_slot: DateTime): + # order matters! + self._update_last_time_slot_data(time_slot) + self._calc_and_set_cop(time_slot) + self._state.set_heat_demand_kJ( + time_slot, self._heat_demand_Q_J.get_value(time_slot) / 1000.0 + ) + self._calc_and_set_energy_demand(time_slot) + + def _calc_and_set_cop(self, time_slot: DateTime): + cop = self._cop_model.calc_cop( + source_temp_C=self._ambient_temp_C.get_value(time_slot), + condenser_temp_C=self._target_temp_C.get_value(time_slot), + ) + self._state.set_cop(time_slot, cop) + + def _decrement_posted_energy(self, time_slot: DateTime, energy_kWh: float): + updated_energy_demand_kWh = max(0.0, self.get_energy_demand_kWh(time_slot) - energy_kWh) + updated_min_energy_demand_kWh = max( + 0.0, self.get_min_energy_demand_kWh(time_slot) - energy_kWh + ) + updated_max_energy_demand_kWh = max( + 0.0, self.get_max_energy_demand_kWh(time_slot) - energy_kWh + ) + self._state.set_energy_demand_kWh(time_slot, updated_energy_demand_kWh) + self._state.set_min_energy_demand_kWh(time_slot, updated_min_energy_demand_kWh) + self._state.set_max_energy_demand_kWh(time_slot, updated_max_energy_demand_kWh) + + self._state.increase_total_traded_energy_kWh(energy_kWh) + + def _calc_and_set_energy_demand(self, time_slot: DateTime): + energy_demand_kWh = convert_kJ_to_kWh( + self._heat_demand_Q_J.get_value(time_slot) / 1000 + ) / self._state.get_cop(time_slot) + self._state.set_energy_demand_kWh(time_slot, energy_demand_kWh) + self._state.set_min_energy_demand_kWh( + time_slot, self._calc_energy_to_buy_minimum(time_slot) + ) + self._state.set_max_energy_demand_kWh( + time_slot, self._calc_energy_to_buy_maximum(time_slot) + ) + + def _get_performance_energy_charge_kWh(self, time_slot: DateTime) -> float: + charge_power_kW = SorTesPerformanceMaps.get_power_charging( + self._ambient_temp_C.get_value(time_slot) + + SorTesConfiguration.AMBIENT_TEMPERATURE_CORRECTION + ) + return convert_kW_to_kWh(charge_power_kW, GlobalConfig.slot_length) + + def _get_performance_energy_discharge_kWh(self, time_slot: DateTime) -> float: + discharge_power_kW = SorTesPerformanceMaps.get_power_discharging( + self._ambient_temp_C.get_value(time_slot) + - SorTesConfiguration.AMBIENT_TEMPERATURE_CORRECTION + ) + return convert_kW_to_kWh(discharge_power_kW, GlobalConfig.slot_length) + + def _calc_condenser_electricity_kWh(self, charging_energy_kWh: float) -> float: + return ( + charging_energy_kWh + / SorTesConfiguration.COP_CONDENSER + * SorTesConfiguration.CONVERSION_CHARGE_CONDENSER_POWER + ) + + def _calc_evaporator_electricity_kWh(self, discharging_energy_kWh: float) -> float: + return ( + discharging_energy_kWh + / SorTesConfiguration.COP_EVAPORATOR + * SorTesConfiguration.CONVERSION_DISCHARGE_EVAPORATOR_POWER + ) + + def _get_total_electricity_demand_for_time_slot_kWh(self, time_slot: DateTime) -> float: + # we need this function in order to also access the demand after trading + return convert_kJ_to_kWh(self._state.get_heat_demand_kJ(time_slot)) / self._state.get_cop( + time_slot + ) + + def _calc_heat_capacity_into_electricity_kWh(self, heat_capacity: float) -> float: + return heat_capacity / SorTesConfiguration.COP_HEAT_SOURCE + + def _calc_available_free_storage_kWh(self, time_slot: DateTime) -> float: + charge_energy_kWh = self._get_performance_energy_charge_kWh(time_slot) + # add the charge energy to the capacity be able to reach the maximum SOC tolerance + available_heat_storage_kWh = ( + (SorTesConfiguration.MAX_SOC_TOLERANCE - self._state.get_soc(time_slot)) / 100 + ) * self._capacity_kWh + charge_energy_kWh + + if available_heat_storage_kWh < charge_energy_kWh: + return 0 + return charge_energy_kWh + + def _calc_available_stored_heat_kWh(self, time_slot: DateTime) -> float: + discharge_energy_kWh = self._get_performance_energy_discharge_kWh(time_slot) + # add the discharge energy to the capacity be able to reach the minimum SOC tolerance + stored_heat_kWh = ( + (self._state.get_soc(time_slot) - SorTesConfiguration.MIN_SOC_TOLERANCE) / 100 + ) * self._capacity_kWh + discharge_energy_kWh + + if stored_heat_kWh < discharge_energy_kWh: + return 0 + return discharge_energy_kWh + + def _calc_energy_to_buy_maximum(self, time_slot: DateTime) -> float: + available_heat_storage_kWh = self._calc_available_free_storage_kWh(time_slot) + return ( + self._get_total_electricity_demand_for_time_slot_kWh(time_slot) + + self._calc_heat_capacity_into_electricity_kWh(available_heat_storage_kWh) + + self._calc_condenser_electricity_kWh(available_heat_storage_kWh) + ) + + def _calc_energy_to_buy_minimum(self, time_slot: DateTime) -> float: + available_stored_heat_kWh = self._calc_available_stored_heat_kWh(time_slot) + electricity_demand_kWh = self._get_total_electricity_demand_for_time_slot_kWh(time_slot) + storage_capacity_electric_kWh = self._calc_heat_capacity_into_electricity_kWh( + available_stored_heat_kWh + ) + energy_to_be_bought_for_heat = electricity_demand_kWh - storage_capacity_electric_kWh + + if energy_to_be_bought_for_heat < FLOATING_POINT_TOLERANCE < storage_capacity_electric_kWh: + # corner case when the demand is lower than the discharging energy + if energy_to_be_bought_for_heat > FLOATING_POINT_TOLERANCE: + log.warning( + "The heat demand is lower than the discharging energy: %s, %s", + electricity_demand_kWh, + storage_capacity_electric_kWh, + ) + return electricity_demand_kWh + + return energy_to_be_bought_for_heat + self._calc_evaporator_electricity_kWh( + available_stored_heat_kWh + ) + + def _rotate_profiles(self, time_slot: Optional[DateTime] = None): + self._state.delete_past_state_values(time_slot) + self._heat_demand_Q_J.read_or_rotate_profiles() + self._ambient_temp_C.read_or_rotate_profiles() + self._target_temp_C.read_or_rotate_profiles() + + def _charge_or_discharge_tank(self, time_slot: DateTime): + electricity_demand_kWh = self._get_total_electricity_demand_for_time_slot_kWh( + self.last_time_slot(time_slot) + ) + net_traded_energy_kWh = self._bought_energy_kWh - electricity_demand_kWh + + def _is_net_traded_energy_zero(): + return isclose(net_traded_energy_kWh, 0, abs_tol=FLOATING_POINT_TOLERANCE) + + if _is_net_traded_energy_zero(): + self._no_charge(time_slot) + return + + if ( + self.soc_management.current_state == HeatPumpChargingState.CHARGE + and net_traded_energy_kWh > FLOATING_POINT_TOLERANCE + ): + self._charge(net_traded_energy_kWh, time_slot) + elif self.soc_management.current_state == HeatPumpChargingState.DISCHARGE: + self._discharge(net_traded_energy_kWh, time_slot) + else: + assert False, "should never reach this point" + + def _charge(self, net_traded_energy_kWh: float, time_slot: DateTime): + charge_energy_kWh = self._get_performance_energy_charge_kWh(self.last_time_slot(time_slot)) + condenser_energy_kWh = self._calc_condenser_electricity_kWh(charge_energy_kWh) + heat_energy_kWh = net_traded_energy_kWh * SorTesConfiguration.COP_HEAT_SOURCE + assert ( + abs(condenser_energy_kWh + charge_energy_kWh - heat_energy_kWh) + < FLOATING_POINT_TOLERANCE + ) + + self._update_soc(time_slot, charge_energy_kWh) + self._state.update_energy_used_for_dis_charging_kWh(time_slot, charge_energy_kWh) + self._state.set_auxiliary_energy_kWh(time_slot, condenser_energy_kWh) + + def _discharge(self, net_traded_energy_kWh: float, time_slot: DateTime): + assert net_traded_energy_kWh < FLOATING_POINT_TOLERANCE + discharge_energy_kWh = self._get_performance_energy_discharge_kWh( + self.last_time_slot(time_slot) + ) + evaporator_energy_kWh = self._calc_evaporator_electricity_kWh(discharge_energy_kWh) + assert (net_traded_energy_kWh - evaporator_energy_kWh) < FLOATING_POINT_TOLERANCE + + self._update_soc(time_slot, -discharge_energy_kWh) + self._state.update_energy_used_for_dis_charging_kWh(time_slot, -discharge_energy_kWh) + self._state.set_auxiliary_energy_kWh(time_slot, evaporator_energy_kWh) + + def _update_soc(self, time_slot: DateTime, heat_energy_kWh: float): + # heat_energy_kWh can be both positive (charging) and negative (discharging) + old_charge = self._state.get_soc(self.last_time_slot(time_slot)) / 100 * self._capacity_kWh + new_charge = old_charge + heat_energy_kWh + new_soc = new_charge / self._capacity_kWh + self._state.set_soc(time_slot, new_soc * 100) + + def _no_charge(self, time_slot: DateTime): + self._state.set_soc(time_slot, self._state.get_soc(self.last_time_slot(time_slot))) + + def _update_last_time_slot_data(self, time_slot: DateTime): + last_time_slot = self.last_time_slot(time_slot) + if last_time_slot not in self._ambient_temp_C.profile: + return + self._charge_or_discharge_tank(time_slot) + + self._bought_energy_kWh = 0.0 + + @staticmethod + def last_time_slot(current_market_slot: DateTime) -> DateTime: + """Calculate the previous time slot from the current one.""" + return current_market_slot - GlobalConfig.slot_length + + +class HeatPumpWithSorTesTankStrategy(HeatPumpStrategyBase): + """Strategy class for a heat pump that is connected to a SorTES tank""" + + def __init__( + self, + heat_demand_Q_profile: Union[str, float, dict], + ambient_temp_C_profile: Union[str, float, dict], + target_temp_C_profile: Union[str, float, dict], + average_trade_rate: Union[str, float, dict], + source_type: HeatPumpSourceType = ConstSettings.HeatPumpSettings.SOURCE_TYPE, + order_updater_parameters: dict[ + AvailableMarketTypes, HeatPumpOrderUpdaterParameters + ] = None, + ): + # pylint: disable=too-many-arguments, too-many-positional-arguments, super-init-not-called + + self._init_price_params(order_updater_parameters) + + self._energy_params = SorTesTankEnergyParameters( + heat_demand_Q_profile=heat_demand_Q_profile, + ambient_temp_C_profile=ambient_temp_C_profile, + target_temp_C_profile=target_temp_C_profile, + average_trade_rate=average_trade_rate, + source_type=source_type, + ) + + def serialize(self): + """Serialize strategy parameters.""" + return { + **self._energy_params.serialize(), + **self._order_updater_params.get(AvailableMarketTypes.SPOT).serialize(), + } + + def post_order( + self, market: "MarketBase", market_slot: DateTime, order_rate: float = None, **kwargs + ): + if not order_rate: + order_rate = self._order_updaters[market][market_slot].get_energy_rate(self.area.now) + else: + order_rate = Decimal(order_rate) + order_energy_kWh = Decimal( + self._energy_params.soc_management.calculate(market_slot, float(order_rate)) + ) + self._post_order(market, market_slot, order_energy_kWh, order_rate) + + @property + def state(self) -> SorTesTankState: + return self._energy_params.state + + def _init_price_params(self, order_updater_parameters): + if not order_updater_parameters: + order_updater_parameters = { + AvailableMarketTypes.SPOT: HeatPumpOrderUpdaterParameters() + } + + super().__init__(order_updater_parameters=order_updater_parameters) diff --git a/src/gsy_e/models/strategy/state/heatpump_state.py b/src/gsy_e/models/strategy/state/heatpump_state.py index 9cfc2cd40..a1419a661 100644 --- a/src/gsy_e/models/strategy/state/heatpump_state.py +++ b/src/gsy_e/models/strategy/state/heatpump_state.py @@ -81,7 +81,8 @@ def set_cop(self, time_slot: DateTime, cop: float): """Set cop for the given time slot.""" self._cop[time_slot] = cop - def _last_time_slot(self, current_market_slot: DateTime) -> DateTime: + @staticmethod + def _last_time_slot(current_market_slot: DateTime) -> DateTime: return current_market_slot - GlobalConfig.slot_length @staticmethod diff --git a/src/gsy_e/resources/average_trading_profile_sortes.csv b/src/gsy_e/resources/average_trading_profile_sortes.csv new file mode 100644 index 000000000..1b76198da --- /dev/null +++ b/src/gsy_e/resources/average_trading_profile_sortes.csv @@ -0,0 +1,87 @@ +slot,rate [ct./kWh] +2026-05-12T00:00,30.0 +2026-05-12T00:15,30.0 +2026-05-12T00:30,30.0 +2026-05-12T00:45,30.0 +2026-05-12T01:00,30.0 +2026-05-12T01:15,30.0 +2026-05-12T01:30,30.0 +2026-05-12T01:45,30.0 +2026-05-12T02:00,30.0 +2026-05-12T02:15,30.0 +2026-05-12T02:30,30.0 +2026-05-12T02:45,30.0 +2026-05-12T03:00,30.0 +2026-05-12T03:15,30.0 +2026-05-12T03:30,30.0 +2026-05-12T03:45,30.0 +2026-05-12T04:00,30.0 +2026-05-12T04:15,30.0 +2026-05-12T04:30,30.0 +2026-05-12T04:45,30.0 +2026-05-12T05:00,30.0 +2026-05-12T05:15,30.0 +2026-05-12T05:30,30.0 +2026-05-12T05:45,30.0 +2026-05-12T06:00,30.0 +2026-05-12T06:15,30.0 +2026-05-12T06:30,30.0 +2026-05-12T06:45,30.0 +2026-05-12T07:00,30.0 +2026-05-12T07:15,30.0 +2026-05-12T07:30,30.0 +2026-05-12T07:45,30.0 +2026-05-12T08:00,22.5 +2026-05-12T08:15,22.5 +2026-05-12T08:30,22.5 +2026-05-12T08:45,22.5 +2026-05-12T09:00,15.0 +2026-05-12T09:15,15.0 +2026-05-12T09:30,15.0 +2026-05-12T09:45,15.0 +2026-05-12T10:00,15.0 +2026-05-12T10:15,15.0 +2026-05-12T10:30,15.0 +2026-05-12T10:45,15.0 +2026-05-12T11:00,15.0 +2026-05-12T11:15,15.0 +2026-05-12T11:30,15.0 +2026-05-12T11:45,15.0 +2026-05-12T12:00,15.0 +2026-05-12T12:15,15.0 +2026-05-12T12:30,15.0 +2026-05-12T12:45,15.0 +2026-05-12T13:00,15.0 +2026-05-12T13:15,15.0 +2026-05-12T13:30,15.0 +2026-05-12T13:45,15.0 +2026-05-12T14:00,15.0 +2026-05-12T14:15,15.0 +2026-05-12T14:30,15.0 +2026-05-12T14:45,15.0 +2026-05-12T15:00,15.0 +2026-05-12T15:15,15.0 +2026-05-12T15:30,15.0 +2026-05-12T15:45,15.0 +2026-05-12T16:00,15.0 +2026-05-12T16:15,15.0 +2026-05-12T16:30,15.0 +2026-05-12T16:45,30.0 +2026-05-12T17:00,30.0 +2026-05-12T17:15,30.0 +2026-05-12T17:30,30.0 +2026-05-12T17:45,30.0 +2026-05-12T18:00,30.0 +2026-05-12T18:15,30.0 +2026-05-12T18:30,30.0 +2026-05-12T18:45,30.0 +2026-05-12T19:00,30.0 +2026-05-12T19:15,30.0 +2026-05-12T19:30,30.0 +2026-05-12T19:45,30.0 +2026-05-12T20:00,30.0 +2026-05-12T20:15,30.0 +2026-05-12T20:30,30.0 +2026-05-12T20:45,30.0 +2026-05-12T21:00,30.0 +2026-05-12T21:15,30.0 diff --git a/src/gsy_e/setup/strategy_tests/heat_pump_sortes.py b/src/gsy_e/setup/strategy_tests/heat_pump_sortes.py new file mode 100644 index 000000000..0b84b61c1 --- /dev/null +++ b/src/gsy_e/setup/strategy_tests/heat_pump_sortes.py @@ -0,0 +1,80 @@ +""" +Copyright 2018 Grid Singularity +This file is part of Grid Singularity Exchange. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +""" + +import os + +from gsy_framework.constants_limits import ConstSettings + +from gsy_e.gsy_e_core.util import gsye_root_path +from gsy_e.models.area import Area +from gsy_e.models.strategy.heatpump_with_sortes_tank import HeatPumpWithSorTesTankStrategy +from gsy_e.models.strategy.infinite_bus import InfiniteBusStrategy +from gsy_e.models.strategy.pv import PVStrategy + +ConstSettings.MASettings.MARKET_TYPE = 2 +ConstSettings.GeneralSettings.DEFAULT_UPDATE_INTERVAL = 5 + +# Attention: Set the start date to 2026-05-12 for the average trading rate profile! + + +def get_setup(config): + area = Area( + "Grid", + [ + Area( + "House 1", + [ + Area( + "Sortes", + strategy=HeatPumpWithSorTesTankStrategy( + heat_demand_Q_profile=20 * 3600 * 1000, + ambient_temp_C_profile=10, + target_temp_C_profile=30, + average_trade_rate=os.path.join( + gsye_root_path, "resources", "average_trading_profile_sortes.csv" + ), + ), + ), + ], + grid_fee_percentage=0, + grid_fee_constant=0, + ), + Area( + "House 2", + [ + Area( + "H2 PV", + strategy=PVStrategy( + capacity_kW=20, + panel_count=1, + initial_selling_rate=24, + final_selling_rate=0, + ), + ), + ], + grid_fee_percentage=0, + grid_fee_constant=0, + ), + Area( + "Infinite Bus", + strategy=InfiniteBusStrategy(energy_sell_rate=30, energy_buy_rate=0), + ), + ], + config=config, + ) + return area diff --git a/tests/strategies/test_sortes_heat_pump.py b/tests/strategies/test_sortes_heat_pump.py new file mode 100644 index 000000000..aadfcd75b --- /dev/null +++ b/tests/strategies/test_sortes_heat_pump.py @@ -0,0 +1,743 @@ +# pylint: disable=protected-access, attribute-defined-outside-init, too-many-public-methods +import math +from unittest.mock import MagicMock, patch + +from gsy_framework.constants_limits import ConstSettings, GlobalConfig +from gsy_framework.enums import AvailableMarketTypes +from pendulum import UTC, duration, today + +from gsy_e.constants import SorTesConfiguration +from gsy_e.models.area import Area +from gsy_e.models.strategy.heat_pump import HeatPumpOrderUpdaterParameters +from gsy_e.models.strategy.heat_pump_soc_management import HeatPumpChargingState +from gsy_e.models.strategy.heatpump_with_sortes_tank import ( + HeatPumpWithSorTesTankStrategy, + SorTesPerformanceMaps, + SorTesTankEnergyParameters, + SorTesTankMinimiseSwitchStrategy, + SorTesTankState, +) +from gsy_e.models.strategy.strategy_profile import global_objects + +START_TIME_SLOT = today(tz=UTC) +SLOT_LENGTH = duration(minutes=15) +NEXT_SLOT = START_TIME_SLOT + SLOT_LENGTH + + +class TestSorTesPerformanceMaps: + + def test_get_power_charging_at_exact_table_temperatures(self): + assert math.isclose(SorTesPerformanceMaps.get_power_charging(5), 3.8) + assert math.isclose(SorTesPerformanceMaps.get_power_charging(20), 3.0) + assert math.isclose(SorTesPerformanceMaps.get_power_charging(45), 0.5) + + def test_get_power_charging_interpolates_between_table_temperatures(self): + # Between 10 (3.5 kW) and 15 (3.3 kW), midpoint at 12.5 → 3.4 + power = SorTesPerformanceMaps.get_power_charging(12.5) + assert math.isclose(power, 3.4, abs_tol=1e-9) + + def test_get_power_charging_extrapolates_below_range(self): + # Slope from (5, 3.8) to (10, 3.5): -0.06/°C; at 0°C: 3.8 + (-0.06) * (0-5) = 4.1 + power = SorTesPerformanceMaps.get_power_charging(0) + assert math.isclose(power, 4.1, abs_tol=1e-9) + + def test_get_power_charging_extrapolates_above_range(self): + # Slope from (40, 1.5) to (45, 0.5): -0.2/°C; at 50°C: 0.5 + (-0.2) * (50-45) = -0.5 + power = SorTesPerformanceMaps.get_power_charging(50) + assert math.isclose(power, -0.5, abs_tol=1e-9) + + def test_get_power_discharging_at_exact_table_temperatures(self): + assert math.isclose(SorTesPerformanceMaps.get_power_discharging(5), 3.0) + assert math.isclose(SorTesPerformanceMaps.get_power_discharging(15), 4.0) + assert math.isclose(SorTesPerformanceMaps.get_power_discharging(25), 4.8) + + def test_get_power_discharging_interpolates_between_table_temperatures(self): + # Between 5 (3.0 kW) and 10 (3.5 kW), midpoint at 7.5 → 3.25 + power = SorTesPerformanceMaps.get_power_discharging(7.5) + assert math.isclose(power, 3.25, abs_tol=1e-9) + + def test_get_power_discharging_extrapolates_below_range(self): + # Slope from (5, 3.0) to (10, 3.5): 0.1/°C; at 0°C: 3.0 + 0.1 * (0-5) = 2.5 + power = SorTesPerformanceMaps.get_power_discharging(0) + assert math.isclose(power, 2.5, abs_tol=1e-9) + + def test_get_power_discharging_extrapolates_above_range(self): + # Slope from (20, 4.3) to (25, 4.8): 0.1/°C; at 30°C: 4.8 + 0.1 * (30-25) = 5.3 + power = SorTesPerformanceMaps.get_power_discharging(30) + assert math.isclose(power, 5.3, abs_tol=1e-9) + + def test_charging_power_decreases_with_temperature(self): + powers = [SorTesPerformanceMaps.get_power_charging(t) for t in range(5, 46, 5)] + assert all(powers[i] > powers[i + 1] for i in range(len(powers) - 1)) + + def test_discharging_power_increases_with_temperature(self): + powers = [SorTesPerformanceMaps.get_power_discharging(t) for t in range(5, 26, 5)] + assert all(powers[i] < powers[i + 1] for i in range(len(powers) - 1)) + + +class TestSorTesTankMinimiseSwitchStrategy: + + def setup_method(self): + self._original_market_maker_rate = GlobalConfig.market_maker_rate + self._original_slot_length = GlobalConfig.slot_length + self._original_start_date = GlobalConfig.start_date + GlobalConfig.slot_length = SLOT_LENGTH + GlobalConfig.market_maker_rate = { + START_TIME_SLOT: 30, + NEXT_SLOT: 20, + } + GlobalConfig.start_date = START_TIME_SLOT + self._energy_params_mock = MagicMock() + self._energy_params_mock.get_soc = MagicMock(return_value=50.0) + + def teardown_method(self): + GlobalConfig.market_maker_rate = self._original_market_maker_rate + GlobalConfig.slot_length = self._original_slot_length + GlobalConfig.start_date = self._original_start_date + + def _create_strategy(self, average_trade_rate=25.0): + return SorTesTankMinimiseSwitchStrategy(self._energy_params_mock, average_trade_rate) + + def test_initial_state_is_maintain_soc(self): + strategy = self._create_strategy() + assert strategy.current_state == HeatPumpChargingState.MAINTAIN_SOC + + def test_current_state_property_returns_current_state(self): + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.CHARGE + assert strategy.current_state == HeatPumpChargingState.CHARGE + + def test_get_tank_soc_delegates_to_energy_params(self): + strategy = self._create_strategy() + self._energy_params_mock.get_soc = MagicMock(return_value=42.0) + soc = strategy._get_tank_soc(START_TIME_SLOT) + assert math.isclose(soc, 42.0) + self._energy_params_mock.get_soc.assert_called_once_with(START_TIME_SLOT) + + def test_is_energy_affordable_true_when_avg_rate_below_preferred_buying_rate(self): + # average_trade_rate=19, preferred_buying_rate=20 → 19 < 20 → True + strategy = self._create_strategy(average_trade_rate=19.0) + strategy.event_activate() + assert strategy._is_energy_affordable(START_TIME_SLOT, 0.0) is True + + def test_is_energy_affordable_false_when_avg_rate_above_preferred_buying_rate(self): + # average_trade_rate=35, preferred_buying_rate=20 → 35 < 20 → False + strategy = self._create_strategy(average_trade_rate=35.0) + strategy.event_activate() + assert strategy._is_energy_affordable(START_TIME_SLOT, 0.0) is False + + def test_event_activate_calls_read_or_rotate_profiles(self): + strategy = self._create_strategy() + with patch.object(strategy._average_trade_rate, "read_or_rotate_profiles") as mock_rotate: + strategy.event_activate() + mock_rotate.assert_called_once() + + def test_event_market_slot_calls_read_or_rotate_profiles(self): + strategy = self._create_strategy() + strategy.event_activate() + with patch.object(strategy._average_trade_rate, "read_or_rotate_profiles") as mock_rotate: + strategy.event_market_slot() + mock_rotate.assert_called_once() + + def test_minutes_before_switch_taken_from_sortes_configuration(self): + assert ( + SorTesTankMinimiseSwitchStrategy.MINUTES_BEFORE_SWITCH_ALLOWED + == SorTesConfiguration.MINUTES_BEFORE_SWITCH_ALLOWED + ) + + def test_soc_tolerances_taken_from_sortes_configuration(self): + assert ( + SorTesTankMinimiseSwitchStrategy.MIN_SOC_TOLERANCE + == SorTesConfiguration.MIN_SOC_TOLERANCE + ) + assert ( + SorTesTankMinimiseSwitchStrategy.MAX_SOC_TOLERANCE + == SorTesConfiguration.MAX_SOC_TOLERANCE + ) + + def test_handle_state_at_soc_limits_charge_at_max_soc_returns_maintain_soc(self): + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.CHARGE + self._energy_params_mock.get_soc = MagicMock( + return_value=SorTesConfiguration.MAX_SOC_TOLERANCE + ) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.CHARGE + ) + assert result == HeatPumpChargingState.MAINTAIN_SOC + + def test_handle_state_at_soc_limits_charge_above_max_soc_returns_maintain_soc(self): + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.CHARGE + self._energy_params_mock.get_soc = MagicMock( + return_value=SorTesConfiguration.MAX_SOC_TOLERANCE + 1 + ) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.CHARGE + ) + assert result == HeatPumpChargingState.MAINTAIN_SOC + + def test_handle_state_at_soc_limits_discharge_at_min_soc_returns_maintain_soc(self): + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.DISCHARGE + self._energy_params_mock.get_soc = MagicMock( + return_value=SorTesConfiguration.MIN_SOC_TOLERANCE + ) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.DISCHARGE + ) + assert result == HeatPumpChargingState.MAINTAIN_SOC + + def test_handle_state_at_soc_limits_discharge_below_min_soc_returns_maintain_soc(self): + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.DISCHARGE + self._energy_params_mock.get_soc = MagicMock( + return_value=SorTesConfiguration.MIN_SOC_TOLERANCE - 1 + ) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.DISCHARGE + ) + assert result == HeatPumpChargingState.MAINTAIN_SOC + + def test_handle_state_at_soc_limits_within_bounds_returns_input_state(self): + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.CHARGE + self._energy_params_mock.get_soc = MagicMock(return_value=50.0) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.CHARGE + ) + assert result == HeatPumpChargingState.CHARGE + + def test_handle_state_at_soc_limits_max_soc_but_not_charging_preserves_state(self): + # At max SOC but internal state is MAINTAIN_SOC → no override + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.MAINTAIN_SOC + self._energy_params_mock.get_soc = MagicMock( + return_value=SorTesConfiguration.MAX_SOC_TOLERANCE + ) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.MAINTAIN_SOC + ) + assert result == HeatPumpChargingState.MAINTAIN_SOC + + def test_handle_state_at_soc_limits_min_soc_but_not_discharging_preserves_state(self): + # At min SOC but internal state is CHARGE → no discharge override fires + strategy = self._create_strategy() + strategy._current_state = HeatPumpChargingState.CHARGE + self._energy_params_mock.get_soc = MagicMock( + return_value=SorTesConfiguration.MIN_SOC_TOLERANCE + ) + result = strategy._handle_state_at_soc_limits( + START_TIME_SLOT, HeatPumpChargingState.CHARGE + ) + assert result == HeatPumpChargingState.CHARGE + + +class TestSorTesTankState: + + def setup_method(self): + self._original_start_date = GlobalConfig.start_date + self._original_slot_length = GlobalConfig.slot_length + GlobalConfig.slot_length = SLOT_LENGTH + GlobalConfig.start_date = START_TIME_SLOT + self._state = SorTesTankState() + + def teardown_method(self): + GlobalConfig.start_date = self._original_start_date + GlobalConfig.slot_length = self._original_slot_length + + def test_activate_sets_initial_soc_to_min_soc_tolerance(self): + self._state.activate() + assert math.isclose( + self._state.get_soc(START_TIME_SLOT), SorTesConfiguration.MIN_SOC_TOLERANCE + ) + + def test_get_soc_returns_zero_for_unknown_time_slot(self): + assert self._state.get_soc(START_TIME_SLOT) == 0.0 + + def test_set_and_get_soc_round_trips(self): + self._state.set_soc(START_TIME_SLOT, 55.0) + assert math.isclose(self._state.get_soc(START_TIME_SLOT), 55.0) + + def test_set_and_get_min_energy_demand_kWh(self): + self._state.set_min_energy_demand_kWh(START_TIME_SLOT, 1.5) + assert math.isclose(self._state.get_min_energy_demand_kWh(START_TIME_SLOT), 1.5) + + def test_get_min_energy_demand_kWh_returns_zero_for_unknown_time_slot(self): + assert self._state.get_min_energy_demand_kWh(START_TIME_SLOT) == 0.0 + + def test_set_and_get_max_energy_demand_kWh(self): + self._state.set_max_energy_demand_kWh(START_TIME_SLOT, 3.0) + assert math.isclose(self._state.get_max_energy_demand_kWh(START_TIME_SLOT), 3.0) + + def test_get_max_energy_demand_kWh_returns_zero_for_unknown_time_slot(self): + assert self._state.get_max_energy_demand_kWh(START_TIME_SLOT) == 0.0 + + def test_set_and_get_auxiliary_energy_kWh(self): + self._state.set_auxiliary_energy_kWh(START_TIME_SLOT, 1.25) + result = self._state.get_results_dict(START_TIME_SLOT) + assert math.isclose(result["auxiliary_energy_kWh"], 1.25) + + def test_get_auxiliary_energy_kWh_returns_zero_for_unknown_time_slot(self): + result = self._state.get_results_dict(START_TIME_SLOT) + assert result["auxiliary_energy_kWh"] == 0.0 + + def test_update_and_get_energy_used_for_dis_charging_kWh(self): + self._state.update_energy_used_for_dis_charging_kWh(START_TIME_SLOT, 2.5) + result = self._state.get_results_dict(START_TIME_SLOT) + assert math.isclose(result["energy_used_for_dis_charging_kWh"], 2.5) + + def test_get_energy_used_for_dis_charging_kWh_returns_zero_for_unknown_time_slot(self): + result = self._state.get_results_dict(START_TIME_SLOT) + assert result["energy_used_for_dis_charging_kWh"] == 0.0 + + def test_update_energy_used_for_dis_charging_kWh_supports_negative_values(self): + # discharging stores negative values + self._state.update_energy_used_for_dis_charging_kWh(START_TIME_SLOT, -1.5) + result = self._state.get_results_dict(START_TIME_SLOT) + assert math.isclose(result["energy_used_for_dis_charging_kWh"], -1.5) + + def test_increase_total_traded_energy_kWh_accumulates_over_calls(self): + self._state.increase_total_traded_energy_kWh(4.0) + self._state.increase_total_traded_energy_kWh(1.0) + assert math.isclose(self._state.get_state()["total_traded_energy_kWh"], 5.0) + + def test_delete_past_state_values_removes_slots_older_than_current(self): + self._state.set_soc(START_TIME_SLOT, 50.0) + self._state.set_soc(NEXT_SLOT, 60.0) + # current_time_slot is two slots ahead; START_TIME_SLOT should be removed + self._state.delete_past_state_values(NEXT_SLOT + SLOT_LENGTH) + assert self._state.get_soc(START_TIME_SLOT) == 0.0 + assert math.isclose(self._state.get_soc(NEXT_SLOT), 60.0) + + def test_delete_past_state_values_is_noop_when_no_time_slot_given(self): + self._state.set_soc(START_TIME_SLOT, 50.0) + self._state.delete_past_state_values(None) + assert math.isclose(self._state.get_soc(START_TIME_SLOT), 50.0) + + def test_get_state_contains_all_required_keys(self): + state = self._state.get_state() + for key in ( + "soc", + "cop", + "energy_demand_kWh", + "min_energy_demand_kWh", + "max_energy_demand_kWh", + "total_traded_energy_kWh", + ): + assert key in state + + def test_restore_state_round_trips_all_numeric_fields(self): + self._state.set_soc(START_TIME_SLOT, 70.0) + self._state.set_cop(START_TIME_SLOT, 4.0) + self._state.set_energy_demand_kWh(START_TIME_SLOT, 1.5) + self._state.set_min_energy_demand_kWh(START_TIME_SLOT, 0.5) + self._state.set_max_energy_demand_kWh(START_TIME_SLOT, 2.0) + self._state.increase_total_traded_energy_kWh(3.0) + + state_dict = self._state.get_state() + restored = SorTesTankState() + restored.restore_state(state_dict) + + assert math.isclose(restored.get_soc(START_TIME_SLOT), 70.0) + assert math.isclose(restored.get_cop(START_TIME_SLOT), 4.0) + assert math.isclose(restored.get_energy_demand_kWh(START_TIME_SLOT), 1.5) + assert math.isclose(restored.get_min_energy_demand_kWh(START_TIME_SLOT), 0.5) + assert math.isclose(restored.get_max_energy_demand_kWh(START_TIME_SLOT), 2.0) + restored_dict = restored.get_state() + assert math.isclose(restored_dict["total_traded_energy_kWh"], 3.0) + + def test_get_results_dict_returns_correct_values(self): + self._state.set_cop(START_TIME_SLOT, 3.2) + self._state.set_soc(START_TIME_SLOT, 55.0) + self._state.set_heat_demand_kJ(START_TIME_SLOT, 100.0) + self._state.update_energy_used_for_dis_charging_kWh(START_TIME_SLOT, 1.0) + self._state.set_auxiliary_energy_kWh(START_TIME_SLOT, 0.5) + result = self._state.get_results_dict(START_TIME_SLOT) + assert math.isclose(result["cop"], 3.2) + assert math.isclose(result["soc"], 55.0) + assert math.isclose(result["heat_demand_kJ"], 100.0) + assert "total_traded_energy_kWh" in result + assert math.isclose(result["energy_used_for_dis_charging_kWh"], 1.0) + assert math.isclose(result["auxiliary_energy_kWh"], 0.5) + + def test_delete_past_state_values_clears_all_time_series_attributes(self): + self._state.set_soc(START_TIME_SLOT, 50.0) + self._state.set_cop(START_TIME_SLOT, 3.0) + self._state.set_heat_demand_kJ(START_TIME_SLOT, 100.0) + self._state.set_energy_demand_kWh(START_TIME_SLOT, 2.0) + self._state.set_min_energy_demand_kWh(START_TIME_SLOT, 1.0) + self._state.set_max_energy_demand_kWh(START_TIME_SLOT, 3.0) + self._state.delete_past_state_values(NEXT_SLOT + SLOT_LENGTH) + assert self._state.get_soc(START_TIME_SLOT) == 0.0 + assert self._state.get_cop(START_TIME_SLOT) == 0.0 + assert self._state.get_heat_demand_kJ(START_TIME_SLOT) == 0.0 + assert self._state.get_energy_demand_kWh(START_TIME_SLOT) == 0.0 + assert self._state.get_min_energy_demand_kWh(START_TIME_SLOT) == 0.0 + assert self._state.get_max_energy_demand_kWh(START_TIME_SLOT) == 0.0 + + +AMBIENT_TEMP_C = 15.0 +TARGET_TEMP_C = 50.0 +# delta = 35 → COP = 6.08 - 0.09*35 + 0.0005*35^2 = 6.08 - 3.15 + 0.6125 = 3.5425 +EXPECTED_COP = 6.08 - 0.09 * 35 + 0.0005 * 35**2 + +# SorTesPerformanceMaps.get_power_charging(AMBIENT_TEMP_C + 5) = get_power_charging(20) = 3.0 kW +# With 15-min slot: 3.0 kW * 0.25 h = 0.75 kWh +EXPECTED_CHARGE_ENERGY_KWH = 3.0 * 0.25 + +# SorTesPerformanceMaps.get_power_discharging(AMBIENT_TEMP_C + 5) +# = get_power_discharging(20) = 4.3 kW +# With 15-min slot: 4.3 kW * 0.25 h = 1.075 kWh +EXPECTED_DISCHARGE_ENERGY_KWH = 4.3 * 0.25 + + +class TestSorTesTankEnergyParameters: + + def setup_method(self): + self._original_start_date = GlobalConfig.start_date + self._original_slot_length = GlobalConfig.slot_length + self._original_market_maker_rate = GlobalConfig.market_maker_rate + self._original_timestamp = global_objects.profiles_handler.current_timestamp + GlobalConfig.slot_length = SLOT_LENGTH + GlobalConfig.start_date = START_TIME_SLOT + GlobalConfig.market_maker_rate = { + START_TIME_SLOT: 30, + NEXT_SLOT: 30, + } + global_objects.profiles_handler._update_current_time(timestamp=START_TIME_SLOT) + + def teardown_method(self): + GlobalConfig.start_date = self._original_start_date + GlobalConfig.slot_length = self._original_slot_length + GlobalConfig.market_maker_rate = self._original_market_maker_rate + global_objects.profiles_handler._update_current_time(timestamp=self._original_timestamp) + + def _create_energy_params( + self, + heat_demand_Q_profile=None, + ambient_temp_C_profile=None, + target_temp_C_profile=None, + average_trade_rate=25.0, + ): + if heat_demand_Q_profile is None: + heat_demand_Q_profile = {START_TIME_SLOT: 3600.0, NEXT_SLOT: 3600.0} + if ambient_temp_C_profile is None: + ambient_temp_C_profile = {START_TIME_SLOT: AMBIENT_TEMP_C, NEXT_SLOT: AMBIENT_TEMP_C} + if target_temp_C_profile is None: + target_temp_C_profile = {START_TIME_SLOT: TARGET_TEMP_C, NEXT_SLOT: TARGET_TEMP_C} + return SorTesTankEnergyParameters( + heat_demand_Q_profile=heat_demand_Q_profile, + ambient_temp_C_profile=ambient_temp_C_profile, + target_temp_C_profile=target_temp_C_profile, + average_trade_rate=average_trade_rate, + ) + + def test_state_property_returns_sortes_tank_state_instance(self): + ep = self._create_energy_params() + assert isinstance(ep.state, SorTesTankState) + + def test_soc_management_property_returns_switch_strategy_instance(self): + ep = self._create_energy_params() + assert isinstance(ep.soc_management, SorTesTankMinimiseSwitchStrategy) + + def test_event_activate_sets_initial_soc_at_start_date(self): + ep = self._create_energy_params() + ep.event_activate() + assert math.isclose( + ep.state.get_soc(START_TIME_SLOT), SorTesConfiguration.MIN_SOC_TOLERANCE + ) + + def test_get_soc_delegates_to_state(self): + ep = self._create_energy_params() + ep.state.set_soc(START_TIME_SLOT, 45.0) + assert math.isclose(ep.get_soc(START_TIME_SLOT), 45.0) + + def test_get_energy_demand_kWh_delegates_to_state(self): + ep = self._create_energy_params() + ep.state.set_energy_demand_kWh(START_TIME_SLOT, 2.0) + assert math.isclose(ep.get_energy_demand_kWh(START_TIME_SLOT), 2.0) + + def test_get_min_energy_demand_kWh_delegates_to_state(self): + ep = self._create_energy_params() + ep.state.set_min_energy_demand_kWh(START_TIME_SLOT, 0.5) + assert math.isclose(ep.get_min_energy_demand_kWh(START_TIME_SLOT), 0.5) + + def test_get_max_energy_demand_kWh_delegates_to_state(self): + ep = self._create_energy_params() + ep.state.set_max_energy_demand_kWh(START_TIME_SLOT, 3.0) + assert math.isclose(ep.get_max_energy_demand_kWh(START_TIME_SLOT), 3.0) + + def test_last_time_slot_returns_slot_minus_slot_length(self): + assert SorTesTankEnergyParameters.last_time_slot(NEXT_SLOT) == START_TIME_SLOT + + def test_event_traded_energy_decrements_all_demand_values(self): + ep = self._create_energy_params() + ep.state.set_energy_demand_kWh(START_TIME_SLOT, 5.0) + ep.state.set_min_energy_demand_kWh(START_TIME_SLOT, 2.0) + ep.state.set_max_energy_demand_kWh(START_TIME_SLOT, 8.0) + ep.event_traded_energy(START_TIME_SLOT, 3.0) + assert math.isclose(ep.get_energy_demand_kWh(START_TIME_SLOT), 2.0) + assert math.isclose(ep.get_min_energy_demand_kWh(START_TIME_SLOT), 0.0) + assert math.isclose(ep.get_max_energy_demand_kWh(START_TIME_SLOT), 5.0) + + def test_event_traded_energy_clamps_demand_to_zero_when_trade_exceeds_demand(self): + ep = self._create_energy_params() + ep.state.set_energy_demand_kWh(START_TIME_SLOT, 1.0) + ep.state.set_min_energy_demand_kWh(START_TIME_SLOT, 0.5) + ep.state.set_max_energy_demand_kWh(START_TIME_SLOT, 1.0) + ep.event_traded_energy(START_TIME_SLOT, 10.0) + assert ep.get_energy_demand_kWh(START_TIME_SLOT) == 0.0 + assert ep.get_min_energy_demand_kWh(START_TIME_SLOT) == 0.0 + assert ep.get_max_energy_demand_kWh(START_TIME_SLOT) == 0.0 + + def test_event_traded_energy_accumulates_total_traded(self): + ep = self._create_energy_params() + ep.state.set_energy_demand_kWh(START_TIME_SLOT, 5.0) + ep.state.set_min_energy_demand_kWh(START_TIME_SLOT, 5.0) + ep.state.set_max_energy_demand_kWh(START_TIME_SLOT, 5.0) + ep.event_traded_energy(START_TIME_SLOT, 2.0) + ep.event_traded_energy(START_TIME_SLOT, 1.5) + assert math.isclose(ep.state.get_state()["total_traded_energy_kWh"], 3.5) + + def test_event_market_cycle_sets_positive_energy_demand_in_state(self): + ep = self._create_energy_params() + ep.event_activate() + ep.event_market_cycle(START_TIME_SLOT) + assert ep.get_energy_demand_kWh(START_TIME_SLOT) > 0 + + def test_event_market_cycle_sets_cop_for_time_slot(self): + ep = self._create_energy_params() + ep.event_activate() + ep.event_market_cycle(START_TIME_SLOT) + assert math.isclose(ep.state.get_cop(START_TIME_SLOT), EXPECTED_COP, abs_tol=1e-6) + + def test_event_market_cycle_sets_heat_demand_in_state(self): + heat_demand_J = 3600.0 # 3.6 kJ = 0.001 kWh + ep = self._create_energy_params( + heat_demand_Q_profile={START_TIME_SLOT: heat_demand_J, NEXT_SLOT: heat_demand_J} + ) + ep.event_activate() + ep.event_market_cycle(START_TIME_SLOT) + assert math.isclose( + ep.state.get_heat_demand_kJ(START_TIME_SLOT), heat_demand_J / 1000.0, abs_tol=1e-9 + ) + + def test_serialize_contains_required_keys(self): + ep = self._create_energy_params() + result = ep.serialize() + for key in ("heat_demand_Q_J", "ambient_temp_C", "target_temp_C", "source_type"): + assert key in result + + def test_serialize_source_type_matches_default(self): + ep = self._create_energy_params() + result = ep.serialize() + assert result["source_type"] == ConstSettings.HeatPumpSettings.SOURCE_TYPE + + def test_serialize_input_profiles_match_constructor_arguments(self): + heat_profile = {START_TIME_SLOT: 1000.0, NEXT_SLOT: 2000.0} + ambient_profile = {START_TIME_SLOT: 10.0, NEXT_SLOT: 12.0} + target_profile = {START_TIME_SLOT: 45.0, NEXT_SLOT: 50.0} + ep = self._create_energy_params( + heat_demand_Q_profile=heat_profile, + ambient_temp_C_profile=ambient_profile, + target_temp_C_profile=target_profile, + ) + result = ep.serialize() + assert result["heat_demand_Q_J"] == heat_profile + assert result["ambient_temp_C"] == ambient_profile + assert result["target_temp_C"] == target_profile + + def test_event_market_cycle_min_demand_not_greater_than_max_demand(self): + ep = self._create_energy_params() + ep.event_activate() + ep.event_market_cycle(START_TIME_SLOT) + assert ep.get_min_energy_demand_kWh(NEXT_SLOT) <= ep.get_max_energy_demand_kWh(NEXT_SLOT) + + def test_event_market_cycle_max_demand_exceeds_base_electricity_demand(self): + # max includes storage charging on top of baseline heat demand + ep = self._create_energy_params() + ep.event_activate() + ep.event_market_cycle(START_TIME_SLOT) + assert ep.get_max_energy_demand_kWh(NEXT_SLOT) >= ep.get_energy_demand_kWh(NEXT_SLOT) + + def test_no_charge_maintains_soc_from_previous_slot(self): + ep = self._create_energy_params() + ep.event_activate() + # State starts at MIN_SOC_TOLERANCE; soc_management defaults to MAINTAIN_SOC + ep.event_market_cycle(NEXT_SLOT) + expected_soc = SorTesConfiguration.MIN_SOC_TOLERANCE + assert math.isclose(ep.get_soc(NEXT_SLOT), expected_soc) + + def test_charge_increases_soc(self): + ep = self._create_energy_params() + ep.event_activate() + ep.soc_management._current_state = HeatPumpChargingState.CHARGE + # Simulate that we already bought more than needed for baseline heat + # charge_energy = 0.75 kWh; condenser = charge * CONVERSION_CHARGE_CONDENSER = 0.625 kWh + # heat_energy = net_traded * COP_HEAT_SOURCE = (condenser + charge) * 1 = 1.375 kWh + charge_kWh = EXPECTED_CHARGE_ENERGY_KWH + condenser_kWh = ( + charge_kWh + * SorTesConfiguration.CONVERSION_CHARGE_CONDENSER_POWER + / SorTesConfiguration.COP_CONDENSER + ) + ep._bought_energy_kWh = charge_kWh + condenser_kWh # net positive after heat demand + # Call _update_last_time_slot_data which also resets _bought_energy_kWh + ep._ambient_temp_C.profile[START_TIME_SLOT] = AMBIENT_TEMP_C + ep._charge_or_discharge_tank(NEXT_SLOT) + assert ep.get_soc(NEXT_SLOT) == 13 + + def test_discharge_decreases_soc(self): + ep = self._create_energy_params() + ep.event_activate() + # Set SOC high enough to discharge + high_soc = 50.0 + ep.state.set_soc(START_TIME_SLOT, high_soc) + ep.soc_management._current_state = HeatPumpChargingState.DISCHARGE + # Simulate negative net energy (bought less than needed) + discharge_kWh = EXPECTED_DISCHARGE_ENERGY_KWH + evaporator_kWh = ( + discharge_kWh + * SorTesConfiguration.CONVERSION_DISCHARGE_EVAPORATOR_POWER + / SorTesConfiguration.COP_EVAPORATOR + ) + ep._bought_energy_kWh = -evaporator_kWh # net negative + ep._ambient_temp_C.profile[START_TIME_SLOT] = AMBIENT_TEMP_C + ep._charge_or_discharge_tank(NEXT_SLOT) + assert ep.get_soc(NEXT_SLOT) == 46.5 + + +class TestHeatPumpWithSorTesTankStrategy: + + def setup_method(self): + self._original_start_date = GlobalConfig.start_date + self._original_slot_length = GlobalConfig.slot_length + self._original_market_maker_rate = GlobalConfig.market_maker_rate + self._original_market_type = ConstSettings.MASettings.MARKET_TYPE + self._original_timestamp = global_objects.profiles_handler.current_timestamp + GlobalConfig.slot_length = SLOT_LENGTH + GlobalConfig.start_date = START_TIME_SLOT + GlobalConfig.market_maker_rate = { + START_TIME_SLOT: 30, + NEXT_SLOT: 30, + } + ConstSettings.MASettings.MARKET_TYPE = 2 + global_objects.profiles_handler._update_current_time(timestamp=START_TIME_SLOT) + + def teardown_method(self): + GlobalConfig.start_date = self._original_start_date + GlobalConfig.slot_length = self._original_slot_length + GlobalConfig.market_maker_rate = self._original_market_maker_rate + ConstSettings.MASettings.MARKET_TYPE = self._original_market_type + global_objects.profiles_handler._update_current_time(timestamp=self._original_timestamp) + + def _create_strategy(self, **kwargs): + defaults = { + "heat_demand_Q_profile": {START_TIME_SLOT: 3600.0, NEXT_SLOT: 3600.0}, + "ambient_temp_C_profile": {START_TIME_SLOT: AMBIENT_TEMP_C, NEXT_SLOT: AMBIENT_TEMP_C}, + "target_temp_C_profile": {START_TIME_SLOT: TARGET_TEMP_C, NEXT_SLOT: TARGET_TEMP_C}, + "average_trade_rate": 19.0, + } + defaults.update(kwargs) + return HeatPumpWithSorTesTankStrategy(**defaults) + + def _create_activated_strategy(self, **kwargs): + strategy = self._create_strategy(**kwargs) + strategy_area = Area("sortes_hp", strategy=strategy) + area = Area("grid", children=[strategy_area]) + area.config.start_date = START_TIME_SLOT + area.config.end_date = area.config.start_date.add(days=1) + area.activate() + return strategy, area + + def test_state_property_returns_sortes_tank_state(self): + strategy = self._create_strategy() + assert isinstance(strategy.state, SorTesTankState) + + def test_energy_params_is_sortes_tank_energy_parameters(self): + strategy = self._create_strategy() + assert isinstance(strategy._energy_params, SorTesTankEnergyParameters) + + def test_init_with_no_order_updater_params_uses_spot_default(self): + strategy = self._create_strategy() + assert AvailableMarketTypes.SPOT in strategy._order_updater_params + assert isinstance( + strategy._order_updater_params[AvailableMarketTypes.SPOT], + HeatPumpOrderUpdaterParameters, + ) + + def test_init_with_custom_order_updater_params_stores_them(self): + custom_params = { + AvailableMarketTypes.SPOT: HeatPumpOrderUpdaterParameters( + initial_rate=5, final_rate=25 + ) + } + strategy = self._create_strategy(order_updater_parameters=custom_params) + assert strategy._order_updater_params == custom_params + + def test_event_market_cycle_creates_order_updater_for_spot_market(self): + strategy, area = self._create_activated_strategy() + strategy._energy_params.get_max_energy_demand_kWh = MagicMock(return_value=1.0) + strategy.event_market_cycle() + market_object = area.spot_market + assert len(strategy._order_updaters[market_object].keys()) == 1 + + def test_event_market_cycle_posts_bid_on_spot_market(self): + strategy, area = self._create_activated_strategy() + energy_to_buy = 2.5 + strategy._energy_params.get_max_energy_demand_kWh = MagicMock(return_value=energy_to_buy) + strategy.event_market_cycle() + bids = list(area.spot_market.bids.values()) + assert len(bids) == 1 + assert math.isclose(bids[0].energy, energy_to_buy) + + def test_post_order_with_explicit_rate_calls_internal_post_order(self): + strategy, area = self._create_activated_strategy() + market = area.spot_market + market_slot = area.spot_market.time_slot + strategy._energy_params.soc_management.calculate = MagicMock(return_value=1.5) + strategy._post_order = MagicMock() + strategy.post_order(market, market_slot, order_rate=15.0) + strategy._post_order.assert_called_once() + + def test_post_order_uses_soc_management_calculate_for_energy(self): + strategy, area = self._create_activated_strategy() + market = area.spot_market + market_slot = area.spot_market.time_slot + expected_energy = 2.0 + strategy._energy_params.soc_management.calculate = MagicMock(return_value=expected_energy) + captured_args = {} + + def capture_post_order(_mkt, _slot, energy, _rate): + captured_args["energy"] = float(energy) + + strategy._post_order = capture_post_order + strategy.post_order(market, market_slot, order_rate=15.0) + assert math.isclose(captured_args["energy"], expected_energy) + + def test_state_is_same_object_as_energy_params_state(self): + strategy = self._create_strategy() + assert strategy.state is strategy._energy_params.state + + def test_serialize_contains_all_energy_param_keys(self): + strategy = self._create_strategy() + result = strategy.serialize() + for key in ("heat_demand_Q_J", "ambient_temp_C", "target_temp_C", "source_type"): + assert key in result + + def test_serialize_contains_all_order_updater_keys(self): + strategy = self._create_strategy() + result = strategy.serialize() + for key in ( + "update_interval", + "initial_buying_rate", + "final_buying_rate", + "use_market_maker_rate", + ): + assert key in result + + def test_serialize_source_type_matches_default(self): + strategy = self._create_strategy() + result = strategy.serialize() + assert result["source_type"] == ConstSettings.HeatPumpSettings.SOURCE_TYPE