Fahrzeug-Module: Warnung bei fahrzeugseitigem Ladelimit

packages/helpermodules/update_config.py

@@ -58,7 +58,7 @@
 
 class UpdateConfig:
 
-    DATASTORE_VERSION = 147
+    DATASTORE_VERSION = 148
 
     valid_topic = [
         "^openWB/bat/config/bat_control_activated$",
@@ -3847,3 +3847,20 @@ def upgrade(topic: str, payload) -> Optional[dict]:
                     return {topic: payload_component}
         self._loop_all_received_topics(upgrade)
         self._append_datastore_version(147)
+
+    def upgrade_datastore_148(self) -> None:
+        """PSA-Modul entfernt: Die Schnittstelle wurde von PSA/Stellantis bereits 2024 abgeschaltet,
+        das Modul lieferte seitdem nur noch eine Fehlermeldung statt eines SoC-Werts."""
+        def upgrade(topic: str, payload) -> Optional[dict]:
+            if re.search("^openWB/vehicle/[0-9]+/soc_module/config$", topic) is not None:
+                payload = decode_payload(payload)
+                if payload.get("type") == "psa":
+                    pub_system_message(
+                        {},
+                        "Das PSA-Fahrzeug-Modul wurde entfernt, da PSA/Stellantis die Schnittstelle bereits "
+                        "2024 abgeschaltet hat. Bitte konfiguriere ein anderes Fahrzeug-Modul, z.B. Tronity.",
+                        MessageType.WARNING,
+                    )
+                    return {topic: NO_MODULE}
+        self._loop_all_received_topics(upgrade)
+        self._append_datastore_version(148)

packages/modules/common/component_state.py

@@ -158,12 +158,15 @@ class CarState:
     def __init__(self, soc: float,
                  range: Optional[float] = None,
                  soc_timestamp: Optional[float] = None,
-                 odometer: Optional[float] = None):
+                 odometer: Optional[float] = None,
+                 warning: Optional[str] = None):
         """Args:
             soc: actual state of charge in percent
             range: actual range in km
             soc_timestamp: timestamp of last request as unix timestamp
             odometer: actual odometer of vehicle in km
+            warning: informational message about a vehicle-side setting affecting charging
+                (eg. a manufacturer battery-care-mode limit), shown to the user but not an error
         """
         self.soc = soc
         self.range = range
@@ -175,6 +178,7 @@ def __init__(self, soc: float,
                 soc_timestamp /= 1000
             self.soc_timestamp = soc_timestamp
         self.odometer = odometer
+        self.warning = warning
 
 
 @auto_str

packages/modules/common/configurable_vehicle.py

@@ -130,6 +130,9 @@ def _get_carstate_by_source(self, vehicle_update_data: VehicleUpdateData, source
             try:
                 _carState = self.__component_updater(vehicle_update_data)
                 if type(_carState) is CarState:
+                    if _carState.warning:
+                        self.fault_state.warning(_carState.warning)
+                        self.fault_state.store_error()
                     _odometer = _carState.odometer
                     _now = int(time.time())
                     _diff = 0

packages/modules/vehicles/bmw_cardata/soc.py

@@ -45,6 +45,7 @@
     # wird nur bei Fahrtende befüllt ("Mileage after last drive")
     "vehicle.trip.segment.end.travelledDistance",
 ]
+FIELD_TARGET_SOC = "vehicle.powertrain.electric.battery.stateOfCharge.target"
 
 CONTAINER_NAME = "ChargeStats"
 CONTAINER_PURPOSE = "openWB"
@@ -53,13 +54,16 @@
     *FIELD_SOC_CANDIDATES,
     *FIELD_RANGE_CANDIDATES,
     *FIELD_ODOMETER_CANDIDATES,
+    FIELD_TARGET_SOC,
+]
+# Manche neueren Fahrzeuge kennen das älteste (erste) SoC-Attribut oder das
+# Lade-Ziel-Attribut nicht mehr. Legt man einen Container mit einem für das
+# Fahrzeug nicht verfügbaren Descriptor an, antwortet die BMW-API dabei
+# offenbar teils mit einem Serverfehler statt einer sauberen 400er. Als
+# Fallback wird die Container-Erstellung ohne diese Attribute wiederholt.
+CONTAINER_DESCRIPTORS_FALLBACK = [
+    d for d in CONTAINER_DESCRIPTORS if d not in (FIELD_SOC_CANDIDATES[0], FIELD_TARGET_SOC)
 ]
-# Manche neueren Fahrzeuge kennen das älteste (erste) SoC-Attribut nicht mehr.
-# Legt man einen Container mit einem für das Fahrzeug nicht verfügbaren
-# Descriptor an, antwortet die BMW-API dabei offenbar teils mit einem
-# Serverfehler statt einer sauberen 400er. Als Fallback wird die
-# Container-Erstellung ohne dieses eine Attribut wiederholt.
-CONTAINER_DESCRIPTORS_FALLBACK = [d for d in CONTAINER_DESCRIPTORS if d != FIELD_SOC_CANDIDATES[0]]
 
 
 def _get_session(token: Optional[str] = None):
@@ -144,7 +148,8 @@ def _create_container(token: str, descriptors: List[str] = None, _is_retry: bool
         if not _is_retry:
             log.warning(
                 "BMW CarData: Container-Erstellung fehlgeschlagen (%s). Versuche erneut ohne "
-                "'%s' (evtl. für dieses Fahrzeug nicht verfügbar).", e, FIELD_SOC_CANDIDATES[0],
+                "'%s'/'%s' (evtl. für dieses Fahrzeug nicht verfügbar).",
+                e, FIELD_SOC_CANDIDATES[0], FIELD_TARGET_SOC,
             )
             return _create_container(token, CONTAINER_DESCRIPTORS_FALLBACK, _is_retry=True)
         raise Exception(f"BMW CarData: Container konnte nicht erstellt werden: {e}")
@@ -282,14 +287,20 @@ def fetch_soc(config: BmwCardataSetup, vehicle: int = 0) -> CarState:
     range_raw = _extract_first_value(td, FIELD_RANGE_CANDIDATES)
     status = _extract_value(td, FIELD_STATUS)
     odometer_raw = _extract_first_value(td, FIELD_ODOMETER_CANDIDATES)
+    target_soc_raw = _extract_value(td, FIELD_TARGET_SOC)
 
     soc = int(float(soc_raw)) if soc_raw is not None else None
     vehicle_range = int(float(range_raw)) if range_raw is not None else None
     odometer = int(float(odometer_raw)) if odometer_raw is not None else None
+    target_soc = int(float(target_soc_raw)) if target_soc_raw is not None else None
 
     if soc is None:
         raise Exception("BMW CarData: Kein SoC-Wert in API-Antwort gefunden!")
 
+    warning = None
+    if target_soc is not None and target_soc < 100:
+        warning = f"Das Fahrzeug begrenzt die Ladung selbst auf {target_soc}%."
+
     if vehicle_range is None and cfg.container_id:
         log.warning(
             "BMW CarData: Kein Reichweitenwert im Container gefunden – "
@@ -304,7 +315,7 @@ def fetch_soc(config: BmwCardataSetup, vehicle: int = 0) -> CarState:
         status,
         odometer,
     )
-    return CarState(soc=soc, range=vehicle_range, odometer=odometer)
+    return CarState(soc=soc, range=vehicle_range, odometer=odometer, warning=warning)
 
 
 def create_vehicle(vehicle_config: BmwCardataSetup, vehicle: int):

packages/modules/vehicles/bmw_cardata/soc_test.py

@@ -6,7 +6,9 @@
 from modules.common import store
 from modules.common.abstract_vehicle import VehicleUpdateData
 from modules.common.component_context import SingleComponentUpdateContext
-from modules.vehicles.bmw_cardata.soc import FIELD_SOC_CANDIDATES, create_vehicle, fetch_soc
+from modules.vehicles.bmw_cardata.soc import (
+    FIELD_SOC_CANDIDATES, FIELD_TARGET_SOC, create_vehicle, fetch_soc,
+)
 from modules.vehicles.bmw_cardata.config import BmwCardataSetup, BmwCardataConfiguration
 
 
@@ -169,6 +171,92 @@ def test_odometer_trip_end_fallback(self, monkeypatch):
         result = fetch_soc(self._make_config())
         assert result.odometer == 45000
 
+    def test_target_soc_below_100_has_warning(self, monkeypatch):
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "telematicData": {
+                "vehicle.drivetrain.electricEngine.charging.level": {"value": "75", "unit": "%"},
+                "vehicle.powertrain.electric.battery.stateOfCharge.target": {"value": "80", "unit": "%"},
+            }
+        }
+        mock_session = Mock()
+        mock_session.get.return_value = mock_response
+        mock_session.headers = {}
+        monkeypatch.setattr("modules.vehicles.bmw_cardata.soc.req.get_http_session", Mock(return_value=mock_session))
+
+        result = fetch_soc(self._make_config())
+        assert result.warning is not None
+        assert "80" in result.warning
+
+    def test_target_soc_100_has_no_warning(self, monkeypatch):
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "telematicData": {
+                "vehicle.drivetrain.electricEngine.charging.level": {"value": "75", "unit": "%"},
+                "vehicle.powertrain.electric.battery.stateOfCharge.target": {"value": "100", "unit": "%"},
+            }
+        }
+        mock_session = Mock()
+        mock_session.get.return_value = mock_response
+        mock_session.headers = {}
+        monkeypatch.setattr("modules.vehicles.bmw_cardata.soc.req.get_http_session", Mock(return_value=mock_session))
+
+        result = fetch_soc(self._make_config())
+        assert result.warning is None
+
+    def test_no_target_soc_has_no_warning(self, monkeypatch):
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "telematicData": {
+                "vehicle.drivetrain.electricEngine.charging.level": {"value": "75", "unit": "%"},
+            }
+        }
+        mock_session = Mock()
+        mock_session.get.return_value = mock_response
+        mock_session.headers = {}
+        monkeypatch.setattr("modules.vehicles.bmw_cardata.soc.req.get_http_session", Mock(return_value=mock_session))
+
+        result = fetch_soc(self._make_config())
+        assert result.warning is None
+
+    def test_container_create_retries_without_target_soc_descriptor(self, monkeypatch):
+        mock_get_response_empty = Mock()
+        mock_get_response_empty.json.return_value = {"containers": []}
+
+        post_call_count = [0]
+
+        def mock_post(url, json=None):
+            if "containers" in url:
+                post_call_count[0] += 1
+                if post_call_count[0] == 1:
+                    mock_resp = Mock()
+                    mock_resp.status_code = 500
+                    raise RequestsHTTPError(response=mock_resp)
+                assert FIELD_TARGET_SOC not in json["technicalDescriptors"]
+                assert FIELD_SOC_CANDIDATES[0] not in json["technicalDescriptors"]
+                return Mock(json=Mock(return_value={"containerId": "fallback-container-id"}))
+            raise AssertionError(f"Unerwarteter POST an {url}")
+
+        mock_session = Mock()
+        mock_session.get.return_value = mock_get_response_empty
+        mock_session.post.side_effect = mock_post
+        mock_session.headers = {}
+        monkeypatch.setattr("modules.vehicles.bmw_cardata.soc.req.get_http_session", Mock(return_value=mock_session))
+        monkeypatch.setattr(
+            "modules.vehicles.bmw_cardata.soc._fetch_telematic_data",
+            Mock(return_value={
+                "telematicData": {
+                    "vehicle.powertrain.electric.battery.stateOfCharge.displayed": {"value": "58", "unit": "%"},
+                }
+            }),
+        )
+
+        config = self._make_config(container_id="")
+        result = fetch_soc(config)
+
+        assert result.soc == 58
+        assert post_call_count[0] == 2
+
     def test_no_soc_raises(self, monkeypatch):
         mock_response = Mock()
         mock_response.json.return_value = {"telematicData": {}}

packages/modules/vehicles/kia/api.py

@@ -885,12 +885,21 @@ def getStatusFull(vehicle_id: str, control_token: str,
             log.warning("kia.getStatusFull: odometer not available in "
                         "vehicle status response")
 
+        warning = None
+        try:
+            target_soc = int(response_dict['resMsg']['state']['Vehicle']['Green']
+                                          ['ChargingInformation']['TargetSoC']['Standard'])
+            if target_soc < 100:
+                warning = f"Das Fahrzeug begrenzt die Ladung selbst auf {target_soc}% (AC-Ladeziel)."
+        except (KeyError, TypeError, ValueError):
+            log.debug("kia.getStatusFull: AC-Ladeziel not available in vehicle status response")
+
     except Exception:
         log.exception("kia.getStatusFull: receiving update error: " +
                       response)
         raise
 
-    return CarState(soc=soc, range=range, odometer=odometer)
+    return CarState(soc=soc, range=range, odometer=odometer, warning=warning)
 
 # ---------- main function ----------
 

packages/modules/vehicles/kia/api_test.py

@@ -0,0 +1,49 @@
+import json
+from unittest.mock import Mock
+
+import pytest
+
+from modules.vehicles.kia import api
+
+
+def _status_response(target_soc=None):
+    vehicle = {
+        "Green": {
+            "BatteryManagement": {"BatteryRemain": {"Ratio": 55}},
+        },
+        "Drivetrain": {
+            "FuelSystem": {"DTE": {"Total": 300}},
+            "Odometer": 12345,
+        },
+    }
+    if target_soc is not None:
+        vehicle["Green"]["ChargingInformation"] = {"TargetSoC": {"Standard": target_soc}}
+    return json.dumps({"resMsg": {"state": {"Vehicle": vehicle}}})
+
+
+class TestKiaApi:
+    @pytest.fixture(autouse=True)
+    def set_up(self, monkeypatch):
+        monkeypatch.setattr(api, "getString", Mock(return_value="x"))
+        monkeypatch.setattr(api, "getStamp", Mock(return_value="stamp"))
+        monkeypatch.setattr(api.time, "sleep", Mock())
+
+    def _call(self, monkeypatch, response_json):
+        monkeypatch.setattr(api, "getHTTP", Mock(return_value=response_json))
+        token = {"deviceId": "d", "gcmClientId": "c", "gcmVehicleId": "v",
+                 "tokenType": "Bearer", "accessToken": "a"}
+        return api.getStatusFull("vehicle-id", "control-token", token, "kia")
+
+    def test_no_target_soc_has_no_warning(self, monkeypatch):
+        result = self._call(monkeypatch, _status_response())
+        assert result.warning is None
+        assert result.soc == 55
+
+    def test_target_soc_below_100_has_warning(self, monkeypatch):
+        result = self._call(monkeypatch, _status_response(target_soc=80))
+        assert result.warning is not None
+        assert "80" in result.warning
+
+    def test_target_soc_100_has_no_warning(self, monkeypatch):
+        result = self._call(monkeypatch, _status_response(target_soc=100))
+        assert result.warning is None

packages/modules/vehicles/myskoda/api.py

@@ -118,3 +118,14 @@ def extract_odometer(data: dict) -> Optional[float]:
     if odometer is None or odometer.get("mileageInKm") is None:
         return None
     return float(odometer["mileageInKm"])
+
+
+def extract_warning(data: dict) -> Optional[str]:
+    vehicle = data.get("vehicle") or {}
+    settings = ((vehicle.get("charging") or {}).get("settings")) or {}
+    if settings.get("chargingCareMode") != "ACTIVATED":
+        return None
+    target = settings.get("targetStateOfChargeInPercent")
+    if target is None or target >= 100:
+        return None
+    return f"Battery Care Mode ist im Fahrzeug aktiv und begrenzt die Ladung selbst auf {target}%."

packages/modules/vehicles/myskoda/api_test.py

@@ -0,0 +1,36 @@
+from modules.vehicles.myskoda.api import extract_warning
+
+
+def _data(charging_care_mode=None, target_soc=None):
+    settings = {}
+    if charging_care_mode is not None:
+        settings["chargingCareMode"] = charging_care_mode
+    if target_soc is not None:
+        settings["targetStateOfChargeInPercent"] = target_soc
+    return {"vehicle": {"charging": {"settings": settings}}}
+
+
+def test_extract_warning_without_settings_returns_none():
+    assert extract_warning({"vehicle": {}}) is None
+
+
+def test_extract_warning_care_mode_deactivated_returns_none():
+    data = _data(charging_care_mode="DEACTIVATED", target_soc=80)
+    assert extract_warning(data) is None
+
+
+def test_extract_warning_care_mode_activated_returns_warning():
+    data = _data(charging_care_mode="ACTIVATED", target_soc=80)
+    warning = extract_warning(data)
+    assert warning is not None
+    assert "80" in warning
+
+
+def test_extract_warning_care_mode_activated_without_target_returns_none():
+    data = _data(charging_care_mode="ACTIVATED")
+    assert extract_warning(data) is None
+
+
+def test_extract_warning_care_mode_activated_target_100_returns_none():
+    data = _data(charging_care_mode="ACTIVATED", target_soc=100)
+    assert extract_warning(data) is None

packages/modules/vehicles/myskoda/soc.py

@@ -27,8 +27,9 @@ def fetch(vehicle_config: Myskoda, vehicle: int) -> CarState:
     soc = api.extract_soc(data)
     range_km = api.extract_range(data)
     odometer_km = api.extract_odometer(data)
+    warning = api.extract_warning(data)
 
-    return CarState(soc=soc, range=range_km, odometer=odometer_km)
+    return CarState(soc=soc, range=range_km, odometer=odometer_km, warning=warning)
 
 
 def create_vehicle(vehicle_config: Myskoda, vehicle: int):