bjorn/esc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: solar forecast charging

Browse Source
This commit is contained in:
Björn Stormwall
2026-06-23 15:22:05 +02:00
parent 27ed07f0c4
commit d2f7c1dc73
7 changed files with 676 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/easee_solar_charging/config_flow.py
+30
View File
@@ -10,8 +10,15 @@ from .const import (
CONF_CHARGER_ID,
CHARGING_PROFILE_ALL_SURPLUS,
CHARGING_PROFILE_CONSERVATIVE,
CHARGING_PROFILE_FORECAST,
CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_CHARGING_PROFILE,
CONF_DEAD_BAND_RESYNC_MINUTES,
CONF_FORECAST_CONFIDENCE,
CONF_FORECAST_LOOKAHEAD_MINUTES,
CONF_FORECAST_SENSOR,
DEFAULT_FORECAST_CONFIDENCE,
DEFAULT_FORECAST_LOOKAHEAD_MINUTES,
CONF_DEVICE_ID,
CONF_EV_CHARGING_SENSOR,
CONF_HOUSE_LOAD_SENSOR,
@@ -43,11 +50,21 @@ _CHARGING_PROFILE_SELECTOR = selector.SelectSelector(
options=[
selector.SelectOptionDict(value=CHARGING_PROFILE_CONSERVATIVE, label="Conservative"),
selector.SelectOptionDict(value=CHARGING_PROFILE_ALL_SURPLUS, label="All Surplus"),
selector.SelectOptionDict(value=CHARGING_PROFILE_FORECAST, label="Forecast look-ahead"),
selector.SelectOptionDict(value=CHARGING_PROFILE_FORECAST_CONSERVATIVE, label="Forecast surplus"),
],
mode=selector.SelectSelectorMode.DROPDOWN,
)
)
_CONFIDENCE_SELECTOR = selector.NumberSelector(
selector.NumberSelectorConfig(min=50, max=100, step=5, unit_of_measurement="%", mode=selector.NumberSelectorMode.SLIDER)
)
_LOOKAHEAD_SELECTOR = selector.NumberSelector(
selector.NumberSelectorConfig(min=5, max=120, step=5, unit_of_measurement="min", mode=selector.NumberSelectorMode.BOX)
)
_POWER_SENSOR_SELECTOR = selector.EntitySelector(
selector.EntitySelectorConfig(domain="sensor", device_class="power")
)
@@ -103,6 +120,19 @@ def _options_schema(hass: HomeAssistant, current: dict) -> vol.Schema:
CONF_DEAD_BAND_RESYNC_MINUTES,
default=current.get(CONF_DEAD_BAND_RESYNC_MINUTES, DEFAULT_DEAD_BAND_RESYNC_MINUTES),
): _DEAD_BAND_RESYNC_SELECTOR,
# ── Forecast profiles ─────────────────────────────────────────────────
vol.Optional(CONF_FORECAST_SENSOR): selector.EntitySelector(
selector.EntitySelectorConfig(integration="forecast_solar")
),
vol.Required(
CONF_FORECAST_CONFIDENCE,
default=current.get(CONF_FORECAST_CONFIDENCE, DEFAULT_FORECAST_CONFIDENCE),
): _CONFIDENCE_SELECTOR,
vol.Required(
CONF_FORECAST_LOOKAHEAD_MINUTES,
default=current.get(CONF_FORECAST_LOOKAHEAD_MINUTES, DEFAULT_FORECAST_LOOKAHEAD_MINUTES),
): _LOOKAHEAD_SELECTOR,
# ─────────────────────────────────────────────────────────────────────
vol.Optional(CONF_NOTIFY_TARGET): _notify_selector(_notify_services(hass)),
})
custom_components/easee_solar_charging/const.py
+10
View File
@@ -32,8 +32,18 @@ DEFAULT_STOP_GRACE_MINUTES = 5
CONF_CHARGING_PROFILE = "charging_profile"
CHARGING_PROFILE_CONSERVATIVE = "conservative"
CHARGING_PROFILE_ALL_SURPLUS = "all_surplus"
CHARGING_PROFILE_FORECAST = "forecast"
CHARGING_PROFILE_FORECAST_CONSERVATIVE = "forecast_conservative"
DEFAULT_CHARGING_PROFILE = CHARGING_PROFILE_CONSERVATIVE
# Forecast profile settings
CONF_FORECAST_SENSOR = "forecast_sensor"
CONF_FORECAST_CONFIDENCE = "forecast_confidence" # percent, 0-100
CONF_FORECAST_LOOKAHEAD_MINUTES = "forecast_lookahead_minutes"
DEFAULT_FORECAST_CONFIDENCE = 80 # percent
DEFAULT_FORECAST_LOOKAHEAD_MINUTES = 30
CONF_DEAD_BAND_RESYNC_MINUTES = "dead_band_resync_minutes"
DEFAULT_DEAD_BAND_RESYNC_MINUTES = 1
custom_components/easee_solar_charging/coordinator.py
+139 -30
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
from collections import deque
from datetime import datetime, timedelta, timezone
import logging
from zoneinfo import ZoneInfo
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
@@ -13,7 +14,14 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import (
CHARGING_PROFILE_ALL_SURPLUS,
CHARGING_PROFILE_CONSERVATIVE,
CHARGING_PROFILE_FORECAST,
CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_CHARGING_PROFILE,
CONF_FORECAST_CONFIDENCE,
CONF_FORECAST_LOOKAHEAD_MINUTES,
CONF_FORECAST_SENSOR,
DEFAULT_FORECAST_CONFIDENCE,
DEFAULT_FORECAST_LOOKAHEAD_MINUTES,
CONF_CHARGER_ID,
CONF_DEAD_BAND_RESYNC_MINUTES,
CONF_DEVICE_ID,
@@ -112,6 +120,75 @@ class EaseeSolarCoordinator(DataUpdateCoordinator):
minutes = self._entry.options.get(CONF_DEAD_BAND_RESYNC_MINUTES, DEFAULT_DEAD_BAND_RESYNC_MINUTES)
return timedelta(minutes=minutes)
def _w_at_lookahead(self, watts: dict) -> float | None:
"""Find the watt value in a {timestamp: W} dict nearest to now + lookahead_minutes.
Handles both naive local-time strings ("YYYY-MM-DD HH:MM:SS") and
ISO strings with timezone offset, plus datetime keys serialised by HA.
"""
lookahead = self._entry.options.get(CONF_FORECAST_LOOKAHEAD_MINUTES, DEFAULT_FORECAST_LOOKAHEAD_MINUTES)
try:
tz = ZoneInfo(self.hass.config.time_zone)
except Exception:
tz = timezone.utc
target = datetime.now(tz) + timedelta(minutes=lookahead)
best_delta: float | None = None
best_w: float | None = None
for ts_key, w in watts.items():
try:
if isinstance(ts_key, datetime):
ts = ts_key if ts_key.tzinfo else ts_key.replace(tzinfo=tz)
else:
ts = datetime.fromisoformat(str(ts_key))
if ts.tzinfo is None:
ts = ts.replace(tzinfo=tz)
except ValueError:
continue
delta = abs((ts - target).total_seconds())
if best_delta is None or delta < best_delta:
best_delta = delta
best_w = float(w)
return best_w
def _get_raw_forecast_w(self) -> float | None:
"""Return raw forecast power from the Forecast.Solar HA integration sensor.
Prefers the 'watts' attribute (dict of period→W) for a true look-ahead.
Falls back to the sensor state value when the attribute is absent.
"""
entity_id = self._entry.options.get(CONF_FORECAST_SENSOR)
if not entity_id:
return None
state = self.hass.states.get(entity_id)
if state is None or state.state in ("unknown", "unavailable"):
return None
watts = state.attributes.get("watts")
if isinstance(watts, dict) and watts:
w = self._w_at_lookahead(watts)
if w is not None:
return w
try:
return float(state.state)
except ValueError:
_LOGGER.warning("Forecast sensor %s has non-numeric state: %s", entity_id, state.state)
return None
def _get_adjusted_forecast_w(self) -> float | None:
raw = self._get_raw_forecast_w()
if raw is None:
return None
confidence = self._entry.options.get(CONF_FORECAST_CONFIDENCE, DEFAULT_FORECAST_CONFIDENCE) / 100
adjusted = raw * confidence
lookahead = self._entry.options.get(CONF_FORECAST_LOOKAHEAD_MINUTES, DEFAULT_FORECAST_LOOKAHEAD_MINUTES)
_LOGGER.debug(
"Forecast %.0f W × %.0f%% confidence = %.0f W (lookahead %d min)",
raw, confidence * 100, adjusted, lookahead,
)
return adjusted
def _get_solar_excess_w(self) -> tuple[float | None, float | None, float | None, float | None]:
profile = self._entry.options.get(CONF_CHARGING_PROFILE, DEFAULT_CHARGING_PROFILE)
production = self._rolling_average(self._production_sensor)
@@ -122,10 +199,35 @@ class EaseeSolarCoordinator(DataUpdateCoordinator):
return None, None, house_load, ev_charging
if profile == CHARGING_PROFILE_ALL_SURPLUS:
# Allocate all solar production to EV; house load draws from grid.
surplus = production
else:
assert profile == CHARGING_PROFILE_CONSERVATIVE
elif profile == CHARGING_PROFILE_FORECAST:
# Conservative surplus first; if below threshold, check whether
# the confidence-adjusted forecast justifies an early start.
actual = production - house_load - ev_charging
forecast = self._get_adjusted_forecast_w()
if forecast is not None and forecast > actual:
_LOGGER.debug(
"Forecast profile: using forecast %.0f W instead of actual %.0f W",
forecast, actual,
)
surplus = forecast
else:
surplus = actual
elif profile == CHARGING_PROFILE_FORECAST_CONSERVATIVE:
# Like Forecast but subtracts current house load and EV draw from the
# forecast value — earlier start than Conservative, more cautious than
# plain Forecast which ignores current consumption entirely.
forecast = self._get_adjusted_forecast_w()
if forecast is not None:
surplus = forecast - house_load - ev_charging
else:
surplus = production - house_load - ev_charging
else: # CHARGING_PROFILE_CONSERVATIVE (default) or unrecognised value
if profile != CHARGING_PROFILE_CONSERVATIVE:
_LOGGER.warning("Unknown charging profile %r — falling back to conservative", profile)
surplus = production - house_load - ev_charging
return surplus, production, house_load, ev_charging
@@ -233,36 +335,43 @@ class EaseeSolarCoordinator(DataUpdateCoordinator):
async def _async_update_data(self) -> dict:
try:
solar_excess_w, production_w, house_load_w, ev_charging_w = self._get_solar_excess_w()
raw_target = (
solar_excess_to_charger_current(solar_excess_w)
if self.solar_charging_enabled and solar_excess_w is not None
else 0
)
target_current_a = self._apply_stop_hysteresis(raw_target)
if self._charger_is_sleeping():
# Session finished/waiting: zero out once then stop adjusting.
if self._last_sent_current != 0:
await self._async_set_charger_current(0)
self._last_sent_current = 0
self._last_sent_at = datetime.now(timezone.utc)
if not self._charger_allows_control() and not self._charger_is_sleeping():
# Disconnected / error / unknown — no amps to calculate.
# Clear any running grace period so it doesn't bleed into the next session.
self._below_threshold_since = None
target_current_a = 0
elif self._should_send(target_current_a) and self._charger_allows_control():
prev = self._last_sent_current
await self._async_set_charger_current(target_current_a)
self._last_sent_current = target_current_a
self._last_sent_at = datetime.now(timezone.utc)
else:
raw_target = (
solar_excess_to_charger_current(solar_excess_w)
if self.solar_charging_enabled and solar_excess_w is not None
else 0
)
target_current_a = self._apply_stop_hysteresis(raw_target)
if prev in (None, 0) and target_current_a > 0:
await self._async_notify(
"Solar charging started",
f"Solar excess is sufficient — charging at {target_current_a} A.",
)
elif prev and prev > 0 and target_current_a == 0:
await self._async_notify(
"Solar charging stopped",
"Solar excess dropped below the minimum threshold.",
)
if self._charger_is_sleeping():
# Session finished/waiting: zero out once then stop adjusting.
if self._last_sent_current != 0:
await self._async_set_charger_current(0)
self._last_sent_current = 0
self._last_sent_at = datetime.now(timezone.utc)
target_current_a = 0
elif self._should_send(target_current_a) and self._charger_allows_control():
prev = self._last_sent_current
await self._async_set_charger_current(target_current_a)
self._last_sent_current = target_current_a
self._last_sent_at = datetime.now(timezone.utc)
if prev in (None, 0) and target_current_a > 0:
await self._async_notify(
"Solar charging started",
f"Solar excess is sufficient — charging at {target_current_a} A.",
)
elif prev and prev > 0 and target_current_a == 0:
await self._async_notify(
"Solar charging stopped",
"Solar excess dropped below the minimum threshold.",
)
return {
"charger_id": self.charger_id,
custom_components/easee_solar_charging/translations/en.json
+7 -1
View File
@@ -17,12 +17,18 @@
"charging_profile": "Charging profile",
"stop_grace_minutes": "Stop grace period",
"dead_band_resync_minutes": "Dead-band resync interval",
"forecast_sensor": "Forecast.Solar sensor",
"forecast_confidence": "Forecast confidence",
"forecast_lookahead_minutes": "Look-ahead window",
"notify_target": "Notification target"
},
"data_description": {
"charging_profile": "Conservative: charge only on true solar surplus (production − house load − current EV draw). All Surplus: allocate all solar production to the EV; house load draws from the grid.",
"charging_profile": "Conservative: charge only on true solar surplus (production − house load − current EV draw). All Surplus: allocate all solar production to the EV; house load draws from the grid. Forecast look-ahead: start charging early using the full confidence-adjusted forecast as the surplus. Forecast surplus: same but subtracts current house load and EV draw — more cautious early start.",
"stop_grace_minutes": "How many minutes to keep charging after solar excess drops below the minimum threshold, before stopping the charger.",
"dead_band_resync_minutes": "How often to re-send the current charging level when within the dead-band, to track gradual solar drift.",
"forecast_sensor": "Pick a power sensor from your Forecast.Solar integration. If the sensor exposes a 'watts' attribute the look-ahead window is used to find the matching period; otherwise the sensor state is used directly.",
"forecast_confidence": "Scale the forecast value by this factor before comparing to the charging threshold. At 80%, a 7000 W forecast is treated as 5600 W.",
"forecast_lookahead_minutes": "How many minutes ahead to look in the Forecast.Solar power curve when deciding to start early.",
"notify_target": "Name of a notify service to receive start/stop alerts (e.g. mobile_app_my_phone). Leave empty to disable notifications."
}
}
tests/conftest.py
+17 -4
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import sys
from collections import deque
from datetime import timedelta
from unittest.mock import MagicMock
@@ -39,6 +40,7 @@ for _mod in [
"homeassistant.helpers",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.entity_platform",
"homeassistant.helpers.event",
"homeassistant.helpers.restore_state",
"homeassistant.helpers.selector",
"voluptuous",
@@ -59,7 +61,9 @@ from custom_components.easee_solar_charging.const import ( # noqa: E402
)
CHARGER_ID = "ehxt9pqp"
SOLAR_SENSOR = "sensor.solar_excess"
PRODUCTION_SENSOR = "sensor.solar_production"
HOUSE_SENSOR = "sensor.house_load"
EV_SENSOR = "sensor.ev_charging"
class FakeCoordinator(EaseeSolarCoordinator):
@@ -68,8 +72,16 @@ class FakeCoordinator(EaseeSolarCoordinator):
def __init__(self, options: dict | None = None) -> None:
# Skip DataUpdateCoordinator.__init__ — not needed for logic-only tests.
self.hass = MagicMock()
self.hass.config.time_zone = "UTC"
self.charger_id = CHARGER_ID
self._solar_excess_sensor = SOLAR_SENSOR
self._production_sensor = PRODUCTION_SENSOR
self._house_load_sensor = HOUSE_SENSOR
self._ev_charging_sensor = EV_SENSOR
self._samples: dict = {
PRODUCTION_SENSOR: deque(),
HOUSE_SENSOR: deque(),
EV_SENSOR: deque(),
}
self._device_id = None
self._last_sent_current = None
self._last_sent_at = None
@@ -87,8 +99,9 @@ def coord() -> FakeCoordinator:
return FakeCoordinator()
def make_state(value: str) -> MagicMock:
"""Return a minimal HA state mock with the given state string."""
def make_state(value: str, attributes: dict | None = None) -> MagicMock:
"""Return a minimal HA state mock with the given state string and optional attributes."""
s = MagicMock()
s.state = value
s.attributes = attributes or {}
return s
tests/test_coordinator.py
-35
View File
@@ -145,41 +145,6 @@ class TestApplyStopHysteresis:
assert result == 0
# ---------------------------------------------------------------------------
# _get_solar_excess_w
# ---------------------------------------------------------------------------
class TestGetSolarExcessW:
def test_valid_numeric_state(self, coord):
coord.hass.states.get.return_value = make_state("3500.5")
assert coord._get_solar_excess_w() == 3500.5
def test_integer_state(self, coord):
coord.hass.states.get.return_value = make_state("5000")
assert coord._get_solar_excess_w() == 5000.0
def test_unavailable_returns_none(self, coord):
coord.hass.states.get.return_value = make_state("unavailable")
assert coord._get_solar_excess_w() is None
def test_unknown_returns_none(self, coord):
coord.hass.states.get.return_value = make_state("unknown")
assert coord._get_solar_excess_w() is None
def test_missing_entity_returns_none(self, coord):
coord.hass.states.get.return_value = None
assert coord._get_solar_excess_w() is None
def test_non_numeric_returns_none(self, coord):
coord.hass.states.get.return_value = make_state("not_a_number")
assert coord._get_solar_excess_w() is None
def test_queries_correct_entity(self, coord):
coord.hass.states.get.return_value = make_state("1000")
coord._get_solar_excess_w()
coord.hass.states.get.assert_called_once_with(coord._solar_excess_sensor)
# ---------------------------------------------------------------------------
# _charger_allows_control
# ---------------------------------------------------------------------------
tests/test_forecast.py
+473
View File
@@ -0,0 +1,473 @@
"""
Unit tests for forecast-related coordinator methods.
Covers: _w_at_lookahead, _get_raw_forecast_w, _get_adjusted_forecast_w,
and the forecast profile branches of _get_solar_excess_w.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import pytest
from freezegun import freeze_time
from tests.conftest import FakeCoordinator, make_state
from custom_components.easee_solar_charging.const import (
CHARGING_PROFILE_ALL_SURPLUS,
CHARGING_PROFILE_CONSERVATIVE,
CHARGING_PROFILE_FORECAST,
CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_CHARGING_PROFILE,
CONF_FORECAST_CONFIDENCE,
CONF_FORECAST_LOOKAHEAD_MINUTES,
CONF_FORECAST_SENSOR,
DEFAULT_FORECAST_CONFIDENCE,
DEFAULT_FORECAST_LOOKAHEAD_MINUTES,
MIN_CHARGER_CURRENT_A,
PHASES,
VOLTAGE_V,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_FREEZE = "2024-06-15 10:00:00" # UTC; used by freezegun-based tests
_FREEZE_DT = datetime(2024, 6, 15, 10, 0, 0, tzinfo=timezone.utc)
_FORECAST_ENTITY = "sensor.forecast_solar_power_production_now"
def _coord(options: dict | None = None) -> FakeCoordinator:
return FakeCoordinator(options=options)
def _utc_str(dt: datetime) -> str:
"""ISO string with explicit UTC offset."""
return dt.strftime("%Y-%m-%dT%H:%M:%S+00:00")
def _naive_str(dt: datetime) -> str:
"""Naive local-time string — the raw Forecast.Solar API format."""
return dt.strftime("%Y-%m-%d %H:%M:%S")
def _set_rolling(
coord: FakeCoordinator,
production: float | None,
house: float = 0.0,
ev: float = 0.0,
) -> None:
"""Patch _rolling_average on coord with fixed values for all three sensors."""
sensor_map = {
coord._production_sensor: production,
coord._house_load_sensor: house,
coord._ev_charging_sensor: ev,
}
coord._rolling_average = lambda entity_id: sensor_map.get(entity_id)
def _set_forecast_state(coord: FakeCoordinator, value: str, attributes: dict | None = None) -> None:
"""Make coord.hass.states.get return the given state for the forecast sensor."""
coord.hass.states.get.return_value = make_state(value, attributes)
# ---------------------------------------------------------------------------
# _w_at_lookahead
# ---------------------------------------------------------------------------
class TestWAtLookahead:
"""Timestamp-nearest lookup in a Forecast.Solar watts dict."""
@freeze_time(_FREEZE)
def test_exact_match_returns_correct_watts(self):
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
watts = {
_utc_str(_FREEZE_DT): 1000.0,
_utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0,
_utc_str(_FREEZE_DT + timedelta(minutes=60)): 7000.0,
}
assert coord._w_at_lookahead(watts) == 5000.0
@freeze_time(_FREEZE)
def test_picks_entry_nearest_to_target(self):
"""When no exact match, the closest timestamp wins."""
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
# +20 min is 10 min before target; +50 min is 20 min after — +20 wins
watts = {
_utc_str(_FREEZE_DT + timedelta(minutes=20)): 3000.0,
_utc_str(_FREEZE_DT + timedelta(minutes=50)): 6000.0,
}
assert coord._w_at_lookahead(watts) == 3000.0
@freeze_time(_FREEZE)
def test_naive_string_keys_treated_as_utc(self):
"""Naive 'YYYY-MM-DD HH:MM:SS' strings get the local timezone (UTC here)."""
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
target = _FREEZE_DT + timedelta(minutes=30)
watts = {
_naive_str(target - timedelta(hours=1)): 1000.0,
_naive_str(target): 5000.0,
_naive_str(target + timedelta(hours=1)): 3000.0,
}
assert coord._w_at_lookahead(watts) == 5000.0
@freeze_time(_FREEZE)
def test_datetime_object_keys(self):
"""datetime objects as dict keys are handled directly."""
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
target = _FREEZE_DT + timedelta(minutes=30)
watts = {
_FREEZE_DT: 1000.0,
target: 8000.0,
_FREEZE_DT + timedelta(hours=1): 5000.0,
}
assert coord._w_at_lookahead(watts) == 8000.0
@freeze_time(_FREEZE)
def test_empty_dict_returns_none(self):
coord = _coord()
assert coord._w_at_lookahead({}) is None
@freeze_time(_FREEZE)
def test_malformed_keys_are_skipped(self):
"""Unparseable timestamp strings are silently skipped."""
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
watts = {
"not-a-timestamp": 9999.0,
_utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0,
}
assert coord._w_at_lookahead(watts) == 5000.0
@freeze_time(_FREEZE)
def test_respects_lookahead_minutes_option(self):
"""Different lookahead values select different entries."""
coord_30 = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
coord_60 = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 60})
watts = {
_utc_str(_FREEZE_DT + timedelta(minutes=30)): 4000.0,
_utc_str(_FREEZE_DT + timedelta(minutes=60)): 8000.0,
}
assert coord_30._w_at_lookahead(watts) == 4000.0
assert coord_60._w_at_lookahead(watts) == 8000.0
@freeze_time(_FREEZE)
def test_default_lookahead_is_30_minutes(self):
"""With no option set, the default lookahead is 30 minutes."""
coord = _coord(options={})
watts = {
_utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0,
_utc_str(_FREEZE_DT + timedelta(minutes=60)): 9000.0,
}
assert coord._w_at_lookahead(watts) == 5000.0
@freeze_time(_FREEZE)
def test_invalid_timezone_falls_back_to_utc(self):
"""An unresolvable timezone string falls back to UTC without crashing."""
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
coord.hass.config.time_zone = "Not/AReal_Timezone"
watts = {_utc_str(_FREEZE_DT + timedelta(minutes=30)): 5000.0}
assert coord._w_at_lookahead(watts) == 5000.0
@freeze_time(_FREEZE)
def test_single_entry_is_always_returned(self):
"""A dict with only one entry always returns that entry."""
coord = _coord(options={CONF_FORECAST_LOOKAHEAD_MINUTES: 30})
watts = {_utc_str(_FREEZE_DT + timedelta(hours=5)): 2500.0}
assert coord._w_at_lookahead(watts) == 2500.0
# ---------------------------------------------------------------------------
# _get_raw_forecast_w
# ---------------------------------------------------------------------------
class TestGetRawForecastW:
"""Reading the raw (pre-confidence) forecast value from a HA sensor."""
def test_returns_none_when_no_sensor_configured(self):
coord = _coord(options={})
assert coord._get_raw_forecast_w() is None
def test_returns_none_when_sensor_state_unavailable(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "unavailable")
assert coord._get_raw_forecast_w() is None
def test_returns_none_when_sensor_state_unknown(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "unknown")
assert coord._get_raw_forecast_w() is None
def test_returns_none_when_sensor_entity_missing(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
coord.hass.states.get.return_value = None
assert coord._get_raw_forecast_w() is None
@freeze_time(_FREEZE)
def test_uses_watts_attribute_for_lookahead(self):
"""When the sensor has a 'watts' attribute, uses it for the lookahead lookup."""
coord = _coord(options={
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_LOOKAHEAD_MINUTES: 30,
})
watts_attr = {_utc_str(_FREEZE_DT + timedelta(minutes=30)): 6500.0}
_set_forecast_state(coord, "3000", {"watts": watts_attr})
# Should return the lookahead value (6500), not the sensor state (3000)
assert coord._get_raw_forecast_w() == 6500.0
def test_falls_back_to_sensor_state_when_no_watts_attribute(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "4200.5")
assert coord._get_raw_forecast_w() == 4200.5
def test_falls_back_to_sensor_state_when_watts_attribute_is_empty(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "3000.0", {"watts": {}})
assert coord._get_raw_forecast_w() == 3000.0
def test_returns_none_for_non_numeric_state_with_no_watts_attr(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "banana")
assert coord._get_raw_forecast_w() is None
def test_ignores_non_dict_watts_attribute(self):
"""A 'watts' attribute that is not a dict is ignored; state is used instead."""
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "5000.0", {"watts": "not-a-dict"})
assert coord._get_raw_forecast_w() == 5000.0
def test_state_value_is_returned_as_float(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "7000")
result = coord._get_raw_forecast_w()
assert result == 7000.0
assert isinstance(result, float)
# ---------------------------------------------------------------------------
# _get_adjusted_forecast_w
# ---------------------------------------------------------------------------
class TestGetAdjustedForecastW:
"""Confidence factor is applied to the raw forecast value."""
def test_default_confidence_is_80_percent(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "10000.0")
# 10 000 × 0.80 = 8 000
assert coord._get_adjusted_forecast_w() == pytest.approx(8000.0)
def test_custom_confidence_is_applied(self):
coord = _coord(options={
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 60,
})
_set_forecast_state(coord, "5000.0")
# 5 000 × 0.60 = 3 000
assert coord._get_adjusted_forecast_w() == pytest.approx(3000.0)
def test_100_percent_confidence_returns_raw_value(self):
coord = _coord(options={
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_forecast_state(coord, "7000.0")
assert coord._get_adjusted_forecast_w() == pytest.approx(7000.0)
def test_returns_none_when_sensor_unavailable(self):
coord = _coord(options={CONF_FORECAST_SENSOR: _FORECAST_ENTITY})
_set_forecast_state(coord, "unavailable")
assert coord._get_adjusted_forecast_w() is None
def test_returns_none_when_no_sensor_configured(self):
coord = _coord(options={})
assert coord._get_adjusted_forecast_w() is None
def test_50_percent_confidence_halves_the_value(self):
coord = _coord(options={
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 50,
})
_set_forecast_state(coord, "8000.0")
assert coord._get_adjusted_forecast_w() == pytest.approx(4000.0)
# ---------------------------------------------------------------------------
# _get_solar_excess_w — forecast profile branches
# ---------------------------------------------------------------------------
class TestForecastLookaheadProfile:
"""CHARGING_PROFILE_FORECAST: use forecast when it exceeds actual surplus."""
def test_uses_forecast_when_higher_than_actual(self):
"""Forecast beats actual → forecast value used as surplus."""
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=2000.0, house=500.0, ev=0.0)
_set_forecast_state(coord, "8000.0") # 8000 > actual (1500) → wins
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(8000.0)
def test_uses_actual_when_higher_than_forecast(self):
"""Actual surplus beats forecast → actual used (avoids reducing charge)."""
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=9000.0, house=500.0, ev=0.0)
_set_forecast_state(coord, "3000.0") # 3000 < actual (8500)
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(8500.0)
def test_falls_back_to_actual_when_no_sensor_configured(self):
"""Without a forecast sensor, behaves identically to Conservative."""
coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST})
_set_rolling(coord, production=5000.0, house=1000.0, ev=500.0)
coord.hass.states.get.return_value = None
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(3500.0)
def test_confidence_reduces_forecast_before_comparison(self):
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 80,
})
_set_rolling(coord, production=1000.0, house=0.0, ev=0.0)
_set_forecast_state(coord, "10000.0") # 10 000 × 0.80 = 8 000 > actual (1 000)
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(8000.0)
def test_when_both_below_threshold_actual_is_returned(self):
"""Below-threshold values flow to hysteresis logic — surplus is still the max."""
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=500.0, house=0.0, ev=0.0)
_set_forecast_state(coord, "300.0") # forecast (300) < actual (500)
surplus, *_ = coord._get_solar_excess_w()
# actual wins because 500 > 300
assert surplus == pytest.approx(500.0)
def test_returns_full_four_tuple(self):
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=6000.0, house=1000.0, ev=200.0)
_set_forecast_state(coord, "4000.0")
surplus, production, house, ev = coord._get_solar_excess_w()
assert production == pytest.approx(6000.0)
assert house == pytest.approx(1000.0)
assert ev == pytest.approx(200.0)
# actual (4800) > forecast (4000) → actual
assert surplus == pytest.approx(4800.0)
class TestForecastSurplusProfile:
"""CHARGING_PROFILE_FORECAST_CONSERVATIVE: forecast_adj − house_load − ev_charging."""
def test_subtracts_loads_from_forecast(self):
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=2000.0, house=1500.0, ev=500.0)
_set_forecast_state(coord, "8000.0")
surplus, *_ = coord._get_solar_excess_w()
# 8 000 − 1 500 − 500 = 6 000
assert surplus == pytest.approx(6000.0)
def test_falls_back_to_conservative_when_no_sensor(self):
coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE})
_set_rolling(coord, production=5000.0, house=1000.0, ev=200.0)
coord.hass.states.get.return_value = None
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(3800.0)
def test_confidence_applied_before_subtracting_loads(self):
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 80,
})
_set_rolling(coord, production=1000.0, house=1000.0, ev=0.0)
_set_forecast_state(coord, "10000.0")
surplus, *_ = coord._get_solar_excess_w()
# (10 000 × 0.80) − 1 000 − 0 = 7 000
assert surplus == pytest.approx(7000.0)
def test_surplus_can_be_negative(self):
"""A large house load can push forecast surplus below zero."""
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=1000.0, house=5000.0, ev=0.0)
_set_forecast_state(coord, "4000.0")
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(-1000.0)
def test_returns_full_four_tuple(self):
coord = _coord(options={
CONF_CHARGING_PROFILE: CHARGING_PROFILE_FORECAST_CONSERVATIVE,
CONF_FORECAST_SENSOR: _FORECAST_ENTITY,
CONF_FORECAST_CONFIDENCE: 100,
})
_set_rolling(coord, production=6000.0, house=1500.0, ev=300.0)
_set_forecast_state(coord, "9000.0")
surplus, production, house, ev = coord._get_solar_excess_w()
assert production == pytest.approx(6000.0)
assert house == pytest.approx(1500.0)
assert ev == pytest.approx(300.0)
assert surplus == pytest.approx(7200.0) # 9 000 − 1 500 − 300
# ---------------------------------------------------------------------------
# _get_solar_excess_w — baseline profiles (sanity checks)
# ---------------------------------------------------------------------------
class TestBaselineProfiles:
"""Conservative and All Surplus profiles still work correctly."""
def test_conservative_subtracts_all_loads(self):
coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_CONSERVATIVE})
_set_rolling(coord, production=8000.0, house=2000.0, ev=1000.0)
surplus, production, house, ev = coord._get_solar_excess_w()
assert surplus == pytest.approx(5000.0)
assert production == pytest.approx(8000.0)
def test_all_surplus_ignores_loads(self):
coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_ALL_SURPLUS})
_set_rolling(coord, production=8000.0, house=3000.0, ev=2000.0)
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(8000.0)
def test_unknown_profile_falls_back_to_conservative(self):
coord = _coord(options={CONF_CHARGING_PROFILE: "not_a_real_profile"})
_set_rolling(coord, production=7000.0, house=1000.0, ev=500.0)
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(5500.0)
def test_production_unavailable_returns_none_surplus(self):
"""When the production sensor is unavailable, surplus is None."""
coord = _coord(options={CONF_CHARGING_PROFILE: CHARGING_PROFILE_CONSERVATIVE})
_set_rolling(coord, production=None, house=1000.0, ev=200.0)
surplus, production, *_ = coord._get_solar_excess_w()
assert surplus is None
assert production is None
def test_default_profile_is_conservative(self):
"""With no charging_profile option, Conservative is used."""
coord = _coord(options={})
_set_rolling(coord, production=5000.0, house=1500.0, ev=0.0)
surplus, *_ = coord._get_solar_excess_w()
assert surplus == pytest.approx(3500.0)