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
19 changes: 19 additions & 0 deletions packages/control/bat_all.py
Original file line number Diff line number Diff line change
Expand Up @@ -654,6 +654,25 @@ 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_activated:
# manueller Modus und keine Eigenregelung oder aktive Speichersteuerung bei Fahrzeugladung
if ((self.data.config.power_limit_condition == BatPowerLimitCondition.MANUAL.value and
self.data.config.manual_mode != ManualMode.MANUAL_DISABLE.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.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
return True


def get_bat_components_by_controllability() -> Tuple[List, List]:
bat_components_controllable, bat_components_not_controllable = [], []
Expand Down
114 changes: 114 additions & 0 deletions packages/control/bat_all_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -344,3 +344,117 @@ 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_activated, condition, limit, manual_mode, expected_result",
[
pytest.param(False,
BatPowerLimitCondition.MANUAL.value,
BatPowerLimitMode.MODE_NO_DISCHARGE.value,
ManualMode.MANUAL_DISABLE.value, True,
id="Speichersteuerung nicht aktiviert, aber aktiviert -> laden"),
pytest.param(True,
BatPowerLimitCondition.MANUAL.value,
BatPowerLimitMode.MODE_NO_DISCHARGE.value,
ManualMode.MANUAL_DISABLE.value, True,
id="Manuell, Eigenregelung, volle Entladesperre -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.MANUAL.value,
BatPowerLimitMode.MODE_DISCHARGE_HOME_CONSUMPTION.value,
ManualMode.MANUAL_LIMIT.value, False,
id="Manuell, Entladung in Fahrzeuge sperren -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.MANUAL.value,
BatPowerLimitMode.MODE_CHARGE_PV_PRODUCTION.value,
ManualMode.MANUAL_CHARGE.value, False,
id="Manuell, PV-Ertrag speichern -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.VEHICLE_CHARGING.value,
BatPowerLimitMode.MODE_NO_DISCHARGE.value,
ManualMode.MANUAL_DISABLE.value, False,
id="Fahrzeuge laden, volle Entladesperre -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.VEHICLE_CHARGING.value,
BatPowerLimitMode.MODE_DISCHARGE_HOME_CONSUMPTION.value,
ManualMode.MANUAL_DISABLE.value, False,
id="Fahrzeuge laden, Entladung in Fahrzeuge sperren -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.VEHICLE_CHARGING.value,
BatPowerLimitMode.MODE_CHARGE_PV_PRODUCTION.value,
ManualMode.MANUAL_DISABLE.value, False,
id="Fahrzeuge laden, PV-Ertrag speichern -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.PRICE_LIMIT.value,
BatPowerLimitMode.MODE_NO_DISCHARGE.value,
ManualMode.MANUAL_DISABLE.value, False,
id="Preislimit, volle Entladesperre -> nicht laden"),
pytest.param(True,
BatPowerLimitCondition.PRICE_LIMIT.value,
BatPowerLimitMode.MODE_DISCHARGE_HOME_CONSUMPTION.value,
ManualMode.MANUAL_DISABLE.value, False,
id="Preislimit, Entladung in Fahrzeuge sperren -> nicht laden"),

]
)
def test_time_charging_min_bat_soc_allowed(control_activated: bool,
condition: str,
limit: str,
manual_mode: str,
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_activated = control_activated
b.data.config.manual_mode = manual_mode

Comment on lines +406 to +412

Copilot AI Apr 17, 2026

Copy link

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: pytest.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_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 +349 to +460

Copilot AI Apr 17, 2026

Copy link

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():

Copilot AI Apr 17, 2026

Copy link

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: 11 additions & 2 deletions 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 = 128
DATASTORE_VERSION = 129

valid_topic = [
"^openWB/bat/config/bat_control_activated$",
Expand Down Expand Up @@ -3265,4 +3265,13 @@ def upgrade_component(topic: str, payload) -> Optional[dict]:
MessageType.INFO
)

self._append_datastore_version(128)
def upgrade_datastore_129(self) -> None:
def upgrade(topic: str, payload) -> None:
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 +3272 to +3275

Copilot AI Apr 17, 2026

Copy link

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(129)
Loading