Fetch wallbox status via HTTP for real-time updates

custom_components/enpal_webparser/api/wallbox_client.py

@@ -281,17 +281,67 @@ async def set_mode(self, mode: str) -> bool:
     async def get_wallbox_data(self) -> Optional[Dict]:
         """Return current wallbox status as a dict (compatible with old addon API).
 
-        Ensures the WebSocket connection is alive so that server-push
-        RenderBatches keep the cached mode/status up to date.
+        Performs a lightweight HTTP GET to /wallbox on every call so that
+        external status changes (e.g. via the Enpal app) are picked up
+        within the normal polling interval.  The existing WebSocket
+        connection is *not* disturbed — it stays open for button clicks.
         """
-        if not await self.ensure_fresh_connection():
-            return None
+        # Lightweight poll: GET /wallbox HTML and parse pre-rendered status
+        mode, status = await self._fetch_status_via_http()
+        if mode:
+            self._mode = mode
+        if status:
+            self._status = status
+
+        # Fallback: if we never got any status yet, ensure WebSocket is up
+        # so the initial RenderBatch seeds the values.
+        if self._mode is None and self._status is None:
+            if not await self.ensure_fresh_connection():
+                return None
+
         return {
             "mode": self._mode.lower() if self._mode else None,
             "status": self._status.lower() if self._status else None,
             "success": True,
         }
 
+    async def _fetch_status_via_http(self) -> tuple:
+        """Fetch current mode/status via a lightweight HTTP GET to /wallbox.
+
+        Blazor Server pre-renders the page with the current component state,
+        so a simple GET returns HTML containing 'Mode Eco' / 'Status Connected'
+        etc.  We reuse the existing ``_extract_status_text`` parser.
+
+        Returns:
+            (mode, status) tuple — either value may be None on failure.
+        """
+        own_session = None
+        try:
+            session = self.session
+            if session is None or session.closed:
+                own_session = aiohttp.ClientSession(
+                    connector=aiohttp.TCPConnector(use_dns_cache=False),
+                )
+                session = own_session
+
+            async with session.get(
+                f"{self.base_url}/wallbox",
+                timeout=aiohttp.ClientTimeout(total=10),
+            ) as resp:
+                if resp.status != 200:
+                    _LOGGER.debug(
+                        "[Enpal Wallbox] HTTP status poll returned %s", resp.status
+                    )
+                    return None, None
+                html = await resp.text()
+                return self._extract_status_text(html.encode("utf-8"))
+        except Exception as e:
+            _LOGGER.debug("[Enpal Wallbox] HTTP status poll failed: %s", e)
+            return None, None
+        finally:
+            if own_session and not own_session.closed:
+                await own_session.close()
+
     # ------------------------------------------------------------------
     # Internal: WebSocket message loop
     # ------------------------------------------------------------------

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)
+
+