App to monitor sensors

Anybody have an app to monitor the sensors?
My setup has a lot of sensors and some of them dies some times and I would not notice it for sometime.
Anyone has an app that monitors the sensors and if the value not changed for say 8 hours creates a persistent notification or send an email?
Any directions how to create one?

Check this out

something like this: (code just out of my head and not correct)

import datetime
  self.updatetime =
  self.listen_state (self.updatetime,self.args["sensor"])
  self.run_hourly (self.check,self.updatetime)

  self.updatetime =

check(self, ...):
  timegoneby = - self.updatetime
  if timegoneby > datetime.timedelta(hours=8):
    self.notify("more then 8 hours have gone by since sensor " + self.args["sensor"] + " has been updated")

now you can create an instance in the cfg for every sensor you want to monitor.

I have this for my battery powered nodes, that are supposed to report every 24 hours

import appdaemon.appapi as appapi

# app to report a battery level when it gets too low
# Args
# battery_sensor: sensor to check
# threshold: threshold below which we send a warning

class BatteryLevel(appapi.AppDaemon):

    # Actually, measure 26 hours between reports, to give it some 
    # wiggle room
    timeout_length = 60 * 60 * 26

    def initialize(self):
        self.log("Initializing BatteryLevel", level = "DEBUG")
            self.listen_state(self.level_changed, self.args["battery_sensor"],
                constrain_input_boolean = "input_boolean.notifications")
            self.listen_state(self.restart_timer, self.args["battery_sensor"])
            self.watchdog = self.run_in(self.timeout_handler, self.timeout_length)
        except KeyError as e:
            self.log("Argument not found : {}".format(e), level="ERROR")

            self.listen_state(self.restart_timer, self.args["sensor"])
        except KeyError:

    def level_changed(self, entity, attribute, old, new, kwargs):
        self.log("{} changed to {}".format(entity, new), level = "DEBUG")
        if new < self.args["threshold"]:
            self.notify("{} battery percentage is {}".format(
                    self.friendly_name(entity), new),
                    title = "Battery Level Low")

    def restart_timer(self, entity, attribute, old, new, kwargs):
        self.log("restart_timer", level="DEBUG")
        self.watchdog = self.run_in(self.timeout_handler, self.timeout_length)

    def timeout_handler(self, kwargs):
        self.log("timeout_handler", level="DEBUG")
        self.watchdog = self.run_in(self.timeout_handler, self.timeout_length)
        message = "{} has not reported for 24 hours".format(self.windowname)
        self.log(message, level="WARNING")
        if self.get_state("input_boolean.notifications") == "on":
            self.notify(message, title = "Window Node Not Reporting")


Thank you for your reply,
What if I use the duration parameter like so

self.handle = self.listen_state(self.my_callback, "light.office_1", new = "on", duration = 60)

I am going to modify one of my script to check.

that way the callback will only be called when the light has been on for 60 seconds.

I took the example battery monitor and beefed it up. Originally it only looked at attributes but I use MQTT so each of my “attributes” is a different sensor. Here is what I am using right now.

Reports battery levels every morning, shows those below threshold and then all

import appdaemon.appapi as appapi
import datetime

# App to send email report for devices running low on battery
# Args:
# threshold = value below which battery levels are reported and email is sent
# always_send = set to 1 to override threshold and force send
# Example Config
# Battery Check:
#   always_send: '0'
#   class: Battery
#   dependencies: utils
#   module: battery_monitor
#   threshold: '25'
# Release Notes
# Version 1.0:
#   Initial Version 

class Battery(appapi.AppDaemon):

    def initialize(self):
        #self.check_batteries({"force": 1})
        time = datetime.time(6, 00, 0)
        self.run_daily(self.check_batteries, time) 
    def check_batteries(self, kwargs):
        devices = self.get_state()
        values = {}
        low = []
        for device in devices:        
            battery = None
            if "group" not in device:
                    if "battery" in devices[device]["attributes"]:
                        battery = devices[device]["attributes"]["battery"]
                    if "battery_level" in devices[device]["attributes"]:
                        battery = devices[device]["attributes"]["battery_level"]
                    if "battery" in device:
                        battery = int(self.get_state(device))
                    if "battery_level" in device:
                        battery = int(self.get_state(device))
                except TypeError:
                    self.error("{} is not scriptable.".format(device))

            if battery != None:
                    friendly_name = self.get_state(device, attribute="group")['group.battery_group']['friendly_name']
                except TypeError:
                    friendly_name = self.get_state(device, attribute="friendly_name")

                if battery < int(self.args["threshold"]):
                values[friendly_name] = battery
        message = ""
        message += "...............\n\n"
        # message += "Battery Level Report\n"
        # message += "...............\n\n\n"
        if low:
            message += "*Below Threshold* (< {}): \n".format(self.args["threshold"])
            for device in low:
                message = message + device + " \n"
            message += "\n\n"
        message += "*All Battery Levels*:\n"
        for device in sorted(values):
            message += "{}: {}\n".format(device, values[device])
        if low or ("always_send" in self.args and self.args["always_send"] == "1") or ("force" in kwargs and kwargs["force"] == 1):
            title = "*Home Assistant Battery Report*"  
            self.call_service('notify/notify', title=title, message=message)

output example:


I have finally whipped up something by adapting the SwitchReset example from AppDaemon.
I keep track of state and lastChanged time from HA on to a database file, and checks if the values have changed in 24 hours.
Sends an email if it hasn’t.

import appdaemon.appapi as appapi
import shelve
from datetime import datetime
# App to notify if a sensor/device_tracker is dead for 24 hours. Adapted from SwitchReset example app.
# Args:
#file: db file to save state and lastchanged to persist HA shutdown
#blacklist: list of sensors/device_trackers not to track.
#check_interval: time interval between check in seconds
# Release Notes
# Version 1.0:
#   Initial Version

class SaveSensorState(appapi.AppDaemon):
  def initialize(self):
    start_time =
    self.timeout_length = self.args["check_interval"]
    self.device_db =["file"])
    self.listen_state(self.state_change, "sensor")
    self.listen_state(self.state_change, "device_tracker")
    self.run_every(self.check_if_updated,start_time, self.timeout_length)
    self.blacklist = self.args["blacklist"].split(",")

  def state_change(self, entity, attribute, old, new, kwargs):
    #self.log("State change: {} to {}".format(entity, new))
    if new != old:
      now =
      data = {'state': new, 'time': now}
      self.device_db[entity] = data
      self.log("Sensor {} value changed to {} at {}".format(entity, new, now))
    #  self.log("Sensor {} value not changed".format(entity))
  def check_if_updated(self, kwargs):
    self.log("Checking sensor health")
    message = ""
    state = self.get_state()
    for entity in state:
      type, id = entity.split(".")
      if type == "sensor" or type == "device_tracker":
        if entity in self.device_db:
          oldState = self.device_db[entity]['state']
          oldTime = self.device_db[entity]['time']
          #self.log("Old state for {} was {} and old time was {}".format(entity, oldState, oldTime))
          if not entity in self.blacklist:
            if (self.device_db[entity]['state'] == state[entity]["state"]):
              now =
              elapsedTime = now - oldTime
              if elapsedTime.seconds > self.timeout_length:
                if message =="":
                  message = entity
                  message = message + ', ' + entity
                self.log("Sensor {} is in {} state for 24 hours.".format(entity, oldState))
          self.log("Adding {}, setting value to current state ({})".format(entity, state[entity]["state"]))
          now =
          data = {'state': state[entity]["state"], 'time': now}
          self.device_db[entity] = data
    if message != "":
      self.notify("Sensors {} have not changed for 24 hours".format(message), name = "as_email", title = "Message from AppDaemon")

Working as expected, now research how to remove all these dead Mysensor nodes:face_with_monocle:


Interesting! This looks like what I search for! It surprised me that there is no official component to do so, as I expected form HASS. I have a couple of sensors that are sometimes disconnecting, and as some automation depends on it, it would by nice to get a notification if there is no change in the values any more!
It would be handy if I could configure a “watchdog-like” component with for every sensor to look after the minimun time I expect it to change. I already use an InfluxDB so this might be possible to reuse perhaps…
In the meantime I might try the example from above, but can anybody tell me how to get this code into my HASS configuration? Thanks!

this code is written for appdaemon.

and it needs a little rewrite because its appdaemon 2.0