Initial commit
This commit is contained in:
Constantin Pascal
+132
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
Custom integration for heat pump monitoring via web interface.
|
||||
Place this code in custom_components/heat_pump_monitor/__init__.py
|
||||
"""
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
|
||||
"""Set up the Heat Pump Monitor component."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
return True
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
return unload_ok
|
||||
import asyncio
|
||||
import logging
|
||||
import aiohttp
|
||||
from datetime import timedelta
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_URL,
|
||||
CONF_SCAN_INTERVAL,
|
||||
UnitOfTemperature,
|
||||
TIME_HOURS,
|
||||
)
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
UpdateFailed,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
DOMAIN = "heat_pump_monitor"
|
||||
DEFAULT_SCAN_INTERVAL = 3
|
||||
|
||||
PLATFORMS = ["sensor", "binary_sensor"]
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up heat pump monitor from a config entry."""
|
||||
url = entry.data[CONF_URL]
|
||||
scan_interval = entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
|
||||
|
||||
coordinator = HeatPumpDataCoordinator(
|
||||
hass,
|
||||
_LOGGER,
|
||||
url=url,
|
||||
update_interval=timedelta(seconds=scan_interval),
|
||||
)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][entry.entry_id] = coordinator
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
class HeatPumpDataCoordinator(DataUpdateCoordinator):
|
||||
"""Class to manage fetching heat pump data."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
logger: logging.Logger,
|
||||
url: str,
|
||||
update_interval: timedelta,
|
||||
) -> None:
|
||||
"""Initialize."""
|
||||
super().__init__(
|
||||
hass,
|
||||
logger,
|
||||
name=DOMAIN,
|
||||
update_interval=update_interval,
|
||||
)
|
||||
self.url = url
|
||||
|
||||
async def _async_update_data(self):
|
||||
"""Fetch data from heat pump."""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(self.url) as response:
|
||||
if response.status != 200:
|
||||
raise UpdateFailed(f"Error fetching data: {response.status}")
|
||||
data = await response.json()
|
||||
|
||||
if not data.get("success", False):
|
||||
raise UpdateFailed("Heat pump returned unsuccessful response")
|
||||
|
||||
status = data.get("RESULT", {}).get("status", {})
|
||||
return {
|
||||
"software_version": data.get("sw", {}).get("V"),
|
||||
"last_update": data.get("RESULT", {}).get("TSLastUpdate"),
|
||||
"datetime": data.get("RESULT", {}).get("info", {}).get("datetime"),
|
||||
# Temperature readings
|
||||
# heating smart setpoint?
|
||||
"inlet_setpoint_temp": float(status.get("ISP", "0").split()[0]),
|
||||
# sanitary setpoint?
|
||||
"system_setpoint_temp": status.get("SSP"),
|
||||
"water_temp_t1": status.get("watert1"),
|
||||
"water_temp_t2": status.get("watert2"),
|
||||
"dhw_temp_t3": status.get("dhwt3"),
|
||||
"outdoor_temp_t4": status.get("t4"),
|
||||
# Setpoints
|
||||
# heating water normal setpoint
|
||||
"water_setpoint": status.get("watersetpoint"),
|
||||
"dhw_setpoint": status.get("dhwsetpoint"),
|
||||
# Operation states
|
||||
"is_heating": status.get("heating", False),
|
||||
"is_working": status.get("pwr", False),
|
||||
"is_cooling": status.get("cooling", False),
|
||||
"is_dhw": status.get("san", False),
|
||||
"is_pump_active": status.get("pump", False),
|
||||
"is_aux_heater" : status.get("res",False),
|
||||
"aux_heater": status.get("auxheater", 0),
|
||||
"is_standby": status.get("standby", False),
|
||||
"disabled_prod": status.get("disableProd",False),
|
||||
"sce": status.get("sce",False),
|
||||
# Operation times
|
||||
"worktime_pdc": status.get("worktime_pdc", 0),
|
||||
"worktime_res": status.get("worktime_res", 0),
|
||||
# Alarm status
|
||||
"has_alarm": status.get("alarm", {}).get("status", False),
|
||||
"alarm_type": status.get("alarm", {}).get("type", ""),
|
||||
"alarm_label": status.get("alarm", {}).get("label", ""),
|
||||
}
|
||||
except aiohttp.ClientError as err:
|
||||
raise UpdateFailed(f"Error communicating with heat pump: {err}")
|
||||
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
Place this code in custom_components/heat_pump_monitor/binary_sensor.py
|
||||
"""
|
||||
from homeassistant.components.binary_sensor import (
|
||||
BinarySensorEntity,
|
||||
BinarySensorDeviceClass,
|
||||
)
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from RDZ_PDC import DOMAIN, HeatPumpDataCoordinator
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
"""Set up binary sensors."""
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
|
||||
entities = [
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_heating", "Heating", BinarySensorDeviceClass.HEAT),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_working", "Working", BinarySensorDeviceClass.RUNNING),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_cooling", "Cooling", BinarySensorDeviceClass.COLD),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_dhw", "DHW Mode", BinarySensorDeviceClass.RUNNING),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_pump_active", "Water Pump", BinarySensorDeviceClass.RUNNING),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_aux_heater", "Auxiliary Heater", BinarySensorDeviceClass.HEAT),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "is_standby", "Standby", BinarySensorDeviceClass.POWER),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "disabled_prod", "Disabled Prod", BinarySensorDeviceClass.POWER),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "sce", "SCE", BinarySensorDeviceClass.POWER),
|
||||
HeatPumpOperatingBinarySensor(coordinator, "has_alarm", "Alarm", BinarySensorDeviceClass.PROBLEM),
|
||||
]
|
||||
async_add_entities(entities)
|
||||
|
||||
class HeatPumpOperatingBinarySensor(CoordinatorEntity, BinarySensorEntity):
|
||||
"""Binary sensor for heat pump operation states."""
|
||||
|
||||
def __init__(self, coordinator, sensor_type, name, device_class):
|
||||
"""Initialize the binary sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.sensor_type = sensor_type
|
||||
self._attr_name = f"Heat Pump {name}"
|
||||
self._attr_unique_id = f"{DOMAIN}_{sensor_type}"
|
||||
self._attr_device_class = device_class
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if the binary sensor is on."""
|
||||
return self.coordinator.data.get(self.sensor_type, False)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""
|
||||
Place this code in custom_components/heat_pump_monitor/config_flow.py
|
||||
"""
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_URL, CONF_SCAN_INTERVAL
|
||||
import voluptuous as vol
|
||||
|
||||
class HeatPumpMonitorConfigFlow(config_entries.ConfigFlow, domain="heat_pump_monitor"):
|
||||
"""Handle a config flow."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(
|
||||
title="Heat Pump Monitor",
|
||||
data={
|
||||
CONF_URL: user_input[CONF_URL],
|
||||
CONF_SCAN_INTERVAL: user_input.get(CONF_SCAN_INTERVAL, 3),
|
||||
},
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_URL): str,
|
||||
vol.Optional(CONF_SCAN_INTERVAL, default=3): int,
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
cards:
|
||||
- type: heading
|
||||
heading: History
|
||||
- title: History
|
||||
type: history-graph
|
||||
hours_to_show: 3
|
||||
entities:
|
||||
- binary_sensor.heat_pump_alarm
|
||||
- sensor.heat_pump_auxiliary_heater_2
|
||||
- binary_sensor.heat_pump_auxiliary_heater
|
||||
- binary_sensor.heat_pump_cooling
|
||||
- binary_sensor.heat_pump_dhw_mode
|
||||
- sensor.heat_pump_dhw_setpoint
|
||||
- sensor.heat_pump_dhw_temperature_t3
|
||||
- binary_sensor.heat_pump_disabled_prod
|
||||
- sensor.heat_pump_water_temperature_t1
|
||||
- sensor.heat_pump_inlet_setpoint_temperature
|
||||
- sensor.heat_pump_outdoor_temperature_t4
|
||||
- sensor.heat_pump_water_temperature_t2
|
||||
- binary_sensor.heat_pump_heating
|
||||
- binary_sensor.heat_pump_sce
|
||||
- binary_sensor.heat_pump_standby
|
||||
- sensor.heat_pump_system_setpoint_temperature
|
||||
- binary_sensor.heat_pump_pump
|
||||
- sensor.heat_pump_water_setpoint
|
||||
- binary_sensor.heat_pump_working
|
||||
Executable
+12
@@ -0,0 +1,12 @@
|
||||
|
||||
{
|
||||
"domain": "heat_pump_monitor",
|
||||
"name": "Heat Pump Monitor",
|
||||
"documentation": "https://github.com/your_username/heat_pump_monitor",
|
||||
"dependencies": [],
|
||||
"codeowners": ["@your_username"],
|
||||
"requirements": [],
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling",
|
||||
"version": "1.0.0"
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
"""
|
||||
Place this code in custom_components/heat_pump_monitor/sensor.py
|
||||
"""
|
||||
from homeassistant.components.sensor import (
|
||||
SensorEntity,
|
||||
SensorDeviceClass,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.const import UnitOfTemperature, UnitOfTime
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from RDZ_PDC import DOMAIN, HeatPumpDataCoordinator
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
"""Set up sensors."""
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
|
||||
entities = [
|
||||
HeatPumpTemperatureSensor(coordinator, "inlet_setpoint_temp", "Inlet Setpoint Temperature"),
|
||||
HeatPumpTemperatureSensor(coordinator, "system_setpoint_temp", "System Setpoint Temperature"),
|
||||
HeatPumpTemperatureSensor(coordinator, "water_temp_t1", "Inflow Temperature T1"),
|
||||
HeatPumpTemperatureSensor(coordinator, "water_temp_t2", "Outflow Temperature T2"),
|
||||
HeatPumpTemperatureSensor(coordinator, "dhw_temp_t3", "DHW Temperature T3"),
|
||||
HeatPumpTemperatureSensor(coordinator, "outdoor_temp_t4", "Outdoor Temperature T4"),
|
||||
HeatPumpTemperatureSensor(coordinator, "water_setpoint", "Water Setpoint"),
|
||||
HeatPumpTemperatureSensor(coordinator, "dhw_setpoint", "DHW Setpoint"),
|
||||
HeatPumpTemperatureSensor(coordinator, "aux_heater", "Auxiliary Heater"),
|
||||
HeatPumpTimeSensor(coordinator, "worktime_pdc", "Runtime"),
|
||||
HeatPumpTimeSensor(coordinator, "worktime_res", "Auxiliary Heater Runtime"),
|
||||
HeatPumpVersionSensor(coordinator, "software_version", "Software Version"),
|
||||
]
|
||||
async_add_entities(entities)
|
||||
|
||||
class HeatPumpTemperatureSensor(CoordinatorEntity, SensorEntity):
|
||||
"""Temperature sensor for heat pump."""
|
||||
|
||||
def __init__(self, coordinator, sensor_type, name):
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.sensor_type = sensor_type
|
||||
self._attr_name = f"Heat Pump {name}"
|
||||
self._attr_unique_id = f"{DOMAIN}_{sensor_type}"
|
||||
self._attr_device_class = SensorDeviceClass.TEMPERATURE
|
||||
self._attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
|
||||
self._attr_state_class = SensorStateClass.MEASUREMENT
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self.coordinator.data.get(self.sensor_type)
|
||||
|
||||
class HeatPumpTimeSensor(CoordinatorEntity, SensorEntity):
|
||||
"""Time sensor for heat pump operation hours."""
|
||||
|
||||
def __init__(self, coordinator, sensor_type, name):
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.sensor_type = sensor_type
|
||||
self._attr_name = f"Heat Pump {name}"
|
||||
self._attr_unique_id = f"{DOMAIN}_{sensor_type}"
|
||||
self._attr_device_class = SensorDeviceClass.DURATION
|
||||
self._attr_native_unit_of_measurement = UnitOfTime.HOURS
|
||||
self._attr_state_class = SensorStateClass.TOTAL_INCREASING
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
# Convert the value from hours to seconds for Home Assistant
|
||||
hours = self.coordinator.data.get(self.sensor_type, 0)
|
||||
return hours
|
||||
|
||||
class HeatPumpVersionSensor(CoordinatorEntity, SensorEntity):
|
||||
"""Software version sensor for heat pump."""
|
||||
|
||||
def __init__(self, coordinator, sensor_type, name):
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.sensor_type = sensor_type
|
||||
self._attr_name = f"Heat Pump {name}"
|
||||
self._attr_unique_id = f"{DOMAIN}_{sensor_type}"
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self.coordinator.data.get(self.sensor_type)
|
||||
@@ -0,0 +1,21 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Heat Pump Monitor",
|
||||
"description": "Set up heat pump monitoring",
|
||||
"data": {
|
||||
"url": "Heat Pump URL",
|
||||
"scan_interval": "Update interval (seconds)"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect",
|
||||
"invalid_url": "Invalid URL"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured"
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user