Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 12 additions & 17 deletions packages/control/load.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,31 +9,26 @@

log = logging.getLogger(__name__)

# Zustände, in denen reserved_surplus/released_surplus gerade eine laufende Reservierung halten (siehe die
# jeweiligen += beim Eintritt und -= beim Verlassen dieser Zustände in counter.py/ev.py/chargepoint.py).
# NO_CHARGING_ALLOWED gehört bewusst nicht dazu: jede Reservierung wird spätestens beim Übergang in diesen
# Zustand bereits wieder freigegeben (siehe zB Counter.reset_switch_on_off()), er wird hier also nie mit einer
# noch offenen Reservierung erreicht.
RESERVATION_HOLDING_STATES = (ChargepointState.PERFORMING_PHASE_SWITCH,
ChargepointState.PHASE_SWITCH_DELAY,
ChargepointState.SWITCH_OFF_DELAY,
ChargepointState.SWITCH_ON_DELAY)


def reset_pv_data_if_no_active_delays() -> None:
""" Wenn keine EV angesteckt sind oder keine Verzögerungen aktiv sind, werden die Algorithmus-Werte
""" Wenn keine Komponente gerade eine Reservierung offen hält, werden die Algorithmus-Werte
zurückgesetzt.
(dient der Robustheit)
"""
try:
for load in list(data.data.cp_data.values()) + list(data.data.consumer_data.values()):
try:
control_parameter = load.data.control_parameter
if isinstance(load, Chargepoint):
if (not load.data.get.plug_state or
(control_parameter.state != ChargepointState.PERFORMING_PHASE_SWITCH and
control_parameter.state != ChargepointState.PHASE_SWITCH_DELAY and
control_parameter.state != ChargepointState.SWITCH_OFF_DELAY and
control_parameter.state != ChargepointState.SWITCH_ON_DELAY and
control_parameter.state != ChargepointState.NO_CHARGING_ALLOWED)):
continue
break
else:
# Verbraucher haben keinen Plug-Status.
# Nur laufende Ein-/Ausschaltverzögerungen müssen den Reset verhindern.
if (control_parameter.state != ChargepointState.SWITCH_OFF_DELAY and
control_parameter.state != ChargepointState.SWITCH_ON_DELAY):
continue
if load.data.control_parameter.state in RESERVATION_HOLDING_STATES:
break
except Exception:
log.exception(f"Fehler beim Bereinigen der Werte für Ladepunkt/Verbraucher {load.num}")
Expand Down
26 changes: 14 additions & 12 deletions packages/control/load_test.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from typing import Dict
from unittest.mock import Mock

import pytest

from control import data, load
from control.chargepoint.chargepoint import Chargepoint
from control.chargepoint.chargepoint_state import ChargepointState
Expand Down Expand Up @@ -38,27 +40,27 @@ def test_do_not_reset_pv_data_when_consumer_switch_off_delay_active():
evu_counter.reset_pv_data.assert_not_called()


def test_do_not_reset_pv_data_with_plugged_chargepoint_no_charging_allowed():
def test_reset_pv_data_with_chargepoint_no_charging_allowed():
# eine wartende, aber nicht (mehr) reservierte Komponente darf den robustness-Reset nicht dauerhaft
# blockieren - jede Reservierung wird bereits beim Übergang in NO_CHARGING_ALLOWED freigegeben.
cp = Mock(spec=Chargepoint, num="1")
cp.data = Mock(
control_parameter=Mock(state=ChargepointState.NO_CHARGING_ALLOWED),
get=Mock(plug_state=True),
)
cp.data = Mock(control_parameter=Mock(state=ChargepointState.NO_CHARGING_ALLOWED))
evu_counter = _setup_control_data({"cp1": cp}, {})

load.reset_pv_data_if_no_active_delays()

evu_counter.reset_pv_data.assert_not_called()
evu_counter.reset_pv_data.assert_called_once()


def test_reset_pv_data_with_unplugged_chargepoint_no_charging_allowed():
@pytest.mark.parametrize("state", [ChargepointState.PERFORMING_PHASE_SWITCH,
ChargepointState.PHASE_SWITCH_DELAY,
ChargepointState.SWITCH_OFF_DELAY,
ChargepointState.SWITCH_ON_DELAY])
def test_do_not_reset_pv_data_with_chargepoint_holding_a_reservation(state):
cp = Mock(spec=Chargepoint, num="1")
cp.data = Mock(
control_parameter=Mock(state=ChargepointState.NO_CHARGING_ALLOWED),
get=Mock(plug_state=False),
)
cp.data = Mock(control_parameter=Mock(state=state))
evu_counter = _setup_control_data({"cp1": cp}, {})

load.reset_pv_data_if_no_active_delays()

evu_counter.reset_pv_data.assert_called_once()
evu_counter.reset_pv_data.assert_not_called()