Skip to content

pytest #3786

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LKuemmel merged 4 commits into from
Aug 10, 2026
Merged

pytest #3786

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4828548
fix pytest
LKuemmel Aug 10, 2026
ec7290a
fixes, pytest
LKuemmel Aug 10, 2026
d3069ec
review
LKuemmel Aug 10, 2026
86c97ae
fix
LKuemmel Aug 10, 2026
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
3 changes: 2 additions & 1 deletion packages/control/algorithm/additional_current.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@ def set_additional_current(self) -> None:
f"counter{counter.num}"))
for preferenced_loads in preferenced_loads_groups:
common.update_raw_data(preferenced_loads)
log.info(f"Zähler {counter.num}, Verbraucher {filtered_loads_to_str(preferenced_loads)}")
if len(preferenced_loads) > 0:
log.info(f"Zähler {counter.num}, Verbraucher {filtered_loads_to_str(preferenced_loads)}")
while len(preferenced_loads):
load = preferenced_loads[0]
missing_currents, counts = common.get_missing_currents_left(preferenced_loads)
Expand Down
2 changes: 2 additions & 0 deletions packages/control/algorithm/integration_test/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,11 +80,13 @@ class ParamsExpectedSetCurrent:
expected_current_cp3: float = 0
expected_current_cp4: float = 0
expected_current_cp5: float = 0
expected_current_consumer7: float = 0


def assert_expected_current(params: ParamsExpectedSetCurrent):
for i in range(3, 6):
assert data.data.cp_data[f"cp{i}"].data.set.current == getattr(params, f"expected_current_cp{i}")
assert data.data.consumer_data["consumer7"].data.set.current == params.expected_current_consumer7


@pytest.fixture()
Expand Down
15 changes: 9 additions & 6 deletions packages/control/algorithm/integration_test/instant_charging_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,9 +108,10 @@ class ParamsLimit(ParamsExpectedSetCurrent, ParamsExpectedCounterSet):
raw_currents_left_counter6=[16, 12, 14],
expected_state_str=(f"Es kann nicht mit der vorgegebenen Stromstärke geladen werden"
f"{LimitingValue.CURRENT.value.format('Garage')}"),
expected_current_cp3=14,
expected_current_cp3=13.5,
expected_current_cp4=12,
expected_current_cp5=14,
expected_current_consumer7=0.5,
expected_raw_power_left=12110,
expected_raw_currents_left_counter0=[0, 18, 17],
expected_raw_currents_left_counter6=[16, 0, 0]),
Expand All @@ -120,12 +121,13 @@ class ParamsLimit(ParamsExpectedSetCurrent, ParamsExpectedCounterSet):
raw_currents_left_counter6=[16]*3,
expected_state_str=(f"Es kann nicht mit der vorgegebenen Stromstärke geladen werden"
f"{LimitingValue.POWER.value.format('Garage')}"),
expected_current_cp3=10.461538461538462,
expected_current_cp4=6.769230769230769,
expected_current_cp5=6.769230769230769,
expected_current_cp3=10.346153846153847,
expected_current_cp4=6.576923076923077,
expected_current_cp5=6.576923076923077,
expected_current_consumer7=0.5,
expected_raw_power_left=0,
expected_raw_currents_left_counter0=[21.53846153846154, 25.23076923076923, 25.23076923076923],
expected_raw_currents_left_counter6=[16, 9.23076923076923, 9.23076923076923]),
expected_raw_currents_left_counter0=[21.153846153846153, 25.423076923076923, 25.423076923076923],
expected_raw_currents_left_counter6=[16, 9.423076923076923, 9.423076923076923]),

# limit by unbalanced load
]
Expand All @@ -144,6 +146,7 @@ def test_instant_charging_limit(params: ParamsLimit, all_cp_instant_charging_1p,

# evaluation
assert_expected_current(params)
assert data.data.consumer_data["consumer7"].data.set.current == 0.5
for i in range(3, 6):
assert data.data.cp_data[
f"cp{i}"].data.get.state_str.replace("\n", "") == params.expected_state_str.replace("\n", "")
Expand Down
34 changes: 24 additions & 10 deletions packages/control/algorithm/integration_test/pv_charging_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ def all_cp_pv_charging_3p():
control_parameter.phases = 1
control_parameter.chargemode = Chargemode.PV_CHARGING
control_parameter.submode = Chargemode.PV_CHARGING
control_parameter.state = ChargepointState.CHARGING_ALLOWED


@pytest.fixture()
Expand All @@ -42,6 +43,10 @@ def all_cp_charging_3p():
data.data.cp_data["cp3"].data.get.power = 6900
data.data.cp_data["cp4"].data.get.power = 5520
data.data.cp_data["cp5"].data.get.power = 5520
data.data.consumer_data["consumer7"].data.get.charge_state = True
data.data.consumer_data["consumer7"].data.get.currents = [0.5, 0, 0]
data.data.consumer_data["consumer7"].data.get.power = 115
data.data.consumer_data["consumer7"].data.set.current = 0.5

for i in range(3, 6):
charging_ev_data = data.data.cp_data[f"cp{i}"].data.set.charging_ev_data
Expand Down Expand Up @@ -153,7 +158,7 @@ def test_pv_delay_expired(all_cp_pv_charging_3p, all_cp_not_charging, monkeypatc
data.data.counter_data["counter0"].data.set.raw_power_left = 31200
data.data.counter_data["counter0"].data.set.raw_currents_left = [32, 30, 31]
data.data.counter_data["counter6"].data.set.raw_currents_left = [16, 12, 14]
data.data.counter_data["counter0"].data.set.reserved_surplus = 9000
data.data.counter_data["counter0"].data.set.reserved_surplus = 9115
data.data.cp_data[
"cp3"].data.control_parameter.timestamp_switch_on_off = 1652683185.0
data.data.cp_data[
Expand All @@ -165,6 +170,10 @@ def test_pv_delay_expired(all_cp_pv_charging_3p, all_cp_not_charging, monkeypatc
"cp4"].data.control_parameter.state = ChargepointState.SWITCH_ON_DELAY
data.data.cp_data[
"cp5"].data.control_parameter.timestamp_switch_on_off = None
data.data.consumer_data[
"consumer7"].data.control_parameter.timestamp_switch_on_off = 1652683185.0
data.data.consumer_data[
"consumer7"].data.control_parameter.state = ChargepointState.SWITCH_ON_DELAY

# execution
Algorithm().calc_current()
Expand All @@ -179,7 +188,7 @@ def test_pv_delay_expired(all_cp_pv_charging_3p, all_cp_not_charging, monkeypatc
"cp4"].data.control_parameter.timestamp_switch_on_off is None
assert data.data.cp_data[
"cp5"].data.control_parameter.timestamp_switch_on_off is None
assert data.data.counter_data["counter0"].data.set.raw_power_left == 24300
assert data.data.counter_data["counter0"].data.set.raw_power_left == 24185
assert data.data.counter_data["counter0"].data.set.surplus_power_left == -690
assert data.data.counter_data["counter0"].data.set.reserved_surplus == 0

Expand All @@ -190,10 +199,11 @@ def test_pv_delay_expired(all_cp_pv_charging_3p, all_cp_not_charging, monkeypatc
raw_power_left=50000,
raw_currents_left_counter0=[40]*3,
raw_currents_left_counter6=[16]*3,
expected_current_cp3=16,
expected_current_cp3=14.375,
expected_current_cp4=8,
expected_current_cp5=8,
expected_raw_power_left=34820,
expected_current_consumer7=5.5,
expected_raw_power_left=34705,
expected_surplus_power_left=1090,
expected_reserved_surplus=0,
expected_released_surplus=0),
Expand All @@ -202,10 +212,11 @@ def test_pv_delay_expired(all_cp_pv_charging_3p, all_cp_not_charging, monkeypatc
raw_power_left=39650,
raw_currents_left_counter0=[40]*3,
raw_currents_left_counter6=[16]*3,
expected_current_cp3=11.248792270531402,
expected_current_cp4=7.248792270531401,
expected_current_cp5=7.2487922705314,
expected_raw_power_left=24470,
expected_current_cp3=10.878199918732223,
expected_current_cp4=6.8827997554740445,
expected_current_cp5=7.201405373552686,
expected_current_consumer7=2.3519152915057484,
expected_raw_power_left=24355,
expected_surplus_power_left=1090,
expected_reserved_surplus=0,
expected_released_surplus=0),
Expand All @@ -217,7 +228,8 @@ def test_pv_delay_expired(all_cp_pv_charging_3p, all_cp_not_charging, monkeypatc
expected_current_cp3=10,
expected_current_cp4=6,
expected_current_cp5=6,
expected_raw_power_left=5635,
expected_current_consumer7=0.5,
expected_raw_power_left=5520.0,
expected_surplus_power_left=-8200,
expected_reserved_surplus=0,
expected_released_surplus=11040),
Expand Down Expand Up @@ -256,7 +268,8 @@ def test_surplus(params: ParamsSurplus, all_cp_pv_charging_3p, all_cp_charging_3
expected_current_cp3=10,
expected_current_cp4=6,
expected_current_cp5=6,
expected_raw_power_left=17400,
expected_current_consumer7=0.5,
expected_raw_power_left=17285,
expected_surplus_power_left=-690,
expected_reserved_surplus=0,
expected_released_surplus=0),
Expand All @@ -270,6 +283,7 @@ def test_surplus(params: ParamsSurplus, all_cp_pv_charging_3p, all_cp_charging_3
expected_current_cp3=32,
expected_current_cp4=6,
expected_current_cp5=6,
expected_current_consumer7=0.5,
expected_raw_power_left=37520.0,
expected_surplus_power_left=3000,
expected_reserved_surplus=0,
Expand Down
3 changes: 2 additions & 1 deletion packages/control/algorithm/min_current.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@ def set_min_current(self) -> None:
preferenced_loads_groups = get_grouped_loads_by_mode_and_counter(CONSIDERED_CHARGE_MODES_MIN_CURRENT,
f"counter{counter.num}")
for preferenced_loads in preferenced_loads_groups:
log.info(f"Zähler {counter.num}, Verbraucher {filtered_loads_to_str(preferenced_loads)}")
if len(preferenced_loads) > 0:
log.info(f"Zähler {counter.num}, Verbraucher {filtered_loads_to_str(preferenced_loads)}")
common.update_raw_data(preferenced_loads, diff_to_zero=True)
while len(preferenced_loads):
load = preferenced_loads[0]
Expand Down
3 changes: 2 additions & 1 deletion packages/control/algorithm/surplus_controlled.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,8 @@ def _set(self,
feed_in_yield: int,
counter: Counter) -> None:
for loads in preferenced_loads_groups:
log.info(f"Zähler {counter.num}, Verbraucher {filtered_loads_to_str(loads)}")
if len(loads) > 0:
log.info(f"Zähler {counter.num}, Verbraucher {filtered_loads_to_str(loads)}")
common.update_raw_data(loads, surplus=True)
while len(loads):
load = loads[0]
Expand Down
2 changes: 2 additions & 0 deletions packages/control/consumer/consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,8 @@ def update(self):
self.is_switch_interval_elapsed()
min_current, required_current, message, mode, submode = self.get_parameter()
self.set_mode_changed(submode, mode)
if self.chargemode_changed or self.submode_changed:
data.data.counter_all_data.get_evu_counter().reset_switch_on_off(self)
self.set_control_parameter(min_current, required_current,
self.data.config.connected_phases, submode, mode)
self.set_state_and_log(message)
Expand Down
55 changes: 32 additions & 23 deletions packages/control/counter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,28 +320,31 @@ def _get_switch_on_texts(self, load: Load) -> SwitchOnTexts:
return self.SWITCH_ON_TEXTS_CP
return self.SWITCH_ON_TEXTS_CONSUMER

def calc_switch_on_power(self, load: Load) -> Tuple[float, float]:
def calc_switch_on_power(self, switch_on_threshold: float) -> Tuple[float, float]:
surplus = self.calc_raw_surplus() - self.data.set.reserved_surplus
control_parameter = load.data.control_parameter
surplus_config = data.data.general_data.data.chargemode_config.surplus

if surplus_config.feed_in_limit:
threshold = surplus_config.feed_in_yield
else:
threshold = surplus_config.vehicle.switch_on_threshold*control_parameter.phases
threshold = switch_on_threshold

return surplus, threshold

def get_pv_config_py_load(self, load: Load) -> Tuple[float, float, bool]:
def get_pv_config_py_load(self, load: Load) -> Tuple[float, float, float]:
surplus_config = data.data.general_data.data.chargemode_config.surplus
if isinstance(load, Chargepoint):
switch_on_delay = data.data.general_data.data.chargemode_config.surplus.vehicle.switch_on_delay
switch_on_threshold = data.data.general_data.data.chargemode_config.surplus.vehicle.switch_on_threshold
control_parameter = load.data.control_parameter
switch_on_delay = surplus_config.vehicle.switch_on_delay
switch_on_threshold = surplus_config.vehicle.switch_on_threshold * control_parameter.phases
power_to_reserve = surplus_config.vehicle.switch_on_threshold*control_parameter.phases
else:
switch_on_delay = data.data.general_data.data.chargemode_config.surplus.consumer.switch_on_delay
switch_on_delay = surplus_config.consumer.switch_on_delay
switch_on_threshold = (load.data.control_parameter.required_current *
230 *
load.data.control_parameter.phases)
feed_in_limit = data.data.general_data.data.chargemode_config.surplus.feed_in_limit
return switch_on_delay, switch_on_threshold, feed_in_limit
power_to_reserve = switch_on_threshold
return switch_on_delay, switch_on_threshold, power_to_reserve

def switch_on_threshold_reached(self, load: Load) -> None:
try:
Expand All @@ -350,10 +353,9 @@ def switch_on_threshold_reached(self, load: Load) -> None:
control_parameter = load.data.control_parameter
surplus_config = data.data.general_data.data.chargemode_config.surplus
timestamp_switch_on_off = control_parameter.timestamp_switch_on_off
switch_on_delay, switch_on_threshold, feed_in_limit = self.get_pv_config_py_load(load)
switch_on_delay, switch_on_threshold, power_to_reserve = self.get_pv_config_py_load(load)

surplus, threshold = self.calc_switch_on_power(load)
power_to_reserve = surplus_config.vehicle.switch_on_threshold*control_parameter.phases
surplus, threshold = self.calc_switch_on_power(switch_on_threshold)
if control_parameter.state == ChargepointState.SWITCH_ON_DELAY:
# Wurde die Einschaltschwelle erreicht? Reservierte Leistung aus all_surplus herausrechnen,
# da diese Leistung ja schon reserviert wurde, als die Einschaltschwelle erreicht wurde.
Expand Down Expand Up @@ -397,7 +399,7 @@ def switch_on_timer_expired(self, load: Load) -> None:
try:
msg = None
texts = self._get_switch_on_texts(load)
switch_on_delay, switch_on_threshold, _ = self.get_pv_config_py_load(load)
switch_on_delay, _, power_to_reserve = self.get_pv_config_py_load(load)
control_parameter = load.data.control_parameter
# Timer ist noch nicht abgelaufen
if timecheck.check_timestamp(control_parameter.timestamp_switch_on_off,
Expand All @@ -407,7 +409,7 @@ def switch_on_timer_expired(self, load: Load) -> None:
# Timer abgelaufen
else:
control_parameter.timestamp_switch_on_off = None
self.data.set.reserved_surplus -= switch_on_threshold*control_parameter.phases
self.data.set.reserved_surplus -= power_to_reserve
msg = texts.expired
control_parameter.state = ChargepointState.WAIT_FOR_USING_PHASES

Expand Down Expand Up @@ -581,22 +583,29 @@ def switch_off_check_threshold(self, load: Load) -> bool:
load.set_state_and_log(msg)
return charge

def reset_switch_on_off(self, chargepoint: Chargepoint):
def reset_switch_on_off(self, load: Load):
try:
if chargepoint.data.control_parameter.timestamp_switch_on_off is not None:
chargepoint.data.control_parameter.timestamp_switch_on_off = None
if load.data.control_parameter.timestamp_switch_on_off is not None:
load.data.control_parameter.timestamp_switch_on_off = None
evu_counter = data.data.counter_all_data.get_evu_counter()
# Wenn bereits geladen wird, lief die Abschaltverzögerung -> Leistung, die nach Abschalten frei
# geworden wäre, nicht mehr zum zur Verfügung stehenden Überschuss zählen.
# Wenn nicht geladen wird, reservierte Leistung freigeben.
surplus_config = data.data.general_data.data.chargemode_config.surplus
if not chargepoint.data.get.charge_state:
evu_counter.data.set.reserved_surplus -= (surplus_config.vehicle.switch_on_threshold
* chargepoint.data.set.phases_to_use)
if isinstance(load, Chargepoint):
if not load.data.get.charge_state:
threshold = surplus_config.vehicle.switch_on_threshold * load.data.set.phases_to_use
else:
threshold = surplus_config.vehicle.switch_on_threshold * load.data.control_parameter.phases
else:
threshold = (load.data.control_parameter.required_current *
230 *
load.data.control_parameter.phases)
if not load.data.get.charge_state:
evu_counter.data.set.reserved_surplus -= threshold
else:
evu_counter.data.set.released_surplus -= (surplus_config.vehicle.switch_on_threshold
* chargepoint.data.control_parameter.phases)
chargepoint.data.control_parameter.state = ChargepointState.NO_CHARGING_ALLOWED
evu_counter.data.set.released_surplus -= threshold
load.data.control_parameter.state = ChargepointState.NO_CHARGING_ALLOWED
except Exception:
log.exception("Fehler im allgemeinen PV-Modul")

Expand Down
Loading