diff --git a/custom_components/easee_solar_charging/config_flow.py b/custom_components/easee_solar_charging/config_flow.py index a17d3f2..96cebfc 100644 --- a/custom_components/easee_solar_charging/config_flow.py +++ b/custom_components/easee_solar_charging/config_flow.py @@ -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)), }) diff --git a/custom_components/easee_solar_charging/const.py b/custom_components/easee_solar_charging/const.py index a47ec34..d503250 100644 --- a/custom_components/easee_solar_charging/const.py +++ b/custom_components/easee_solar_charging/const.py @@ -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 diff --git a/custom_components/easee_solar_charging/coordinator.py b/custom_components/easee_solar_charging/coordinator.py index 67274a2..0b2b3da 100644 --- a/custom_components/easee_solar_charging/coordinator.py +++ b/custom_components/easee_solar_charging/coordinator.py @@ -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, diff --git a/custom_components/easee_solar_charging/translations/en.json b/custom_components/easee_solar_charging/translations/en.json index 4db9bb9..6b3765b 100644 --- a/custom_components/easee_solar_charging/translations/en.json +++ b/custom_components/easee_solar_charging/translations/en.json @@ -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." } } diff --git a/tests/conftest.py b/tests/conftest.py index 3dc0bdc..a48e27e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py index acdfda5..9111c04 100644 --- a/tests/test_coordinator.py +++ b/tests/test_coordinator.py @@ -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 # --------------------------------------------------------------------------- diff --git a/tests/test_forecast.py b/tests/test_forecast.py new file mode 100644 index 0000000..3d3d5d3 --- /dev/null +++ b/tests/test_forecast.py @@ -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)