EV-SoC-Fallback to calculation when Online Query fails

packages/modules/common/abstract_vehicle.py

@@ -5,6 +5,7 @@
 @dataclass
 class VehicleUpdateData:
     plug_state: bool = False
+    plug_time: float = 0.0
     charge_state: bool = False
     imported: float = 0
     battery_capacity: float = 82

packages/modules/common/configurable_vehicle.py

@@ -1,5 +1,6 @@
 from enum import Enum
 import logging
+# import time
 from typing import Optional, TypeVar, Generic, Callable
 from helpermodules import timecheck
 
@@ -14,6 +15,7 @@
 from modules.vehicles.manual.config import ManualSoc
 from modules.vehicles.mqtt.config import MqttSocSetup
 
+
 T_VEHICLE_CONFIG = TypeVar("T_VEHICLE_CONFIG")
 
 log = logging.getLogger(__name__)
@@ -125,7 +127,34 @@ def _get_carstate_source(self, vehicle_update_data: VehicleUpdateData) -> SocSou
 
     def _get_carstate_by_source(self, vehicle_update_data: VehicleUpdateData, source: SocSource) -> CarState:
         if source == SocSource.API:
-            return self.__component_updater(vehicle_update_data)
+            try:
+                _carState = self.__component_updater(vehicle_update_data)
+            except Exception as e:
+                if vehicle_update_data.plug_state and\
+                   vehicle_update_data.last_soc and\
+                   vehicle_update_data.last_soc_timestamp >= vehicle_update_data.plug_time and\
+                   (self.calculated_soc_state.last_imported or vehicle_update_data.imported):
+                    _txt1 = "SoC FALLBACK: SoC wird berechnet, da ein Fehler bei der Abfrage aufgetreten ist:"
+                    self.fault_state.warning(f"{_txt1} {e}")
+                    _carState = CarState(soc=calc_soc.calc_soc(
+                                         vehicle_update_data,
+                                         vehicle_update_data.efficiency,
+                                         self.calculated_soc_state.last_imported or vehicle_update_data.imported,
+                                         vehicle_update_data.battery_capacity))
+                else:
+                    if not vehicle_update_data.plug_state:
+                        reason = ", weil kein Fahrzeug eingesteckt ist."
+                    elif not vehicle_update_data.last_soc:
+                        reason = ", weil kein SOC-Wert verfügbar ist."
+                    elif vehicle_update_data.last_soc_timestamp < vehicle_update_data.plug_time:
+                        reason = ", da der SOC-Zeitstempel vor dem Einstecken liegt."
+                    elif not (self.calculated_soc_state.last_imported or vehicle_update_data.imported):
+                        reason = ", weil Daten zum Berechnen des SOC fehlen."
+                    else:
+                        reason = ""
+                    _txt1 = "Die Berechnung vom letzten bekannten Soc ist nicht möglich"
+                    raise Exception(f"Der SoC kann nicht ausgelesen werden: {e}. {_txt1}{reason}")
+            return _carState
         elif source == SocSource.CALCULATION:
             return CarState(soc=calc_soc.calc_soc(
                 vehicle_update_data,

packages/modules/vehicles/evnotify/EVNotify_test.py

@@ -32,16 +32,20 @@ def test_update_updates_value_store(self, monkeypatch):
 
     def test_update_passes_errors_to_context(self, monkeypatch):
         # setup
-        dummy_error = Exception()
+        dummy_error = Exception("Der SoC kann nicht ausgelesen werden")
         self.mock_fetch_soc.side_effect = dummy_error
 
         # execution
         create_vehicle(EVNotify(configuration=EVNotifyConfiguration(
             1, "someKey", "someToken")), 0).update(VehicleUpdateData())
 
         # evaluation
-        self.assert_context_manager_called_with(dummy_error)
+        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])

packages/modules/vehicles/skoda/soc_test.py

@@ -38,20 +38,25 @@ def test_update_updates_value_store(self):
 
     def test_update_passes_errors_to_context(self):
         # setup
-        dummy_error = Exception("API Error")
+        dummy_error = Exception("Der SoC kann nicht ausgelesen werden")
         self.mock_fetch_soc.side_effect = dummy_error
         config = Skoda(configuration=SkodaConfiguration(user_id="test_user", password="test_password", vin="test_vin"))
 
         # execution
         create_vehicle(config, 1).update(VehicleUpdateData())
 
         # evaluation
-        self.assert_context_manager_called_with(dummy_error)
+        # self.assert_context_manager_called_with(dummy_error)
+        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):

packages/modules/vehicles/tesla/soc_test.py

@@ -51,16 +51,20 @@ def test_update_updates_value_store_not_charging(self, monkeypatch):
 
     def test_update_passes_errors_to_context(self, monkeypatch):
         # setup
-        dummy_error = Exception()
+        dummy_error = Exception("Der SoC kann nicht ausgelesen werden")
         self.mock_request_soc_range.side_effect = dummy_error
 
         # execution
         create_vehicle(TeslaSoc(configuration=TeslaSocConfiguration(
             tesla_ev_num=0, token=self.token)), 0).update(VehicleUpdateData())
 
         # evaluation
-        self.assert_context_manager_called_with(dummy_error)
+        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])