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
7 changes: 6 additions & 1 deletion packages/control/consumer/consumer.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from control.chargepoint.chargepoint_state import CHARGING_STATES, ChargepointState
from control.consumer.consumer_data import ConsumerData, ConsumerUsage, ResetModes, WaitForStartStates
from control.load_protocol import Load
from control.text import format_next_time_charging_start
from helpermodules import timecheck
from helpermodules.abstract_plans import ScheduledPlanConsumer
from helpermodules.phase_handling import convert_single_evu_phase_to_cp_phase, voltages_mean
Expand Down Expand Up @@ -350,7 +351,11 @@ def time_charging(self) -> Tuple[float, Optional[str], Chargemode]:
self._convert_power_to_current(self.data.config.max_power))
submode = Chargemode.TIME_CHARGING
else:
message = self.TIME_CHARGING_NO_PLAN_ACTIVE
next_start = timecheck.get_next_timeframe_plan_start(self.data.usage.time_charging.plans)
if next_start is not None:
message = format_next_time_charging_start(next_start)
else:
message = self.TIME_CHARGING_NO_PLAN_ACTIVE
else:
message = self.TIME_CHARGING_NO_PLAN_CONFIGURED
return required_current, message, submode
Expand Down
33 changes: 32 additions & 1 deletion packages/control/consumer/consumer_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,8 @@ def test_wait_for_start_handler(
pytest.param({}, None, (0,
Consumer.TIME_CHARGING_NO_PLAN_CONFIGURED, Chargemode.STOP), id="no plan defined"),
pytest.param({"0": TimeChargingPlanConsumer(id=0)}, None,
(0, Consumer.TIME_CHARGING_NO_PLAN_ACTIVE, Chargemode.STOP), id="no plan active"),
(0, Consumer.TIME_CHARGING_NO_PLAN_ACTIVE, Chargemode.STOP),
id="no plan active without next start"),
pytest.param({"0": TimeChargingPlanConsumer(id=0)}, TimeChargingPlanConsumer(id=0),
(10, None, Chargemode.TIME_CHARGING), id="plan active"),
pytest.param({"0": TimeChargingPlanConsumer(id=0)}, None,
Expand All @@ -166,6 +167,7 @@ def test_time_charging(plans: Dict[int, TimeChargingPlanConsumer],
consumer.data.usage.time_charging.plans = plans
check_plans_timeframe_mock = Mock(return_value=plan_found)
monkeypatch.setattr(timecheck, "check_plans_timeframe", check_plans_timeframe_mock)
monkeypatch.setattr(timecheck, "get_next_timeframe_plan_start", Mock(return_value=None))

# execution
ret = consumer.time_charging()
Expand All @@ -174,6 +176,35 @@ def test_time_charging(plans: Dict[int, TimeChargingPlanConsumer],
assert ret == expected


def test_time_charging_no_plan_active_includes_next_start(
consumer: Consumer,
monkeypatch: pytest.MonkeyPatch):
# setup
consumer.data.usage.time_charging.plans = [TimeChargingPlanConsumer(id=0)]
next_start = datetime.datetime(2026, 1, 2, 14, 30)

monkeypatch.setattr(timecheck, "check_plans_timeframe", Mock(return_value=None))
monkeypatch.setattr(
timecheck,
"get_next_timeframe_plan_start",
Mock(return_value=next_start),
)
monkeypatch.setattr(
"control.consumer.consumer.format_next_time_charging_start",
Mock(return_value="Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr."),
)

# execution
ret = consumer.time_charging()

# evaluation
assert ret == (
0,
"Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr.",
Chargemode.STOP,
)


@pytest.mark.parametrize(
"min_bat_soc, charging_allowed_mock, soc, expected",
[
Expand Down
8 changes: 6 additions & 2 deletions packages/control/ev/charge_template.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from control.chargepoint.charging_type import ChargingType
from control.chargepoint.control_parameter import ControlParameter
from control.ev.ev_template import EvTemplate
from control.text import BidiState
from control.text import BidiState, format_next_time_charging_start
from dataclass_utils import asdict
from dataclass_utils.factories import empty_list_factory
from helpermodules.abstract_plans import Limit, limit_factory, ScheduledChargingPlan, TimeChargingPlan
Expand Down Expand Up @@ -187,7 +187,11 @@ def time_charging(self,
current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current
sub_mode = ChargemodeEnum.TIME_CHARGING
else:
message = self.TIME_CHARGING_NO_PLAN_ACTIVE
next_start = timecheck.get_next_timeframe_plan_start(self.data.time_charging.plans)
if next_start is not None:
message = format_next_time_charging_start(next_start)
else:
message = self.TIME_CHARGING_NO_PLAN_ACTIVE
else:
message = self.TIME_CHARGING_NO_PLAN_CONFIGURED
return current, sub_mode, message, id, phases
Expand Down
37 changes: 35 additions & 2 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 import text
from control.bat_all import BatAll
from control.chargemode import Chargemode
from control.chargepoint.control_parameter import ControlParameter
Expand All @@ -14,7 +15,7 @@
from control.ev.charge_template import ChargeTemplate
from control.ev.ev_template import EvTemplate, EvTemplateData
from control.general import General
from control.text import BidiState
from control.text import BidiState, format_next_time_charging_start
from helpermodules import timecheck
from helpermodules.abstract_plans import Limit, ScheduledChargingPlan, TimeChargingPlan, ScheduledLimit

Expand Down Expand Up @@ -62,12 +63,13 @@ def data_module() -> None:
def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amount_time_charging: float,
plan_found: TimeChargingPlan,
expected: Tuple[int, Chargemode, Optional[str], Optional[str], Optional[int]],
monkeypatch):
monkeypatch: pytest.MonkeyPatch):
# setup
ct = ChargeTemplate()
ct.data.time_charging.plans = plans
check_plans_timeframe_mock = Mock(return_value=plan_found)
monkeypatch.setattr(timecheck, "check_plans_timeframe", check_plans_timeframe_mock)
monkeypatch.setattr(timecheck, "get_next_timeframe_plan_start", Mock(return_value=None))

# execution
ret = ct.time_charging(soc, used_amount_time_charging, ChargingType.AC.value)
Expand All @@ -76,6 +78,37 @@ def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amou
assert ret == expected


def test_time_charging_no_plan_active_includes_next_start(monkeypatch: pytest.MonkeyPatch):
# setup
ct = ChargeTemplate()
ct.data.time_charging.plans = [TimeChargingPlan(id=0)]
next_start = datetime.datetime(2026, 1, 2, 14, 30)

monkeypatch.setattr(timecheck, "check_plans_timeframe", Mock(return_value=None))
monkeypatch.setattr(
timecheck,
"get_next_timeframe_plan_start",
Mock(return_value=next_start),
)
monkeypatch.setattr(
text,
"format_next_time_charging_start",
Mock(return_value="Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr."),
)

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

# evaluation
assert ret == (
0,
Chargemode.STOP,
format_next_time_charging_start(next_start),
None,
None,
)


@pytest.mark.parametrize(
"min_bat_soc, charging_allowed_mock, soc, expected",
[
Expand Down
10 changes: 10 additions & 0 deletions packages/control/text.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import datetime
from enum import Enum


Expand All @@ -6,3 +7,12 @@ class BidiState(Enum):
CP_NOT_BIDI_CAPABLE = "Bidirektionales Laden ist nur mit einer openWB Pro oder Pro+ möglich. "
CP_WRONG_PROTOCOL = "Bitte in den Einstellungen der openWB Pro/Pro+ die Charging Version auf 'HLC' stellen. "
EV_NOT_BIDI_CAPABLE = "Das Fahrzeug unterstützt kein bidirektionales Laden. "


def format_next_time_charging_start(next_start: datetime.datetime) -> str:
today = datetime.datetime.today().date()
if next_start.date() == today:
return f"Nächster Zeitladen-Plan startet heute um {next_start.strftime('%-H:%M')} Uhr."
if next_start.date() == today + datetime.timedelta(days=1):
return f"Nächster Zeitladen-Plan startet morgen um {next_start.strftime('%-H:%M')} Uhr."
return f"Nächster Zeitladen-Plan startet am {next_start.strftime('%d.%m.')} um {next_start.strftime('%-H:%M')} Uhr."
125 changes: 106 additions & 19 deletions packages/helpermodules/timecheck.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,19 @@ def is_now_in_locking_time(now: datetime.datetime,
T = TypeVar("T", AutolockPlan, TimeChargingPlan, TimeChargingPlanConsumer)


def _parse_plan_time(time_value: str) -> Tuple[int, int]:
parsed_time = datetime.datetime.strptime(time_value, "%H:%M")
return parsed_time.hour, parsed_time.minute


def _is_timeframe_valid(now: datetime.datetime, begin: datetime.datetime, end: datetime.datetime) -> bool:
return (not now < begin) and now < end


def _is_weekday_enabled(weekly: List[bool], weekday: int) -> bool:
return weekly[weekday % 7]


def check_plans_timeframe(plans: List[T]) -> Optional[T]:
""" gibt den ersten aktiven Plan zurück. None, falls kein Plan aktiv ist.
"""
Expand All @@ -68,52 +81,126 @@ def check_plans_timeframe(plans: List[T]) -> Optional[T]:
def check_timeframe(plan: Union[AutolockPlan, TimeChargingPlan]) -> bool:
""" Returns: True -> Zeitfenster gültig, False -> Zeitfenster nicht gültig
"""
def is_timeframe_valid(now: datetime.datetime, begin: datetime.datetime, end: datetime.datetime) -> bool:
return True if (not now < begin) and now < end else False

state = False
try:
now = datetime.datetime.today()
begin = datetime.datetime.strptime(plan.time[0], '%H:%M')
end = datetime.datetime.strptime(plan.time[1], '%H:%M')
begin_hour, begin_minute = _parse_plan_time(plan.time[0])
end_hour, end_minute = _parse_plan_time(plan.time[1])

if plan.frequency.selected == "once":
beginDate = datetime.datetime.strptime(plan.frequency.once[0], "%Y-%m-%d")
begin = begin.replace(beginDate.year, beginDate.month, beginDate.day)
endDate = datetime.datetime.strptime(plan.frequency.once[1], "%Y-%m-%d")
end = end.replace(endDate.year, endDate.month, endDate.day)
state = is_timeframe_valid(now, begin, end)
begin_date = datetime.datetime.strptime(plan.frequency.once[0], "%Y-%m-%d")
begin = now.replace(
year=begin_date.year,
month=begin_date.month,
day=begin_date.day,
hour=begin_hour,
minute=begin_minute,
second=0,
microsecond=0,
)
end_date = datetime.datetime.strptime(plan.frequency.once[1], "%Y-%m-%d")
end = now.replace(
year=end_date.year,
month=end_date.month,
day=end_date.day,
hour=end_hour,
minute=end_minute,
second=0,
microsecond=0,
)
state = _is_timeframe_valid(now, begin, end)

else:
begin = begin.replace(now.year, now.month, now.day)
end = end.replace(now.year, now.month, now.day)
begin = now.replace(hour=begin_hour, minute=begin_minute, second=0, microsecond=0)
end = now.replace(hour=end_hour, minute=end_minute, second=0, microsecond=0)
day_change = begin > end
if day_change:
# Endzeit ist am nächsten Tag, in Zeitabschnitt vor und nach Mitternacht einteilen
next_day = now + datetime.timedelta(days=1)
next_day_midnight = next_day.replace(hour=0, minute=0)
state_after_midnight = is_timeframe_valid(now, begin, next_day_midnight)
state_before_midnight = is_timeframe_valid(now, now.replace(hour=0, minute=0), end)
state_after_midnight = _is_timeframe_valid(now, begin, next_day_midnight)
state_before_midnight = _is_timeframe_valid(now, now.replace(hour=0, minute=0), end)

if plan.frequency.selected == "daily":
if day_change:
state = state_before_midnight or state_after_midnight
else:
state = is_timeframe_valid(now, begin, end)
state = _is_timeframe_valid(now, begin, end)

elif plan.frequency.selected == "weekly":
if day_change:
state = ((state_after_midnight and plan.frequency.weekly[now.weekday()]) or
(state_before_midnight and plan.frequency.weekly[now.weekday() - 1]))
state = ((state_after_midnight and _is_weekday_enabled(plan.frequency.weekly, now.weekday())) or
(state_before_midnight and _is_weekday_enabled(plan.frequency.weekly, now.weekday() - 1)))
else:
if plan.frequency.weekly[now.weekday()]:
state = is_timeframe_valid(now, begin, end)
if _is_weekday_enabled(plan.frequency.weekly, now.weekday()):
state = _is_timeframe_valid(now, begin, end)
except Exception:
log.exception("Fehler im System-Modul")
finally:
return state


def get_next_timeframe_plan_start(plans: List[T],
now: Optional[datetime.datetime] = None) -> Optional[datetime.datetime]:
if now is None:
now = datetime.datetime.today()

next_starts: List[datetime.datetime] = []
for plan in plans:
if not getattr(plan, "active", False):
continue
try:
next_start = _get_next_start_for_timeframe_plan(plan, now)
if next_start is not None:
next_starts.append(next_start)
except Exception:
log.exception("Fehler im System-Modul")

if not next_starts:
return None
return min(next_starts)


def _get_next_start_for_timeframe_plan(plan: T, now: datetime.datetime) -> Optional[datetime.datetime]:
start_hour, start_minute = _parse_plan_time(plan.time[0])

if plan.frequency.selected == "once":
begin_date = datetime.datetime.strptime(plan.frequency.once[0], "%Y-%m-%d")
next_start = now.replace(
year=begin_date.year,
month=begin_date.month,
day=begin_date.day,
hour=start_hour,
minute=start_minute,
second=0,
microsecond=0,
)
return next_start if next_start > now else None

if plan.frequency.selected == "daily":
next_start = now.replace(hour=start_hour, minute=start_minute, second=0, microsecond=0)
if next_start <= now:
next_start += datetime.timedelta(days=1)
return next_start

if plan.frequency.selected == "weekly":
if not any(plan.frequency.weekly):
return None
for day_offset in range(0, 8):
weekday = (now.weekday() + day_offset) % 7
if not plan.frequency.weekly[weekday]:
continue
candidate = (now + datetime.timedelta(days=day_offset)).replace(
hour=start_hour,
minute=start_minute,
second=0,
microsecond=0,
)
if candidate > now:
return candidate
return None


def check_end_time(plan: Union[ScheduledPlanConsumer, ScheduledChargingPlan, TimeChargingPlanConsumer],
buffer: Optional[float]) -> float:
""" gibt die verbleibende Zeit in Sekunden zurück.
Expand Down
Loading