Sure, here it is. I’ve ended up just using blacklists for some sensors I don’t want to track, but as said before, this has worked well for me so far (even after moving house recently).
import hassapi as hass
import re
class LowBatteries(hass.Hass):
def initialize(self):
self.battery_sensors = self.get_battery_entities()
self.low_batteries = set()
self.empty_batteries = set()
self.import_low_batteries()
self.import_empty_batteries()
# Check if there's new low batteries that were not in the imported list.
# Don't check empty batteries, chances are that we incorrectly see an
# 'unavailable' as empty at this stage.
for sensor, value in self.battery_sensors.items():
if value > 0 and value < self.args["threshold"]:
self.check_low_battery(sensor, "state", None, value, None)
# Check if any low batteries have become empty during the downtime
for sensor in self.low_batteries:
if self.convert_state(self.battery_sensors[sensor]) == 0:
self.battery_empty(sensor, "state", None, 0, None)
# Check if any low or empty batteries have been replaced during the downtime
for sensor in self.low_batteries | self.empty_batteries:
self.check_new_battery(
sensor, "state", None, self.battery_sensors[sensor], None
)
# schedule callbacks for checking low, empty and new batteries
for sensor in self.battery_sensors:
self.listen_state(self.check_low_battery, sensor)
self.listen_state(
self.battery_empty, sensor, new="unavailable", duration=600
)
self.listen_state(self.battery_empty, sensor, new="0")
self.listen_state(self.check_new_battery, sensor)
# Check for potential new devices or renaming during runtime
self.listen_event(
self.new_entity_added, event="entity_registry_updated", action="create"
)
self.listen_event(
self.new_entity_added, event="entity_registry_updated", action="update"
)
def check_low_battery(self, entity, attribute, old, new, kwargs):
state = self.convert_state(new)
if (
state > 0
and state < self.args["threshold"]
and entity not in self.low_batteries
):
self.warn_low_battery(entity, state)
self.low_batteries.add(entity)
self.update_low_battery_list()
def battery_empty(self, entity, attribute, old, new, kwargs):
self.low_batteries.discard(entity)
self.update_low_battery_list()
self.warn_empty_battery(entity)
self.empty_batteries.add(entity)
self.update_empty_battery_list()
def check_new_battery(self, entity, attribute, old, new, kwargs):
state = self.convert_state(new)
if state > self.args["threshold"] and (
entity in self.low_batteries or entity in self.empty_batteries
):
self.log("new or recharged battery detected for {}".format(entity))
self.low_batteries.discard(entity)
self.update_low_battery_list()
self.empty_batteries.discard(entity)
self.update_empty_battery_list()
def new_entity_added(self, event_name, data, kwargs):
sensor = self.get_state(data["entity_id"], attribute="all", copy=False)
if not sensor:
return
if (
self.is_battery_entity(sensor)
and sensor["entity_id"] not in self.battery_sensors
):
name = sensor["entity_id"]
self.battery_sensors[name] = self.convert_state(sensor["state"])
self.listen_state(self.check_low_battery, name)
self.listen_state(self.battery_unavailable, name, new="unavailable")
self.listen_state(self.check_new_battery, name)
self.log("new battery sensor '{}' added".format(name))
def get_battery_entities(self):
batteries = {}
sensors = self.get_state("sensor", copy=False)
# remove excluded sensors
for battery in self.args["excluded_sensors"]:
sensors.pop(battery, None)
for sensor in sensors.values():
if self.is_battery_entity(sensor):
batteries.update(
{sensor["entity_id"]: self.convert_state(sensor["state"])}
)
self.log("Battery sensors found: {}".format(batteries))
return batteries
def is_battery_entity(self, sensor):
check = sensor["attributes"].get("device_class") == "battery"
return check
def warn_empty_battery(self, sensor):
name = self.rewrite_name(sensor)
message = "Battery {} is empty".format(name)
title = "Empty Battery"
for device in self.args["notification_services"]:
self.call_service(
device, message=message, title=title,
)
self.call_service(
"persistent_notification/create", message=message, title=title,
)
def warn_low_battery(self, sensor, value):
name = self.rewrite_name(sensor)
message="Battery {} is almost empty at {}".format(name, value)
title="Low Battery"
for device in self.args["notification_services"]:
self.call_service(
device,
message=message,
title=title,
)
self.call_service(
"persistent_notification/create", message=message, title=title,
)
def update_low_battery_list(self):
self.call_service(
"input_text/set_value",
entity_id=self.args["low_battery_list"],
value=",".join(self.low_batteries),
)
def update_empty_battery_list(self):
self.call_service(
"input_text/set_value",
entity_id=self.args["empty_battery_list"],
value=",".join(self.empty_batteries),
)
def import_low_batteries(self):
batteries = self.get_state(self.args["low_battery_list"])
if batteries == "" or batteries == "unknown":
return
else:
self.low_batteries.update(batteries.split(","))
self.log("low batteries: [{}] imported".format(batteries))
def import_empty_batteries(self):
batteries = self.get_state(self.args["empty_battery_list"])
if batteries == "" or batteries == "unknown":
return
else:
self.empty_batteries.update(batteries.split(","))
self.log("empty batteries: [{}] imported".format(batteries))
def rewrite_name(self, name):
name = re.sub(r"^.+?\.", "", name)
name = re.sub(r"_battery_level(_\d+)?$", "", name)
return name
def convert_state(self, value):
if value == "unavailable":
value = 0
try:
value = int(float(value))
except ValueError:
self.log(
"battery state was '{}' and not an int or 'unavailable' as expected".format(
value, level="ERROR"
)
)
return value
An example of my yaml file:
low_battery:
module: low_batteries
class: LowBatteries
threshold: 10
low_battery_list: input_text.low_batteries
empty_battery_list: input_text.empty_batteries
notification_services:
- notify/mobile_app_telefoon_sjoerd
excluded_sensors:
- sensor.hal9000_battery_level
- sensor.hal9000_battery_remaining
- sensor.telefoon_sjoerd_battery_level
- sensor.telefoon_judith_batterijniveau
- sensor.knop_hobbykamer_power
