Fix odometer issue in cupra and skoda SoC modules

packages/modules/vehicles/cupra/api.py

@@ -2,32 +2,35 @@
 
 import aiohttp
 from asyncio import new_event_loop, set_event_loop
-from typing import Union
+from typing import Optional, Tuple
 from modules.vehicles.cupra import libcupra
 from modules.vehicles.cupra.config import Cupra
 from modules.vehicles.vwgroup.vwgroup import VwGroup
 
 
+SoCResult = Tuple[int, float, str, float, Optional[float]]
+
+
 class api(VwGroup):
 
     def __init__(self, conf: Cupra, vehicle: int):
         super().__init__(conf, vehicle)
 
     # async method, called from sync fetch_soc, required because libvwid/libskoda/libcupra expect async environment
-    async def _fetch_soc(self) -> Union[int, float, str]:
+    async def _fetch_soc(self) -> SoCResult:
         async with aiohttp.ClientSession() as self.session:
             cupra = libcupra.cupra(self.session)
             return await super().request_data(cupra)
 
 
-def fetch_soc(conf: Cupra, vehicle: int) -> Union[int, float, str]:
+def fetch_soc(conf: Cupra, vehicle: int) -> SoCResult:
 
     # prepare and call async method
     loop = new_event_loop()
     set_event_loop(loop)
 
-    # get soc, range from server
+    # get soc, range, timestamp, and odometer from server
     a = api(conf, vehicle)
-    soc, range, soc_ts, soc_tsX = loop.run_until_complete(a._fetch_soc())
+    soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc())
 
-    return soc, range, soc_ts, soc_tsX
+    return soc, range, soc_ts, soc_tsX, odometer

packages/modules/vehicles/cupra/libcupra.py

@@ -253,6 +253,7 @@ async def refresh_tokens(self):
 
     async def get_status(self):
         status_url = f"{API_BASE}/vehicles/{self.vin}/charging/status"
+        mileage_url = f"{API_BASE}/v1/vehicles/{self.vin}/mileage"
         response = await self.session.get(status_url, headers=self.headers)
 
         # If first attempt fails, try to refresh tokens
@@ -277,10 +278,27 @@ async def get_status(self):
         status_data = await response.json()
         self.log.debug(f"Status data from Cupra API: {status_data}")
 
+        # Fetch mileage data
+        response = await self.session.get(mileage_url, headers=self.headers)
+        if response.status >= 400:
+            self.log.error("Get mileage failed")
+            odometer = None
+        else:
+            mileage_data = await response.json()
+            self.log.debug(f"Mileage data from Cupra API: {mileage_data}")
+            odometer = mileage_data.get('mileageKm', None)
+
+        battery_value = {
+            'currentSOC_pct': status_data['status']['battery']['currentSOC_pct'],
+            'cruisingRangeElectric_km': status_data['status']['battery']['cruisingRangeElectric_km'],
+            'carCapturedTimestamp': status_data['status']['battery']['carCapturedTimestamp'],
+            'odometer': odometer,
+        }
+
         return {
             'charging': {
                 'batteryStatus': {
-                    'value': status_data['status']['battery']
+                    'value': battery_value,
                 }
             }
         }

packages/modules/vehicles/cupra/soc.py

@@ -16,9 +16,9 @@
 
 
 def fetch(vehicle_update_data: VehicleUpdateData, config: Cupra, vehicle: int) -> CarState:
-    soc, range, soc_ts, soc_tsX = api.fetch_soc(config, vehicle)
-    log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts)
-    return CarState(soc=soc, range=range, soc_timestamp=soc_tsX)
+    soc, range, soc_ts, soc_tsX, odometer = api.fetch_soc(config, vehicle)
+    log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts + ", odometer=" + str(odometer))
+    return CarState(soc=soc, range=range, soc_timestamp=soc_tsX, odometer=odometer)
 
 
 def create_vehicle(vehicle_config: Cupra, vehicle: int):

packages/modules/vehicles/cupra/soc_test.py

@@ -0,0 +1,170 @@
+from unittest.mock import Mock, MagicMock
+import pytest
+import asyncio
+from modules.common import store
+from modules.common.abstract_vehicle import VehicleUpdateData
+from modules.common.component_context import SingleComponentUpdateContext
+from modules.vehicles.cupra import api
+from modules.vehicles.cupra.soc import create_vehicle
+from modules.vehicles.cupra.config import Cupra, CupraConfiguration
+from modules.vehicles.cupra.libcupra import cupra as CupraApi
+
+
+class TestCupra:
+    @pytest.fixture(autouse=True)
+    def set_up(self, monkeypatch):
+        self.mock_context_exit = Mock(return_value=True)
+        self.mock_fetch_soc = Mock(
+            name="fetch_soc",
+            return_value=(60, 320, "2026-03-29T11:00:00Z", 1774782000.0, 45678),
+        )
+        self.mock_value_store = Mock(name="value_store")
+        monkeypatch.setattr(api, "fetch_soc", self.mock_fetch_soc)
+        monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=self.mock_value_store))
+        monkeypatch.setattr(SingleComponentUpdateContext, '__exit__', self.mock_context_exit)
+
+    def test_update_updates_value_store(self):
+        # setup
+        config = Cupra(configuration=CupraConfiguration(user_id="test_user", password="test_password", vin="test_vin"))
+
+        # execution
+        create_vehicle(config, 1).update(VehicleUpdateData())
+
+        # evaluation
+        self.assert_context_manager_called_with(None)
+        self.mock_fetch_soc.assert_called_once_with(config, 1)
+        assert self.mock_value_store.set.call_count == 1
+        call_args = self.mock_value_store.set.call_args[0][0]
+        assert call_args.soc == 60
+        assert call_args.range == 320
+        assert call_args.soc_timestamp == 1774782000.0
+        assert call_args.odometer == 45678
+
+    def test_update_passes_errors_to_context(self):
+        # setup
+        dummy_error = Exception("Der SoC kann nicht ausgelesen werden")
+        self.mock_fetch_soc.side_effect = dummy_error
+        config = Cupra(configuration=CupraConfiguration(user_id="test_user", password="test_password", vin="test_vin"))
+
+        # execution
+        create_vehicle(config, 1).update(VehicleUpdateData())
+
+        # evaluation
+        self.assert_context_manager_called_with_substr(dummy_error)
+
+    def assert_context_manager_called_with(self, error):
+        assert self.mock_context_exit.call_count == 1
+        assert self.mock_context_exit.call_args[0][1] is error
+
+    def assert_context_manager_called_with_substr(self, error):
+        assert self.mock_context_exit.call_count == 1
+        assert str(error) in str(self.mock_context_exit.call_args[0][1])
+
+
+class MockAiohttpResponse:
+    def __init__(self, json_data, status_code):
+        self._json_data = json_data
+        self.status = status_code
+
+    async def json(self):
+        return self._json_data
+
+    def release(self):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+
+class TestCupraGetStatus:
+    @pytest.fixture
+    def mock_session(self):
+        session = MagicMock()
+
+        async def get_side_effect(*args, **kwargs):
+            return session.get.return_value
+        session.get = MagicMock(side_effect=get_side_effect)
+        return session
+
+    @pytest.fixture
+    def cupra_instance(self, mock_session):
+        instance = CupraApi(mock_session)
+        instance.set_vin("test_vin")
+        instance.headers = {"Authorization": "Bearer test_token"}
+        return instance
+
+    def test_get_status_success(self, cupra_instance, mock_session):
+        # setup
+        status_response_data = {
+            "status": {
+                "battery": {
+                    "currentSOC_pct": 67,
+                    "cruisingRangeElectric_km": 305,
+                    "carCapturedTimestamp": "2026-03-29T11:20:00Z"
+                }
+            }
+        }
+        mileage_response_data = {
+            "mileageKm": 77889
+        }
+        responses = [
+            MockAiohttpResponse(status_response_data, 200),
+            MockAiohttpResponse(mileage_response_data, 200)
+        ]
+
+        async def side_effect_func(*args, **kwargs):
+            return responses.pop(0)
+        mock_session.get.side_effect = side_effect_func
+
+        # execution
+        status = asyncio.run(cupra_instance.get_status())
+
+        # evaluation
+        assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 67
+        assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305
+        assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z"
+        assert status['charging']['batteryStatus']['value']['odometer'] == 77889
+        assert mock_session.get.call_count == 2
+        mock_session.get.assert_any_call(
+            "https://ola.prod.code.seat.cloud.vwgroup.com/vehicles/test_vin/charging/status",
+            headers=cupra_instance.headers
+        )
+        mock_session.get.assert_any_call(
+            "https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/test_vin/mileage",
+            headers=cupra_instance.headers
+        )
+
+    def test_get_status_mileage_error_sets_odometer_none(self, cupra_instance, mock_session):
+        # setup
+        status_response_data = {
+            "status": {
+                "battery": {
+                    "currentSOC_pct": 67,
+                    "cruisingRangeElectric_km": 305,
+                    "carCapturedTimestamp": "2026-03-29T11:20:00Z"
+                }
+            }
+        }
+        mileage_error_response = {
+            "error": "service_unavailable"
+        }
+        responses = [
+            MockAiohttpResponse(status_response_data, 200),
+            MockAiohttpResponse(mileage_error_response, 503)
+        ]
+
+        async def side_effect_func(*args, **kwargs):
+            return responses.pop(0)
+        mock_session.get.side_effect = side_effect_func
+
+        # execution
+        status = asyncio.run(cupra_instance.get_status())
+
+        # evaluation
+        assert status['charging']['batteryStatus']['value']['currentSOC_pct'] == 67
+        assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305
+        assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z"
+        assert status['charging']['batteryStatus']['value']['odometer'] is None

packages/modules/vehicles/skoda/api.py

@@ -2,32 +2,35 @@
 
 import aiohttp
 from asyncio import new_event_loop, set_event_loop
-from typing import Union
+from typing import Optional, Tuple
 from modules.vehicles.skoda import libskoda
 from modules.vehicles.skoda.config import Skoda
 from modules.vehicles.vwgroup.vwgroup import VwGroup
 
 
+SoCResult = Tuple[int, float, str, float, Optional[float]]
+
+
 class api(VwGroup):
 
     def __init__(self, conf: Skoda, vehicle: int):
         super().__init__(conf, vehicle)
 
     # async method, called from sync fetch_soc, required because libvwid/libskoda expect async environment
-    async def _fetch_soc(self) -> Union[int, float, str]:
+    async def _fetch_soc(self) -> SoCResult:
         async with aiohttp.ClientSession() as self.session:
             skoda = libskoda.skoda(self.session)
             return await super().request_data(skoda)
 
 
-def fetch_soc(conf: Skoda, vehicle: int) -> Union[int, float, str]:
+def fetch_soc(conf: Skoda, vehicle: int) -> SoCResult:
 
     # prepare and call async method
     loop = new_event_loop()
     set_event_loop(loop)
 
-    # get soc, range from server
+    # get soc, range, timestamp, and odometer from server
     a = api(conf, vehicle)
-    soc, range, soc_ts, soc_tsX = loop.run_until_complete(a._fetch_soc())
+    soc, range, soc_ts, soc_tsX, odometer = loop.run_until_complete(a._fetch_soc())
 
-    return soc, range, soc_ts, soc_tsX
+    return soc, range, soc_ts, soc_tsX, odometer

packages/modules/vehicles/skoda/libskoda.py

@@ -238,29 +238,38 @@ async def refresh_tokens(self):
     async def get_status(self):
         vehicle_status_url = f"{API_BASE}/v2/vehicle-status/{self.vin}/driving-range"
         charging_url = f"{API_BASE}/v1/charging/{self.vin}"
+        maintenance_report_url = f"{API_BASE}/v3/vehicle-maintenance/vehicles/{self.vin}/report"
         response = await self.session.get(vehicle_status_url, headers=self.headers)
 
-        # If first attempt fails, try to refresh tokens
-        if response.status >= 400:
-            self.log.debug("Refreshing tokens")
-            if await self.refresh_tokens():
-                response = await self.session.get(vehicle_status_url, headers=self.headers)
-
-        # If refreshing tokens failed, try a full reconnect
-        if response.status >= 400:
-            self.log.info("Reconnecting")
-            if await self.reconnect():
-                response = await self.session.get(vehicle_status_url, headers=self.headers)
-            else:
-                self.log.error("Reconnect failed")
-                return {}
+        if response.status == 403:
+            self.log.info("vehicle-status returned 403, trying charging_url")
+            status_data = {}
+        else:
+            # If first attempt fails, try to refresh tokens
+            if response.status >= 400:
+                self.log.debug("Refreshing tokens")
+                if await self.refresh_tokens():
+                    response = await self.session.get(vehicle_status_url, headers=self.headers)
 
-        if response.status >= 400:
-            self.log.error("Get status failed")
-            return {}
+            # If refreshing tokens failed, try a full reconnect
+            if response.status >= 400:
+                self.log.info("Reconnecting")
+                if await self.reconnect():
+                    response = await self.session.get(vehicle_status_url, headers=self.headers)
+                else:
+                    self.log.error("Reconnect failed")
+                    return {}
 
-        status_data = await response.json()
-        self.log.debug(f"Status data from Skoda API (vehicle-status): {status_data}")
+            if response.status >= 400:
+                if response.status == 403:
+                    self.log.info("vehicle-status returned 403, trying charging_url")
+                    status_data = {}
+                else:
+                    self.log.error("Get status failed")
+                    return {}
+            else:
+                status_data = await response.json()
+                self.log.debug(f"Status data from Skoda API (vehicle-status): {status_data}")
 
         # check if all values are valid, otherwise use charging_url
         electric_engine_range = {}
@@ -293,13 +302,23 @@ async def get_status(self):
         if not timestamp.endswith('Z'):
             timestamp += 'Z'
 
+        response = await self.session.get(maintenance_report_url, headers=self.headers)
+        if response.status >= 400:
+            self.log.error("Get status from maintenance_report_url failed")
+            odometer = None
+        else:
+            maintenance_data = await response.json()
+            self.log.debug(f"Status data from Skoda API (maintenance report): {maintenance_data}")
+            odometer = maintenance_data.get('mileageInKm', None)
+
         return {
             'charging': {
                 'batteryStatus': {
                     'value': {
                         'currentSOC_pct': soc,
                         'cruisingRangeElectric_km': range_km,
                         'carCapturedTimestamp': timestamp,
+                        'odometer': odometer,
                     }
                 }
             }

packages/modules/vehicles/skoda/soc.py

@@ -16,9 +16,9 @@
 
 
 def fetch(vehicle_update_data: VehicleUpdateData, config: Skoda, vehicle: int) -> CarState:
-    soc, range, soc_ts, soc_tsX = api.fetch_soc(config, vehicle)
-    log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts)
-    return CarState(soc=soc, range=range, soc_timestamp=soc_tsX)
+    soc, range, soc_ts, soc_tsX, odometer = api.fetch_soc(config, vehicle)
+    log.info("Result: soc=" + str(soc)+", range=" + str(range) + "@" + soc_ts + ", odometer=" + str(odometer))
+    return CarState(soc=soc, range=range, soc_timestamp=soc_tsX, odometer=odometer)
 
 
 def create_vehicle(vehicle_config: Skoda, vehicle: int):