fix: implement wallbox power sensor with zero-override for non-charging state

Author: derolli1976

custom_components/enpal_webparser/entity_factory.py

@@ -124,16 +124,59 @@ async def async_added_to_hass(self):
         self._handle_coordinator_update()
 
 
-def build_sensor_entity(sensor: dict, coordinator: DataUpdateCoordinator) -> SensorEntity:
+# Wallbox sensors whose value must be forced to 0 when not actively charging.
+# Works around an Enpal firmware bug where these values freeze after charging ends.
+_WALLBOX_ZERO_OVERRIDE_IDS = {
+    "power_wallbox_connector_1_charging",
+    "current_wallbox_connector_1_phase_a",
+    "current_wallbox_connector_1_phase_b",
+    "current_wallbox_connector_1_phase_c",
+}
+
+
+def build_sensor_entity(
+    sensor: dict,
+    coordinator: DataUpdateCoordinator,
+    use_wallbox: bool = False,
+) -> SensorEntity:
     """
     Factory function: Builds the appropriate sensor entity.
     Extendable for special cases or subclasses.
     """
     if sensor.get("device_class") == "energy":
         return EnpalEnergySensor(sensor, coordinator)
+    if use_wallbox and make_id(sensor.get("name", "")) in _WALLBOX_ZERO_OVERRIDE_IDS:
+        return EnpalWallboxPowerSensor(sensor, coordinator)
     return EnpalBaseSensor(sensor, coordinator)
 
 
+class EnpalWallboxPowerSensor(EnpalBaseSensor):
+    """Wallbox power/current sensor that reports 0 when not charging.
+
+    Works around an Enpal firmware bug where power and current values
+    freeze at their last reading after a charging session ends.
+    """
+
+    _WALLBOX_STATUS_ENTITY = "sensor.wallbox_status"
+
+    @property
+    def native_value(self):
+        raw = self._sensor.get("value")
+        status_state = self.hass.states.get(self._WALLBOX_STATUS_ENTITY)
+        if status_state is not None and status_state.state != "charging":
+            return 0
+        return raw
+
+    @property
+    def extra_state_attributes(self):
+        attrs = super().extra_state_attributes
+        status_state = self.hass.states.get(self._WALLBOX_STATUS_ENTITY)
+        if status_state is not None and status_state.state != "charging":
+            attrs["enpal_raw_value"] = self._sensor.get("value")
+            attrs["enpal_zero_reason"] = "wallbox not charging"
+        return attrs
+
+
 class EnpalEnergySensor(EnpalBaseSensor):
     def __init__(self, sensor: dict, coordinator: DataUpdateCoordinator):
         super().__init__(sensor, coordinator)

custom_components/enpal_webparser/sensor.py

@@ -150,10 +150,12 @@ async def _on_push_data(result: dict) -> None:
     for sensor in coordinator.data:
         _LOGGER.info("[Enpal]   Name: %s -> UID: %s", sensor["name"], make_id(sensor["name"]))
 
+    use_wallbox = entry.options.get("use_wallbox", False)
+
     entities = []
     for sensor_dict in coordinator.data:
         _LOGGER.debug("[Enpal] Adding sensor entity: %s", sensor_dict["name"])
-        entities.append(build_sensor_entity(sensor_dict, coordinator))
+        entities.append(build_sensor_entity(sensor_dict, coordinator, use_wallbox=use_wallbox))
 
 
     # Create cumulative energy sensor with smart fallback for different inverter types

custom_components/enpal_webparser/tests/test_entity_factory.py

@@ -21,6 +21,7 @@
     build_sensor_entity,
     EnpalBaseSensor,
     EnpalEnergySensor,
+    EnpalWallboxPowerSensor,
 )
 
 class DummyCoordinator(DataUpdateCoordinator):
@@ -117,3 +118,98 @@ async def test_build_energy_sensor_full(hass: HomeAssistant, mock_sensor_dict, h
     assert "enpal_last_update" in sensor.extra_state_attributes
 
 
+# ---------- Wallbox power zero-override tests ----------
+
+class _FakeState:
+    """Minimal stand-in for homeassistant.core.State."""
+    def __init__(self, state: str):
+        self.state = state
+
+
+class _FakeHass:
+    """Minimal hass stub with a states registry."""
+    def __init__(self, states_map: dict):
+        self._map = states_map
+
+    class _States:
+        def __init__(self, m):
+            self._m = m
+        def get(self, entity_id):
+            return self._m.get(entity_id)
+
+    @property
+    def states(self):
+        return self._States(self._map)
+
+
+def _wallbox_power_sensor(status_state: str | None):
+    """Create an EnpalWallboxPowerSensor with a faked wallbox status."""
+    sensor_dict = {
+        "name": "Power Wallbox Connector 1 Charging",
+        "value": "4500",
+        "unit": "W",
+        "device_class": "power",
+        "enabled": True,
+        "group": "Wallbox",
+    }
+    entity = build_sensor_entity(sensor_dict, DummyCoordinator(), use_wallbox=True)
+    assert isinstance(entity, EnpalWallboxPowerSensor)
+
+    if status_state is not None:
+        entity.hass = _FakeHass({"sensor.wallbox_status": _FakeState(status_state)})
+    else:
+        entity.hass = _FakeHass({})
+    return entity
+
+
+def test_wallbox_power_zero_when_not_charging():
+    """Power sensor returns 0 when wallbox status is not 'charging'."""
+    entity = _wallbox_power_sensor("connected")
+    assert entity.native_value == 0
+    assert entity.extra_state_attributes.get("enpal_raw_value") == "4500"
+    assert entity.extra_state_attributes.get("enpal_zero_reason") == "wallbox not charging"
+
+
+def test_wallbox_power_passthrough_when_charging():
+    """Power sensor returns raw value when wallbox status is 'charging'."""
+    entity = _wallbox_power_sensor("charging")
+    assert entity.native_value == "4500"
+    assert "enpal_raw_value" not in entity.extra_state_attributes
+
+
+def test_wallbox_power_passthrough_when_status_missing():
+    """Power sensor returns raw value when wallbox_status entity doesn't exist."""
+    entity = _wallbox_power_sensor(None)
+    assert entity.native_value == "4500"
+    assert "enpal_raw_value" not in entity.extra_state_attributes
+
+
+def test_wallbox_current_sensor_zero_override():
+    """Current sensor also gets zero-override treatment."""
+    sensor_dict = {
+        "name": "Current Wallbox Connector 1 Phase (A)",
+        "value": "12.31",
+        "unit": "A",
+        "device_class": "current",
+        "enabled": True,
+        "group": "Wallbox",
+    }
+    entity = build_sensor_entity(sensor_dict, DummyCoordinator(), use_wallbox=True)
+    assert isinstance(entity, EnpalWallboxPowerSensor)
+    entity.hass = _FakeHass({"sensor.wallbox_status": _FakeState("connected")})
+    assert entity.native_value == 0
+
+
+def test_wallbox_power_not_used_without_flag():
+    """Without use_wallbox=True, the regular base sensor is returned."""
+    sensor_dict = {
+        "name": "Power Wallbox Connector 1 Charging",
+        "value": "4500",
+        "unit": "W",
+        "device_class": "power",
+    }
+    entity = build_sensor_entity(sensor_dict, DummyCoordinator(), use_wallbox=False)
+    assert isinstance(entity, EnpalBaseSensor)
+    assert not isinstance(entity, EnpalWallboxPowerSensor)
+
+