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?
something like this: (code just out of my head and not correct)
import datetime
initialize(...):
self.updatetime = datetime.datetime.now()
self.listen_state (self.updatetime,self.args["sensor"])
self.run_hourly (self.check,self.updatetime)
updatetime(self,...):
self.updatetime = datetime.datetime.now()
check(self, ...):
timegoneby = datetime.datetime.now() - 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")
try:
self.args["threshold"]
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")
try:
self.listen_state(self.restart_timer, self.args["sensor"])
except KeyError:
pass
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.cancel_timer(self.watchdog)
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")
2 Likes
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:
try:
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:
try:
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"]):
low.append(friendly_name)
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:
4 Likes
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 = datetime.now()
self.timeout_length = self.args["check_interval"]
self.device_db = shelve.open(self.args["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 = datetime.now()
data = {'state': new, 'time': now}
self.device_db[entity] = data
self.log("Sensor {} value changed to {} at {}".format(entity, new, now))
#else:
# 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 = datetime.now()
elapsedTime = now - oldTime
if elapsedTime.seconds > self.timeout_length:
if message =="":
message = entity
else:
message = message + ', ' + entity
self.log("Sensor {} is in {} state for 24 hours.".format(entity, oldState))
else:
self.log("Adding {}, setting value to current state ({})".format(entity, state[entity]["state"]))
now = datetime.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:
5 Likes
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.
https://appdaemon.readthedocs.io/en/latest/index.html
and it needs a little rewrite because its appdaemon 2.0