Skip to content
Open
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
22 changes: 22 additions & 0 deletions packages/control/bat_all.py
Original file line number Diff line number Diff line change
Expand Up @@ -608,6 +608,28 @@ def get_power_limit(self):
else:
self.data.set.current_state = CurrentState.ACTIVE.value

def time_charging_min_bat_soc_allowed(self) -> bool:
if (self.data.config.configured and
self.data.config.bat_control_permitted and
self.data.config.bat_control_activated):
if (self.data.config.power_limit_condition == BatPowerLimitCondition.MANUAL.value or
self.data.config.power_limit_condition == BatPowerLimitCondition.VEHICLE_CHARGING.value):
return False
elif self.data.config.power_limit_condition == BatPowerLimitCondition.PRICE_LIMIT.value:
if self.data.config.power_limit_mode == BatPowerLimitMode.MODE_CHARGE_PV_PRODUCTION.value:
if ((self.data.config.price_limit_activated and
data.data.optional_data.ep_is_charging_allowed_price_threshold(
self.data.config.price_limit))
or (self.data.config.price_charge_activated and
data.data.optional_data.ep_is_charging_allowed_price_threshold(
self.data.config.charge_limit))):
return True
else:
return False
else:
return False
return True


def get_controllable_bat_components() -> List:
bat_components = []
Expand Down
104 changes: 104 additions & 0 deletions packages/control/bat_all_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -367,3 +367,107 @@ def test_control_price_limit(params: BatControlParams, data_, monkeypatch):
data.data.bat_all_data._set_bat_power_active_control(data.data.bat_all_data.data.set.power_limit)

assert data.data.bat_data["bat2"].data.set.power_limit == params.expected_power_limit_bat


@pytest.mark.parametrize(
"control_permitted, control_activated, condition, limit, expected_result",
[
pytest.param(False, True,
BatPowerLimitCondition.MANUAL.value, BatPowerLimitMode.MODE_NO_DISCHARGE.value, True,
id="Speichersteuerung nicht erlaubt, aber aktiviert -> laden"),
pytest.param(True, False,
BatPowerLimitCondition.MANUAL.value, BatPowerLimitMode.MODE_NO_DISCHARGE.value, True,
id="Speichersteuerung erlaubt, aber nicht aktiviert -> laden"),
pytest.param(True, True,
BatPowerLimitCondition.MANUAL.value, BatPowerLimitMode.MODE_NO_DISCHARGE.value, False,
id="Manuell, volle Entladesperre -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.MANUAL.value,
BatPowerLimitMode.MODE_DISCHARGE_HOME_CONSUMPTION.value, False,
id="Manuell, Entladung in Fahrzeuge sperren -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.MANUAL.value, BatPowerLimitMode.MODE_CHARGE_PV_PRODUCTION.value, False,
id="Manuell, PV-Ertrag speichern -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.VEHICLE_CHARGING.value, BatPowerLimitMode.MODE_NO_DISCHARGE.value, False,
id="Fahrzeuge laden, volle Entladesperre -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.VEHICLE_CHARGING.value,
BatPowerLimitMode.MODE_DISCHARGE_HOME_CONSUMPTION.value, False,
id="Fahrzeuge laden, Entladung in Fahrzeuge sperren -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.VEHICLE_CHARGING.value,
BatPowerLimitMode.MODE_CHARGE_PV_PRODUCTION.value, False,
id="Fahrzeuge laden, PV-Ertrag speichern -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.PRICE_LIMIT.value, BatPowerLimitMode.MODE_NO_DISCHARGE.value, False,
id="Preislimit, volle Entladesperre -> nicht laden"),
pytest.param(True, True,
BatPowerLimitCondition.PRICE_LIMIT.value,
BatPowerLimitMode.MODE_DISCHARGE_HOME_CONSUMPTION.value, False,
id="Preislimit, Entladung in Fahrzeuge sperren -> nicht laden"),

]
)
def test_time_charging_min_bat_soc_allowed(control_permitted: bool,
control_activated: bool,
condition: BatPowerLimitCondition,
limit: BatPowerLimitMode,
Comment on lines +414 to +415
Copy link

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In test_time_charging_min_bat_soc_allowed, the parameters condition and limit are annotated as BatPowerLimitCondition / BatPowerLimitMode, but the parametrized values passed in are the .value strings. This makes the type hints incorrect and can confuse readers/tools; either annotate them as str or pass the enum instances instead of .value.

Suggested change
condition: BatPowerLimitCondition,
limit: BatPowerLimitMode,
condition: str,
limit: str,

Copilot uses AI. Check for mistakes.
expected_result: bool):
# setup
b = BatAll()
b.data.config.configured = True
b.data.config.power_limit_condition = condition
b.data.config.power_limit_mode = limit
b.data.config.bat_control_permitted = control_permitted
b.data.config.bat_control_activated = control_activated

Comment on lines +418 to +424
Copy link

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Test setup does not set b.data.config.configured = True, but BatAll.time_charging_min_bat_soc_allowed() gates its logic on self.data.config.configured. With the default configured=False, this test will always get True and cannot validate the intended branches.

Copilot uses AI. Check for mistakes.
# execution
result = b.time_charging_min_bat_soc_allowed()

# evaluation
assert result == expected_result


@pytest.mark.parametrize(
"ep_configured, price_limit_activated, price_charge_activated, price_threshold_mock, expected_result",
[
pytest.param(False, True, True, [True, True], True,
id="Preislimit aktiviert, aber kein Preis konfiguriert -> Eigenregelung -> laden"),
pytest.param(True, True, False, [True], True,
id="Strompreis für Regelmodus, Preis unter Limit -> laden"),
pytest.param(True, True, False, [False], False,
id="Strompreis für Regelmodus, Preis über Limit -> nicht laden"),
pytest.param(True, False, True, [True], True,
id="Strompreis für aktives Laden, Preis unter Limit -> laden"),
pytest.param(True, False, True, [False], False,
id="Strompreis für aktives Laden, Preis unter Limit -> nicht laden"),
pytest.param(True, False, False, [], False,
id="beide Strompreise deaktiviert -> nicht laden"),
]
)
def test_time_charging_min_bat_soc_allowed_pricing(ep_configured: bool,
price_limit_activated: bool,
price_charge_activated: bool,
price_threshold_mock: List[bool],
expected_result: bool,
monkeypatch):
# setup
b = BatAll()
b.data.config.configured = True
b.data.config.power_limit_condition = BatPowerLimitCondition.PRICE_LIMIT.value
b.data.config.power_limit_mode = BatPowerLimitMode.MODE_CHARGE_PV_PRODUCTION.value
b.data.config.price_limit_activated = price_limit_activated
b.data.config.price_charge_activated = price_charge_activated
data.data.optional_data.data.electricity_pricing.configured = ep_configured
b.data.config.bat_control_permitted = True
b.data.config.bat_control_activated = True

monkeypatch.setattr(data.data.optional_data, "ep_is_charging_allowed_price_threshold",
Mock(side_effect=price_threshold_mock))

# execution
result = b.time_charging_min_bat_soc_allowed()

# evaluation
assert result == expected_result
Comment on lines +372 to +473
Copy link

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are two test functions named test_time_charging_min_bat_soc_allowed in this module. The second definition overwrites the first one at import time, so the first parametrized test set will never be collected/executed by pytest; rename one of them.

Copilot uses AI. Check for mistakes.
32 changes: 21 additions & 11 deletions packages/control/ev/charge_template.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,10 @@ class ChargeTemplate:
TIME_CHARGING_NO_PLAN_ACTIVE = "Kein Zeitfenster für Zeitladen aktiv."
TIME_CHARGING_SOC_REACHED = "Das Ladeziel für das Zeitladen wurde erreicht."
TIME_CHARGING_AMOUNT_REACHED = "Die gewünschte Energiemenge für das Zeitladen wurde geladen."
TIME_CHARGING_CONFLICT_ACTIVE_BAT_CONTROL = ("Laden mit Zeitladen nach Speicher-SoC nicht möglich, da dies im "
"Konflikt mit der aktiven Speichersteuerung steht.")
TIME_CHARGING_MIN_BAT_SOC_REACHED = ("Laden mit Zeitladen nach Speicher-SoC nicht möglich, da der SoC des"
" Speichers unter dem minimalen SoC liegt.")

def time_charging(self,
soc: Optional[float],
Expand All @@ -151,34 +155,40 @@ def time_charging(self,
""" prüft, ob ein Zeitfenster aktiv ist und setzt entsprechend den Ladestrom
"""
message = None
sub_mode = "time_charging"
sub_mode = "stop"
current = 0
id = None
phases = None
try:
if self.data.time_charging.plans:
plan = timecheck.check_plans_timeframe(self.data.time_charging.plans)
if plan is not None:
current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current
phases = plan.phases_to_use
id = plan.id
phases = plan.phases_to_use
if plan.limit.selected == "soc" and soc and soc >= plan.limit.soc:
# SoC-Limit erreicht
current = 0
sub_mode = "stop"
message = self.TIME_CHARGING_SOC_REACHED
elif plan.limit.selected == "amount" and used_amount_time_charging >= plan.limit.amount:
# Energie-Limit erreicht
current = 0
sub_mode = "stop"
message = self.TIME_CHARGING_AMOUNT_REACHED
elif plan.min_bat_soc is not None and data.data.bat_all_data.data.config.configured:
if data.data.bat_all_data.time_charging_min_bat_soc_allowed():
Copy link

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In time_charging(), when plan.min_bat_soc is set you always compare data.data.bat_all_data.data.get.soc against it. If no storage is configured (bat_all_data.data.config.configured == False), bat_all_data.data.get.soc defaults to 0, so any configured min_bat_soc will block time charging even though there is no battery to evaluate. Consider skipping the min-battery-SoC logic (or treating it as satisfied) when the battery system is not configured.

Suggested change
if data.data.bat_all_data.time_charging_min_bat_soc_allowed():
if not data.data.bat_all_data.data.config.configured:
log.debug("Zeitladen: kein Speicher konfiguriert, minimaler Speicher-SoC wird ignoriert.")
current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current
sub_mode = "time_charging"
elif data.data.bat_all_data.time_charging_min_bat_soc_allowed():

Copilot uses AI. Check for mistakes.
if data.data.bat_all_data.data.get.soc < plan.min_bat_soc:
message = self.TIME_CHARGING_MIN_BAT_SOC_REACHED
else:
log.debug(
"Zeitladen: minimaler Speicher-SoC überschritten, Laden mit Zeitladen möglich.")
current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current
sub_mode = "time_charging"
else:
message = self.TIME_CHARGING_CONFLICT_ACTIVE_BAT_CONTROL
else:
current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current
sub_mode = "time_charging"
else:
message = self.TIME_CHARGING_NO_PLAN_ACTIVE
current = 0
sub_mode = "stop"
else:
message = self.TIME_CHARGING_NO_PLAN_CONFIGURED
current = 0
sub_mode = "stop"
return current, sub_mode, message, id, phases
except Exception:
log.exception("Fehler im ev-Modul "+str(self.data.id))
Expand Down
39 changes: 39 additions & 0 deletions packages/control/ev/charge_template_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from control import data
from control import optional
from control.bat_all import BatAll
from control.chargepoint.control_parameter import ControlParameter
from control.ev.charge_template import SelectedPlan
from control.chargepoint.charging_type import ChargingType
Expand Down Expand Up @@ -73,6 +74,44 @@ def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amou
assert ret == expected


@pytest.mark.parametrize(
"min_bat_soc, charging_allowed_mock, soc, expected",
[
pytest.param(80, False, 81, (0, "stop", ChargeTemplate.TIME_CHARGING_CONFLICT_ACTIVE_BAT_CONTROL, 0, 1),
id="Konflikt mit aktiver Speichersteuerung -> nicht laden"),
pytest.param(80, True, 79, (0, "stop", ChargeTemplate.TIME_CHARGING_MIN_BAT_SOC_REACHED, 0, 1),
id="Mindest-SoC des Speichers unterschritten -> nicht laden"),
pytest.param(80, True, 80, (16, "time_charging", None, 0, 1), id="laden erlaubt"),
pytest.param(None, True, 80, (16, "time_charging", None, 0, 1),
id="Mindest-SoC-Beachtung nicht konfiguriert, laden erlaubt"),
]
)
def test_time_charging_min_bat_soc(min_bat_soc: Optional[int],
charging_allowed_mock: bool,
soc: float,
expected: Tuple[int, str, Optional[str], Optional[str], int],
monkeypatch):
# setup
ct = ChargeTemplate()
plan = TimeChargingPlan(id=0)
plan.min_bat_soc = min_bat_soc
ct.data.time_charging.plans = [plan]
check_plans_timeframe_mock = Mock(return_value=plan)
monkeypatch.setattr(timecheck, "check_plans_timeframe", check_plans_timeframe_mock)

data.data.bat_all_data = BatAll()
data.data.bat_all_data.data.config.configured = True
data.data.bat_all_data.data.get.soc = soc
monkeypatch.setattr(data.data.bat_all_data, "time_charging_min_bat_soc_allowed",
Mock(return_value=charging_allowed_mock))

# execution
ret = ct.time_charging(soc, 100, ChargingType.AC.value)

# evaluation
assert ret == expected


@pytest.mark.parametrize(
"selected, current_soc, used_amount, expected",
[
Expand Down
1 change: 1 addition & 0 deletions packages/helpermodules/abstract_plans.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ class TimeChargingPlan(TimeframePlan):
dc_current: float = 145
id: Optional[int] = None
limit: Limit = field(default_factory=limit_factory)
min_bat_soc: Optional[int] = None
name: str = "neuer Zeitladen-Plan"
phases_to_use: int = 1

Expand Down
13 changes: 12 additions & 1 deletion packages/helpermodules/update_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@

class UpdateConfig:

DATASTORE_VERSION = 121
DATASTORE_VERSION = 122

valid_topic = [
"^openWB/bat/config/bat_control_permitted$",
Expand Down Expand Up @@ -3068,3 +3068,14 @@ def upgrade(topic: str, payload) -> Optional[dict]:
return {topic: payload}
self._loop_all_received_topics(upgrade)
self._append_datastore_version(121)

def upgrade_datastore_122(self) -> None:
def upgrade(topic: str, payload) -> Optional[dict]:
if re.search("openWB/vehicle/template/charge_template/[0-9]+$", topic) is not None:
payload = decode_payload(payload)
for plan in payload["time_charging"]["plans"]:
if plan.get("min_bat_soc") is None:
plan.update({"min_bat_soc": None})
return {topic: payload}
Comment on lines +3076 to +3079
Copy link

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

upgrade_datastore_122 assumes payload["time_charging"]["plans"] always exists and is iterable. For older/partial datastores this can raise KeyError/TypeError during upgrade and abort the whole migration. Please make the migration defensive (e.g., payload.get("time_charging", {}).get("plans", []), setdefault, and/or type checks) and only return an updated topic when a change was actually made.

Suggested change
for plan in payload["time_charging"]["plans"]:
if plan.get("min_bat_soc") is None:
plan.update({"min_bat_soc": None})
return {topic: payload}
if not isinstance(payload, dict):
return None
time_charging = payload.get("time_charging")
if not isinstance(time_charging, dict):
return None
plans = time_charging.get("plans", [])
if not isinstance(plans, list):
return None
changed = False
for plan in plans:
if isinstance(plan, dict) and "min_bat_soc" not in plan:
plan["min_bat_soc"] = None
changed = True
if changed:
return {topic: payload}

Copilot uses AI. Check for mistakes.
self._loop_all_received_topics(upgrade)
self._append_datastore_version(122)
Loading