Skip to content

Feature/tibber/quarter hourly prices #2801

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 19 commits into from
Nov 4, 2025
Merged

Feature/tibber/quarter hourly prices #2801

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
3aca822
request quarter hourly prices
tpd-opitz Sep 30, 2025
21c6c28
Update timecheck_test.py
tpd-opitz Oct 1, 2025
e6d4889
expect shorter price time slots in charge time calculation
tpd-opitz Oct 4, 2025
46f7daf
unittests for 15 minuts charging time slots selection
tpd-opitz Oct 6, 2025
8e9a90b
enable shorter update cycle of ET price data
tpd-opitz Oct 8, 2025
7486511
fix missing parameter in et_charging_allowed()
tpd-opitz Oct 16, 2025
ac22545
do not overwrite known prices if list from ET provider is empty
tpd-opitz Oct 16, 2025
06199c7
enhance logging
tpd-opitz Oct 21, 2025
a54dffd
requested changes
tpd-opitz Oct 21, 2025
cd68a4e
improve schedule charging messages
tpd-opitz Oct 21, 2025
5172eba
avoid resetting price list
tpd-opitz Oct 21, 2025
b5aa349
fix price slot selection
tpd-opitz Oct 22, 2025
8b53dea
fix next fetch date
tpd-opitz Oct 23, 2025
09aaa8f
update prices on construction of Optional, track (re)creation of Et/E…
tpd-opitz Oct 28, 2025
3918d07
enhance logging for Tarif update target time
tpd-opitz Oct 28, 2025
5d61668
fix persistence of TariffState
tpd-opitz Oct 31, 2025
35e44b4
move et price update call in main.py
tpd-opitz Nov 3, 2025
baba301
refactoring: rename optional.et_is_charging_allowed_*
tpd-opitz Nov 3, 2025
d9e3006
completely remove instance_id
tpd-opitz Nov 3, 2025
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
11 changes: 9 additions & 2 deletions packages/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,11 @@
from modules.common.store._api import LoggingValueStore


def pytest_configure(config):
config.addinivalue_line("markers", "no_mock_full_hour: mark test to disable full_hour mocking.")
config.addinivalue_line("markers", "no_mock_quarter_hour: mark test to disable quarter_hour mocking.")


@pytest.fixture(autouse=True)
def mock_open_file(monkeypatch) -> None:
mock_config = Mock(return_value={"dc_charging": False, "openwb-version": 1, "max_c_socket": 32})
Expand All @@ -35,8 +40,10 @@ def mock_today(monkeypatch) -> None:
datetime_mock.today.return_value = datetime.datetime(2022, 5, 16, 8, 40, 52)
datetime_mock.now.return_value = datetime.datetime(2022, 5, 16, 8, 40, 52)
monkeypatch.setattr(datetime, "datetime", datetime_mock)
mock_today_timestamp = Mock(return_value=1652683252)
monkeypatch.setattr(timecheck, "create_timestamp", mock_today_timestamp)
now_timestamp = Mock(return_value=1652683252)
monkeypatch.setattr(timecheck, "create_timestamp", now_timestamp)
full_hour_timestamp = Mock(return_value=int(datetime.datetime(2022, 5, 16, 8, 0, 0).timestamp()))
monkeypatch.setattr(timecheck, "create_unix_timestamp_current_full_hour", full_hour_timestamp)
Comment on lines +45 to +46
Copy link
Contributor

@LKuemmel LKuemmel Oct 20, 2025

Choose a reason for hiding this comment

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

Suggested change
full_hour_timestamp = Mock(return_value=int(datetime.datetime(2022, 5, 16, 8, 0, 0).timestamp()))
monkeypatch.setattr(timecheck, "create_unix_timestamp_current_full_hour", full_hour_timestamp)

Müsste der Mock nicht gerade wegfallen, wenn es unabhängig von festen Zeitintervallen sein soll?

Copy link
Contributor Author

@tpd-opitz tpd-opitz Oct 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

Ja, müsste er.
Hab mich nur nicht getraut, die Funktion komplett aus timecheck raus zu nehmen, daher ist auch der Mock noch drin.

Copy link
Contributor Author

@tpd-opitz tpd-opitz Oct 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

Ich habe noch Referenzen hier:
https://github.com/tpd-opitz/openwb-core/blob/78880eeb7ebb0415cea2bf6993d5a4fc52675262/packages/control/chargelog/chargelog.py#L413
und hier
https://github.com/tpd-opitz/openwb-core/blob/78880eeb7ebb0415cea2bf6993d5a4fc52675262/packages/control/chargelog/chargelog.py#L459-L462
gefunden, für deren Anpassungen wohl umfangreichere Änderungen nötig sind.

Weitere Referenzen hier
https://github.com/tpd-opitz/openwb-core/blob/78880eeb7ebb0415cea2bf6993d5a4fc52675262/packages/modules/electricity_tariffs/energycharts/tariff.py#L25
und hier
https://github.com/tpd-opitz/openwb-core/blob/78880eeb7ebb0415cea2bf6993d5a4fc52675262/packages/modules/electricity_tariffs/voltego/tariff.py#L58
Sind leichter aufzulösen.

Copy link
Contributor Author

@tpd-opitz tpd-opitz Oct 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

Im Voltego-Tarif hat create_unix_timestamp_current_full_hour() noch seine Berechtigung, so lange die noch nicht auf kürzere Auflösungen umstellen

Copy link
Contributor Author

@tpd-opitz tpd-opitz Oct 21, 2025

Choose a reason for hiding this comment

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

Auch be energy-charts ist der Aufruf von create_unix_timestamp_current_full_hour() noch sinnvoll, weil man mit dem aktuellen Timestamp nicht den aktuellen Timeslot, sonder erst den nächsten zurück bekommt.

Copy link
Contributor Author

@tpd-opitz tpd-opitz Oct 21, 2025

Choose a reason for hiding this comment

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

Bezüglich dem Chargelog: Wenn dort einfach immer eine Zwischensumme berechnet wird brauchen wir den Aufruf dort nicht.

Copy link
Contributor

@LKuemmel LKuemmel Oct 21, 2025

Choose a reason for hiding this comment

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

Ist in Arbeit, wird diese Woche fertig.



@pytest.fixture(autouse=True)
Expand Down
65 changes: 44 additions & 21 deletions packages/control/ev/charge_template.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -293,7 +293,7 @@ def eco_charging(self,
sub_mode = "stop"
message = self.AMOUNT_REACHED
elif data.data.optional_data.et_provider_available():
if data.data.optional_data.et_charging_allowed(eco_charging.max_price):
if data.data.optional_data.et_is_charging_allowed_price_threshold(eco_charging.max_price):
sub_mode = "instant_charging"
message = self.CHARGING_PRICE_LOW
phases = max_phases_hw
Expand Down Expand Up @@ -498,28 +498,34 @@ def _calculate_duration(self,
duration = missing_amount/(current * phases*230) * 3600
return duration, missing_amount

SCHEDULED_REACHED_LIMIT_SOC = ("Kein Zielladen, da noch Zeit bis zum Zieltermin ist. "
"Kein Zielladen mit Überschuss, da das SoC-Limit für Überschuss-Laden " +
"erreicht wurde. ")
SCHEDULED_CHARGING_REACHED_LIMIT_SOC = ("Kein Zielladen, da das Limit für Fahrzeug Laden mit Überschuss (SoC-Limit)"
" sowie der Fahrzeug-SoC (Ziel-SoC) bereits erreicht wurde. ")
SCHEDULED_CHARGING_REACHED_AMOUNT = "Kein Zielladen, da die Energiemenge bereits erreicht wurde. "
SCHEDULED_REACHED_MAX_SOC = ("Zielladen ausstehend, da noch Zeit bis zum Zieltermin ist. "
"Kein Zielladen mit Überschuss, da das SoC-Limit für Überschuss-Laden " +
"erreicht wurde. ")
SCHEDULED_CHARGING_REACHED_MAX_AND_LIMIT_SOC = (
"Zielladen abgeschlossen, da das Limit für Fahrzeug Laden mit Überschuss (SoC-Limit)"
" sowie der Fahrzeug-SoC (Ziel-SoC) bereits erreicht wurde. ")
SCHEDULED_CHARGING_REACHED_AMOUNT = "Zielladen abgeschlossen, da die Energiemenge bereits erreicht wurde. "
SCHEDULED_CHARGING_REACHED_SCHEDULED_SOC = ("Falls vorhanden wird mit EVU-Überschuss geladen, da der Ziel-Soc "
"für Zielladen bereits erreicht wurde. ")
SCHEDULED_CHARGING_BIDI = ("Der Ziel-Soc für Zielladen wurde bereits erreicht. Das Auto wird "
"bidirektional ge-/entladen, sodass möglichst weder Bezug noch "
"Einspeisung erfolgt. ")
SCHEDULED_CHARGING_NO_PLANS_CONFIGURED = "Keine Ladung, da keine Ziel-Termine konfiguriert sind."
SCHEDULED_CHARGING_NO_PLANS_CONFIGURED = "Kein Zielladen, da keine Ziel-Termine konfiguriert sind."
SCHEDULED_CHARGING_NO_DATE_PENDING = "Kein Zielladen, da kein Ziel-Termin ansteht. "
SCHEDULED_CHARGING_USE_PV = "Laden startet {}. Falls vorhanden, wird mit Überschuss geladen. "
SCHEDULED_CHARGING_USE_PV = "Zielladen startet {}. Falls vorhanden, wird mit Überschuss geladen. "
SCHEDULED_CHARGING_MAX_CURRENT = "Zielladen mit {}A. Der Ladestrom wurde erhöht, um das Ziel zu erreichen. "
SCHEDULED_CHARGING_LIMITED_BY_SOC = 'einen SoC von {}%'
SCHEDULED_CHARGING_LIMITED_BY_AMOUNT = '{}kWh geladene Energie'
SCHEDULED_CHARGING_IN_TIME = ('Zielladen mit mindestens {}A, um {} um {} zu erreichen. Falls vorhanden wird '
'zusätzlich EVU-Überschuss geladen. ')
SCHEDULED_CHARGING_CHEAP_HOUR = "Zielladen, da ein günstiger Zeitpunkt zum preisbasierten Laden ist. {}"
SCHEDULED_CHARGING_EXPENSIVE_HOUR = ("Zielladen ausstehend, da jetzt kein günstiger Zeitpunkt zum preisbasierten "
"Laden ist. {} Falls vorhanden, wird mit Überschuss geladen. ")
SCHEDULED_CHARGING_EXPENSIVE_HOUR = (
"Zielladen ausstehend, da jetzt kein günstiger Zeitpunkt zum preisbasierten "
"Laden ist. {} Falls vorhanden, wird mit Überschuss geladen. ")
SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC = (
"Zielladen ausstehend, da jetzt kein günstiger Zeitpunkt zum preisbasierten "
"Laden ist. {} " +
"Kein Zielladen mit Überschuss, da das SoC-Limit für Überschuss-Laden erreicht wurde.")

def scheduled_charging_calc_current(self,
selected_plan: Optional[SelectedPlan],
Expand Down Expand Up @@ -554,7 +560,7 @@ def scheduled_charging_calc_current(self,
(soc > limit.soc_limit if (plan.bidi_charging_enabled and bidi_state == BidiState.BIDI_CAPABLE)
else soc >= limit.soc_limit) and
soc >= limit.soc_scheduled):
message = self.SCHEDULED_CHARGING_REACHED_LIMIT_SOC
message = self.SCHEDULED_CHARGING_REACHED_MAX_AND_LIMIT_SOC
elif limit.selected == "soc" and limit.soc_scheduled <= soc <= limit.soc_limit:
if plan.bidi_charging_enabled and bidi_state == BidiState.BIDI_CAPABLE:
message = self.SCHEDULED_CHARGING_BIDI
Expand Down Expand Up @@ -596,29 +602,46 @@ def scheduled_charging_calc_current(self,
# Wenn dynamische Tarife aktiv sind, prüfen, ob jetzt ein günstiger Zeitpunkt zum Laden
# ist.
if plan.et_active:
def get_hours_message() -> str:
def is_loading_hour(hour: int) -> bool:
return data.data.optional_data.et_is_charging_allowed_hours_list(hour)
return ("Geladen wird "+("jetzt und "
if is_loading_hour(hour_list)
else '') +
"zu folgenden Uhrzeiten: " +
", ".join([tomorrow(hour) +
datetime.datetime.fromtimestamp(hour).strftime('%-H:%M')
for hour in (sorted(hour_list)
if not is_loading_hour(hour_list)
else (sorted(hour_list)[1:] if len(hour_list) > 1 else []))])
+ ".")

def end_of_today_timestamp() -> int:
return datetime.datetime.now().replace(
hour=23, minute=59, second=59, microsecond=999000).timestamp()

def tomorrow(timestamp: int) -> str:
return 'morgen ' if end_of_today_timestamp() < timestamp else ''
hour_list = data.data.optional_data.et_get_loading_hours(
selected_plan.duration, selected_plan.remaining_time)
hours_message = ("Geladen wird zu folgenden Uhrzeiten: " +
", ".join([datetime.datetime.fromtimestamp(hour).strftime('%-H:%M')
for hour in sorted(hour_list)])
+ ".")

log.debug(f"Günstige Ladezeiten: {hour_list}")
if timecheck.is_list_valid(hour_list):
message = self.SCHEDULED_CHARGING_CHEAP_HOUR.format(hours_message)
if data.data.optional_data.et_is_charging_allowed_hours_list(hour_list):
message = self.SCHEDULED_CHARGING_CHEAP_HOUR.format(get_hours_message())
current = plan_current
submode = "instant_charging"
elif ((limit.selected == "soc" and soc <= limit.soc_limit) or
(limit.selected == "amount" and used_amount < limit.amount)):
message = self.SCHEDULED_CHARGING_EXPENSIVE_HOUR.format(hours_message)
message = self.SCHEDULED_CHARGING_EXPENSIVE_HOUR.format(get_hours_message())
current = min_current
submode = "pv_charging"
phases = plan.phases_to_use_pv
else:
message = self.SCHEDULED_REACHED_LIMIT_SOC
message = self.SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC.format(get_hours_message())
else:
# Wenn SoC-Limit erreicht wurde, soll nicht mehr mit Überschuss geladen werden
if limit.selected == "soc" and soc >= limit.soc_limit:
message = self.SCHEDULED_REACHED_LIMIT_SOC
message = self.SCHEDULED_REACHED_MAX_SOC
else:
now = datetime.datetime.today()
start_time = now + datetime.timedelta(seconds=selected_plan.remaining_time)
Expand Down
96 changes: 80 additions & 16 deletions packages/control/ev/charge_template_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from control.general import General
from control.text import BidiState
from helpermodules import timecheck
from helpermodules.abstract_plans import Limit, ScheduledChargingPlan, TimeChargingPlan
from helpermodules.abstract_plans import Limit, ScheduledChargingPlan, TimeChargingPlan, ScheduledLimit


@pytest.fixture(autouse=True)
Expand Down Expand Up @@ -233,7 +233,7 @@ def test_scheduled_charging_recent_plan(end_time_mock,
pytest.param(None, 0, 0, "none", False, (0, "stop",
ChargeTemplate.SCHEDULED_CHARGING_NO_DATE_PENDING, 3), id="no date pending"),
pytest.param(SelectedPlan(duration=3600), 90, 0, "soc", False, (0, "stop",
ChargeTemplate.SCHEDULED_CHARGING_REACHED_LIMIT_SOC, 1), id="reached limit soc"),
ChargeTemplate.SCHEDULED_CHARGING_REACHED_MAX_AND_LIMIT_SOC, 1), id="reached limit soc"),
pytest.param(SelectedPlan(duration=3600), 80, 0, "soc", False, (6, "pv_charging",
ChargeTemplate.SCHEDULED_CHARGING_REACHED_SCHEDULED_SOC, 0), id="reached scheduled soc"),
pytest.param(SelectedPlan(duration=3600), 80, 0, "soc", True, (6, "bidi_charging",
Expand Down Expand Up @@ -296,32 +296,96 @@ def test_scheduled_charging_calc_current_no_plans():
assert ret == (0, "stop", ChargeTemplate.SCHEDULED_CHARGING_NO_PLANS_CONFIGURED, 3)


LOADING_HOURS_TODAY = [datetime.datetime(
year=2022, month=5, day=16, hour=8, minute=0).timestamp()]

LOADING_HOURS_TOMORROW = [datetime.datetime(
year=2022, month=5, day=17, hour=8, minute=0).timestamp()]


@pytest.mark.parametrize(
"loading_hour, expected",
"is_loading_hour, current_soc, soc_scheduled, sco_limit, loading_hours, expected",
[
pytest.param(True, (14, "instant_charging", ChargeTemplate.SCHEDULED_CHARGING_CHEAP_HOUR.format(
"Geladen wird zu folgenden Uhrzeiten: 8:00."), 3)),
pytest.param(False, (6, "pv_charging", ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR.format(
"Geladen wird zu folgenden Uhrzeiten: 8:00."), 0)),
pytest.param(True, 79, 80, 90, LOADING_HOURS_TODAY + LOADING_HOURS_TOMORROW,
(
14,
"instant_charging",
ChargeTemplate.SCHEDULED_CHARGING_CHEAP_HOUR.format(
"Geladen wird jetzt und zu folgenden Uhrzeiten: morgen 8:00."),
3),
id="cheap_hour_charge_with_instant_charging"),
pytest.param(True, 79, 80, 70, LOADING_HOURS_TODAY,
(
14,
"instant_charging",
ChargeTemplate.SCHEDULED_CHARGING_CHEAP_HOUR.format(
"Geladen wird jetzt und zu folgenden Uhrzeiten: ."),
3),
id="SOC limit reached but scheduled SOC not, no further loading hours"),
pytest.param(False, 79, 80, 90, LOADING_HOURS_TODAY,
(
6,
"pv_charging",
ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR.format(
"Geladen wird zu folgenden Uhrzeiten: 8:00."),
0),
id="expensive_hour_charge_with_pv"),
pytest.param(False, 79, 80, 70, LOADING_HOURS_TODAY,
(
0,
"stop",
ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC.format(
"Geladen wird zu folgenden Uhrzeiten: 8:00."),
3),
id="expensive_hour_no_charge_with_pv "),
pytest.param(False, 79, 80, 70, LOADING_HOURS_TODAY + LOADING_HOURS_TOMORROW,
(
0,
"stop",
ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC.format(
"Geladen wird zu folgenden Uhrzeiten: 8:00, morgen 8:00."),
3),
id="expensive_hour_no_charge_with_pv scheduled for tomorrow"),
pytest.param(False, 79, 60, 80, LOADING_HOURS_TODAY,
(
6,
"pv_charging",
ChargeTemplate.SCHEDULED_CHARGING_REACHED_SCHEDULED_SOC.format(
""),
0),
id="expensive_hour_pv_charging"),
pytest.param(False, 79, 60, 50, LOADING_HOURS_TODAY,
(
0,
"stop",
ChargeTemplate.SCHEDULED_CHARGING_REACHED_MAX_AND_LIMIT_SOC.format(
""),
3),
id="scheduled and limit SOC reached"),
])
def test_scheduled_charging_calc_current_electricity_tariff(loading_hour, expected, monkeypatch):
def test_scheduled_charging_calc_current_electricity_tariff(
is_loading_hour, current_soc, soc_scheduled, sco_limit, loading_hours, expected, monkeypatch):
# setup
datetime_mock = Mock(wraps=datetime.datetime)
datetime_mock.now.return_value = datetime.datetime.fromtimestamp(LOADING_HOURS_TODAY[0])
monkeypatch.setattr(datetime, "datetime", datetime_mock)

ct = ChargeTemplate()
plan = ScheduledChargingPlan(active=True)
plan = ScheduledChargingPlan(active=True,
limit=ScheduledLimit(selected="soc", soc_scheduled=soc_scheduled, soc_limit=sco_limit))
plan.et_active = True
plan.limit.selected = "soc"
ct.data.chargemode.scheduled_charging.plans = [plan]
# für Github-Test keinen Zeitstempel verwenden
mock_et_get_loading_hours = Mock(return_value=[datetime.datetime(
year=2022, month=5, day=16, hour=8, minute=0).timestamp()])
mock_et_get_loading_hours = Mock(return_value=loading_hours)
monkeypatch.setattr(data.data.optional_data, "et_get_loading_hours", mock_et_get_loading_hours)
mock_is_list_valid = Mock(return_value=loading_hour)
monkeypatch.setattr(timecheck, "is_list_valid", mock_is_list_valid)
mock_is_list_valid = Mock(return_value=is_loading_hour)
monkeypatch.setattr(data.data.optional_data, "et_is_charging_allowed_hours_list", mock_is_list_valid)

# execution
ret = ct.scheduled_charging_calc_current(SelectedPlan(
plan=plan, remaining_time=301, phases=3, duration=3600),
79, 0, 3, 6, 0, ChargingType.AC.value, EvTemplate(), BidiState.BIDI_CAPABLE)
ret = ct.scheduled_charging_calc_current(
SelectedPlan(plan=plan, remaining_time=301, phases=3, duration=3600),
current_soc, 0, 3, 6, 0, ChargingType.AC.value, EvTemplate(), BidiState.BIDI_CAPABLE)

# evaluation
assert ret == expected
Loading