soc_calculation_threshold, range calculation

packages/modules/common/abstract_vehicle.py

@@ -14,6 +14,7 @@ class VehicleUpdateData:
     timestamp_soc_from_cp: Optional[int] = None
     last_soc_timestamp: Optional[int] = None
     last_soc: float = None
+    average_consump: int = None
 
 
 @dataclass

packages/modules/common/configurable_vehicle.py

@@ -1,6 +1,6 @@
 from enum import Enum
 import logging
-# import time
+import time
 from typing import Optional, TypeVar, Generic, Callable
 from helpermodules import timecheck
 
@@ -11,7 +11,7 @@
 from modules.common.component_context import SingleComponentUpdateContext
 from modules.common.component_state import CarState
 from modules.common.fault_state import ComponentInfo, FaultState
-from modules.vehicles.common.calc_soc import calc_soc
+from modules.vehicles.common.calc_vehicle_data import calc_vehicle_data
 from modules.vehicles.manual.config import ManualSoc
 from modules.vehicles.mqtt.config import MqttSocSetup
 
@@ -129,18 +129,34 @@ def _get_carstate_by_source(self, vehicle_update_data: VehicleUpdateData, source
         if source == SocSource.API:
             try:
                 _carState = self.__component_updater(vehicle_update_data)
+                _now = int(time.time())
+                _diff = 0
+                if _carState.soc_timestamp:
+                    _diff = int(_now - _carState.soc_timestamp)
+                if _diff > self.general_config.request_interval_charging and\
+                   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):
+                    _age = int(self.general_config.request_interval_charging / 60)
+                    _txt1 = "Ladestand und Reichweite sind berechnet, da der von der Online-Abfrage "
+                    _txt1 = _txt1 + f"gelieferte Zeitstempel mehr als {_age} min alt ist."
+                    self.fault_state.warning(f"{_txt1}")
+                    self.fault_state.store_error()
+                    _carState = calc_vehicle_data.calc_vehicle_data(vehicle_update_data,
+                                                                    self.calculated_soc_state.last_imported or
+                                                                    vehicle_update_data.imported)
             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))
+                    self.fault_state.store_error()
+                    _carState = calc_vehicle_data.calc_vehicle_data(vehicle_update_data,
+                                                                    self.calculated_soc_state.last_imported or
+                                                                    vehicle_update_data.imported)
                 else:
                     if not vehicle_update_data.plug_state:
                         reason = ", weil kein Fahrzeug eingesteckt ist."
@@ -156,11 +172,10 @@ def _get_carstate_by_source(self, vehicle_update_data: VehicleUpdateData, source
                     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,
-                vehicle_update_data.efficiency,
-                self.calculated_soc_state.last_imported or vehicle_update_data.imported,
-                vehicle_update_data.battery_capacity))
+            _carState = calc_vehicle_data.calc_vehicle_data(vehicle_update_data,
+                                                            self.calculated_soc_state.last_imported or
+                                                            vehicle_update_data.imported)
+            return _carState
         elif source == SocSource.CP:
             return CarState(soc=vehicle_update_data.soc_from_cp,
                             soc_timestamp=vehicle_update_data.timestamp_soc_from_cp)

packages/modules/common/configurable_vehicle_test.py

@@ -6,7 +6,7 @@
 from modules.common.configurable_vehicle import ConfigurableVehicle, SocSource
 from modules.common import store
 from modules.devices.tesla.tesla.config import Tesla
-from modules.vehicles.common.calc_soc import calc_soc
+from modules.vehicles.common.calc_vehicle_data import calc_vehicle_data
 from modules.vehicles.manual.config import ManualSoc
 from modules.vehicles.mqtt.config import MqttSocSetup
 
@@ -141,10 +141,15 @@ def test_get_carstate_source(conf_vehicle: ConfigurableVehicle,
 @pytest.mark.parametrize(
     "vehicle_update_data, use_soc_from_cp, expected_calculated_soc_state, expected_call_count",
     [
-        pytest.param(VehicleUpdateData(last_soc=42), False, CalculatedSocState(), 1, id="request only from api"),
-        pytest.param(VehicleUpdateData(imported=150, last_soc=42), True, CalculatedSocState(
+        pytest.param(VehicleUpdateData(last_soc=42, last_soc_timestamp=1, plug_time=0, average_consump=18),
+                     False, CalculatedSocState(), 1, id="request only from api"),
+        pytest.param(VehicleUpdateData(imported=150, last_soc=42, last_soc_timestamp=1, plug_time=0,
+                                       average_consump=18),
+                     True, CalculatedSocState(
             last_imported=150), 1, id="request from api, not plugged"),
-        pytest.param(VehicleUpdateData(imported=200, plug_state=True, last_soc=42), True,
+        pytest.param(VehicleUpdateData(imported=200, plug_state=True, last_soc=42, last_soc_timestamp=1,
+                                       plug_time=0, average_consump=18),
+                     True,
                      CalculatedSocState(last_imported=200), 1, id="request from api, recently plugged"),
     ])
 def test_update_api(vehicle_update_data,
@@ -197,19 +202,21 @@ def test_1(monkeypatch):
 def test_2(monkeypatch):
     # Anstecken -> Eintragen -> aktueller SoC von EV -> SoC von Ev hochrechnen
     # setup
-    mock_calc_soc = Mock(name="calc_soc", return_value=47)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=47, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
     c = conf_vehicle_manual_from_cp()
 
     # execution
-    c.update(VehicleUpdateData(plug_state=True))
+    c.update(VehicleUpdateData(plug_state=True, last_soc_timestamp=0))
     c.calculated_soc_state.manual_soc = 42
-    c.update(VehicleUpdateData(plug_state=True))
-    c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_VALID))
-    # im nächsten Zyklus wird mit calc_soc hochgerechnet
-    c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_INVALID))
+    c.update(VehicleUpdateData(plug_state=True, last_soc_timestamp=0))
+    c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_VALID,
+                               average_consump=18, last_soc_timestamp=0))
+    # im nächsten Zyklus wird mit calc_vehicle_data hochgerechnet
+    c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_INVALID,
+                               average_consump=18, last_soc_timestamp=0))
 
     # evaluation
     assert mock_value_store.set.call_args_list[3][0][0].soc == 47
@@ -218,8 +225,8 @@ def test_2(monkeypatch):
 def test_3(monkeypatch):
     # Anstecken -> aktueller SoC von EV -> Eintragen -> Manuellen SoC hochrechnen
     # setup
-    mock_calc_soc = Mock(name="calc_soc", return_value=44)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=44, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
     c = conf_vehicle_manual_from_cp()
@@ -229,19 +236,19 @@ def test_3(monkeypatch):
     c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_VALID))
     c.calculated_soc_state.manual_soc = 42
     c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_INVALID))
-    # im nächsten Zyklus wird mit calc_soc hochgerechnet
+    # im nächsten Zyklus wird mit calc_vehicle_data hochgerechnet
     c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_INVALID))
 
     # evaluation
     assert mock_value_store.set.call_args_list[3][0][0].soc == 44
-    assert mock_calc_soc.call_count == 2
+    assert mock_calc_vehicle_data.call_count == 2
 
 
 def test_4(monkeypatch):
     # Anstecken -> kein SoC von EV -> Eintragen -> Manuellen SoC hochrechnen
     # setup
-    mock_calc_soc = Mock(name="calc_soc", return_value=44)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=44, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
     c = conf_vehicle_manual_from_cp()
@@ -250,19 +257,19 @@ def test_4(monkeypatch):
     c.update(VehicleUpdateData(plug_state=True))
     c.calculated_soc_state.manual_soc = 42
     c.update(VehicleUpdateData(plug_state=True))
-    # im nächsten Zyklus wird mit calc_soc hochgerechnet
+    # im nächsten Zyklus wird mit calc_vehicle_data hochgerechnet
     c.update(VehicleUpdateData(plug_state=True))
 
     # evaluation
     assert mock_value_store.set.call_args_list[2][0][0].soc == 44
-    assert mock_calc_soc.call_count == 2
+    assert mock_calc_vehicle_data.call_count == 2
 
 
 def test_5(monkeypatch):
     # Anstecken -> kein aktueller SoC von EV -> Eintragen -> Manuellen SoC hochrechnen
     # setup
-    mock_calc_soc = Mock(name="calc_soc", return_value=44)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=44, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
     c = conf_vehicle_manual_from_cp()
@@ -271,19 +278,19 @@ def test_5(monkeypatch):
     c.update(VehicleUpdateData(plug_state=True, soc_from_cp=45, timestamp_soc_from_cp=TIMESTAMP_SOC_VALID))
     c.calculated_soc_state.manual_soc = 42
     c.update(VehicleUpdateData(plug_state=True))
-    # im nächsten Zyklus wird mit calc_soc hochgerechnet
+    # im nächsten Zyklus wird mit calc_vehicle_data hochgerechnet
     c.update(VehicleUpdateData(plug_state=True))
 
     # evaluation
     assert mock_value_store.set.call_args_list[2][0][0].soc == 44
-    assert mock_calc_soc.call_count == 1
+    assert mock_calc_vehicle_data.call_count == 1
 
 
 def test_6(monkeypatch):
     # Abgesteckt -> vom letzten Manuellen SoC hochrechnen
     # setup
-    mock_calc_soc = Mock(name="calc_soc", return_value=44)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=44, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
     c = conf_vehicle_manual_from_cp()
@@ -293,7 +300,7 @@ def test_6(monkeypatch):
 
     # evaluation
     assert mock_value_store.set.call_args_list[0][0][0].soc == 44
-    assert mock_calc_soc.call_count == 1
+    assert mock_calc_vehicle_data.call_count == 1
 
 
 def test_7(monkeypatch):
@@ -335,8 +342,8 @@ def test_9(monkeypatch):
     # setup
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
-    mock_calc_soc = Mock(name="calc_soc", return_value=46)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=46, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     c = conf_vehicle_api_from_cp()
 
     # execution
@@ -345,7 +352,7 @@ def test_9(monkeypatch):
 
     # evaluation
     assert mock_value_store.set.call_args_list[1][0][0].soc == 46
-    assert mock_calc_soc.call_count == 1
+    assert mock_calc_vehicle_data.call_count == 1
 
 
 def test_10(monkeypatch):
@@ -365,17 +372,17 @@ def test_10(monkeypatch):
 def test_11(monkeypatch):
     # Anstecken -> kein SoC von EV -> API -> von API hochrechnen
     # setup
-    mock_calc_soc = Mock(name="calc_soc", return_value=44)
-    monkeypatch.setattr(calc_soc, "calc_soc", mock_calc_soc)
+    mock_calc_vehicle_data = Mock(name="calc_vehicle_data", return_value=CarState(soc=44, range=250))
+    monkeypatch.setattr(calc_vehicle_data, "calc_vehicle_data", mock_calc_vehicle_data)
     mock_value_store = Mock(name="value_store")
     monkeypatch.setattr(store, "get_car_value_store", Mock(return_value=mock_value_store))
     c = conf_vehicle_api_while_charging()
 
     # execution
     c.update(VehicleUpdateData(plug_state=True))
-    # im nächsten Zyklus wird mit calc_soc hochgerechnet
+    # im nächsten Zyklus wird mit calc_vehicle_data hochgerechnet
     c.update(VehicleUpdateData(plug_state=True, charge_state=True))
 
     # evaluation
     assert mock_value_store.set.call_args_list[1][0][0].soc == 44
-    assert mock_calc_soc.call_count == 1
+    assert mock_calc_vehicle_data.call_count == 1

packages/modules/update_soc.py

@@ -119,6 +119,7 @@ def _get_vehicle_update_data(self, ev_num: int) -> VehicleUpdateData:
             timestamp_soc_from_cp = None
         battery_capacity = ev_template.data.battery_capacity
         efficiency = ev_template.data.efficiency
+        average_consump = ev_template.data.average_consump
         soc_timestamp = ev.data.get.soc_timestamp
         return VehicleUpdateData(plug_state=plug_state,
                                  charge_state=charge_state,
@@ -128,7 +129,8 @@ def _get_vehicle_update_data(self, ev_num: int) -> VehicleUpdateData:
                                  soc_from_cp=soc_from_cp,
                                  timestamp_soc_from_cp=timestamp_soc_from_cp,
                                  last_soc_timestamp=soc_timestamp,
-                                 last_soc=ev.data.get.soc if ev.data.get.soc is not None else soc_from_cp)
+                                 last_soc=ev.data.get.soc if ev.data.get.soc is not None else soc_from_cp,
+                                 average_consump=average_consump)
 
     def _filter_failed_store_threads(self, threads_store: List[Thread]) -> List[Thread]:
         ev_data = copy.deepcopy(subdata.SubData.ev_data)

packages/modules/vehicles/common/calc_vehicle_data/calc_vehicle_data.py

@@ -0,0 +1,25 @@
+import logging
+
+from modules.common.abstract_vehicle import VehicleUpdateData
+from modules.common.component_state import CarState
+
+log = logging.getLogger(__name__)
+
+
+def calc_vehicle_data(vehicle_update_data: VehicleUpdateData,
+                      last_imported: float) -> CarState:
+    imported_since_last_soc = vehicle_update_data.imported - last_imported
+    energy_battery_gain = imported_since_last_soc * vehicle_update_data.efficiency / 100
+    battery_soc_gain = (energy_battery_gain / vehicle_update_data.battery_capacity) * 100
+    _soc = vehicle_update_data.last_soc + battery_soc_gain
+    _eff = vehicle_update_data.efficiency
+    log.debug(
+        f"SoC-Gain: (({imported_since_last_soc/1000}kWh charged * {_eff}% _eff) / "
+        f"{vehicle_update_data.battery_capacity/1000}kWh battery-size) * 100 = {battery_soc_gain}%",
+    )
+    log.debug(f"{vehicle_update_data.last_soc}% + {energy_battery_gain / 1000}kWh = {_soc}%")
+    if _soc > 100:
+        log.warning(f"Calculated SoC of {_soc}% exceeds maximum and is limited to 100%! Check your settings!")
+        _soc = 100
+    _range = int(vehicle_update_data.battery_capacity * _soc / vehicle_update_data.average_consump)
+    return CarState(soc=_soc, range=_range)

packages/modules/vehicles/common/calc_vehicle_data/calc_vehicle_data_test.py

@@ -0,0 +1,19 @@
+
+from modules.common.abstract_vehicle import VehicleUpdateData
+from modules.common.component_state import CarState
+from modules.vehicles.common.calc_vehicle_data.calc_vehicle_data import calc_vehicle_data
+
+
+def test_calc_vehicle_data():
+    # setup & execution
+    carState = calc_vehicle_data(VehicleUpdateData(imported=10000, last_soc=12.6,
+                                                   efficiency=90,
+                                                   battery_capacity=100000,
+                                                   average_consump=18000),
+                                 last_imported=0)
+
+    # evaluation
+    # print(f"soc={carState.soc}, range={carState.range}")
+    assert isinstance(carState, CarState)
+    assert carState.soc == 21.6
+    assert carState.range == 120