Hello,
Another example
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
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)