feat: add SOC kWh fallback for vehicles without realtimePower

pyproject.toml

@@ -41,6 +41,7 @@ pytest-asyncio = "^1.2.0"
 pytest-mock = "^3.14.0"
 mypy = "^1.15.0"
 pylint = "^4.0.0"
+pre-commit = "^4.6.0"
 
 [tool.poetry.dependencies]
 saic-ismart-client-ng = { develop = true }

src/extractors/__init__.py

@@ -39,6 +39,39 @@ def extract_electric_range(
     return None
 
 
+def extract_soc_kwh(
+    charge_status: ChrgMgmtDataRespProcessingResult | None,
+    soc: float | None,
+) -> float | None:
+    if (
+        charge_status is not None
+        and (raw_soc_kwh := charge_status.soc_kwh) is not None
+        and (soc_kwh := __validate_and_convert_soc_kwh(raw_soc_kwh)) is not None
+    ):
+        LOG.debug("SoC kWh derived from realtimePower")
+        return soc_kwh
+
+    if (
+        soc is not None
+        and charge_status is not None
+        and (
+            capacity := __validate_and_convert_soc_kwh(
+                charge_status.real_total_battery_capacity
+            )
+        )
+        is not None
+    ):
+        LOG.debug(
+            "SoC kWh computed from SoC%%=%s and capacity=%s kWh",
+            soc,
+            capacity,
+        )
+        return round(soc / 100.0 * capacity, 2)
+
+    LOG.warning("Could not extract a valid SoC kWh")
+    return None
+
+
 def extract_soc(
     vehicle_status: VehicleStatusRespProcessingResult,
     charge_status: ChrgMgmtDataRespProcessingResult | None,
@@ -71,3 +104,9 @@ def __validate_and_convert_soc(raw_value: float) -> float | None:
     if value_in_range(raw_value, 0, 100.0, is_max_excl=False):
         return raw_value
     return None
+
+
+def __validate_and_convert_soc_kwh(raw_value: float) -> float | None:
+    if raw_value > 0:
+        return raw_value
+    return None

src/handlers/message.py

@@ -54,10 +54,13 @@ async def __polling(self) -> None:
             if (
                 latest_message is not None
                 and latest_message.messageId != self.last_message_id
-                and ensure_datetime_aware(latest_message.message_time) > self.last_message_ts
+                and ensure_datetime_aware(latest_message.message_time)
+                > self.last_message_ts
             ):
                 self.last_message_id = latest_message.messageId
-                self.last_message_ts = ensure_datetime_aware(latest_message.message_time)
+                self.last_message_ts = ensure_datetime_aware(
+                    latest_message.message_time
+                )
                 LOG.info(
                     f"{latest_message.title} detected at {latest_message.message_time}"
                 )
@@ -107,7 +110,8 @@ async def __get_all_alarm_messages(self) -> list[MessageEntity]:
                 oldest_message = self.__get_oldest_message(all_messages)
                 if (
                     oldest_message is not None
-                    and ensure_datetime_aware(oldest_message.message_time) < self.last_message_ts
+                    and ensure_datetime_aware(oldest_message.message_time)
+                    < self.last_message_ts
                 ):
                     return all_messages
             except SaicLogoutException:
@@ -121,7 +125,9 @@ async def __get_all_alarm_messages(self) -> list[MessageEntity]:
                 return all_messages
             finally:
                 idx = idx + 1
-        LOG.warning("Reached max page limit (%d) while fetching alarm messages", max_pages)
+        LOG.warning(
+            "Reached max page limit (%d) while fetching alarm messages", max_pages
+        )
         return all_messages
 
     async def __delete_message(self, message: MessageEntity) -> None:
@@ -174,12 +180,16 @@ def __get_latest_message(
     ) -> MessageEntity | None:
         if len(vehicle_start_messages) == 0:
             return None
-        return max(vehicle_start_messages, key=lambda m: ensure_datetime_aware(m.message_time))
+        return max(
+            vehicle_start_messages, key=lambda m: ensure_datetime_aware(m.message_time)
+        )
 
     @staticmethod
     def __get_oldest_message(
         vehicle_start_messages: list[MessageEntity],
     ) -> MessageEntity | None:
         if len(vehicle_start_messages) == 0:
             return None
-        return min(vehicle_start_messages, key=lambda m: ensure_datetime_aware(m.message_time))
+        return min(
+            vehicle_start_messages, key=lambda m: ensure_datetime_aware(m.message_time)
+        )

src/integrations/openwb/__init__.py

@@ -65,7 +65,9 @@ def update_openwb(
             soc_ts_topic = self.__charging_station.soc_ts_topic
             if soc_ts_topic is not None:
                 soc_ts = int(datetime.datetime.now(tz=datetime.UTC).timestamp())
-                LOG.info("OpenWB Integration published SoC timestamp to %s", soc_ts_topic)
+                LOG.info(
+                    "OpenWB Integration published SoC timestamp to %s", soc_ts_topic
+                )
                 self.__publisher.publish_int(
                     key=soc_ts_topic,
                     value=soc_ts,

src/status_publisher/charge/chrg_mgmt_data_resp.py

@@ -41,6 +41,7 @@ class ChrgMgmtDataRespProcessingResult:
     real_total_battery_capacity: float
     raw_soc: int | None
     raw_fuel_range_elec: int | None
+    soc_kwh: float | None
 
 
 class ChrgMgmtDataRespPublisher(
@@ -112,4 +113,7 @@ def publish(
             raw_fuel_range_elec=charge_status_result.raw_fuel_range_elec
             if charge_status_result is not None
             else None,
+            soc_kwh=charge_status_result.soc_kwh
+            if charge_status_result is not None
+            else None,
         )

src/status_publisher/charge/rvs_charge_status.py

@@ -21,6 +21,7 @@
 class RvsChargeStatusProcessingResult:
     real_total_battery_capacity: float
     raw_fuel_range_elec: int | None
+    soc_kwh: float | None
 
 
 class RvsChargeStatusPublisher(
@@ -38,7 +39,10 @@ def update_total_mileage(self, raw_mileage: int) -> None:
     def _is_valid_partial_mileage(self, raw_value: int) -> bool:
         if not value_in_range(raw_value, 0, 65535):
             return False
-        if self._last_total_mileage_raw is not None and raw_value > self._last_total_mileage_raw:
+        if (
+            self._last_total_mileage_raw is not None
+            and raw_value > self._last_total_mileage_raw
+        ):
             LOG.warning(
                 "Partial mileage %d exceeds total mileage %d, skipping",
                 raw_value,
@@ -98,13 +102,13 @@ def publish(
             validator=lambda x: x > 0,
         )
 
-        self._transform_and_publish(
-            topic=mqtt_topics.DRIVETRAIN_SOC_KWH,
-            value=charge_status.realtimePower,
-            transform=lambda p: round(
-                (battery_capacity_correction_factor * p) / 10.0, 2
-            ),
-        )
+        soc_kwh: float | None = None
+        if charge_status.realtimePower:
+            soc_kwh = round(
+                (battery_capacity_correction_factor * charge_status.realtimePower)
+                / 10.0,
+                2,
+            )
 
         self._transform_and_publish(
             topic=mqtt_topics.DRIVETRAIN_LAST_CHARGE_ENDING_POWER,
@@ -136,6 +140,7 @@ def publish(
         return RvsChargeStatusProcessingResult(
             real_total_battery_capacity=real_total_battery_capacity,
             raw_fuel_range_elec=charge_status.fuelRangeElec,
+            soc_kwh=soc_kwh,
         )
 
     def get_actual_battery_capacity(

src/vehicle.py

@@ -15,7 +15,7 @@
     TargetBatteryCode,
 )
 
-from extractors import extract_electric_range, extract_soc
+from extractors import extract_electric_range, extract_soc, extract_soc_kwh
 import mqtt_topics
 from publisher.core import Publishable
 from status_publisher.charge.chrg_mgmt_data_resp import (
@@ -659,6 +659,13 @@ def update_data_conflicting_in_vehicle_and_bms(
                 value=soc,
             )
 
+        soc_kwh = extract_soc_kwh(charge_status, soc)
+        if soc_kwh is not None:
+            self.__publish(
+                topic=mqtt_topics.DRIVETRAIN_SOC_KWH,
+                value=soc_kwh,
+            )
+
     @property
     def user_timezone(self) -> ZoneInfo | None:
         return self.__user_timezone

src/vehicle_info.py

@@ -64,12 +64,14 @@ def get_ac_temperature_idx(self, remote_ac_temperature: int) -> int:
     def min_ac_temperature(self) -> int:
         if self.series.startswith("EH32"):
             return 17
+        if self.series.startswith("MZS3E"):
+            return 16
         return 16
 
     @property
     def max_ac_temperature(self) -> int:
-        if self.series.startswith("EH32"):
-            return 33
+        if self.series.startswith("EH32") or self.series.startswith("MZS3E"):
+            return 31
         return 28
 
     def __get_property_by_code(self, property_name: str) -> str | None:
@@ -122,6 +124,8 @@ def real_battery_capacity(self) -> float | None:
             result = self.__mg5_real_battery_capacity
         elif self.series.startswith("ZS EV"):
             result = self.__zs_ev_real_battery_capacity
+        elif self.series.startswith("MZS3E"):
+            result = self.__mgs5_real_battery_capacity
 
         if result is None:
             LOG.warning(
@@ -141,6 +145,16 @@ def __mg4_real_battery_capacity(self) -> float | None:
         # MG4 with LFP battery
         return 51.0
 
+    @property
+    def __mgs5_real_battery_capacity(self) -> float | None:
+        # From the datasheet
+        # Battery pack type 1 (49kWh)
+        # Battery pack type 2 (62.2kWh)
+        # Battery pack type 3 (64kWh)
+        if self.supports_target_soc:
+            return 64.0
+        return 49.0
+
     @property
     def __cyberster_real_battery_capacity(self) -> float | None:
         # Model: MG Cyberster

tests/status_publisher/test_rvs_charge_status.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import unittest
+
+import pytest
+from saic_ismart_client_ng.api.vehicle.schema import VehicleModelConfiguration, VinInfo
+from saic_ismart_client_ng.api.vehicle_charging.schema import RvsChargeStatus
+
+from configuration import Configuration
+from status_publisher.charge.rvs_charge_status import RvsChargeStatusPublisher
+from tests.common_mocks import VIN
+from tests.mocks import MessageCapturingConsolePublisher
+from vehicle_info import VehicleInfo
+
+# EH32 S with BType=1 → real_battery_capacity = 64.0, raw=72.5 kWh
+REAL_CAPACITY = 64.0
+RAW_CAPACITY = 72.5
+CORRECTION = REAL_CAPACITY / RAW_CAPACITY
+
+
+def _make_publisher() -> tuple[
+    RvsChargeStatusPublisher, MessageCapturingConsolePublisher
+]:
+    config = Configuration()
+    config.anonymized_publishing = False
+    pub = MessageCapturingConsolePublisher(config)
+    vin_info = VinInfo()
+    vin_info.vin = VIN
+    vin_info.series = "EH32 S"
+    vin_info.modelName = "MG4 Electric"
+    vin_info.modelYear = "2022"
+    vin_info.vehicleModelConfiguration = [
+        VehicleModelConfiguration("BType", "Battery", "1"),
+    ]
+    vehicle_info = VehicleInfo(vin_info, None)
+    return RvsChargeStatusPublisher(vehicle_info, pub, f"/vehicles/{VIN}"), pub
+
+
+class TestRvsChargeStatusSocKwh(unittest.TestCase):
+    def setUp(self) -> None:
+        self.publisher, _ = _make_publisher()
+
+    def test_soc_kwh_present_when_realtime_power_set(self) -> None:
+        charge_status = RvsChargeStatus(
+            realtimePower=int((42.0 / CORRECTION) * 10),
+            totalBatteryCapacity=int(RAW_CAPACITY * 10),
+        )
+        result = self.publisher.publish(charge_status)
+
+        assert result.soc_kwh is not None
+        assert result.soc_kwh == pytest.approx(42.0, abs=0.1)
+
+    def test_soc_kwh_none_when_realtime_power_is_none(self) -> None:
+        charge_status = RvsChargeStatus(
+            realtimePower=None,
+            totalBatteryCapacity=int(RAW_CAPACITY * 10),
+        )
+        result = self.publisher.publish(charge_status)
+
+        assert result.soc_kwh is None
+
+    def test_soc_kwh_none_when_realtime_power_is_zero(self) -> None:
+        charge_status = RvsChargeStatus(
+            realtimePower=0,
+            totalBatteryCapacity=int(RAW_CAPACITY * 10),
+        )
+        result = self.publisher.publish(charge_status)
+
+        assert result.soc_kwh is None

tests/test_extractors.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import pytest
+
+from extractors import extract_soc_kwh
+from status_publisher.charge.chrg_mgmt_data_resp import ChrgMgmtDataRespProcessingResult
+
+
+def _make_charge_result(
+    *,
+    soc_kwh: float | None = None,
+    real_total_battery_capacity: float = 64.0,
+) -> ChrgMgmtDataRespProcessingResult:
+    return ChrgMgmtDataRespProcessingResult(
+        charge_current_limit=None,
+        target_soc=None,
+        scheduled_charging=None,
+        is_charging=None,
+        remaining_charging_time=None,
+        power=None,
+        real_total_battery_capacity=real_total_battery_capacity,
+        raw_soc=None,
+        raw_fuel_range_elec=None,
+        soc_kwh=soc_kwh,
+    )
+
+
+class TestExtractSocKwh:
+    def test_prefers_realtime_power_soc_kwh(self) -> None:
+        result = extract_soc_kwh(_make_charge_result(soc_kwh=42.0), soc=80.0)
+        assert result == pytest.approx(42.0)
+
+    def test_fallback_to_soc_times_capacity(self) -> None:
+        # 80% of 64 kWh = 51.2 kWh
+        result = extract_soc_kwh(_make_charge_result(soc_kwh=None), soc=80.0)
+        assert result == pytest.approx(51.2)
+
+    def test_fallback_returns_none_when_soc_is_none(self) -> None:
+        result = extract_soc_kwh(_make_charge_result(soc_kwh=None), soc=None)
+        assert result is None
+
+    def test_fallback_returns_none_when_charge_status_is_none(self) -> None:
+        result = extract_soc_kwh(None, soc=80.0)
+        assert result is None
+
+    def test_fallback_returns_none_when_capacity_is_zero(self) -> None:
+        result = extract_soc_kwh(
+            _make_charge_result(soc_kwh=None, real_total_battery_capacity=0.0), soc=80.0
+        )
+        assert result is None
+
+    def test_fallback_used_when_soc_kwh_is_zero(self) -> None:
+        # soc_kwh=0 is not a valid primary reading; fall back to soc * capacity
+        result = extract_soc_kwh(_make_charge_result(soc_kwh=0.0), soc=80.0)
+        assert result == pytest.approx(51.2)