AppDaemon App - Thermal Power Meter

thermal_power_meter.py

import appdaemon.plugins.hass.hassapi as hass

class ThermalPowerMeter(hass.Hass):
    def initialize(self):
        # Configuration from app arguments
        self._supply_temperature = self.args["supply_temperature"]
        self._return_temperature = self.args.get("return_temperature", None)
        self._return_difference = self.args.get("return_difference", 10)

        self.power_loss = self.args.get("power_loss", 11) # in percent

        self.hydronic = self.args.get("hydronic", [])

        self._power = self.args["power"]
        self._friendly_name = self.args["friendly_name"]

        self.water_specific_heat = 4186  # J/(kg*K)
        self.water_specific_density = 1000 # kg/m³

        if self.get_state(self._power) is None:
            self.power = 0

        # Listen for changes in supply and return temperature sensors
        self.listen_state(self.state_changed, self._supply_temperature)
        
        if self._return_temperature is not None:
            self.listen_state(self.state_changed, self._return_temperature)

        for pump in self.hydronic:
            self.listen_state(self.state_changed, pump['entity_id'])


        self.calculate_power()

    @property
    def supply_temperature(self):
        return self.get_float_state(self.get_state(self._supply_temperature))

    @property
    def return_temperature(self):
        if self._return_temperature is None:
            return self.supply_temperature - self._return_difference

        return self.get_float_state(self.get_state(self._return_temperature))

    # power output
    @property
    def power(self):
        return self.get_float_state(self.get_state(self._power), 0.0)

    @power.setter
    def power(self, value):
        value = round(value, 3) if value is not None else 0.00
        self.set_state(self._power, state=value, attributes={"unit_of_measurement": "W", "friendly_name": self._friendly_name})

    # calculated delta
    @property
    def delta(self):
        return self.get_float_state(self.get_state(self._power, attribute="delta"), 0.0)

    @delta.setter
    def delta(self, value):
        value = round(value, 3) if value is not None else 0.00
        self.set_state(self._power, attributes={"delta": value})

    #calculated flow rate
    @property
    def flow_rate(self):
        return self.get_float_state(self.get_state(self._power, attribute="flow_rate"), 0.0)

    @flow_rate.setter
    def flow_rate(self, value):
        value = round(value, 6) if value is not None else 0.00
        self.set_state(self._power, attributes={"flow_rate": value})


    def update_flow_rate(self):
        flow_rate = 0
        # first all series
        for pump in self.hydronic:
            if self.get_state(pump['entity_id']) == "on":
                pump_flow_rate = pump['flow_rate']
                if 'mixer' in pump:
                    mixer = self.get_float_state(self.get_state(pump['mixer']), 100)
                    mixer = max(self.power_loss, mixer)
                    pump_flow_rate = pump_flow_rate * (mixer / 100)

                if 'mode' not in pump or pump['mode'] == "series":
                    flow_rate = flow_rate + pump_flow_rate

        # then parallel
        for pump in self.hydronic:
            if self.get_state(pump['entity_id']) == "on":
                pump_flow_rate = pump['flow_rate']
                if 'mixer' in pump:
                    mixer = self.get_float_state(self.get_state(pump['mixer']), 100)
                    mixer = max(self.power_loss, mixer)
                    pump_flow_rate = pump_flow_rate * (mixer / 100)

                if 'mode' in pump and pump['mode'] == "parallel":
                    flow_rate = max(flow_rate, pump_flow_rate)


        self.flow_rate = flow_rate

    def state_changed(self, entity, attribute, old, new, kwargs):
        self.calculate_power()

    def calculate_power(self):
        self.update_flow_rate()

        supply_temp = self.supply_temperature
        return_temp = self.return_temperature

        if supply_temp is not None and return_temp is not None:
            # Calculate power: P = V_dot * p * c * delta_T
            self.delta = (supply_temp - return_temp)
            self.power = (self.flow_rate / 3600) * self.water_specific_heat * self.water_specific_density * self.delta
        else:
            self.power = 0

    def get_float_state(self, value, default=None):
        try:
            return float(value)
        except (TypeError, ValueError):
            return default

thermal_power_meter.yaml

living_heating_power_meter:
  module: thermal_power_meter
  class: ThermalPowerMeter
  supply_temperature: sensor.wohnen_vl
  return_temperature: sensor.wohnen_rl
  hydronic:
    - entity_id: binary_sensor.heizkreis_wohnen
      flow_rate: 1.5
  power: sensor.living_heating_power
  friendly_name: Heizkreis Wohnen Leistung