diff --git a/packages/modules/common/evse.py b/packages/modules/common/evse.py index 187a72cfd0..c889169d1f 100644 --- a/packages/modules/common/evse.py +++ b/packages/modules/common/evse.py @@ -48,7 +48,8 @@ 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) @@ -56,6 +57,15 @@ def get_plug_charge_state(self) -> Tuple[bool, bool, float]: 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 " + diff --git a/packages/modules/internal_chargepoint_handler/chargepoint_module.py b/packages/modules/internal_chargepoint_handler/chargepoint_module.py index 751c7cd544..30d7cc0707 100644 --- a/packages/modules/internal_chargepoint_handler/chargepoint_module.py +++ b/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() diff --git a/packages/modules/internal_chargepoint_handler/chargepoint_module_test.py b/packages/modules/internal_chargepoint_handler/chargepoint_module_test.py new file mode 100644 index 0000000000..2512bd4d5c --- /dev/null +++ b/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