透過WebSocket批次更改Home Assistant實體ID的方式

作者: /

我在Home Assistant中，透過XiaomiGateway3來管理100個以上的Zigbee裝置。每一個Zigbee裝置都會有一個以_zigbee結尾的sensor實體，裡面有這個裝置關於zigbee通訊的資訊。新配對的Zigbee裝置在配對成功後，XiaomiGateway3會用它的IEEE 64位址（或稱MAC位址）來做所有實體的預設命名，但在XiaomiGateway3某個版本之後，_zigbee結尾的sensor實體會在某種狀況下回到預設值，目前原因不明。

如上所示，車庫門的磁簧感應器都以garage_door_contactsensor開頭，但_zigbee這個實體卻變回預設值0x00158d0（下略）。

如果沒有用到這個實體，其實也不是什麼太大的問題，但我會拿它來計算每一種感應器類型的數量，例如計算磁簧感應器的數量

- platform: template
  sensors:
    zigbee_contactsensor:
      friendly_name: Zigbee_ContactSensor
      value_template: >-
        {% set ns = namespace(counter=0) %}
        {% for entity in states.sensor %}
          {% if "ContactSensor Zigbee" in entity.name %}
            {% set ns.counter = ns.counter + 1 %}
          {% endif %}
        {% endfor %}
        {{ ns.counter }}

以及超過180分鐘未更新的磁簧感應器數量，這樣可以得知有多少磁簧感應器可能已經失去連線，需要確認一下連線狀態。

- platform: template
  sensors:
    zigbee_contactsensor_timeout:
      friendly_name: Zigbee_ContactSensor_Timeout
      value_template: >-
        {% set ns = namespace(counter=0) %}
        {% for entity in states.sensor if "ContactSensor Zigbee" in entity.name %}
          {% if as_timestamp(now()) - as_timestamp(entity.last_changed) > 180 * 60 %}
            {% set ns.counter = ns.counter + 1 %}
          {% endif %}
        {% endfor %}
        {{ ns.counter }}

但100個以上的_zigbee實體要改名字是一件很麻煩的事，不太可能手動一筆一筆去修改。Home Assistant的WebSocket則提供了這種功能。可以透過類型””config/entity_registry/update”來修改實體的ID，其做法是

  • （第一次使用）申請一個永久權杖，在http://homeassistant.local:8123/profile/security最下方的「永久有效存取權杖」，點選「創建權杖」並將權杖代碼複製下來。
  • 連接Home Assistant的WebSocket API，網址是ws://homeassistant.local:8123/api/websocket。
  • 透過永久權杖做認證，取得WebSocket的權限。
  • 透過類型”config/entity_registry/update”修改實體ID

可以透過python來做這件事

import json
import websocket

# 填入Home Assistant內網IP與永久權杖代碼
HOST = "192.168.1.10:8123"
TOKEN = "xxx"

# 要更改的實體ID與新的命名
old_entity_id = "sensor.0x00......._zigbee" 
new_entity_id = "sensor.garage_door_contactsensor_zigbee"

def rename_entity(old_entity_id, new_entity_id):
    websocket_url = f'ws://{HOST}/api/websocket'
    ws = websocket.WebSocket()

    try:
        ws.connect(websocket_url)

        # 確認WebSocket連線
        auth_req = ws.recv()

        # 傳送認證請求
        auth_msg = json.dumps({"type": "auth", "access_token": TOKEN})
        ws.send(auth_msg)

        # 確認認證結果
        auth_result = ws.recv()
        auth_result = json.loads(auth_result)
        if auth_result["type"] != "auth_ok":
            print("Authentication failed. Check your access token.")
            return

        # 更改實體ID
        entity_registry_update_msg = json.dumps({
            "id": 1, 
            "type": "config/entity_registry/update",
            "entity_id": old_entity_id,
            "new_entity_id": new_entity_id
        })
        ws.send(entity_registry_update_msg)

        # 確認更新结果
        update_result = ws.recv()
        update_result = json.loads(update_result)

        if update_result.get("success"):
            print(f"Entity '{old_entity_id}' renamed to '{new_entity_id}' successfully!")
        else:
            error_message = update_result.get("error", {}).get("message", "Unknown error")
            print(f"Failed to rename entity '{old_entity_id}': {error_message}")

    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        ws.close()

if __name__ == "__main__":
    rename_entity(old_entity_id, new_entity_id)

執行後得到成功將sensor.0x00……._zigbee重命名為sensor.garage_door_contactsensor_zigbee的提示

Entity 'sensor.0x00......._zigbee' renamed to 'sensor.garage_door_contactsensor_zigbee' successfully!

如果要大量且有規則地修改以符合我前述的需求，讓_zigbee的實體ID重新命名為sensor.{Friendly name}_zigbee的話，我就得先將所有sensor.0x(任意字元)_zigbee的實體ID及其對應的Friendly Name全部找出來，再一個一個自動去做修改。找出符合的實體ID可以透過API及regex實作。整段程式碼如下

import requests
import json
import re
import websocket

# 填入Home Assistant內網IP與永久權杖代碼
HOST = "192.168.1.10:8123"
TOKEN = "xxx"

# API認證表頭
headers = {
    'Authorization': f'Bearer {TOKEN}',
    'Content-Type': 'application/json'
}

# 列出符合sensor.0x(任意字元)_zigbee的實體ID及其對應的Friendly Name，得到entity_data。
def list_entities():
    api_endpoint = f'http://{HOST}/api/states'
    response = requests.get(api_endpoint, headers=headers)
    if response.status_code == 200:
        data = json.loads(response.text)
        entity_data = [(entity['attributes'].get('friendly_name', ''), entity['entity_id']) for entity in data]
        entity_data = [(friendly_name, entity_id) for friendly_name, entity_id in entity_data if re.search(r'^sensor\.0x.[a-z0-9]+_zigbee$', entity_id)]
        return entity_data
    else:
        print(f'Error: {response.status_code} - {response.text}')
        return None

# 根據entity_data裡的entity_id與friendly name，依據規則建立新的entity_id並將表格合併。
def process_entities(entity_data):
    rename_data = []
    for friendly_name, entity_id in entity_data:
        new_entity_id = re.sub(r'^sensor\.0x.[a-z0-9]+_zigbee$', 'sensor.' + friendly_name.lower().replace(" ", "_"), entity_id)
        rename_data.append((friendly_name, entity_id, new_entity_id))
    rename(rename_data)

# 透過WebSocket做重新命名的動作，命名前會有提示問句確認是否修改。
def rename(rename_data):
    websocket_url = f'ws://{HOST}/api/websocket'
    ws = websocket.WebSocket()

    try:
        ws.connect(websocket_url)
        auth_req = ws.recv()
        auth_msg = json.dumps({"type": "auth", "access_token": TOKEN})
        ws.send(auth_msg)

        auth_result = ws.recv()
        auth_result = json.loads(auth_result)
        if auth_result["type"] != "auth_ok":
            print("Authentication failed. Check your access token.")
            return

        for index, (friendly_name, entity_id, new_entity_id) in enumerate(rename_data, start=1):
            print(f"\nEntity ID: {entity_id}")
            print(f"New Entity ID: {new_entity_id}")
            confirm = input(f"Do you want to rename '{entity_id}' to '{new_entity_id}'? (y/n): ").strip().lower()

            if confirm == 'y':
                entity_registry_update_msg = json.dumps({
                    "id": index,
                    "type": "config/entity_registry/update",
                    "entity_id": entity_id,
                    "new_entity_id": new_entity_id
                })
                ws.send(entity_registry_update_msg)
                update_result = ws.recv()
                update_result = json.loads(update_result)

                if update_result.get("success"):
                    print(f"Entity '{entity_id}' renamed to '{new_entity_id}' successfully!")
                else:
                    error_message = update_result.get("error", {}).get("message", "Unknown error")
                    print(f"Failed to rename entity '{entity_id}': {error_message}")
            else:
                print(f"Skipping renaming of '{entity_id}'.")

    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        ws.close()

if __name__ == "__main__":
    entity_data = list_entities()
    if entity_data:
        process_entities(entity_data)
    else:
        print("No entities found matching the search regex.")

執行當下我有46個sensor.0x(任意字元)_zigbee的實體要重新命名，執行前會詢問是否修改，若要放成自動化修改的話可以將提示相關的程式碼移除即可。以其中一個為例

Entity ID: sensor.0x00158d00054158c5_zigbee
New Entity ID: sensor.livingroom_tv_plug_zigbee
Do you want to rename 'sensor.0x00158d00054158c5_zigbee' to 'sensor.livingroom_tv_plug_zigbee'? (y/n): y
Entity 'sensor.0x00158d00054158c5_zigbee' renamed to 'sensor.livingroom_tv_plug_zigbee' successfully!

相關文章

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *

返回頂端