Merge pull request #1431 from LKuemmel/fix

Bugfixes

packages/control/algorithm/filter_chargepoints.py

@@ -92,7 +92,7 @@ def _get_preferenced_chargepoint(valid_chargepoints: List[Chargepoint]) -> List:
                         for cp in chargepoints.keys())
                 elif condition_types[condition] == "soc":
                     chargepoints.update(
-                        (cp, cp.data.set.charging_ev_data.data.get.soc) for cp in chargepoints.keys())
+                        (cp, cp.data.set.charging_ev_data.data.get.soc or 0) for cp in chargepoints.keys())
                 elif condition_types[condition] == "plug_in":
                     chargepoints.update((cp, cp.data.set.plug_time)
                                         for cp in chargepoints.keys())

packages/control/ev.py

@@ -166,10 +166,10 @@ def set_factory() -> Set:
 
 @dataclass
 class Get:
-    soc: int = 0
-    soc_timestamp: float = 0
+    soc: Optional[int] = None
+    soc_timestamp: Optional[float] = None
     force_soc_update: bool = False
-    range: float = 0
+    range: Optional[float] = None
     fault_state: int = 0
     fault_str: str = ""
 
@@ -568,7 +568,7 @@ def __init__(self, index):
     TIME_CHARGING_AMOUNT_REACHED = "Kein Zeitladen, da die Energiemenge bereits geladen wurde."
 
     def time_charging(self,
-                      soc: float,
+                      soc: Optional[float],
                       used_amount_time_charging: float) -> Tuple[int, str, Optional[str], Optional[str]]:
         """ prüft, ob ein Zeitfenster aktiv ist und setzt entsprechend den Ladestrom
         """
@@ -583,10 +583,13 @@ def time_charging(self,
                     if plan.limit.selected == "none":  # kein Limit konfiguriert, mit konfigurierter Stromstärke laden
                         return plan.current, "time_charging", message, plan.name
                     elif plan.limit.selected == "soc":  # SoC Limit konfiguriert
-                        if soc < plan.limit.soc:
-                            return plan.current, "time_charging", message, plan.name  # Limit nicht erreicht
+                        if soc:
+                            if soc < plan.limit.soc:
+                                return plan.current, "time_charging", message, plan.name  # Limit nicht erreicht
+                            else:
+                                return 0, "stop", self.TIME_CHARGING_SOC_REACHED, plan.name  # Limit erreicht
                         else:
-                            return 0, "stop", self.TIME_CHARGING_SOC_REACHED, plan.name  # Limit erreicht
+                            return plan.current, "time_charging", message, plan.name
                     elif plan.limit.selected == "amount":  # Energiemengenlimit konfiguriert
                         if used_amount_time_charging < plan.limit.amount:
                             return plan.current, "time_charging", message, plan.name  # Limit nicht erreicht
@@ -608,7 +611,7 @@ def time_charging(self,
     INSTANT_CHARGING_AMOUNT_REACHED = "Kein Sofortladen, da die Energiemenge bereits geladen wurde."
 
     def instant_charging(self,
-                         soc: float,
+                         soc: Optional[float],
                          imported_instant_charging: float) -> Tuple[int, str, Optional[str]]:
         """ prüft, ob die Lademengenbegrenzung erreicht wurde und setzt entsprechend den Ladestrom.
         """
@@ -621,10 +624,13 @@ def instant_charging(self,
             if instant_charging.limit.selected == "none":
                 return instant_charging.current, "instant_charging", message
             elif instant_charging.limit.selected == "soc":
-                if soc < instant_charging.limit.soc:
-                    return instant_charging.current, "instant_charging", message
+                if soc:
+                    if soc < instant_charging.limit.soc:
+                        return instant_charging.current, "instant_charging", message
+                    else:
+                        return 0, "stop", self.INSTANT_CHARGING_SOC_REACHED
                 else:
-                    return 0, "stop", self.INSTANT_CHARGING_SOC_REACHED
+                    return instant_charging.current, "instant_charging", message
             elif instant_charging.limit.selected == "amount":
                 if imported_instant_charging < self.data.chargemode.instant_charging.limit.amount:
                     return instant_charging.current, "instant_charging", message
@@ -638,14 +644,14 @@ def instant_charging(self,
 
     PV_CHARGING_SOC_REACHED = "Keine Ladung, da der maximale Soc bereits erreicht wurde."
 
-    def pv_charging(self, soc: float, min_current: int) -> Tuple[int, str, Optional[str]]:
+    def pv_charging(self, soc: Optional[float], min_current: int) -> Tuple[int, str, Optional[str]]:
         """ prüft, ob Min-oder Max-Soc erreicht wurden und setzt entsprechend den Ladestrom.
         """
         message = None
         try:
             pv_charging = self.data.chargemode.pv_charging
-            if soc < pv_charging.max_soc:
-                if pv_charging.min_soc != 0:
+            if soc is None or soc < pv_charging.max_soc:
+                if pv_charging.min_soc != 0 and soc is not None:
                     if soc < pv_charging.min_soc:
                         return pv_charging.min_soc_current, "instant_charging", message
                 if pv_charging.min_current == 0:
@@ -739,12 +745,15 @@ def search_plan(self,
 
     def calculate_duration(self,
                            plan: ScheduledChargingPlan,
-                           soc: float,
+                           soc: Optional[float],
                            battery_capacity: float,
                            used_amount: float,
                            phases: int) -> Tuple[float, float]:
         if plan.limit.selected == "soc":
-            missing_amount = ((plan.limit.soc_scheduled - soc) / 100) * battery_capacity
+            if soc:
+                missing_amount = ((plan.limit.soc_scheduled - soc) / 100) * battery_capacity
+            else:
+                raise ValueError("Um Zielladen mit SoC-Ziel nutzen zu können, bitte ein SoC-Modul konfigurieren.")
         else:
             missing_amount = plan.limit.amount - used_amount
         duration = missing_amount/(plan.current * phases*230) * 3600

packages/control/ev_charge_template_test.py

@@ -9,7 +9,7 @@
 from control.ev import ChargeTemplate, EvTemplate, EvTemplateData, SelectedPlan
 from control.general import General
 from helpermodules import timecheck
-from helpermodules.abstract_plans import ScheduledChargingPlan, TimeChargingPlan
+from helpermodules.abstract_plans import Limit, ScheduledChargingPlan, TimeChargingPlan
 
 
 @pytest.fixture(autouse=True)
@@ -21,13 +21,35 @@ def data_module() -> None:
 
 @pytest.mark.parametrize(
     "plans, soc, used_amount_time_charging, plan_found, expected",
-    [pytest.param({}, 0, 0, None, (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_CONFIGURED, None),
-                  id="no plan defined"),
-     pytest.param({"0": TimeChargingPlan()}, 0, 0,  None,
-                  (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None), id="no plan active"),
-     pytest.param({"0": TimeChargingPlan()}, 0, 0,  TimeChargingPlan(),
-                  (16, "time_charging", None, "Zeitladen-Standard"), id="plan active")
-     ])
+    [
+        pytest.param({}, 0, 0, None, (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_CONFIGURED, None),
+                     id="no plan defined"),
+        pytest.param({"0": TimeChargingPlan()}, 0, 0,  None,
+                     (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None), id="no plan active"),
+        pytest.param({"0": TimeChargingPlan()}, 0, 0,  TimeChargingPlan(),
+                     (16, "time_charging", None, "Zeitladen-Standard"), id="plan active"),
+        pytest.param({"0": TimeChargingPlan(limit=Limit(selected="soc"))}, 100, 0,
+                     TimeChargingPlan(limit=Limit(selected="soc")),
+                     (0, "stop", ChargeTemplate.TIME_CHARGING_SOC_REACHED, "Zeitladen-Standard"),
+                     id="plan active, soc is reached"),
+        pytest.param({"0": TimeChargingPlan(limit=Limit(selected="soc"))}, 40, 0,
+                     TimeChargingPlan(limit=Limit(selected="soc")),
+                     (16, "time_charging", None, "Zeitladen-Standard"), id="plan active, soc is not reached"),
+        pytest.param({"0": TimeChargingPlan(limit=Limit(selected="soc"))}, None, 0,
+                     TimeChargingPlan(limit=Limit(selected="soc")),
+                     (16, "time_charging", None, "Zeitladen-Standard"), id="plan active, soc is not defined"),
+        pytest.param({"0": TimeChargingPlan(limit=Limit(selected="amount"))}, 0, 1500,
+                     TimeChargingPlan(limit=Limit(selected="amount")),
+                     (0, "stop", ChargeTemplate.TIME_CHARGING_AMOUNT_REACHED, "Zeitladen-Standard"),
+                     id="plan active, used_amount_time_charging is reached"),
+        pytest.param({"0": TimeChargingPlan(limit=Limit(selected="amount"))}, 0, 500,
+                     TimeChargingPlan(limit=Limit(selected="amount")),
+                     (16, "time_charging", None, "Zeitladen-Standard"),
+                     id="plan active, used_amount_time_charging is not reached"),
+        pytest.param({"0": TimeChargingPlan()}, 0, 0,  None,
+                     (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None), id="plan defined but not found"),
+    ]
+)
 def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amount_time_charging: float,
                        plan_found: TimeChargingPlan,
                        expected: Tuple[int, str, Optional[str], Optional[str]],
@@ -49,6 +71,7 @@ def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amou
     "selected, current_soc, used_amount, expected",
     [
         pytest.param("none", 0, 0, (10, "instant_charging", None), id="without limit"),
+        pytest.param("soc", None, 0, (10, "instant_charging", None), id="limit soc: soc not defined"),
         pytest.param("soc", 49, 0, (10, "instant_charging", None), id="limit soc: soc not reached"),
         pytest.param("soc", 50, 0, (0, "stop", ChargeTemplate.INSTANT_CHARGING_SOC_REACHED),
                      id="limit soc: soc reached"),
@@ -77,6 +100,7 @@ def test_instant_charging(selected: str, current_soc: float, used_amount: float,
                      ChargeTemplate.PV_CHARGING_SOC_REACHED), id="max soc reached"),
         pytest.param(15, 0, 14, SwitchOnBatState.CHARGE_FROM_BAT,
                      (10, "instant_charging", None), id="min soc not reached"),
+        pytest.param(15, 0, None, SwitchOnBatState.CHARGE_FROM_BAT, (6, "pv_charging", None), id="soc not defined"),
         pytest.param(15, 8, 15, SwitchOnBatState.CHARGE_FROM_BAT,
                      (8, "instant_charging", None), id="min current configured"),
         pytest.param(15, 8, 15, SwitchOnBatState.SWITCH_OFF_SOC_REACHED, (0, "stop",

packages/helpermodules/scale_metric.py

@@ -19,6 +19,8 @@
 
 
 def scale_metric(value: Union[float, int], metric: Optional[str], base_metric: str):
+    if value is None:
+        return value
 
     # If no unit or too short unit or the base unit is not matching, return the unmodified value
     # This is crucial for fallback cases where no unit is given at all

packages/modules/update_soc.py

@@ -72,6 +72,9 @@ def _get_threads(self) -> Tuple[List[Thread], List[Thread]]:
                     if ev.data.get.fault_state != 0 or ev.data.get.fault_str != NO_ERROR:
                         Pub().pub(f"openWB/set/vehicle/{ev.num}/get/fault_state", 0)
                         Pub().pub(f"openWB/set/vehicle/{ev.num}/get/fault_str", NO_ERROR)
+                        Pub().pub(f"openWB/set/vehicle/{ev.num}/get/soc", None)
+                        Pub().pub(f"openWB/set/vehicle/{ev.num}/get/soc_timestamp", None)
+                        Pub().pub(f"openWB/set/vehicle/{ev.num}/get/range", None)
             except Exception:
                 log.exception("Fehler im update_soc-Modul")
         return threads_update, threads_store