This commit is contained in:
Björn Stormwall
2026-06-06 18:42:30 +02:00
commit 108d154b3d
15 changed files with 1203 additions and 0 deletions
@@ -0,0 +1,32 @@
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .coordinator import EaseeSolarCoordinator
PLATFORMS = ["sensor", "switch"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
coordinator = EaseeSolarCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_options_updated))
return True
async def _async_options_updated(hass: HomeAssistant, entry: ConfigEntry) -> None:
coordinator: EaseeSolarCoordinator = hass.data[DOMAIN][entry.entry_id]
await coordinator.async_request_refresh()
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
@@ -0,0 +1,108 @@
from __future__ import annotations
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr, selector
from .const import (
CONF_CHARGER_ID,
CONF_DEVICE_ID,
CONF_NOTIFY_TARGET,
CONF_SITE_ID,
CONF_SOLAR_EXCESS_SENSOR,
CONF_STOP_GRACE_MINUTES,
DEFAULT_STOP_GRACE_MINUTES,
DOMAIN,
EASEE_DOMAIN,
)
_NOTIFY_TARGET_SELECTOR = selector.TextSelector(
selector.TextSelectorConfig(type=selector.TextSelectorType.TEXT)
)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_SITE_ID): str,
vol.Required(CONF_SOLAR_EXCESS_SENSOR): selector.EntitySelector(
selector.EntitySelectorConfig(domain="sensor")
),
vol.Required(CONF_DEVICE_ID): selector.DeviceSelector(
selector.DeviceSelectorConfig(integration="easee")
),
vol.Optional(CONF_NOTIFY_TARGET): _NOTIFY_TARGET_SELECTOR,
}
)
OPTIONS_SCHEMA = vol.Schema(
{
vol.Required(CONF_STOP_GRACE_MINUTES, default=DEFAULT_STOP_GRACE_MINUTES): selector.NumberSelector(
selector.NumberSelectorConfig(min=1, max=60, step=1, unit_of_measurement="min", mode=selector.NumberSelectorMode.BOX)
),
vol.Optional(CONF_NOTIFY_TARGET): _NOTIFY_TARGET_SELECTOR,
}
)
def _charger_serial_from_device(hass, device_id: str) -> str | None:
"""Extract the Easee charger serial number from the device registry entry."""
registry = dr.async_get(hass)
device = registry.async_get(device_id)
if device is None:
return None
for domain, identifier in device.identifiers:
if domain == EASEE_DOMAIN:
return identifier
return None
class EaseeSolarChargingConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> EaseeSolarChargingOptionsFlow:
return EaseeSolarChargingOptionsFlow()
async def async_step_user(self, user_input: dict | None = None) -> ConfigFlowResult:
errors: dict[str, str] = {}
if user_input is not None:
device_id = user_input[CONF_DEVICE_ID]
charger_id = _charger_serial_from_device(self.hass, device_id)
if charger_id is None:
errors[CONF_DEVICE_ID] = "cannot_resolve_serial"
else:
await self.async_set_unique_id(charger_id)
self._abort_if_unique_id_configured()
notify_target = user_input.pop(CONF_NOTIFY_TARGET, None)
initial_options = {CONF_NOTIFY_TARGET: notify_target} if notify_target else {}
return self.async_create_entry(
title=f"Easee {charger_id}",
data={**user_input, CONF_CHARGER_ID: charger_id},
options=initial_options,
)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
class EaseeSolarChargingOptionsFlow(OptionsFlow):
async def async_step_init(self, user_input: dict | None = None) -> ConfigFlowResult:
if user_input is not None:
return self.async_create_entry(data=user_input)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
OPTIONS_SCHEMA,
self.config_entry.options,
),
)
@@ -0,0 +1,46 @@
DOMAIN = "easee_solar_charging"
CONF_CHARGER_ID = "charger_id"
CONF_DEVICE_ID = "device_id"
CONF_SITE_ID = "site_id"
CONF_SOLAR_EXCESS_SENSOR = "solar_excess_sensor"
EASEE_DOMAIN = "easee"
EASEE_SERVICE_SET_DYNAMIC_LIMIT = "set_charger_dynamic_limit"
EASEE_CONTROLLABLE_STATES = {"awaiting_start", "ready_to_charge", "charging"}
# States where the session is finished/sleeping: set current to 0 once, then idle.
EASEE_SLEEPING_STATES = {"completed", "awaiting_authorization", "awaiting_schedule"}
DEFAULT_SCAN_INTERVAL = 30 # seconds
# Charging current constraints
VOLTAGE_V = 230
PHASES = 3
MIN_CHARGER_CURRENT_A = 6
MAX_CHARGER_CURRENT_A = 16
# Hysteresis
HYSTERESIS_A = 2 # dead-band within the charging range (fixed)
CONF_STOP_GRACE_MINUTES = "stop_grace_minutes"
DEFAULT_STOP_GRACE_MINUTES = 5
CONF_NOTIFY_TARGET = "notify_target"
# Watts available at each valid current step on a 3-phase 230V system (P = 3 × V × I)
AMPS_TO_POWER_W: dict[int, int] = {
a: PHASES * VOLTAGE_V * a
for a in range(MIN_CHARGER_CURRENT_A, MAX_CHARGER_CURRENT_A + 1)
}
def solar_excess_to_charger_current(excess_w: float) -> int:
"""Return the highest valid charger current (A) that fits within excess_w.
Returns 0 if excess is below the minimum charging threshold (6 A).
Steps are 1 A increments between 6 A and 16 A on a 3-phase 230 V system.
"""
amps = int(excess_w / (PHASES * VOLTAGE_V))
if amps < MIN_CHARGER_CURRENT_A:
return 0
return min(amps, MAX_CHARGER_CURRENT_A)
@@ -0,0 +1,210 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
CONF_CHARGER_ID,
CONF_DEVICE_ID,
CONF_NOTIFY_TARGET,
CONF_SOLAR_EXCESS_SENSOR,
CONF_STOP_GRACE_MINUTES,
DEFAULT_SCAN_INTERVAL,
DEFAULT_STOP_GRACE_MINUTES,
DOMAIN,
EASEE_CONTROLLABLE_STATES,
EASEE_SLEEPING_STATES,
EASEE_DOMAIN,
EASEE_SERVICE_SET_DYNAMIC_LIMIT,
HYSTERESIS_A,
solar_excess_to_charger_current,
)
_LOGGER = logging.getLogger(__name__)
_DEAD_BAND_RESYNC = timedelta(minutes=1)
class EaseeSolarCoordinator(DataUpdateCoordinator):
solar_charging_enabled: bool = True
stop_grace_enabled: bool = True
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
self._entry = entry
self.charger_id = entry.data[CONF_CHARGER_ID]
self._solar_excess_sensor = entry.data[CONF_SOLAR_EXCESS_SENSOR]
self._device_id: str = entry.data[CONF_DEVICE_ID]
self._last_sent_current: int | None = None
self._last_sent_at: datetime | None = None
self._below_threshold_since: datetime | None = None
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
)
@property
def _stop_grace(self) -> timedelta:
minutes = self._entry.options.get(CONF_STOP_GRACE_MINUTES, DEFAULT_STOP_GRACE_MINUTES)
return timedelta(minutes=minutes)
def _get_solar_excess_w(self) -> float | None:
state = self.hass.states.get(self._solar_excess_sensor)
if state is None or state.state in ("unknown", "unavailable"):
return None
try:
return float(state.state)
except ValueError:
_LOGGER.warning(
"Solar excess sensor %s has non-numeric state: %s",
self._solar_excess_sensor,
state.state,
)
return None
def _get_charger_current_a(self) -> float | None:
entity_id = f"sensor.{self.charger_id.lower()}_current"
state = self.hass.states.get(entity_id)
if state is None or state.state in ("unknown", "unavailable"):
return None
try:
return float(state.state)
except ValueError:
return None
def _charger_is_sleeping(self) -> bool:
entity_id = f"sensor.{self.charger_id.lower()}_status"
state = self.hass.states.get(entity_id)
return state is not None and state.state in EASEE_SLEEPING_STATES
def _charger_allows_control(self) -> bool:
entity_id = f"sensor.{self.charger_id.lower()}_status"
state = self.hass.states.get(entity_id)
if state is None:
_LOGGER.warning("Charger status entity %s not found", entity_id)
return False
return state.state in EASEE_CONTROLLABLE_STATES
def cancel_stop_grace(self) -> None:
self._below_threshold_since = None
def _apply_stop_hysteresis(self, raw_target: int) -> int:
"""Hold current charging level for up to stop_grace_minutes before stopping.
Once excess drops below the 6 A threshold, start a timer. Only send 0 A
(stop charging) after the grace period expires without recovery.
"""
if not self.stop_grace_enabled:
self._below_threshold_since = None
return raw_target
now = datetime.now(timezone.utc)
if raw_target == 0:
if self._below_threshold_since is None:
self._below_threshold_since = now
_LOGGER.debug(
"Solar excess below threshold — grace period started (%s)",
self._stop_grace,
)
elapsed = now - self._below_threshold_since
if elapsed < self._stop_grace:
return self._last_sent_current or 0
_LOGGER.debug("Grace period expired — stopping charger")
return 0
if self._below_threshold_since is not None:
_LOGGER.debug("Solar excess recovered — grace period cancelled")
self._below_threshold_since = None
return raw_target
def _should_send(self, target: int) -> bool:
"""Return True only when the target warrants a new service call.
- First run: always send.
- Transitions to/from 0: always send (start/stop events).
- Outside dead-band: send immediately.
- Inside dead-band: send once per minute so gradual solar drift is applied.
"""
if self._last_sent_current is None:
return True
if target == self._last_sent_current:
return False
if target == 0 or self._last_sent_current == 0:
return True
if abs(target - self._last_sent_current) >= HYSTERESIS_A:
return True
# Within dead-band: resync once per minute
return (
self._last_sent_at is None
or datetime.now(timezone.utc) - self._last_sent_at >= _DEAD_BAND_RESYNC
)
async def _async_notify(self, title: str, message: str) -> None:
target = self._entry.options.get(CONF_NOTIFY_TARGET)
if not target:
return
await self.hass.services.async_call(
"notify",
target,
{"title": title, "message": message},
blocking=False,
)
async def _async_set_charger_current(self, amps: int) -> None:
await self.hass.services.async_call(
EASEE_DOMAIN,
EASEE_SERVICE_SET_DYNAMIC_LIMIT,
{"device_id": self._device_id, "current": amps},
blocking=True,
)
_LOGGER.debug("Set Easee dynamic limit to %d A", amps)
async def _async_update_data(self) -> dict:
try:
solar_excess_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)
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,
"solar_excess_w": solar_excess_w,
"target_current_a": target_current_a,
"charger_current_a": self._get_charger_current_a(),
"charger_status": (s := self.hass.states.get(f"sensor.{self.charger_id.lower()}_status")) and s.state,
}
except Exception as err:
raise UpdateFailed(f"Error fetching Easee data: {err}") from err
@@ -0,0 +1,12 @@
{
"domain": "easee_solar_charging",
"name": "Easee Solar Charging",
"version": "0.1.0",
"config_flow": true,
"documentation": "https://github.com/bjorn/easee-solar-charging-custom-component",
"issue_tracker": "https://github.com/bjorn/easee-solar-charging-custom-component/issues",
"requirements": [],
"dependencies": [],
"codeowners": ["@bjorn"],
"iot_class": "local_polling"
}
@@ -0,0 +1,87 @@
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfElectricCurrent, UnitOfPower
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import EaseeSolarCoordinator
@dataclass(frozen=True, kw_only=True)
class EaseeSolarSensorDescription(SensorEntityDescription):
data_key: str
SENSOR_DESCRIPTIONS: tuple[EaseeSolarSensorDescription, ...] = (
EaseeSolarSensorDescription(
key="solar_excess",
data_key="solar_excess_w",
name="Solar Excess Power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
),
EaseeSolarSensorDescription(
key="target_current",
data_key="target_current_a",
name="Target Charging Current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
),
EaseeSolarSensorDescription(
key="charger_current",
data_key="charger_current_a",
name="Charger Current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
),
EaseeSolarSensorDescription(
key="charger_status",
data_key="charger_status",
name="Charger Status",
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
coordinator: EaseeSolarCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
EaseeSolarSensor(coordinator, description)
for description in SENSOR_DESCRIPTIONS
)
class EaseeSolarSensor(CoordinatorEntity[EaseeSolarCoordinator], SensorEntity):
entity_description: EaseeSolarSensorDescription
_attr_has_entity_name = True
def __init__(
self,
coordinator: EaseeSolarCoordinator,
description: EaseeSolarSensorDescription,
) -> None:
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.charger_id}_{description.key}"
self.entity_id = f"sensor.esc_{coordinator.charger_id.lower()}_{description.key}"
@property
def native_value(self):
return self.coordinator.data.get(self.entity_description.data_key)
@@ -0,0 +1,85 @@
from __future__ import annotations
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from .const import DOMAIN
from .coordinator import EaseeSolarCoordinator
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
coordinator: EaseeSolarCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([
SolarChargingSwitch(coordinator),
StopGraceSwitch(coordinator),
])
class SolarChargingSwitch(RestoreEntity, SwitchEntity):
_attr_has_entity_name = True
_attr_name = "Solar Charging"
_attr_icon = "mdi:solar-power"
def __init__(self, coordinator: EaseeSolarCoordinator) -> None:
self._coordinator = coordinator
self._attr_unique_id = f"{coordinator.charger_id}_solar_charging_enabled"
self.entity_id = f"switch.esc_{coordinator.charger_id.lower()}_solar_charging"
async def async_added_to_hass(self) -> None:
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if last_state is not None:
self._coordinator.solar_charging_enabled = last_state.state == "on"
@property
def is_on(self) -> bool:
return self._coordinator.solar_charging_enabled
async def async_turn_on(self, **kwargs) -> None:
self._coordinator.solar_charging_enabled = True
await self._coordinator.async_request_refresh()
self.async_write_ha_state()
async def async_turn_off(self, **kwargs) -> None:
self._coordinator.solar_charging_enabled = False
await self._coordinator.async_request_refresh()
self.async_write_ha_state()
class StopGraceSwitch(RestoreEntity, SwitchEntity):
_attr_has_entity_name = True
_attr_name = "Stop Grace"
_attr_icon = "mdi:timer-pause"
def __init__(self, coordinator: EaseeSolarCoordinator) -> None:
self._coordinator = coordinator
self._attr_unique_id = f"{coordinator.charger_id}_stop_grace_enabled"
self.entity_id = f"switch.esc_{coordinator.charger_id.lower()}_stop_grace"
async def async_added_to_hass(self) -> None:
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if last_state is not None:
self._coordinator.stop_grace_enabled = last_state.state == "on"
@property
def is_on(self) -> bool:
return self._coordinator.stop_grace_enabled
async def async_turn_on(self, **kwargs) -> None:
self._coordinator.stop_grace_enabled = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs) -> None:
self._coordinator.stop_grace_enabled = False
# Reset any running timer so the charger stops on the next poll
self._coordinator.cancel_stop_grace()
await self._coordinator.async_request_refresh()
self.async_write_ha_state()
@@ -0,0 +1,48 @@
{
"entity": {
"switch": {
"solar_charging_enabled": {
"name": "Solar Charging"
},
"stop_grace_enabled": {
"name": "Stop Grace"
}
}
},
"options": {
"step": {
"init": {
"title": "Easee Solar Charging options",
"data": {
"stop_grace_minutes": "Stop grace period",
"notify_target": "Notification target"
},
"data_description": {
"stop_grace_minutes": "How many minutes to keep charging after solar excess drops below the minimum threshold, before stopping the charger.",
"notify_target": "Name of a notify service to receive start/stop alerts (e.g. mobile_app_my_phone). Leave empty to disable notifications."
}
}
}
},
"config": {
"step": {
"user": {
"title": "Set up Easee Solar Charging",
"data": {
"site_id": "Site ID",
"solar_excess_sensor": "Solar Excess Power Sensor",
"device_id": "Easee Charger",
"notify_target": "Notification target"
},
"data_description": {
"solar_excess_sensor": "Sensor reporting the momentarily available solar power surplus in Watts.",
"notify_target": "Name of a notify service to receive start/stop alerts (e.g. mobile_app_my_phone). Leave empty to disable notifications."
}
}
},
"error": {},
"abort": {
"already_configured": "Charger is already configured."
}
}
}
+40
View File
@@ -0,0 +1,40 @@
# Easee Solar Charging — Dashboard Card
#
# Entity IDs follow the pattern: esc_{charger_id}_{key}
# Replace CHARGER_ID with your charger serial (e.g. ehxt9pqp).
#
# How to add to your dashboard:
# Edit dashboard → Add Card → Manual card → paste this YAML
type: vertical-stack
cards:
- type: glance
title: Easee Solar Charging
columns: 2
show_name: true
show_icon: true
show_state: true
entities:
- entity: sensor.esc_CHARGER_ID_solar_excess
name: Solar Excess
icon: mdi:solar-power
- entity: sensor.esc_CHARGER_ID_target_current
name: Target Current
icon: mdi:lightning-bolt
- entity: sensor.esc_CHARGER_ID_charger_current
name: Charger Current
icon: mdi:current-ac
- entity: sensor.esc_CHARGER_ID_charger_status
name: Status
icon: mdi:ev-station
- type: entities
entities:
- type: section
label: Controls
- entity: switch.esc_CHARGER_ID_solar_charging
name: Solar Charging
icon: mdi:solar-power
- entity: switch.esc_CHARGER_ID_stop_grace
name: Stop Grace
icon: mdi:timer-pause
+15
View File
@@ -0,0 +1,15 @@
[project]
name = "easee-solar-charging"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = []
[dependency-groups]
dev = [
"pytest>=8.0",
"freezegun>=1.4",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]
View File
+94
View File
@@ -0,0 +1,94 @@
"""Stub out homeassistant before any local imports so tests run without HA installed."""
from __future__ import annotations
import sys
from datetime import timedelta
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# HA module stubs — must happen before coordinator.py is imported
# ---------------------------------------------------------------------------
class _DataUpdateCoordinator:
"""Minimal stand-in for DataUpdateCoordinator."""
def __init__(self, hass, logger, *, name, update_interval):
self.hass = hass
self.name = name
self.data: dict = {}
class _UpdateFailed(Exception):
pass
_update_coordinator_stub = MagicMock()
_update_coordinator_stub.DataUpdateCoordinator = _DataUpdateCoordinator
_update_coordinator_stub.UpdateFailed = _UpdateFailed
for _mod in [
"homeassistant",
"homeassistant.config_entries",
"homeassistant.core",
"homeassistant.components",
"homeassistant.components.sensor",
"homeassistant.components.switch",
"homeassistant.const",
"homeassistant.helpers",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.entity_platform",
"homeassistant.helpers.restore_state",
"homeassistant.helpers.selector",
"voluptuous",
]:
sys.modules.setdefault(_mod, MagicMock())
sys.modules["homeassistant.helpers.update_coordinator"] = _update_coordinator_stub
# ---------------------------------------------------------------------------
# Shared test fixtures
# ---------------------------------------------------------------------------
from custom_components.easee_solar_charging.coordinator import EaseeSolarCoordinator # noqa: E402
from custom_components.easee_solar_charging.const import ( # noqa: E402
CONF_STOP_GRACE_MINUTES,
DEFAULT_STOP_GRACE_MINUTES,
)
CHARGER_ID = "ehxt9pqp"
SOLAR_SENSOR = "sensor.solar_excess"
class FakeCoordinator(EaseeSolarCoordinator):
"""Coordinator with HA wiring stripped out so logic methods can be unit-tested."""
def __init__(self, options: dict | None = None) -> None:
# Skip DataUpdateCoordinator.__init__ — not needed for logic-only tests.
self.hass = MagicMock()
self.charger_id = CHARGER_ID
self._solar_excess_sensor = SOLAR_SENSOR
self._device_id = None
self._last_sent_current = None
self._last_sent_at = None
self._below_threshold_since = None
self.solar_charging_enabled = True
self.stop_grace_enabled = True
self._entry = MagicMock()
self._entry.options = options if options is not None else {
CONF_STOP_GRACE_MINUTES: DEFAULT_STOP_GRACE_MINUTES,
}
@pytest.fixture
def coord() -> FakeCoordinator:
return FakeCoordinator()
def make_state(value: str) -> MagicMock:
"""Return a minimal HA state mock with the given state string."""
s = MagicMock()
s.state = value
return s
+71
View File
@@ -0,0 +1,71 @@
import pytest
from custom_components.easee_solar_charging.const import (
AMPS_TO_POWER_W,
MAX_CHARGER_CURRENT_A,
MIN_CHARGER_CURRENT_A,
PHASES,
VOLTAGE_V,
solar_excess_to_charger_current,
)
# ---------------------------------------------------------------------------
# solar_excess_to_charger_current
# ---------------------------------------------------------------------------
def test_zero_excess_returns_zero():
assert solar_excess_to_charger_current(0) == 0
def test_just_below_minimum_threshold_returns_zero():
min_power = PHASES * VOLTAGE_V * MIN_CHARGER_CURRENT_A
assert solar_excess_to_charger_current(min_power - 1) == 0
def test_exactly_at_minimum_threshold():
min_power = PHASES * VOLTAGE_V * MIN_CHARGER_CURRENT_A
assert solar_excess_to_charger_current(min_power) == MIN_CHARGER_CURRENT_A
def test_fractional_watts_floor():
# 3 * 230 * 8 = 5520 W → 8 A; adding 229 W is still not enough for 9 A
assert solar_excess_to_charger_current(PHASES * VOLTAGE_V * 8 + 229) == 8
def test_intermediate_value():
assert solar_excess_to_charger_current(PHASES * VOLTAGE_V * 10) == 10
def test_clamps_at_maximum():
assert solar_excess_to_charger_current(999_999) == MAX_CHARGER_CURRENT_A
def test_negative_excess_returns_zero():
assert solar_excess_to_charger_current(-500) == 0
@pytest.mark.parametrize("amps", range(MIN_CHARGER_CURRENT_A, MAX_CHARGER_CURRENT_A + 1))
def test_round_trip(amps):
"""Power for N amps should convert back to exactly N amps."""
assert solar_excess_to_charger_current(AMPS_TO_POWER_W[amps]) == amps
# ---------------------------------------------------------------------------
# AMPS_TO_POWER_W table
# ---------------------------------------------------------------------------
def test_table_covers_full_range():
assert set(AMPS_TO_POWER_W) == set(range(MIN_CHARGER_CURRENT_A, MAX_CHARGER_CURRENT_A + 1))
def test_table_minimum_entry():
assert AMPS_TO_POWER_W[MIN_CHARGER_CURRENT_A] == PHASES * VOLTAGE_V * MIN_CHARGER_CURRENT_A
def test_table_maximum_entry():
assert AMPS_TO_POWER_W[MAX_CHARGER_CURRENT_A] == PHASES * VOLTAGE_V * MAX_CHARGER_CURRENT_A
def test_table_is_strictly_increasing():
values = [AMPS_TO_POWER_W[a] for a in sorted(AMPS_TO_POWER_W)]
assert values == sorted(values) and len(set(values)) == len(values)
+239
View File
@@ -0,0 +1,239 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock
import pytest
from freezegun import freeze_time
from tests.conftest import FakeCoordinator, make_state
from custom_components.easee_solar_charging.const import (
CONF_STOP_GRACE_MINUTES,
DEFAULT_STOP_GRACE_MINUTES,
EASEE_CONTROLLABLE_STATES,
EASEE_SLEEPING_STATES,
HYSTERESIS_A,
MIN_CHARGER_CURRENT_A,
)
_NOW = "2024-06-01 12:00:00"
# ---------------------------------------------------------------------------
# _should_send
# ---------------------------------------------------------------------------
class TestShouldSend:
def test_first_run_always_sends(self, coord):
assert coord._should_send(8) is True
def test_same_value_no_send(self, coord):
coord._last_sent_current = 8
assert coord._should_send(8) is False
def test_zero_to_nonzero_sends(self, coord):
coord._last_sent_current = 0
assert coord._should_send(8) is True
def test_nonzero_to_zero_sends(self, coord):
coord._last_sent_current = 8
assert coord._should_send(0) is True
def test_outside_deadband_sends_immediately(self, coord):
coord._last_sent_current = 8
assert coord._should_send(8 + HYSTERESIS_A) is True
def test_inside_deadband_suppressed_when_recent(self, coord):
coord._last_sent_current = 8
coord._last_sent_at = datetime.now(timezone.utc)
# 1 A delta — inside dead-band, timestamp is fresh
assert coord._should_send(9) is False
def test_inside_deadband_sends_after_one_minute(self, coord):
coord._last_sent_current = 8
coord._last_sent_at = datetime.now(timezone.utc) - timedelta(minutes=1, seconds=1)
assert coord._should_send(9) is True
def test_inside_deadband_sends_when_no_timestamp(self, coord):
coord._last_sent_current = 8
coord._last_sent_at = None
assert coord._should_send(9) is True
# ---------------------------------------------------------------------------
# _apply_stop_hysteresis
# ---------------------------------------------------------------------------
class TestApplyStopHysteresis:
def test_nonzero_target_passes_through(self, coord):
assert coord._apply_stop_hysteresis(10) == 10
def test_nonzero_target_clears_timer(self, coord):
coord._below_threshold_since = datetime.now(timezone.utc)
coord._apply_stop_hysteresis(10)
assert coord._below_threshold_since is None
def test_zero_starts_grace_period(self, coord):
coord._last_sent_current = 10
with freeze_time(_NOW):
result = coord._apply_stop_hysteresis(0)
assert result == 10
assert coord._below_threshold_since is not None
def test_zero_held_during_grace(self, coord):
coord._last_sent_current = 8
with freeze_time(_NOW):
coord._apply_stop_hysteresis(0)
with freeze_time("2024-06-01 12:04:00"):
result = coord._apply_stop_hysteresis(0)
assert result == 8
def test_zero_applied_after_grace_expires(self, coord):
coord._last_sent_current = 8
with freeze_time(_NOW):
coord._apply_stop_hysteresis(0)
with freeze_time("2024-06-01 12:05:01"):
result = coord._apply_stop_hysteresis(0)
assert result == 0
def test_recovery_before_grace_expires_cancels_timer(self, coord):
coord._last_sent_current = 8
with freeze_time(_NOW):
coord._apply_stop_hysteresis(0)
with freeze_time("2024-06-01 12:02:00"):
result = coord._apply_stop_hysteresis(10)
assert result == 10
assert coord._below_threshold_since is None
def test_disabled_passes_through_immediately(self, coord):
coord.stop_grace_enabled = False
coord._last_sent_current = 8
assert coord._apply_stop_hysteresis(0) == 0
assert coord._below_threshold_since is None
def test_disabled_clears_running_timer(self, coord):
coord._below_threshold_since = datetime.now(timezone.utc)
coord.stop_grace_enabled = False
coord._apply_stop_hysteresis(0)
assert coord._below_threshold_since is None
def test_respects_configured_grace_duration(self):
coord = FakeCoordinator(options={CONF_STOP_GRACE_MINUTES: 2})
coord._last_sent_current = 8
with freeze_time(_NOW):
coord._apply_stop_hysteresis(0)
with freeze_time("2024-06-01 12:02:01"):
result = coord._apply_stop_hysteresis(0)
assert result == 0
def test_no_last_sent_during_grace_returns_zero(self, coord):
# Edge case: grace period started but charger was never commanded
coord._last_sent_current = None
with freeze_time(_NOW):
result = coord._apply_stop_hysteresis(0)
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
# ---------------------------------------------------------------------------
class TestChargerAllowsControl:
@pytest.mark.parametrize("state", sorted(EASEE_CONTROLLABLE_STATES))
def test_allows_controllable_states(self, coord, state):
coord.hass.states.get.return_value = make_state(state)
assert coord._charger_allows_control() is True
def test_rejects_disconnected(self, coord):
coord.hass.states.get.return_value = make_state("disconnected")
assert coord._charger_allows_control() is False
def test_rejects_error_state(self, coord):
coord.hass.states.get.return_value = make_state("error")
assert coord._charger_allows_control() is False
def test_rejects_missing_entity(self, coord):
coord.hass.states.get.return_value = None
assert coord._charger_allows_control() is False
def test_queries_correct_status_entity(self, coord):
coord.hass.states.get.return_value = make_state("charging")
coord._charger_allows_control()
coord.hass.states.get.assert_called_once_with(
f"sensor.{coord.charger_id.lower()}_status"
)
# ---------------------------------------------------------------------------
# cancel_stop_grace
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# _charger_is_sleeping
# ---------------------------------------------------------------------------
class TestChargerIsSleeping:
@pytest.mark.parametrize("state", sorted(EASEE_SLEEPING_STATES))
def test_detects_sleeping_states(self, coord, state):
coord.hass.states.get.return_value = make_state(state)
assert coord._charger_is_sleeping() is True
@pytest.mark.parametrize("state", sorted(EASEE_CONTROLLABLE_STATES))
def test_not_sleeping_when_controllable(self, coord, state):
coord.hass.states.get.return_value = make_state(state)
assert coord._charger_is_sleeping() is False
def test_not_sleeping_when_entity_missing(self, coord):
coord.hass.states.get.return_value = None
assert coord._charger_is_sleeping() is False
def test_not_sleeping_when_disconnected(self, coord):
coord.hass.states.get.return_value = make_state("disconnected")
assert coord._charger_is_sleeping() is False
def test_cancel_stop_grace_clears_timer(coord):
coord._below_threshold_since = datetime.now(timezone.utc)
coord.cancel_stop_grace()
assert coord._below_threshold_since is None
def test_cancel_stop_grace_is_idempotent(coord):
coord._below_threshold_since = None
coord.cancel_stop_grace()
assert coord._below_threshold_since is None
Generated
+116
View File
@@ -0,0 +1,116 @@
version = 1
revision = 3
requires-python = ">=3.12"
[[package]]
name = "colorama"
version = "0.4.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
]
[[package]]
name = "easee-solar-charging"
version = "0.1.0"
source = { virtual = "." }
[package.dev-dependencies]
dev = [
{ name = "freezegun" },
{ name = "pytest" },
]
[package.metadata]
[package.metadata.requires-dev]
dev = [
{ name = "freezegun", specifier = ">=1.4" },
{ name = "pytest", specifier = ">=8.0" },
]
[[package]]
name = "freezegun"
version = "1.5.5"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "python-dateutil" },
]
sdist = { url = "https://files.pythonhosted.org/packages/95/dd/23e2f4e357f8fd3bdff613c1fe4466d21bfb00a6177f238079b17f7b1c84/freezegun-1.5.5.tar.gz", hash = "sha256:ac7742a6cc6c25a2c35e9292dfd554b897b517d2dec26891a2e8debf205cb94a", size = 35914, upload-time = "2025-08-09T10:39:08.338Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5e/2e/b41d8a1a917d6581fc27a35d05561037b048e47df50f27f8ac9c7e27a710/freezegun-1.5.5-py3-none-any.whl", hash = "sha256:cd557f4a75cf074e84bc374249b9dd491eaeacd61376b9eb3c423282211619d2", size = 19266, upload-time = "2025-08-09T10:39:06.636Z" },
]
[[package]]
name = "iniconfig"
version = "2.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
]
[[package]]
name = "packaging"
version = "26.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" },
]
[[package]]
name = "pluggy"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]]
name = "pygments"
version = "2.20.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
]
[[package]]
name = "pytest"
version = "9.0.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/7d/0d/549bd94f1a0a402dc8cf64563a117c0f3765662e2e668477624baeec44d5/pytest-9.0.3.tar.gz", hash = "sha256:b86ada508af81d19edeb213c681b1d48246c1a91d304c6c81a427674c17eb91c", size = 1572165, upload-time = "2026-04-07T17:16:18.027Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" },
]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "six" },
]
sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" },
]
[[package]]
name = "six"
version = "1.17.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" },
]