Guten Abend, ich habe es gerade in zusammenarbeit mit Claude geschafft!!! Hat aber jetzt auch gut 1 Jahr gebraucht und es musste erst Sonnet 3.7 veröffentlicht werden, damit es funktioniert. Ich möchte euch an der Lösung Teilgaben lassen, vll. ist diese auch für andere hilfreich.
trompreis-Optimierung für Home Assistant
Diese Lösung ermöglicht die automatische Optimierung des Waschmaschinen-Starts basierend auf Tibber-Strompreisen und PV-Erzeugungsprognosen. Sie findet den günstigsten Zeitpunkt in den kommenden 4 Stunden, wobei auch Slots nach Mitternacht berücksichtigt werden.
Voraussetzungen
Diese Lösung benötigt folgende Komponenten:
- Home Assistant mit aktiviertem
python_script integration
- Tibber-Integration für Strompreisdaten
- Solcast-Integration für PV-Erzeugungsprognosen
- Fingerbot oder ein anderes schaltbares Gerät zur Steuerung der Waschmaschine
Installation
1. Python Script einrichten
Erstelle eine Datei fixed_washing_machine_optimizer.py im Ordner /config/python_scripts/ mit folgendem Inhalt:
# fixed_washing_machine_optimizer.py
#
# Version mit Minuten-Berücksichtigung und 4-Stunden-Zeitfenster inkl. Tageswechsel
tibber_sensor = "sensor.tibber_prices"
pv_sensor_heute = "sensor.solcast_pv_forecast_prognose_heute"
pv_sensor_morgen = "sensor.solcast_pv_forecast_prognose_morgen"
fingerbot_switch = "switch.DEINE_FINGERBOT_ID" # <-- Hier deine Fingerbot-ID einsetzen
input_datetime_entity = "input_datetime.waschmaschine_startzeit"
# Waschmaschinen-Verbrauch (kWh pro Stunde für einen 2-Stunden-Waschgang)
wash_consumption_per_hour = 0.5 # <-- An deinen Waschmaschinenverbrauch anpassen
# Dynamische Berechnung der aktuellen Zeit
current_time = dt_util.now()
AKTUELLE_STUNDE = current_time.hour
AKTUELLE_MINUTE = current_time.minute
# Zeitfenster-Konfiguration
MAX_STUNDEN_ZEITFENSTER = 4 # Zeitfenster von 4 Stunden für die Optimierung
MIN_MINUTEN_ABSTAND = 15 # Minimaler Abstand in Minuten zur aktuellen Zeit
# Berücksichtigung der Tibber-Aktualisierungszeit (13 Uhr)
TIBBER_UPDATE_HOUR = 13
# Vor 13 Uhr könnten die Daten für morgen noch nicht vollständig sein
MORGEN_DATEN_VERFUEGBAR = (AKTUELLE_STUNDE >= TIBBER_UPDATE_HOUR)
# Startmeldung zur Kontrolle (nur kurz anzeigen, dass der Vorgang startet)
hass.services.call("persistent_notification", "create", {
"title": "Waschmaschine - Optimierung gestartet",
"message": f"Optimierung für die nächsten {MAX_STUNDEN_ZEITFENSTER} Stunden läuft...\nAktuelle Zeit: {AKTUELLE_STUNDE}:{AKTUELLE_MINUTE:02d} Uhr"
})
try:
# Hilfsfunktion zur Normalisierung von Zeitstempeln
def normalize_date_hour(date_str):
"""Extrahiert Datum und Stunde aus verschiedenen Zeitstempelformaten."""
try:
normalized = date_str.replace("T", " ")
date_part = normalized[0:10]
hour_part = normalized[11:13]
minute_part = normalized[14:16] # Auch Minuten extrahieren
return date_part, hour_part, minute_part
except:
return None, None, None
# 1. Tibber-Daten abrufen
tibber_state = hass.states.get(tibber_sensor)
if tibber_state is None:
hass.services.call("persistent_notification", "create", {
"title": "Fehler - Tibber-Sensor",
"message": f"Tibber-Sensor '{tibber_sensor}' nicht gefunden!"
})
raise Exception(f"Tibber-Sensor '{tibber_sensor}' nicht gefunden!")
# Preisdaten laden
price_data_today = tibber_state.attributes.get("today", [])
price_data_tomorrow = tibber_state.attributes.get("tomorrow", [])
# Wenn keine Daten vorhanden
if not price_data_today and not price_data_tomorrow:
hass.services.call("persistent_notification", "create", {
"title": "Warnung - Tibber-Daten",
"message": "Keine Preisdaten verfügbar!"
})
raise Exception("Keine Tibber-Preisdaten verfügbar!")
# Warnung nur, wenn relevant
if not MORGEN_DATEN_VERFUEGBAR and (AKTUELLE_STUNDE + MAX_STUNDEN_ZEITFENSTER > 24) and price_data_tomorrow:
hass.services.call("persistent_notification", "create", {
"title": "Hinweis - Tibber-Daten",
"message": "Daten für morgen werden berücksichtigt, könnten aber noch unvollständig sein (Tibber aktualisiert erst ab 13 Uhr)."
})
# Wichtig: Aktuelles und morgiges Datum aus Tibber extrahieren
aktuelles_datum = None
morgen_datum = None
if price_data_today and len(price_data_today) > 0:
first_slot = price_data_today[0]
if first_slot and "startsAt" in first_slot:
aktuelles_datum, _, _ = normalize_date_hour(first_slot["startsAt"])
if price_data_tomorrow and len(price_data_tomorrow) > 0:
first_slot = price_data_tomorrow[0]
if first_slot and "startsAt" in first_slot:
morgen_datum, _, _ = normalize_date_hour(first_slot["startsAt"])
# Preisdaten kombinieren
price_data = []
for item in price_data_today:
price_data.append(item)
for item in price_data_tomorrow:
price_data.append(item)
# 2. Solcast-Daten abrufen
pv_state_heute = hass.states.get(pv_sensor_heute)
pv_state_morgen = hass.states.get(pv_sensor_morgen)
if (pv_state_heute is None) or (pv_state_morgen is None):
hass.services.call("persistent_notification", "create", {
"title": "Fehler - Solcast-Sensoren",
"message": "Mindestens einer der Solcast-Sensoren nicht gefunden!"
})
raise Exception("Mindestens einer der Solcast-Sensoren nicht gefunden!")
# PV-Daten laden
pv_data_today = pv_state_heute.attributes.get("detailedForecast", [])
pv_data_tomorrow = pv_state_morgen.attributes.get("detailedForecast", [])
# Sicherstellen, dass es Listen sind
pv_data = []
for entry in pv_data_today:
pv_data.append(entry)
for entry in pv_data_tomorrow:
pv_data.append(entry)
# 3. Optimalen Zeitraum berechnen
best_start_time = None
lowest_cost = 9999999.9
best_average_price = 0.0
best_debug_info = None
# Debug-Hilfsvariablen
total_slots = 0
future_slots = 0
matching_slots = 0
# Solcast-Daten in ein Lookup-Dictionary umwandeln für schnelleren Zugriff
pv_lookup = {}
for entry in pv_data:
if entry and "period_start" in entry:
ps_str = str(entry["period_start"])
date_part, hour_part, _ = normalize_date_hour(ps_str)
if date_part and hour_part:
pv_lookup[f"{date_part} {hour_part}"] = entry.get("pv_estimate", 0)
# Hilfsfunktion zum Prüfen, ob ein Zeitpunkt im Zeitfenster liegt und nicht zu nah an der aktuellen Zeit
def is_in_time_window(date_part, hour_str, minute_str):
"""Prüft, ob ein Zeitpunkt im definierten Zeitfenster liegt und genügend in der Zukunft."""
try:
hour_part = int(hour_str)
minute_part = int(minute_str)
if not aktuelles_datum:
return True # Keine Datumsprüfung möglich
# Fall 1: Der Slot ist am aktuellen Tag
if date_part == aktuelles_datum:
# Wenn die Stunde gleich der aktuellen ist, prüfen wir den Minutenabstand
if hour_part == AKTUELLE_STUNDE:
# Nur erlauben, wenn genügend Minuten in der Zukunft (mindestens MIN_MINUTEN_ABSTAND)
return minute_part >= AKTUELLE_MINUTE + MIN_MINUTEN_ABSTAND
# Ansonsten prüfen, ob die Stunde im Zeitfenster liegt
if hour_part > AKTUELLE_STUNDE and hour_part < AKTUELLE_STUNDE + MAX_STUNDEN_ZEITFENSTER:
return True
return False
# Fall 2: Tageswechsel - das Fenster geht über Mitternacht
if morgen_datum and date_part == morgen_datum:
# Wenn die aktuelle Zeit + Zeitfenster über 24 Stunden hinausgeht,
# dann sind frühe Stunden des nächsten Tages auch relevant
max_heute = AKTUELLE_STUNDE + MAX_STUNDEN_ZEITFENSTER
if max_heute > 24:
# Wie viele Stunden gehen in den nächsten Tag?
stunden_morgen = max_heute - 24
# Ist die Stunde im nächsten Tag innerhalb dieses Fensters?
if hour_part < stunden_morgen:
return True
return False
except:
return False # Im Zweifelsfall ablehnen
# Über alle 2-Stunden-Blöcke iterieren
for i in range(len(price_data) - 1):
try:
total_slots += 1
hour1_info = price_data[i]
hour2_info = price_data[i+1]
# Nur fortfahren, wenn es sich um gültige Objekte handelt
if not hour1_info or not hour2_info:
continue
# Prüfen, ob die notwendigen Schlüssel existieren
if "total" not in hour1_info or "total" not in hour2_info:
continue
if "startsAt" not in hour1_info or "startsAt" not in hour2_info:
continue
hour1_price = hour1_info.get("total", 0)
hour2_price = hour2_info.get("total", 0)
hour1_str = hour1_info.get("startsAt")
hour2_str = hour2_info.get("startsAt")
if (hour1_str is None) or (hour2_str is None):
continue
# Datum+Stunde+Minute extrahieren
tibber_date1, tibber_hour1, tibber_minute1 = normalize_date_hour(hour1_str)
tibber_date2, tibber_hour2, tibber_minute2 = normalize_date_hour(hour2_str)
if (not tibber_date1) or (not tibber_date2):
continue
# NUR ZEITPUNKTE IM ZEITFENSTER BERÜCKSICHTIGEN UND MINUTENABSTAND PRÜFEN
time_in_window = is_in_time_window(tibber_date1, tibber_hour1, tibber_minute1)
if not time_in_window:
continue
future_slots += 1
# Normalisierte Zeitschlüssel erstellen
time_key1 = f"{tibber_date1} {tibber_hour1}"
time_key2 = f"{tibber_date2} {tibber_hour2}"
# PV-Produktion für die beiden Stunden aus dem Lookup-Dictionary holen
hour1_pv = pv_lookup.get(time_key1, 0)
hour2_pv = pv_lookup.get(time_key2, 0)
# Wenn wir einen Match gefunden haben, zählen
if time_key1 in pv_lookup or time_key2 in pv_lookup:
matching_slots += 1
# Kostenberechnung
energy_needed1 = max(0, wash_consumption_per_hour - hour1_pv)
energy_needed2 = max(0, wash_consumption_per_hour - hour2_pv)
cost1 = hour1_price * energy_needed1
cost2 = hour2_price * energy_needed2
total_cost = cost1 + cost2
# Debug-Info für diesen Zeitraum
current_debug = {
"start": time_key1,
"end": time_key2,
"prices": [hour1_price, hour2_price],
"pv": [hour1_pv, hour2_pv],
"energy_needed": [energy_needed1, energy_needed2],
"costs": [cost1, cost2],
"total": total_cost
}
if total_cost < lowest_cost:
lowest_cost = total_cost
best_start_time = hour1_str
best_average_price = (hour1_price + hour2_price) / 2.0
best_debug_info = current_debug
except Exception as e:
# Fehler in dieser Iteration - mit nächstem Zeitpunkt fortfahren
continue
# 4. Ergebnis verarbeiten
if best_start_time is None:
hass.services.call("persistent_notification", "create", {
"title": "Warnung - Keine Startzeit",
"message": f"Kein geeigneter Zeitraum in den nächsten {MAX_STUNDEN_ZEITFENSTER} Stunden gefunden. Bitte versuche es später erneut."
})
else:
# Zeitformat für input_datetime konvertieren
new_dt = best_start_time.replace("T", " ")
plus_idx = new_dt.find("+")
if plus_idx > 0:
new_dt = new_dt[:plus_idx]
# Timestamp für richtiges Anzeigeformat generieren
timestamp = dt_util.parse_datetime(new_dt).timestamp()
# Startzeit setzen
hass.services.call("input_datetime", "set_datetime", {
"entity_id": input_datetime_entity,
"datetime": new_dt
})
# Kalkulierten Preis als Attribut speichern (mit Timestamp für Anzeige)
hass.states.set(
input_datetime_entity,
new_dt,
{
"friendly_name": "Waschmaschine Startzeit",
"timestamp": timestamp,
"calculated_price": f"{best_average_price:.3f}",
"calculated_cost": f"{lowest_cost:.3f}"
}
)
# Erfolgsbenachrichtigung - nur das Wichtigste anzeigen
hass.services.call("persistent_notification", "create", {
"title": "Waschmaschine - Optimaler Zeitraum",
"message": (
f"Start: {new_dt}\n"
f"Durchschnittspreis: {best_average_price:.3f} €/kWh\n"
f"Kosten: {lowest_cost:.3f} €"
)
})
# Fingerbot einschalten
hass.services.call("switch", "turn_on", {
"entity_id": fingerbot_switch
})
except Exception as e:
# Hauptfehlerbehandlung
hass.services.call("persistent_notification", "create", {
"title": "Fehler bei der Berechnung",
"message": f"Fehler: {str(e)}"
})
2. Erforderliche Sensoren einrichten
Füge folgende Konfiguration zu deiner configuration.yaml oder einer separaten sensors.yaml hinzu:
# Tibber Preis-Sensor
sensor:
# Sensor für Tibber Preise
- platform: rest
name: Tibber Prices
resource: https://api.tibber.com/v1-beta/gql
method: POST
payload: '{ "query": "{ viewer { homes { currentSubscription { priceInfo { current { total, startsAt } today { total, startsAt } tomorrow { total, startsAt } } } } } }" }'
json_attributes_path: "$.data.viewer.homes[0].currentSubscription.priceInfo"
json_attributes:
- today
- tomorrow
value_template: "{{ value_json.data.viewer.homes[0].currentSubscription.priceInfo.current.total | float }}"
scan_interval: 30
headers:
Authorization: Bearer DEIN_TIBBER_API_TOKEN # <-- Ersetze durch deinen Tibber API-Token
Content-Type: application/json
User-Agent: REST
unit_of_measurement: EUR/kWh
# Template-Sensor für bessere Darstellung der Startzeit
- platform: template
sensors:
waschmaschine_startzeit_anzeige:
friendly_name: "Waschmaschine Startzeit"
value_template: >
{% if state_attr('input_datetime.waschmaschine_startzeit', 'timestamp') %}
{% set timestamp = state_attr('input_datetime.waschmaschine_startzeit', 'timestamp') | int %}
{% set time = timestamp | timestamp_custom('%d.%m.%Y %H:%M Uhr') %}
{{ time }}
{% else %}
{{ states('input_datetime.waschmaschine_startzeit') }}
{% endif %}
icon_template: "mdi:clock-outline"
# Optional: Sensor für Waschmaschine Preis
- platform: template
sensors:
waschmaschine_preis:
friendly_name: "Waschmaschine Strompreis"
unit_of_measurement: "€/kWh"
value_template: "{{ state_attr('input_datetime.waschmaschine_startzeit', 'calculated_price') | default(0) }}"
icon_template: "mdi:currency-eur"
# Erforderliches Datum/Zeit-Entity
input_datetime:
waschmaschine_startzeit:
name: Waschmaschine Startzeit
has_date: true
has_time: true
icon: mdi:washing-machine
3. Dashboard-Karte erstellen
Erstelle eine neue Karte in deinem Dashboard mit diesem Code:
alias: Waschmaschine Optimal Starten
description: >-
Berechnet den optimalen Zeitpunkt zum Starten der Waschmaschine basierend auf
Tibber-Preisen und PV-Produktion (nur manuell per Button)
triggers:
- entity_id: input_button.waschmaschine_optimieren
trigger: state
conditions:
- condition: template
value_template: "{{ states('sensor.tibber_prices') != 'unavailable' }}"
- condition: template
value_template: >-
{{ states('sensor.solcast_pv_forecast_prognose_heute') != 'unavailable'
and states('sensor.solcast_pv_forecast_prognose_morgen') != 'unavailable'
}}
actions:
- action: python_script.fixed_washing_machine_optimizer
data: {}
mode: single
max_exceeded: silent
Anpassung
Ersetze folgende Werte in den Konfigurationsdateien:
DEINE_FINGERBOT_ID - Die Entity-ID deines Fingerbot-SchaltersDEIN_TIBBER_API_TOKEN - Dein persönlicher API-Token von Tibberwash_consumption_per_hour - Der stündliche Stromverbrauch deiner Waschmaschine (Standard: 0.5 kWh)
Solcast-Integration
Für die PV-Prognose wird die Solcast-Integration benötigt. Stelle sicher, dass die folgenden Sensoren verfügbar sind (oder passe die Namen im Python-Script an):
sensor.solcast_pv_forecast_prognose_heutesensor.solcast_pv_forecast_prognose_morgen
Funktionsweise
- Dieses Script analysiert die Strompreise von Tibber für die nächsten 4 Stunden
- Es berücksichtigt die prognostizierte PV-Erzeugung von Solcast
- Es berechnet die optimale Startzeit für die Waschmaschine (günstigster Gesamtpreis)
- Es aktiviert dann automatisch den Fingerbot zur ermittelten Zeit
Features
Dynamische Berechnung des 4-Stunden-Zeitfensters- Berücksichtigung von Tageswechseln (über Mitternacht hinaus)
- PV-Überschuss wird in die Kostenberechnung einbezogen
- Automatisches Ein-/Ausschalten des Fingerbots
- Übersichtliche Darstellung im Dashboard
Hinweise
- Tibber aktualisiert die Preise für den nächsten Tag erst gegen 13 Uhr
- Das Script berücksichtigt einen 2-stündigen Waschzyklus
- Der tatsächliche Energieverbrauch kann je nach Waschmaschine variieren
