Script and config to disable Home Assistant entities and to "template" configurate an MQTT device

disable-entities

#!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)

disable-entities.yaml

# 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

@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.