App #4: Boiler Alert

Tags: #<Tag:0x00007f7c6e18f2d0>

Hello,

Another example :grinning:

This automation is not final or complete, and probably there is a way to make more generic. My aim is to share examples that maybe can help new members starting with HA+AppDaemon.

Suggestions or recommendations to improve the implementation are welcome!

App #4: Boiler Alert
Notify me if the boiler fires three Floor Heating Failures and one Internal Failure within one hour. Do not notify me more than once per hour.

For this automation, I use the same approach used in App #1: Doorbell notification to keep track of the last notification. I also use timers to create a sliding window effect and correlate failures occurred within the last hour.

Entities

sensor.boiler_alarm  

boiler_alert.yaml

boiler_alert:
  module: boiler_alert
  class: BoilerAlert
  sensor: sensor.boiler_alarm  
  failures:
    - name: fh_sensor_failure
      amount: 3
    - name: i_sensor_failure
      amount: 1

boiler_alert.py

import appdaemon.plugins.hass.hassapi as hass
from datetime import datetime, timedelta

class BoilerAlert(hass.Hass):

    def initialize(self):
        self.log('initializing ...')
        self.events = {}
        self.last_notification = datetime.now() - timedelta(hours= 2)
        sensor = self.args["sensor"]
        for failure in self.args["failures"]:
            failure_name = failure["name"]
            self.events[failure_name] = 0
            self.log(f'subscribing to <{failure_name}> from <{sensor}> ')
            self.listen_state(self.on_boiler_failure, sensor, new=failure_name) 


    def on_boiler_failure(self, entity, attribute, old, new, kwargs):
        currCount = self.events[new]
        self.events[new] = currCount+1
        self.run_in(self.__update_counter, 3.600, event = new)
        
        notify_now = True
        for failure in self.args["failures"]:
             failure_name = failure["name"]
             failure_amount = failure["amount"]
             if self.events[failure_name] < failure_amount :
                 notify_now = False

        # only check the time condition if we have the amount the failures needed.
        if notify_now and self.last_notification > (datetime.now() - timedelta(hours= 1)):
            notify_now = False

        if notify_now:
            self.last_notification = datetime.now()
            self.log('Send Boiler Alert Notification')
        else: # only used for debugging purposes
            self.log('Waiting ...')   


    def __update_counter(self, kwargs):
        event = kwargs["event"]
        currCount = self.events[event]
        self.events[event] = currCount-1    

Did I use the right approach to keep track of the failures? If you have a better approach, please share it :slightly_smiling_face:

Happy coding!
Humberto

Edit #1 Avoid hardcode entities names in the automation code. Thanks @swiftlyfalling @Burningstone, and @ReneTode for your help on the discord server
Edit #2 Update code based on the feedback received by @ReneTode (see his code below)

Previous post: App #3: Smart Radiator
Next post: App #5: Smart Radiator (Generic)

if you really like it flexible i would get the failure events out of the code.
and name the yaml the same as the code file, that would save others the trouble from renaming.
boiler_alert.yaml (and you got a double entry in your yaml)

boiler_alert:
  module: boiler_alert
  class: BoilerAlert
  sensor: sensor.boiler_alarm  
  failures:
    - {"fh_sensor_failure": 3}
    - {"i_sensor_failure": 1}

boiler_alert.py

import appdaemon.plugins.hass.hassapi as hass
from datetime import datetime, timedelta

class BoilerAlert(hass.Hass):

    def initialize(self):
        self.log('initializing ...')
        self.events = {}
        self.last_notification = datetime.now() - timedelta(hours= 1)
        sensor = self.args["sensor"]
        for failure in self.args["failures"]:
            for failure_name, failure_amount in failure.items():
                self.log(f'subscribing to failure {failure_name} from {sensor} ')
                self.events[failure_name] = 0
                self.listen_state(self.on_boiler_failure, sensor, new=failure_name) 

    def on_boiler_failure(self, entity, attribute, old, new, kwargs):
        currCount = self.events[new]
        self.events[new] = currCount+1
        self.run_in(self.__update_counter, 3.600, event = new)

        notify_now = True
        for failure in self.args["failures"]:
            for failure_name, failure_amount in failure.items():
                if self.events[failure_name] < failure_amount:
                    notify_now = False
        if self.last_notification < (datetime.now() - timedelta(hours= 1)):
            notify_now = False

        if notify_now:
            self.last_notification = datetime.now()
            self.log('Send Boiler Alert Notification')
        else: # only used for debugging purposes
            self.log('Waiting ...')   


    def __update_counter(self, kwargs):
        event = kwargs["event"]
        self.log('event: ' +event)
        currCount = self.events[event]
        self.events[event] = currCount-1

keep up the good work

1 Like

Thanks again for your feedback @ReneTode!

The right operator for this version is > instead of <

if self.last_notification > (datetime.now() - timedelta(hours= 1)):
    notify_now = False

That small change and everything worked perfectly :+1:

I also did other minor changes to your solution to avoid the two nested for. I used a list instead of a dictionary as an alternative to a tuple as they are not supported by YAML. I will update the code of my initial solution based on your feedback and my changes.

@ReneTode :+1::+1:

1 Like

yup, didnt look at it correctly and copy pasted :wink:
its always about choices.
i mostly use dicts. thats because i know that yaml is 1 big dict anyway.

in this case it was a list with dicts which off course also could be made more readable like:

boiler_alert:
  module: boiler_alert
  class: BoilerAlert
  sensor: sensor.boiler_alarm  
  failures:
    - failure: fh_sensor_failure
      failure_amount: 3
    - failure: i_sensor_failure
      failure_amount: 1

it all depends on what you like more and what you find more readable.

1 Like

@ReneTode once again many thanks for your feedback!

I definitely have to read more about YAML, your last suggestion is indeed more readable (at least for me)

Iā€™m updating my code now :+1:

1 Like