|
| 1 | +"""Effective grid-charge SoC target calculation.""" |
| 2 | + |
| 3 | +from dataclasses import dataclass |
| 4 | +from typing import Optional, Sequence |
| 5 | + |
| 6 | +from .logic_interface import CalculationInput |
| 7 | + |
| 8 | +GRID_CHARGE_TARGET_STRATEGY_FIXED = 'fixed' |
| 9 | +GRID_CHARGE_TARGET_STRATEGY_FORECAST = 'forecast' |
| 10 | +GRID_CHARGE_TARGET_STRATEGIES = ( |
| 11 | + GRID_CHARGE_TARGET_STRATEGY_FIXED, |
| 12 | + GRID_CHARGE_TARGET_STRATEGY_FORECAST, |
| 13 | +) |
| 14 | + |
| 15 | + |
| 16 | +@dataclass(frozen=True) |
| 17 | +class GridChargeTargetConfig: |
| 18 | + """Configuration for effective grid-charge target calculation.""" |
| 19 | + |
| 20 | + strategy: str = GRID_CHARGE_TARGET_STRATEGY_FIXED |
| 21 | + pv_forecast_factor: float = 1.0 |
| 22 | + |
| 23 | + @classmethod |
| 24 | + def from_battery_control_config(cls, config: dict) -> 'GridChargeTargetConfig': |
| 25 | + """Create target strategy configuration from battery_control config.""" |
| 26 | + return cls( |
| 27 | + strategy=_parse_grid_charge_target_strategy( |
| 28 | + config.get( |
| 29 | + 'grid_charge_target_strategy', |
| 30 | + GRID_CHARGE_TARGET_STRATEGY_FIXED, |
| 31 | + ) |
| 32 | + ), |
| 33 | + pv_forecast_factor=_parse_ratio( |
| 34 | + config.get('grid_charge_forecast_pv_factor', 1.0), |
| 35 | + 'battery_control.grid_charge_forecast_pv_factor', |
| 36 | + ), |
| 37 | + ) |
| 38 | + |
| 39 | + |
| 40 | +def _parse_ratio(value, config_key: str) -> float: |
| 41 | + """Parse a required 0..1 ratio config value.""" |
| 42 | + try: |
| 43 | + ratio = float(value) |
| 44 | + except (TypeError, ValueError) as exc: |
| 45 | + raise ValueError( |
| 46 | + f"{config_key} must be numeric between 0 and 1, got {value!r}" |
| 47 | + ) from exc |
| 48 | + if not 0 <= ratio <= 1: |
| 49 | + raise ValueError( |
| 50 | + f"{config_key} must be between 0 and 1, got {ratio}" |
| 51 | + ) |
| 52 | + return ratio |
| 53 | + |
| 54 | + |
| 55 | +def _parse_grid_charge_target_strategy(value) -> str: |
| 56 | + """Parse the grid-charge target strategy config value.""" |
| 57 | + strategy = str(value).strip().lower() |
| 58 | + if strategy not in GRID_CHARGE_TARGET_STRATEGIES: |
| 59 | + raise ValueError( |
| 60 | + f"battery_control.grid_charge_target_strategy must be one of " |
| 61 | + f"{GRID_CHARGE_TARGET_STRATEGIES}, got {value!r}" |
| 62 | + ) |
| 63 | + return strategy |
| 64 | + |
| 65 | + |
| 66 | +def _ordered_values(values) -> Sequence[float]: |
| 67 | + if isinstance(values, dict): |
| 68 | + if not values: |
| 69 | + return [] |
| 70 | + keys = set(values.keys()) |
| 71 | + error_message = ( |
| 72 | + "forecast dict values must use consecutive integer " |
| 73 | + "indices starting at 0" |
| 74 | + ) |
| 75 | + if not all(isinstance(index, int) for index in keys): |
| 76 | + raise ValueError(error_message) |
| 77 | + expected_keys = set(range(max(keys) + 1)) |
| 78 | + if keys != expected_keys: |
| 79 | + raise ValueError(error_message) |
| 80 | + return [values[index] for index in range(max(keys) + 1)] |
| 81 | + return list(values) |
| 82 | + |
| 83 | + |
| 84 | +def _calculate_min_dynamic_price_difference( |
| 85 | + current_price: float, |
| 86 | + min_price_difference: float, |
| 87 | + min_price_difference_rel: float) -> float: |
| 88 | + return max(min_price_difference, |
| 89 | + min_price_difference_rel * abs(current_price)) |
| 90 | + |
| 91 | + |
| 92 | +def calculate_effective_min_grid_charge_soc( |
| 93 | + config: GridChargeTargetConfig, |
| 94 | + calc_input: CalculationInput, |
| 95 | + configured_min_grid_charge_soc: Optional[float], |
| 96 | + max_charging_from_grid_limit: float, |
| 97 | + max_capacity: float, |
| 98 | + min_price_difference: float, |
| 99 | + min_price_difference_rel: float = 0.0) -> Optional[float]: |
| 100 | + """Calculate the effective minimum grid-charge SoC from logic inputs.""" |
| 101 | + min_soc_energy = max( |
| 102 | + 0.0, |
| 103 | + calc_input.stored_energy - calc_input.stored_usable_energy, |
| 104 | + ) |
| 105 | + return calculate_effective_grid_charge_soc( |
| 106 | + strategy=config.strategy, |
| 107 | + configured_min_grid_charge_soc=configured_min_grid_charge_soc, |
| 108 | + max_charging_from_grid_limit=max_charging_from_grid_limit, |
| 109 | + max_capacity=max_capacity, |
| 110 | + min_soc_energy=min_soc_energy, |
| 111 | + production=calc_input.production, |
| 112 | + consumption=calc_input.consumption, |
| 113 | + prices=calc_input.prices, |
| 114 | + min_price_difference=min_price_difference, |
| 115 | + min_price_difference_rel=min_price_difference_rel, |
| 116 | + pv_forecast_factor=config.pv_forecast_factor, |
| 117 | + ) |
| 118 | + |
| 119 | + |
| 120 | +def calculate_effective_grid_charge_soc( |
| 121 | + strategy: str, |
| 122 | + configured_min_grid_charge_soc: Optional[float], |
| 123 | + max_charging_from_grid_limit: float, |
| 124 | + max_capacity: float, |
| 125 | + min_soc_energy: float, |
| 126 | + production, |
| 127 | + consumption, |
| 128 | + prices, |
| 129 | + min_price_difference: float, |
| 130 | + min_price_difference_rel: float = 0.0, |
| 131 | + pv_forecast_factor: float = 1.0) -> Optional[float]: |
| 132 | + """Calculate the effective minimum grid-charge SoC for this evaluation. |
| 133 | +
|
| 134 | + ``fixed`` returns the configured target unchanged. ``forecast`` treats the |
| 135 | + configured target as a floor and raises it when future expensive-slot net |
| 136 | + demand implies a higher target. Forecast PV can be discounted with |
| 137 | + ``pv_forecast_factor`` to account for uncertain PV ramps. |
| 138 | + """ |
| 139 | + if configured_min_grid_charge_soc is None: |
| 140 | + return None |
| 141 | + if strategy not in GRID_CHARGE_TARGET_STRATEGIES: |
| 142 | + raise ValueError( |
| 143 | + f"grid_charge_target_strategy must be one of " |
| 144 | + f"{GRID_CHARGE_TARGET_STRATEGIES}, got '{strategy}'" |
| 145 | + ) |
| 146 | + if strategy == GRID_CHARGE_TARGET_STRATEGY_FIXED: |
| 147 | + return configured_min_grid_charge_soc |
| 148 | + if max_capacity <= 0: |
| 149 | + raise ValueError("max_capacity must be greater than 0") |
| 150 | + if not 0 <= pv_forecast_factor <= 1: |
| 151 | + raise ValueError( |
| 152 | + "grid_charge_forecast_pv_factor must be between 0 and 1, " |
| 153 | + f"got {pv_forecast_factor}" |
| 154 | + ) |
| 155 | + |
| 156 | + production_values = _ordered_values(production) |
| 157 | + consumption_values = _ordered_values(consumption) |
| 158 | + price_values = _ordered_values(prices) |
| 159 | + max_slot = min(len(production_values), len(consumption_values), len(price_values)) |
| 160 | + if max_slot < 2: |
| 161 | + return configured_min_grid_charge_soc |
| 162 | + |
| 163 | + current_price = price_values[0] |
| 164 | + min_dynamic_price_difference = _calculate_min_dynamic_price_difference( |
| 165 | + current_price, |
| 166 | + min_price_difference, |
| 167 | + min_price_difference_rel, |
| 168 | + ) |
| 169 | + |
| 170 | + # Evaluate until the next price slot that is no more expensive than the |
| 171 | + # current slot. This keeps the target tied to the current cheap/economical |
| 172 | + # charging window rather than charging for the whole forecast horizon. |
| 173 | + for slot in range(1, max_slot): |
| 174 | + if price_values[slot] <= current_price: |
| 175 | + max_slot = slot |
| 176 | + break |
| 177 | + |
| 178 | + forecast_need = 0.0 |
| 179 | + for slot in range(1, max_slot): |
| 180 | + if price_values[slot] <= current_price + min_dynamic_price_difference: |
| 181 | + continue |
| 182 | + discounted_pv = production_values[slot] * pv_forecast_factor |
| 183 | + forecast_need += max(0.0, consumption_values[slot] - discounted_pv) |
| 184 | + |
| 185 | + forecast_target = (min_soc_energy + forecast_need) / max_capacity |
| 186 | + forecast_target = min(forecast_target, max_charging_from_grid_limit) |
| 187 | + return max(configured_min_grid_charge_soc, forecast_target) |
0 commit comments