-
-
Save gwww/1c959a00ec2bcf59f7041fe731b33a3c to your computer and use it in GitHub Desktop.
#!python3 | |
# Based on: https://community.home-assistant.io/t/api-for-changing-entities/282114/3 | |
import asyncio | |
import json | |
import os | |
import re | |
import sys | |
import yaml # type: ignore | |
import websockets | |
async def send(ws, data): | |
"""Send on websocket.""" | |
json_str = json.dumps(data) | |
# print(f">>>>>> {json_str}") | |
await ws.send(json_str) | |
async def recv(ws): | |
"""Return data received on websocket.""" | |
str = await ws.recv() | |
# print(f"<<<<<< {str}") | |
data = json.loads(str) | |
return data | |
def error(str): | |
print(str) | |
exit(1) | |
def next_id(): | |
"""Return a unique monotonically incrementing ID.""" | |
try: | |
next_id.id += 1 | |
except AttributeError: | |
next_id.id = 100 | |
return next_id.id | |
def read_config(): | |
with open("disable-entities.yaml", "r") as stream: | |
try: | |
config = yaml.safe_load(stream) | |
except yaml.YAMLError as exc: | |
print(exc) | |
exit(2) | |
if not config["token"]: | |
config["token"] = os.environ.get("HA_ACCESS_TOKEN") | |
return config | |
def success(result): | |
return f"{'Success' if result['success'] else 'Failed!!!'}" | |
async def get_entities(ws): | |
"""Returns a list of enabled HA entities.""" | |
# await send(ws, {"id": next_id(), "type": "config/entity_registry/list"}) | |
await send(ws, {"id": next_id(), "type": "get_states"}) | |
result = await recv(ws) | |
states = [entity["entity_id"] for entity in result["result"]] | |
return states | |
async def auth(ws, config): | |
auth_required = await recv(ws) | |
if auth_required["type"] != "auth_required": | |
error(f"Protocol error, auth_required expected, got: {auth_required}") | |
await send(ws, {"type": "auth", "access_token": config["token"]}) | |
auth_status = await recv(ws) | |
if auth_status["type"] != "auth_ok": | |
error(f"Auth failed. {auth_status}") | |
async def disable_ha_entities(ws, config): | |
"""Disable the entities specified in the configuration YAML.""" | |
ha_existing_entities = await get_entities(ws) | |
entities_to_update = [] | |
for entity in config.get("disable_entities", []): | |
entity = entity.replace("%", config["_device_name"]).lower() | |
regex = re.compile(entity) | |
entities_to_update += list(filter(regex.match, ha_existing_entities)) | |
entities_to_update = list(set(entities_to_update)) # Filter duplicates | |
entities_to_update.sort() | |
if entities_to_update: | |
print("List of Home Assistant entities that will be disabled:") | |
for entity in entities_to_update: | |
print(f" {entity}") | |
print('To confirm disabling type "yes": ', end="") | |
if input() != "yes": | |
print("Cancelling") | |
exit(4) | |
else: | |
print("No Home Assistant entities found to disable.") | |
for entity in entities_to_update: | |
update_msg = { | |
"type": "config/entity_registry/update", | |
"id": next_id(), | |
"disabled_by": "user", | |
"entity_id": entity, | |
} | |
await send(ws, update_msg) | |
result = await recv(ws) | |
print(f"{success(result)} updating {entity}") | |
print() | |
async def ha_mqtt_service(ws, topic, payload): | |
msg = { | |
"type": "call_service", | |
"id": next_id(), | |
"domain": "mqtt", | |
"service": "publish", | |
"service_data": { | |
"topic": topic, | |
"payload": json.dumps(payload), | |
}, | |
} | |
await send(ws, msg) | |
return await recv(ws) | |
async def set_attribute(ws, device, attr, value): | |
result = await ha_mqtt_service(ws, f"zigbee2mqtt/{device}/set", {attr: value}) | |
print(f"{success(result)} setting attribute {attr}={value}") | |
async def set_reporting(ws, device, report): | |
report["id"] = f"{device}/{report['endpoint']}" | |
del report["endpoint"] | |
result = await ha_mqtt_service( | |
ws, "zigbee2mqtt/bridge/request/device/configure_reporting", report | |
) | |
print(f"{success(result)} setting report threshold {report['attribute']}") | |
async def configure_device(ws, config): | |
for attr, value in config.get("device_configuration", {}).items(): | |
await set_attribute(ws, config["_device_name"], attr, value) | |
for report in config.get("reporting_configuration", []): | |
await set_reporting(ws, config["_device_name"], report) | |
async def main(): | |
config = read_config() | |
config["_device_name"] = sys.argv[1] | |
async with websockets.connect(f"ws://{config['server']}/api/websocket") as ws: # type: ignore | |
await auth(ws, config) | |
await disable_ha_entities(ws, config) | |
await configure_device(ws, config) | |
if len(sys.argv) != 2: | |
print(f"Usage: {sys.argv[0]} <name of device>") | |
exit(1) | |
try: | |
asyncio.run(main()) | |
except KeyboardInterrupt: | |
exit(3) |
# Server IP or name | |
server: localhost:8123 | |
token: <token from Home Assistant here> | |
# Disable Home Assistant entities that match the regular expression. | |
# The device name is substituted in for the '%' character in the entity_name. | |
disable_entities: | |
- '(number|sensor).%_default.*' | |
- '(number|select).%' | |
- 'sensor.%_remoteprotection' | |
- 'sensor.%_powertype' | |
- 'binary_sensor.%_update_available' | |
# Configure MQTT device attributes. | |
# All values are integer; to find name to value mapping see here: | |
# https://github.com/Koenkk/zigbee-herdsman-converters/blob/master/devices/inovelli.js | |
device_configuration: | |
outputMode: 0 # Dimmer (vs On/Off switch) | |
relayClick: 1 # Disable the relay click, oddly 1 is disable | |
stateAfterPowerRestored: 255 # Restore to previous level after power restore | |
defaultLevelLocal: 254 # Turn on full when paddle pressed | |
defaultLevelRemote: 254 # Turn on full when controlled over Zigbee | |
# Configure MQTT reporting entries. | |
reporting_configuration: | |
- "endpoint": 1 | |
"cluster": "haElectricalMeasurement" | |
"attribute": "activePower" | |
"minimum_report_interval": 1 | |
"maximum_report_interval": 3600 | |
"reportable_change": 10 | |
- "endpoint": 1 | |
"cluster": "seMetering" | |
"attribute": "currentSummDelivered" | |
"minimum_report_interval": 1 | |
"maximum_report_interval": 3600 | |
"reportable_change": 10 |
More entities to add - I think these were added in either v2.11 or v2.14 of the Inovelli Blue firmware.
- 'select.%_auxswitchuniquescenes' - 'select.%_bindingofftoonsynclevel' - 'number.%_brightnesslevelfordoubletap(up|down)' - 'select.%_doubletapdowntoparam56' - 'select.%_doubletapuptoparam55' - 'select.%_higheroutputinnonneutral' - 'select.%_ledbarscaling'
Updated my example
Can someone point me to a tutorial on how to actually use this? I'm in entity hell with my install over 120 Inovelli blue switches.
I have home assistant, MQTT and Zigbee2MQTT all running in separate dockers on Unraid.
I assume I need to create a disabled-entities.py file with the code above and a disabled-entities.yaml file with the configuration variables set for Server and Token... and put them both in the same directory????
Is this right?
And then what? run >python disable-entities.py ????
Yep!
You can run this from any computer that has python installed.
Install some pip packages first:
pip install pyyaml websockets
Then run the script with the name of the device
python disable-entities.py "<device name>"
The only part I don't recall is what the device name should be. It was either the entity_id
or the friendly name.
@zanix thanks!
So I have to do this for every device individually?
I was hoping I could do it to them all at once. Right now, I've got over 100 Inovelli blue switches. Doing this 1x at a time is time consuming. But I guess not as time consuming as disabling all of the entities individually or in groups.
This worked great! @chansearrington you can replace %
in the patterns with *
to have it match anything up to the suffix for each pattern. That will modify all the entities for all the devices at once. Take a look at the entities that it lists before saying "yes" to make sure that it looks right.
More entities to add - I think these were added in either v2.11 or v2.14 of the Inovelli Blue firmware.