Last active
March 4, 2024 18:45
-
-
Save gwww/1c959a00ec2bcf59f7041fe731b33a3c to your computer and use it in GitHub Desktop.
Script and config to disable Home Assistant entities and to "template" configurate an MQTT device
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.