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
Thank you very much for sharing your script @erkr. This looks exactly like what I need.
How are you running them in/through HA to show the results there?
Hi @Quaternion
Prerequisite to run this script is that you have installed the custom integration pyscript (e.g. via HACS), where you probably have to enable these two options in the config:
The script will run autonomously by itself (the examples in the posts above runs every 30 minutes). The script does two things:
pyscript.device_availability_checker. The state represents the number of unavaliable devices. An attribute lists the detailsNote: In my post above I only improved the detection of the original script, no personal tailoring, in order to stay close to the original script.
Then I tailored it fully to my use case:
ignore_entities; When evaluating the entities of each device, the state of those entities is ignored, not the whole device!My personalized version of the script:
ignore_devices = [
# Add ids of devices to be ignored here.
# The template {{ device_id('entity id') }} can be used to find device id's
]
ignore_entities = [
# Add entities that should be ignored when evaluating their device state
]
extend_integrations = [
# Add integrations whose entities are added to devices but can remain available
"powercalc","nmap_tracker"]
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
@service(supports_response="only")
def device_availability_test(device_id: str):
"""yaml
description: Test the device_availability_checker result for a specific device
fields:
device_id:
description: device id to be tested
required: true
default: ''
example: 2e6ec7645070322e572e1e161306c2f7
selector:
text:
"""
if device_id in ignore_devices:
return {"result": "ignored device" }
er = erm.async_get(hass)
dr = drm.async_get(hass)
import json
result=[]
for e in erm.async_entries_for_device(er, device_id):
useForCheck = "Yes"
if ( # check if entity should be ignored
(e.original_device_class == "connectivity") or
(e.platform in extend_integrations) or
(e.entity_id in ignore_entities)
):
useForCheck = "Ignored"
#log.warning(f"avaliablity test entity {e.entity_id}")
result.append( {"entity": e.entity_id, "part_of_check": useForCheck, "available": "No" if hass.states.is_state(e.entity_id, STATE_UNAVAILABLE) else "Yes", "platform":e.platform, "device_class": e.original_device_class} )
return {"result": result }
@time_trigger('cron(* * * * *)')
def device_availability_checker(doNotify=False):
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". Except some added entities by other integrations, and connectivity types
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 ( # check if entity should be ignored
(e.original_device_class == "connectivity") or
(e.platform in extend_integrations) or
(e.entity_id in ignore_entities)
):
continue
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():
name = d.name if d.name_by_user is None else d.name_by_user
manufacturer = d.manufacturer
model = d.model
area = d.area_id or ""
integration="unknown"
#ids = d.identifiers
#if ids is not None and len(ids):
# integration = str(list(ids)[0][0])
entry = hass.config_entries.async_get_entry(entry_id=d.primary_config_entry)
if entry is not None:
integration = entry.domain
if doNotify:
text = f'- {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
if area is not None:
text += f"\n - Area: {area}"
text += f"\n - Integration: {integration}"
text += f"\n - Since: {unavailable_since[k]}"
text += f"\n - ID: {d.id}"
notifs.append(text)
devices[k] = {
"integration": integration,
"area": area,
"name": name,
"manufacturer": manufacturer,
"model": model,
"since": unavailable_since[k]
}
if doNotify:
# 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": f"List of {len(devices)} 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)
I created a straight forward dashboard to display them. I used a combi of the mark down and card custom:auto-entities to list the devices. Further I styled the card using card_mod (you can safely remove that ) and used the custom relative_time_plus template to display how long devices are unavailable. You optionally can use the standard available relative_time instead. All these custom items can be installed via HACS.
Note: In this card you can tap on a device to open the device page!
YAML of my card
type: custom:auto-entities
card:
type: entities
card_mod:
style: |
:host {
--ha-card-border-width: 0px;
}
.card-content {
margin-left: -50px;
}
filter:
template: >
{%- set devlist =
state_attr('pyscript.device_availability_checker','devices') or [] %}
{%- set ns=namespace(rows=[]) -%} {%- from 'relative_time_plus.jinja'
import relative_time_plus -%} {%- for k in devlist -%}
{%- set info=devlist[k] -%}
{%- set sec_info = 'Area:'~info.area~', Mdl:'~info.model -%}
{%- set entry = {
'type': 'custom:template-entity-row',
'name': info.integration~': '~info.name,
'state': relative_time_plus(info.since,abbr=true),
'secondary': sec_info,
'since': info.since.timestamp(),
'tap_action': {
'action': 'navigate',
'navigation_path': '/config/devices/device/'~k },
} -%}
{%- set ns.rows= ns.rows + [entry] -%}
{%- endfor -%}
{{ ns.rows | sort(attribute='since', reverse = True)}}
include: []
exclude: []
sort:
method: none
Thank you so much @erkr for taking your time to share all of this, including the detailed explanation.
It works like a charm on my setup and I have been trying exactly to do this through many other options for a while 