Hi there!
I was a silent reader till now, but now i finally registered and thought i maybe can give something back to the community. I have a Kermi Heat Pump, but no Modbus. I’ve read a thread here where some other people searched for it too, but no solution.
So here’s what i did in summary:
- In installed the MQTT Integration in HA and configured it
- I have a Python Script on my HA-Server that logs in to Heat Pump Web Interface and grabs the Data from the API and writes it to MQTT (Autodiscovery and States)
- This script is run through automation and the shell_command integration every minute
The downsides of this integration are:
- You have to have specific items in your favorites in the heat pump web interface
- If the Web-Interface in any way changes (URLs, API-JSON-Strucutre, …) the script may break
- It’s only one-way, you only can read data, not write
- The Heat Pumps may be different in version and functions and some adjustments in the code may be neccessary, but the script should give you a good start to understand everything
If this is fine for you and you know about the things i talked about, here’s my scraper python-script:
import json
import requests
import paho.mqtt.client as mqtt
import time
import unicodedata
import re
# Konfigurationswerte
DEBUG = False # Auf True setzen, um Log-Ausgaben zu erhalten; sonst auf False
LOGIN_URL = "http://<ip-of-your-hp>/api/Security/Login"
DATA_URL = "http://<ip-of-your-hp>/api/Favorite/GetFavorites/00000000-0000-0000-0000-000000000000"
MQTT_BROKER = "core-mosquitto"
MQTT_PORT = 1883
MQTT_USERNAME = "<mqtt-username>" # Kann auch ein HA-Benutzer sein
MQTT_PASSWORD = "<mqtt-password>" # Kann auch ein HA-Benutzer sein
MQTT_TOPIC_PREFIX = "mqtt" # Sensorwerte
DISCOVERY_PREFIX = "mqtt-ad" # Auto-Discovery Prefix für HA
PASSWORD = "<webinterface-login-password>" # i.d.R. letzte 4 Stellen der Seriennummer
WP_DEVICE_NAME = "wp_kermi"
def normalize_key(name):
"""Erzeugt einen konsistenten Schlüssel: kleingeschrieben, ohne Umlaute und Sonderzeichen."""
name = name.lower()
name = unicodedata.normalize('NFKD', name).encode('ascii', 'ignore').decode('ascii')
name = re.sub(r'[\s-]+', '_', name)
name = re.sub(r'[^a-z0-9_]', '', name)
return name
def log_extracted_data(extracted):
"""Schreibt die vom Scraper verarbeiteten Sensorwerte (Name, Wert, Einheit) in die Log-Datei."""
if DEBUG:
try:
with open("kermi_scraper.log", "a", encoding="utf-8") as f:
f.write(json.dumps(extracted, ensure_ascii=False, indent=2))
f.write("\n\n")
except Exception as e:
print(f"Fehler beim Schreiben ins Log: {e}")
def login():
session = requests.Session()
payload = {"Password": PASSWORD}
headers = {"Content-Type": "application/json"}
response = session.post(LOGIN_URL, json=payload, headers=headers)
print(f"Login-Response Code: {response.status_code}")
if response.status_code != 200:
print(f"Fehler beim Login: {response.text}")
response.raise_for_status()
try:
data = response.json()
if not data.get("isValid", False):
print("Fehler: Login war nicht erfolgreich!")
return None
except json.JSONDecodeError:
print("Fehler: Keine gültige JSON-Antwort beim Login.")
return None
return session
def fetch_data(session):
payload = {"WithDetails": True, "OnlyHomeScreen": True}
headers = {"Content-Type": "application/json"}
response = session.post(DATA_URL, json=payload, headers=headers)
print(f"Datenabruf-Response Code: {response.status_code}")
if response.status_code != 200:
print(f"Fehler beim Abrufen der Daten: {response.text}")
response.raise_for_status()
try:
return response.json()
except json.JSONDecodeError:
print("Fehler: Die API hat keine gültige JSON-Antwort zurückgegeben.")
print(f"Antworttext: {response.text}")
return None
def extract_data(api_response):
"""Extrahiert Daten aus `ResponseData` und `VisualizationDatapoints`."""
all_sensors = []
# Daten aus `ResponseData`
for item in api_response.get("ResponseData", []):
if "$type" not in item or "DatapointValue" not in item or "DatapointConfig" not in item:
continue
config = item["DatapointConfig"]
value = round(item["DatapointValue"]["Value"], 2) if isinstance(item["DatapointValue"]["Value"], float) else item["DatapointValue"]["Value"]
name = config["DisplayName"]
unit = config.get("Unit", "")
all_sensors.append({
"name": name,
"value": value,
"unit": unit
})
# Zusätzliche Daten aus `VisualizationDatapoints`
for item in api_response.get("ResponseData", []):
if "VisualizationDatapoints" in item:
for datapoint in item["VisualizationDatapoints"].get("$values", []):
config = datapoint["Config"]
value = round(datapoint["DatapointValue"]["Value"], 2) if isinstance(datapoint["DatapointValue"]["Value"], float) else datapoint["DatapointValue"]["Value"]
name = config["DisplayName"]
unit = config.get("Unit", "")
all_sensors.append({
"name": name,
"value": value,
"unit": unit
})
return all_sensors
def send_mqtt(data):
if not data:
print("Keine Daten vorhanden, MQTT wird nicht gesendet.")
return
#client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) # Funktioniert nicht im HA docker container
client = mqtt.Client()
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
client.connect(MQTT_BROKER, MQTT_PORT, 60)
published = set()
extracted_sensors = [] # Hier werden die extrahierten Sensorwerte gesammelt
for item in data:
name = item["name"]
sensor_key = normalize_key(name)
if sensor_key in published:
continue
published.add(sensor_key)
value = item["value"]
unit = item.get("unit", "")
# Discovery-Nachricht erstellen
discovery_topic = f"{DISCOVERY_PREFIX}/sensor/{sensor_key}/config"
state_topic = f"{MQTT_TOPIC_PREFIX}/{sensor_key}"
config_payload = {
"name": f"Kermi: {name}",
"state_topic": state_topic,
"value_template": "{{ value_json.value }}",
"unit_of_measurement": unit,
"device_class": None,
"icon": "mdi:flash",
"unique_id": f"kermi_{sensor_key}",
"availability_topic": "mqtt/status",
"device": {
"name": WP_DEVICE_NAME,
"identifiers": [WP_DEVICE_NAME]
}
}
client.publish(discovery_topic, json.dumps(config_payload), retain=True)
print(f"Discovery gesendet: {discovery_topic}")
# Kurze Pause, damit HA die Discovery-Nachrichten verarbeiten kann
time.sleep(3)
# State-Nachrichten senden und extrahierte Sensorwerte sammeln
for item in data:
name = item["name"]
sensor_key = normalize_key(name)
state_topic = f"{MQTT_TOPIC_PREFIX}/{sensor_key}"
value = item["value"]
unit = item.get("unit", "")
client.publish(state_topic, json.dumps({"value": value, "unit": unit}), retain=False)
print(f"Sende MQTT: {state_topic} -> {value} {unit}")
extracted_sensors.append({"name": name, "value": value, "unit": unit})
log_extracted_data(extracted_sensors)
client.disconnect()
try:
session = login()
if session:
print("Login erfolgreich!")
api_response = fetch_data(session)
if api_response:
print("Daten erfolgreich abgerufen!")
extracted_data = extract_data(api_response)
send_mqtt(extracted_data)
print("Daten erfolgreich gesendet!")
else:
print("Fehler: Die API hat keine verwertbaren Daten geliefert.")
else:
print("Fehler: Kein gültiges Session-Objekt erhalten.")
except Exception as e:
print(f"Fehler: {e}")
Here’s an example how this could look in HA:
If you need help setting up, just ask 
Otherwise, have fun!
Greetings