Allow charging.level to re-anchor PHEV prediction downward during charging

The PHEV sync-down protection from 7d5e4bd only updated display values,
which the monotonicity guard in get_predicted_soc() silently overrode.
Pass a from_charging_level flag so update_bmw_soc can distinguish trusted
charging.level (full re-anchor) from stale header (display-only, preserved
by the existing skip-header logic).

Fixes #281

Author: renaudallard

README.md

@@ -240,6 +240,7 @@ For Plug-in Hybrid Electric Vehicles (PHEVs), the predicted SOC has special hand
 - **Automatic PHEV detection**: Vehicles with both an HV battery and fuel system are detected as PHEVs, unless metadata (driveTrain/propulsionType) or the model name (e.g. i4, iX, i5) identifies them as a known BEV
 - **Sync down on battery depletion**: If the actual BMW SOC is lower than the predicted value, the prediction syncs down immediately. This handles scenarios where the hybrid system depletes the battery (e.g., battery recovery mode, engine-priority driving)
 - **Stale header filtering during charging**: When `charging.level` is available and fresh, the stale `batteryManagement.header` value is skipped to avoid corrupting the predicted SOC display
+- **Charging level re-anchor**: During charging, `charging.level` can re-anchor the prediction downward (correcting overshoot). Stale header values are still blocked from doing this, preserving session integrity
 - **BEVs**: For pure electric vehicles, the predicted SOC only syncs when not actively charging (standard behavior)
 
 This ensures the predicted SOC stays accurate for PHEVs even when the hybrid system uses battery power in ways that don't register as "discharging" in the BMW API.

custom_components/cardata/soc_prediction.py

@@ -296,7 +296,9 @@ def update_power_reading(self, vin: str, power_kw: float | None = None, aux_powe
                     session.last_predicted_soc,
                 )
 
-    def update_bmw_soc(self, vin: str, soc: float, timestamp: datetime | None = None) -> None:
+    def update_bmw_soc(
+        self, vin: str, soc: float, timestamp: datetime | None = None, *, from_charging_level: bool = False
+    ) -> None:
         """Record BMW SOC update for staleness tracking.
 
         Also updates last_predicted_soc when not charging (passthrough mode).
@@ -337,7 +339,15 @@ def update_bmw_soc(self, vin: str, soc: float, timestamp: datetime | None = None
                         session.anchor_soc = soc
                         session.total_energy_kwh = 0.0
                         session.last_energy_update = time.time()
-                    # During charging: preserve anchor/energy for learning
+                    elif from_charging_level:
+                        # Charging, from trusted charging.level: full re-anchor downward
+                        old_anchor = session.anchor_soc
+                        ref_time = session.last_energy_update or session.anchor_timestamp.timestamp()
+                        session.anchor_soc = soc
+                        session.total_energy_kwh = 0.0
+                        session.last_energy_update = time.time()
+                        self._derive_power_from_soc_change(vin, session, old_anchor, soc, ref_time)
+                    # During charging from header: preserve anchor/energy for learning
             elif not is_charging:
                 # Not charging: snap to actual BMW SOC
                 self._last_predicted_soc[vin] = soc

custom_components/cardata/soc_wiring.py

@@ -641,7 +641,7 @@ def process_soc_descriptors(
                 if not skip_stale_level:
                     try:
                         level_val = float(value)
-                        soc_predictor.update_bmw_soc(vin, level_val)
+                        soc_predictor.update_bmw_soc(vin, level_val, from_charging_level=True)
                         if not soc_predictor.has_active_session(vin):
                             _LOGGER.debug(
                                 "Late anchor attempt for %s (charging.level arrived after charging started)",