Phasenumschaltung series2: auf Zählerstrom 0 A und EVSE-Status B warten (#4289)

packages/modules/common/evse.py

@@ -48,14 +48,24 @@ def __init__(self, modbus_id: int, client: modbus.ModbusSerialClient_) -> None:
                 self._precise_current = self.is_precise_current_active()
         self.evse_current = 0
 
-    def get_plug_charge_state(self) -> Tuple[bool, bool, float]:
+    def _read_state(self) -> int:
+        """Liest Soll-Strom (Register 1000) und Status (Register 1002), gibt den Rohstatus zurück."""
         time.sleep(0.1)
         raw_set_current, _, state_number = self.client.read_holding_registers(
             1000, [ModbusDataType.UINT_16]*3, unit=self.id)
         # remove leading zeros
         self.evse_current = int(raw_set_current)
         log.debug("Gesetzte Stromstärke EVSE: "+str(self.evse_current) +
                   ", Status: "+str(state_number)+", Modbus-ID: "+str(self.id))
+        return int(state_number)
+
+    def get_state_number(self) -> int:
+        """Rohstatus (EvseStatusCode-Nummer) dieser Lesung, z. B. für die Phasenumschaltung:
+        1 = State A, 2 = State B, 3/4 = State C/D (Fahrzeug lädt), 5 = Fehler."""
+        return self._read_state()
+
+    def get_plug_charge_state(self) -> Tuple[bool, bool, float]:
+        state_number = self._read_state()
         state = EvseStatusCode(state_number)
         if state == EvseStatusCode.FAILURE:
             raise ValueError("Unbekannter Zustand der EVSE: State " +

packages/modules/internal_chargepoint_handler/chargepoint_module.py

@@ -1,4 +1,5 @@
 import logging
+from typing import List, Optional
 
 import time
 from helpermodules.broker import BrokerClient
@@ -9,6 +10,7 @@
 from modules.common.abstract_chargepoint import AbstractChargepoint
 from modules.common.component_context import SingleComponentUpdateContext
 from modules.common.component_state import ChargepointState
+from modules.common.evse import EvseStatusCode
 from modules.common.fault_state import ComponentInfo, FaultState
 from modules.common.store import get_internal_chargepoint_value_store, get_chargepoint_value_store
 from modules.internal_chargepoint_handler.clients import ClientHandler
@@ -141,27 +143,69 @@ def store_state(chargepoint_state: ChargepointState) -> None:
         self.old_chargepoint_state = chargepoint_state
         return chargepoint_state
 
+    # Issue #4289: Vor dem CP-Trennen muss der Zähler 0 A zeigen UND die EVSE das Fahrzeug in State A/B melden
+    # (Statusregister 1002 = 1/2). Der EVSE-Sollwert (Register 1000) taugt nicht als Kriterium, er steht
+    # sofort nach set_current(0) auf 0 (#3326). Ein einzelner Busfehler beim Rücklesen bricht nicht ab;
+    # erst ohne Bestätigung im Zeitfenster wird abgebrochen (Exception -> FaultState, keine GPIO-Aktion).
+    PHASE_SWITCH_CURRENT_THRESHOLD_A = 0.5   # MPM3PM zeigt bei offenem EV-Schütz 0,0 A, SDM/B23 rauschen
+    PHASE_SWITCH_METER_POLLS = 20            # 20 x (0,5 s + Lesezeit) = 10-20 s
+
+    @classmethod
+    def _vehicle_stopped_charging(cls, currents: Optional[List[Optional[float]]], evse_state: int) -> bool:
+        # Status 3/4 = State C/D (lädt), 5 = Fehler: beides keine Bestätigung
+        if evse_state not in (EvseStatusCode.READY, EvseStatusCode.EV_PRESENT):
+            return False
+        if currents is None:  # kein Zähler vorhanden, nur der EVSE-Status zählt
+            return True
+        # jede Phase braucht einen gültigen Messwert unter der Schwelle; None (z. B. NaN vom B23) zählt nicht als 0 A,
+        # Vorzeichen (Zähler verkehrt herum eingebaut) spielt keine Rolle
+        return len(currents) == 3 and all(
+            c is not None and abs(c) < cls.PHASE_SWITCH_CURRENT_THRESHOLD_A for c in currents)
+
     def perform_phase_switch(self, phases_to_use: int) -> None:
         gpio_cp, gpio_relay = self._client.get_pins_phase_switch(phases_to_use)
         evse = self._client.evse_client
+        meter = self._client.meter_client
+        num = self.local_charge_point_num
+        start = time.monotonic()
         with SingleComponentUpdateContext(self.fault_state, update_always=False, reraise=True):
-            evse.set_current(0)
-            for _ in range(20):  # poll up to 10s (20 × 0.5s) for EVSE to confirm 0 A
-                _, _, evse_current = evse.get_plug_charge_state()
-                if evse_current == 0:
-                    break
+            if meter is None:
+                log.warning(f"Phasenumschaltung LP{num}: kein Zähler gefunden, es wird nur der EVSE-Status geprüft.")
+            evse.set_current(0)  # Schreibfehler -> Abbruch: ohne 0-A-Vorgabe wird nicht geschaltet (#3326)
+            for poll in range(1, self.PHASE_SWITCH_METER_POLLS + 1):
+                try:
+                    # Status aus dieser Lesung verwenden, nicht aus einem geteilten Attribut: der reguläre
+                    # Zyklus liest die EVSE parallel und könnte sonst einen älteren Wert hinterlassen
+                    state = evse.get_state_number()
+                    currents = meter.get_counter_state().currents if meter is not None else None
+                    log.debug(f"Phasenumschaltung LP{num}: Durchlauf {poll}, Ströme {currents} A, EVSE-Status {state}")
+                    if self._vehicle_stopped_charging(currents, state):
+                        log.warning(f"Phasenumschaltung LP{num}: 0 A und EVSE-Status {state} nach {poll} Durchläufen"
+                                    f" / {time.monotonic() - start:.1f} s bestätigt, Ströme {currents} A.")
+                        break
+                except Exception as e:
+                    # einzelner Busfehler (EVSE und Zähler am selben RS485-Bus): weiter pollen
+                    log.warning(f"Phasenumschaltung LP{num}: Rücklesen fehlgeschlagen ({e}), weiter.")
                 time.sleep(0.5)
-                evse.set_current(0)
+                try:
+                    evse.set_current(0)
+                except Exception as e:
+                    log.warning(f"Phasenumschaltung LP{num}: 0 A setzen fehlgeschlagen ({e}), weiter.")
             else:
-                raise Exception("Ladung konnte nicht gestoppt werden - Phasenumschaltung abgebrochen.")
-        GPIO.output(gpio_cp, GPIO.HIGH)  # CP off
-        GPIO.output(gpio_relay, GPIO.HIGH)  # 3 on/off
-        time.sleep(5)
-        GPIO.output(gpio_relay, GPIO.LOW)  # 3 on/off
-        time.sleep(5)
-        GPIO.output(gpio_cp, GPIO.LOW)  # CP on
+                raise Exception("Fahrzeug hat die Ladung nicht beendet (Zählerstrom > 0 oder EVSE-Status C)"
+                                " - Phasenumschaltung abgebrochen.")
+        try:
+            GPIO.output(gpio_cp, GPIO.HIGH)  # CP off
+            GPIO.output(gpio_relay, GPIO.HIGH)  # 3 on/off
+            time.sleep(5)
+            GPIO.output(gpio_relay, GPIO.LOW)  # 3 on/off
+            time.sleep(5)
+        finally:
+            GPIO.output(gpio_cp, GPIO.LOW)  # CP on - in jedem Fall wieder verbinden (#4037)
         time.sleep(1)
         self.old_phases_in_use = phases_to_use
+        log.warning(f"Phasenumschaltung LP{num}: auf {phases_to_use} Phasen umgeschaltet "
+                    f"({time.monotonic() - start:.1f} s gesamt).")
 
     def perform_cp_interruption(self, duration: int) -> None:
         gpio_cp = self._client.get_pins_cp_interruption()

packages/modules/internal_chargepoint_handler/chargepoint_module_test.py

@@ -0,0 +1,94 @@
+from typing import List, Optional
+from unittest.mock import MagicMock
+
+import pytest
+
+from modules.internal_chargepoint_handler import chargepoint_module
+from modules.internal_chargepoint_handler.chargepoint_module import ChargepointModule
+
+
+@pytest.mark.parametrize(
+    "currents, evse_state, expected",
+    [
+        pytest.param([0.0, 0.0, 0.0], 2, True, id="0 A, State B"),
+        pytest.param([0.0, 0.0, 0.0], 1, True, id="0 A, State A (abgesteckt)"),
+        pytest.param([0.0, 0.857, 0.013], 2, False, id="Restlast auf L2 bei State B"),
+        pytest.param([15.6, 9.7, 15.9], 3, False, id="Last, State C"),
+        pytest.param([0.0, 0.0, 0.0], 3, False, id="0 A, aber noch State C"),
+        pytest.param([0.0, 0.0, 0.0], 4, False, id="0 A, State D"),
+        pytest.param([0.0, 0.0, 0.0], 5, False, id="EVSE-Fehler"),
+        pytest.param([None, None, None], 2, False, id="keine Messwerte (B23 NaN)"),
+        pytest.param([0.0, None, 0.0], 2, False, id="eine Phase ohne Messwert"),
+        pytest.param([-0.2, 0.0, 0.0], 2, True, id="Rauschen negativ"),
+        pytest.param([-3.0, 0.0, 0.0], 2, False, id="Strom negativ (Zähler verdreht)"),
+        pytest.param(None, 2, True, id="kein Zähler, State B"),
+        pytest.param(None, 3, False, id="kein Zähler, State C"),
+    ])
+def test_vehicle_stopped_charging(currents: Optional[List[Optional[float]]], evse_state: int, expected: bool):
+    assert ChargepointModule._vehicle_stopped_charging(currents, evse_state) == expected
+
+
+@pytest.fixture
+def module(monkeypatch) -> ChargepointModule:
+    monkeypatch.setattr("time.sleep", lambda seconds: None)
+    monkeypatch.setattr(chargepoint_module, "GPIO", MagicMock(), raising=False)
+    cp = ChargepointModule.__new__(ChargepointModule)
+    cp.local_charge_point_num = 0
+    cp.old_phases_in_use = 3
+    cp.fault_state = MagicMock()
+    cp.fault_state.component_info.name = "Interner Ladepunkt 0"
+    cp._client = MagicMock()
+    cp._client.get_pins_phase_switch.return_value = (7, 11)
+    return cp
+
+
+def counter_state(currents):
+    state = MagicMock()
+    state.currents = currents
+    return state
+
+
+def test_perform_phase_switch_waits_for_meter_and_evse(module: ChargepointModule):
+    module._client.evse_client.get_state_number.side_effect = [3, 2, 2]
+    module._client.meter_client.get_counter_state.side_effect = [
+        counter_state([15.6, 9.7, 15.9]), counter_state([0.0, 0.857, 0.013]), counter_state([0.0, 0.0, 0.0])]
+
+    module.perform_phase_switch(1)
+
+    assert module._client.evse_client.get_state_number.call_count == 3
+    assert module._client.evse_client.set_current.call_count == 3  # einmal vorab, je Wartedurchlauf erneut
+    assert chargepoint_module.GPIO.output.call_count == 4
+    assert module.old_phases_in_use == 1
+
+
+def test_perform_phase_switch_continues_after_read_error(module: ChargepointModule):
+    module._client.evse_client.get_state_number.side_effect = [Exception("No response received"), 2]
+    module._client.meter_client.get_counter_state.return_value = counter_state([0.0, 0.0, 0.0])
+
+    module.perform_phase_switch(1)
+
+    assert chargepoint_module.GPIO.output.call_count == 4
+    assert module.old_phases_in_use == 1
+
+
+def test_perform_phase_switch_aborts_without_confirmation(module: ChargepointModule, monkeypatch):
+    monkeypatch.setattr(ChargepointModule, "PHASE_SWITCH_METER_POLLS", 3)
+    module._client.evse_client.get_state_number.return_value = 3
+    module._client.meter_client.get_counter_state.return_value = counter_state([15.6, 9.7, 15.9])
+
+    with pytest.raises(Exception, match="Phasenumschaltung abgebrochen"):
+        module.perform_phase_switch(1)
+
+    assert module._client.evse_client.get_state_number.call_count == 3
+    chargepoint_module.GPIO.output.assert_not_called()
+    assert module.old_phases_in_use == 3
+
+
+def test_perform_phase_switch_without_meter_uses_evse_state(module: ChargepointModule):
+    module._client.meter_client = None
+    module._client.evse_client.get_state_number.side_effect = [3, 2]
+
+    module.perform_phase_switch(1)
+
+    assert chargepoint_module.GPIO.output.call_count == 4
+    assert module.old_phases_in_use == 1