Energieverbrauchssensoren - Technische Dokumentation¶

# Elektrische Sensoren
for hp_idx in range(1, num_hps + 1):
    for mode in ENERGY_CONSUMPTION_MODES:
        for period in ENERGY_CONSUMPTION_PERIODS:
            sensor_id = f"{mode}_energy_{period}"
            sensor = LambdaEnergyConsumptionSensor(...)

# Thermische Sensoren
for hp_idx in range(1, num_hps + 1):
    for mode in ENERGY_CONSUMPTION_MODES:
        for period in ENERGY_CONSUMPTION_PERIODS:
            sensor_id = f"{mode}_thermal_energy_{period}"
            if sensor_template.get("data_type") == "thermal_calculated":
                sensor = LambdaEnergyConsumptionSensor(...)

ENERGY_CONSUMPTION_SENSOR_TEMPLATES = {
    # Elektrische Sensoren
    "heating_energy_total": {
        "name": "Heating Energy Total",
        "unit": "kWh",
        "data_type": "energy_calculated",
        "state_class": "total_increasing",
        "device_class": "energy",
        "operating_state": "heating",
        "period": "total",
    },
    # Thermische Sensoren
    "heating_thermal_energy_total": {
        "name": "Heating Thermal Energy Total",
        "unit": "kWh",
        "data_type": "thermal_calculated",
        "state_class": "total_increasing",
        "device_class": "energy",
        "operating_state": "heating",
        "period": "total",
    },
    # ...
}

async def _track_hp_energy_consumption(self, hp_idx, current_state, data):
    # Elektrische Energie
    await self._track_hp_energy_type_consumption(
        hp_idx, current_state, data,
        sensor_type="electrical",
        default_sensor_id_template="sensor.{name_prefix}_hp{hp_idx}_compressor_power_consumption_accumulated",
        increment_fn=self._increment_energy_consumption
    )

    # Thermische Energie
    await self._track_hp_energy_type_consumption(
        hp_idx, current_state, data,
        sensor_type="thermal",
        default_sensor_id_template="sensor.{name_prefix}_hp{hp_idx}_compressor_thermal_energy_output_accumulated",
        increment_fn=self._increment_thermal_energy_consumption
    )

async def _track_hp_energy_type_consumption(
    self, hp_idx, current_state, data, sensor_type, 
    default_sensor_id_template, unit_check_fn, convert_to_kwh_fn,
    last_reading_dict, first_value_seen_dict, increment_fn
):
    # 1. Lese Quellsensor
    current_energy_state = self.hass.states.get(sensor_entity_id)
    current_energy = float(current_energy_state.state)

    # 2. Konvertiere zu kWh
    current_energy_kwh = convert_to_kwh_fn(current_energy, unit)

    # 3. Berechne Delta
    last_energy = last_reading_dict.get(f"hp{hp_idx}", None)
    energy_delta = calculate_energy_delta(current_energy_kwh, last_energy, max_delta=100.0)

    # 4. Bestimme Betriebsmodus
    mode = mode_mapping[current_state]  # heating, hot_water, cooling, defrost

    # 5. Aktualisiere Sensoren bei Moduswechsel oder kontinuierlich
    if current_state != last_state or (mode == "stby" or energy_delta > 0):
        await increment_fn(hp_idx, mode, energy_delta)

async def increment_energy_consumption_counter(
    hass, mode, hp_index, energy_delta, name_prefix,
    use_legacy_modbus_names=True, energy_offsets=None,
    sensor_type="electrical"  # oder "thermal"
):
    for period in ["total", "daily", "monthly", "yearly", "2h", "4h"]:
        # Bestimme sensor_id basierend auf sensor_type
        if sensor_type == "thermal":
            sensor_id = f"{mode}_thermal_energy_{period}"
        else:
            sensor_id = f"{mode}_energy_{period}"

        # Finde Entity
        energy_entity = find_energy_entity(hass, entity_id)

        # Berechne neuen Wert
        current_value = float(state_obj.state)
        new_value = current_value + energy_delta

        # Wende Offset an (nur für total)
        if period == "total" and energy_offsets:
            new_value += offset

        # Aktualisiere Entity
        energy_entity.set_energy_value(new_value)

class LambdaEnergyConsumptionSensor(RestoreEntity, SensorEntity):
    def __init__(self, hass, entry, sensor_id, name, entity_id, 
                 unique_id, unit, state_class, device_class, 
                 device_type, hp_index, mode, period):
        self._energy_value = 0.0  # Total-Wert
        self._period = period
        # ...

    @property
    def native_value(self):
        """Berechnet Wert basierend auf period."""
        if self._period == "total":
            return self._energy_value
        elif self._period == "daily":
            return self._energy_value - self._yesterday_value
        elif self._period == "monthly":
            return self._energy_value - self._previous_monthly_value
        # ...

    def set_energy_value(self, value):
        """Wird von increment_energy_consumption_counter aufgerufen."""
        self._energy_value = value
        self.async_write_ha_state()

mode_mapping = {
    0: "stby",      # Standby
    1: "heating",   # CH - Heizen
    2: "hot_water", # DHW - Warmwasser
    3: "cooling",   # CC - Kühlen
    4: "stby",      # Standby (alternativ)
    5: "defrost",   # DEFROST - Abtauen
}

def _convert_energy_to_kwh_cached(self, value, unit):
    if unit == "Wh":
        return value / 1000.0
    elif unit == "kWh":
        return value
    elif unit == "MWh":
        return value * 1000.0
    else:
        return value  # Fallback

if current_energy_kwh == 0.0:
    first_value_seen_dict[f"hp{hp_idx}"] = False
    return  # Warte auf ersten gültigen Wert

if current_energy_kwh < last_energy:
    # Sensor wurde zurückgesetzt oder gewechselt
    first_value_seen_dict[f"hp{hp_idx}"] = False
    last_reading_dict[f"hp{hp_idx}"] = None
    return