To assist in my maintenance of a remote installation, I would like a dashboard panel that lists all the devices that are unavailable, preferably sorted according to integration (in my case Tuya, Local Tuya, Sonoff/eWeLink and ZHA).
Has anyone done this?
I did not quite exactly what you are searching for but something very similar.
I have a script written in pyscript that checks for unavailable devices every 30 minutes and uses persistent notifications to notify me when it finds any. It’s quick and dirty and can be improved, but it does its job well enough.
It basically works by getting a list of all devices from the device registry (the huge advantage of pyscript compared to other solutions like AppDaemon or Node Red is, that pyscript allows you to access internal APIs of home assistant like the device registry) and then checking all entities associated with a device. When all entities are in state “unavailable” I then assume that the whole device is unavailable.
ignore_devices = [
# Add ids of devices to be ignored here
]
ignore_entities = [
# Add entities whose devices should be ignored here
]
CHECKER_ID = "pyscript.device_availability_checker"
from homeassistant.helpers import entity_registry as erm
from homeassistant.helpers import device_registry as drm
from homeassistant.const import STATE_UNAVAILABLE
@time_trigger('cron(*/30 * * * *)')
def device_availability_checker_cron():
er = erm.async_get(hass)
dr = drm.async_get(hass)
unavailable_devices = {}
unavailable_since = {}
# Iterate over all devices and check whether they are unavailable
# A device is supposed to be unavailable if all related entities are in
# state "unavailable"
for d in dr.devices:
if d in ignore_devices:
continue
unavailable = False
since = None
for e in erm.async_entries_for_device(er, d):
if e.entity_id in ignore_entities:
unavailable = False
break
elif hass.states.is_state(e.entity_id, STATE_UNAVAILABLE):
unavailable = True
since = hass.states.get(e.entity_id).last_changed
else:
unavailable = False
break
if unavailable:
unavailable_devices[d] = dr.async_get(d)
unavailable_since[d] = since
notifs = []
devices = {}
# Iterate over all unavailable devices and construct notification text
# for a persistent_notification
for k, d in unavailable_devices.items():
manufacturer = d.manufacturer
model = d.model
text = f'- {d.name}'
desc = ""
if model is not None:
desc += model
if manufacturer is not None:
if desc != "":
desc += f" [{manufacturer}]"
else:
desc += manufacturer
if desc != "":
text += "\n - " + desc
text += f"\n - Since: {unavailable_since[k]}"
text += f"\n - ID: {d.id}"
devices[k] = {
"name": d.name,
"manufacturer": manufacturer,
"model": model,
"since": unavailable_since[k]
}
notifs.append(text)
# Show a persistent notification or dismiss an old one if there is nothing
# to show
ntext = "\n".join(notifs)
if ntext != "":
hass.services.async_call("persistent_notification", "create", {
"notification_id": "device_availability_warning",
"title": "List of Unavailable Devices",
"message": ntext
}, False)
else:
hass.services.async_call("persistent_notification", "dismiss", {
"notification_id": "device_availability_warning"
}, False)
state.set(CHECKER_ID, len(devices), devices = devices)
@time_trigger('startup')
def device_availability_checker_startup_trigger():
log.info("device_availability_checker.startup_trigger()")
task.sleep(60)
device_availability_checker_cron()
Hi
Nice to get this done with pyscript. I saw some opportunities to improve the code,
Below an extended version that takes these corner cases also into account:
ignore_devices = [
# Add ids of devices to be ignored here
]
ignore_entities = [
# Add entities whose devices should be ignored here
]
extend_integrations = [
# Add integration whose entities are added to devices and therefor to be ignored
"powercalc"]
CHECKER_ID = "pyscript.device_availability_checker"
from homeassistant.helpers import entity_registry as erm
from homeassistant.helpers import device_registry as drm
from homeassistant.const import STATE_UNAVAILABLE
@time_trigger('cron(*/30 * * * *)')
def device_availability_checker_cron():
er = erm.async_get(hass)
dr = drm.async_get(hass)
unavailable_devices = {}
unavailable_since = {}
# Iterate over all devices and check whether they are unavailable
# A device is supposed to be unavailable if all related entities are in
# state "unavailable"
for d in dr.devices:
if d in ignore_devices:
continue
unavailable = False
since = None
for e in erm.async_entries_for_device(er, d):
if e.entity_id in ignore_entities:
unavailable = False
break
elif hass.states.is_state(e.entity_id, STATE_UNAVAILABLE):
unavailable = True
since = hass.states.get(e.entity_id).last_changed
elif ( # check it is not a connection status entity
(e.original_device_class != "connectivity") and
# check it is not an entity added by integrations to ignore
(e.platform not in extend_integrations)
):
unavailable = False
break
if unavailable:
unavailable_devices[d] = dr.async_get(d)
unavailable_since[d] = since
notifs = []
devices = {}
# Iterate over all unavailable devices and construct notification text
# for a persistent_notification
for k, d in unavailable_devices.items():
name = d.name if d.name_by_user is None else d.name_by_user
manufacturer = d.manufacturer
model = d.model
text = f'- {name}'
integration="unknown"
entry = hass.config_entries.async_get_entry(entry_id=d.primary_config_entry)
if entry is not None:
integration = entry.domain
desc = ""
if model is not None:
desc += model
if manufacturer is not None:
if desc != "":
desc += f" [{manufacturer}]"
else:
desc += manufacturer
if desc != "":
text += "\n - " + desc
text += f"\n - Integration: {integration}"
text += f"\n - Since: {unavailable_since[k]}"
text += f"\n - ID: {d.id}"
devices[k] = {
"integration": integration,
"name": name,
"manufacturer": manufacturer,
"model": model,
"since": unavailable_since[k]
}
notifs.append(text)
# Show a persistent notification or dismiss an old one if there is nothing
# to show
ntext = "\n".join(notifs)
if ntext != "":
hass.services.async_call("persistent_notification", "create", {
"notification_id": "device_availability_warning",
"title": "List of Unavailable Devices",
"message": ntext
}, False)
else:
hass.services.async_call("persistent_notification", "dismiss", {
"notification_id": "device_availability_warning"
}, False)
state.set(CHECKER_ID, len(devices), devices = devices)
Note: @matzman666 I realy loved to build further on your code. It solved several obstacles I didn’t know how to tackle them. I wrote a template version in the past. This is much better.
Edit: simplified the code as above