Zigbee friendly_names automatisieren

Hier ist ein Skript (python), der die Eigenschaften von eingelernten Zigbee devices erkennt und als “friendly names” in Z2MQTT einträgt.

#!/usr/bin/env python3
"""
Zigbee2MQTT Batch Renamer (DB-driven, MQTT rename)

Purpose
-------
Rename Zigbee2MQTT devices to a standardized convention:

    <type>_<hex>_<location>

Where:
- <hex> is the IEEE address (stable identity), e.g. 0xa4c138...
- <location> is derived from your existing friendly name table
  by stripping embedded hex strings and common device words, then normalizing.
- <type> is inferred from:
    1) modelId/manufName (most reliable, close to Z2M identification)
    2) endpoint cluster signatures (best-effort)
    3) fallback keywords from the existing name

Special rules (as requested)
----------------------------
1) EndDevice with known kind:
     use kind directly (no "enddevice_" prefix), e.g.
       button_0x..._OfficeDesk
       temperaturesensor_0x..._WashRoom
       motion_0x..._Hall

2) EndDevice with unknown kind:
     enddevice_unknown_<hex>
     (no location because we do not pretend to know it).

3) Non-EndDevice/Non-Router unknown + raw IEEE-only old name:
     dev_<hex>
     (parser-safe fallback, avoids Z2M choking on names starting with 0x.)

4) Routers:
     router_<hex>_<location>
     and optionally router_<kind>_<hex>_<location> when "kind" is meaningful.

Exclusions (recommended hygiene)
-------------------------------
Some entries in database.db can represent infrastructure "router-mode sticks"
or gateway proxies. You typically don't want to rename these.
We skip:
- SONOFF DONGLE-E_R
- SMLIGHT SLZB-06P10
(You can extend SKIP_MODELS.)

How it works
------------
Inputs:
- /opt/zigbee2mqtt/data/database.db
    NDJSON lines with keys such as:
      ieeeAddr, type, manufName, modelId, endpoints, ...

- /srv/z2m_vault_2026-01-25/device_exports_v3/DEVICE_RENAME_MAP_keep65.json
    Your authoritative IEEE -> current friendly name table.
    This script uses it to extract a location string and keyword hints.

Output:
- Dry-run table to stdout (default).
- If --apply is passed: publishes MQTT rename requests to:
    <base_topic>/bridge/request/device/rename
  with payload:
    {"from":"<ieeeAddr>","to":"<new_friendly_name>"}

Important notes
---------------
- This script does NOT require Z2M to answer bridge/request/devices.
- Z2M will accept "from" as the IEEE address (you already verified this).
- Use --dry-run first; only use --apply once satisfied.
- Use --force to rename everything (otherwise only raw IEEE/dev_ names).
"""

import argparse
import json
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple


# ---------------------------------------------------------------------------
# Regex and normalization helpers
# ---------------------------------------------------------------------------

IEEE_RE = re.compile(r"^0x[0-9a-fA-F]{16}$")
HEX_RE = re.compile(r"0x[0-9a-fA-F]{16}", re.IGNORECASE)

def norm_token(s: str) -> str:
    """
    Normalize to a Zigbee2MQTT-friendly token:
    - Replace umlauts (so you can keep German locations safely)
    - Replace any non-alphanumeric sequences with underscores
    - Collapse multiple underscores
    """
    s = (s or "").strip()
    s = s.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue")
    s = s.replace("Ä", "Ae").replace("Ö", "Oe").replace("Ü", "Ue")
    s = re.sub(r"[^A-Za-z0-9]+", "_", s)
    s = re.sub(r"_+", "_", s).strip("_")
    return s


def is_raw_ieee_name(name: str) -> bool:
    """True if name is exactly an IEEE address like 0x00158d..."""
    return bool(IEEE_RE.match((name or "").strip()))


def ieee_hex(ieee: str, keep_0x: bool) -> str:
    """Return IEEE as hex token, optionally removing the 0x prefix."""
    ieee = (ieee or "").lower()
    return ieee if keep_0x else ieee.replace("0x", "")


# ---------------------------------------------------------------------------
# Location extraction from your existing friendly name table
# ---------------------------------------------------------------------------

# Words removed from location extraction (expand freely)
STRIP_WORDS = {
    "dev", "devc", "device",
    "switch", "button", "remote",
    "radar", "pir", "motion", "sensor",
    "temp", "temperature", "thermo", "thermometer", "humidity",
    "door", "window", "contact",
    "blind", "blinds", "shade", "shades",
    "router", "repeater", "coordinator",
    "aqara", "tuya",
    "inside", "outside",
}

def extract_location(old_name: str) -> str:
    """
    Extract a location token from existing friendly name:
    - remove dev_/devc_ prefix
    - remove embedded 0x... IEEE strings
    - remove common device words (STRIP_WORDS)
    - normalize to underscores
    """
    s = (old_name or "").strip()
    s = re.sub(r"^(devc_|dev_)", "", s, flags=re.IGNORECASE)
    s = HEX_RE.sub(" ", s)

    parts = re.split(r"[\s_]+", s)
    kept: List[str] = []
    for p in parts:
        if not p:
            continue
        pl = p.lower()
        if IEEE_RE.match(pl):
            continue
        if pl in STRIP_WORDS:
            continue
        kept.append(p)

    return norm_token(" ".join(kept))


# ---------------------------------------------------------------------------
# Device metadata from database.db
# ---------------------------------------------------------------------------

@dataclass
class Dev:
    ieee: str
    role: str           # Router / EndDevice / ...
    manuf: str          # manufName
    model: str          # modelId
    endpoints: Any      # endpoints (dict/list), may contain clusters
    old_name: str       # from your mapping table if present else ieee


def load_name_map(path: Path) -> Dict[str, str]:
    """
    Load your authoritative IEEE->friendly_name JSON table.
    Keys are normalized to lowercase IEEE.
    """
    raw = json.loads(path.read_text())
    out: Dict[str, str] = {}
    for k, v in raw.items():
        if isinstance(k, str) and isinstance(v, str) and k and v:
            out[k.lower()] = v
    return out


def load_devices_from_db(db: Path, name_map: Dict[str, str]) -> Dict[str, Dev]:
    """
    Parse NDJSON Zigbee2MQTT database.db and build a dict ieee->Dev.

    Your observed DB keys include:
      type, ieeeAddr, manufName, modelId, endpoints, ...

    We trust ieeeAddr/type/manufName/modelId/endpoints when present.
    """
    out: Dict[str, Dev] = {}

    for line in db.read_text(errors="ignore").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            o = json.loads(line)
        except Exception:
            continue
        if not isinstance(o, dict):
            continue

        ieee = (o.get("ieeeAddr") or "").lower()
        if not ieee:
            continue

        role = str(o.get("type") or "")
        if role.lower() == "coordinator":
            continue

        manuf = str(o.get("manufName") or "")
        model = str(o.get("modelId") or "")
        endpoints = o.get("endpoints") or {}

        old_name = name_map.get(ieee, ieee)

        if ieee not in out:
            out[ieee] = Dev(
                ieee=ieee,
                role=role,
                manuf=manuf,
                model=model,
                endpoints=endpoints,
                old_name=old_name,
            )

    return out


# ---------------------------------------------------------------------------
# Cluster signature (best-effort)
# ---------------------------------------------------------------------------

def clusters_from_endpoints(endpoints: Any) -> List[int]:
    """
    Try to extract Zigbee cluster IDs from database.db endpoints structure.

    This is best-effort because endpoint structures differ by adapter/converter.
    If clusters can't be extracted, returns [].

    Common clusters (hex):
      0x0402 temperature
      0x0405 humidity
      0x0406 occupancy
      0x0500 IAS zone (often contact sensors)
      0x0006 on/off (switches)
      0x0008 level control (dimmers/remotes)
    """
    out: List[int] = []
    try:
        if isinstance(endpoints, dict):
            for _, epd in endpoints.items():
                if not isinstance(epd, dict):
                    continue
                for k in ("inClusterList", "outClusterList", "inClusters", "outClusters"):
                    v = epd.get(k)
                    if isinstance(v, list):
                        for c in v:
                            if isinstance(c, int):
                                out.append(c)

        elif isinstance(endpoints, list):
            for epd in endpoints:
                if not isinstance(epd, dict):
                    continue
                for k in ("inClusterList", "outClusterList", "inClusters", "outClusters"):
                    v = epd.get(k)
                    if isinstance(v, list):
                        for c in v:
                            if isinstance(c, int):
                                out.append(c)
    except Exception:
        pass

    return sorted(set(out))


# ---------------------------------------------------------------------------
# Type (kind) inference
# ---------------------------------------------------------------------------

# Infrastructure models to skip renaming
SKIP_MODELS = {
    "dongle-e_r",
    "slzb-06p10",
}

def infer_kind(d: Dev) -> str:
    """
    Infer semantic kind token (used as <type> in the final name).
    Returns one of:
      router, plug, button, motion, radar, temperaturesensor, doorsensor,
      blinds, vibration, watersensor, unknown, skip

    Priority:
      1) Explicit model rules (very reliable)
      2) Model/manufacturer family rules (Aqara/LUMI, Sonoff, Tuya)
      3) Cluster signatures (if endpoints include cluster lists)
      4) Fallback keywords in old_name
    """
    role = (d.role or "").lower()
    model = (d.model or "").lower()
    manuf = (d.manuf or "").lower()
    old = (d.old_name or "").lower()

    # Skip infrastructure router sticks/gateways
    if model in SKIP_MODELS:
        return "skip"

    # Router baseline
    if role == "router":
        # Some routers are actually semantic sensors (e.g. Tuya TS0601 presence),
        # we override below with model rules.
        return "router"

    # --- Model-specific overrides (high confidence) ---

    # Sonoff SNZB-03 is a PIR motion sensor
    if model == "snzb-03":
        return "motion"

    # Aqara/LUMI common families
    if model.startswith("lumi.remote"):
        return "button"
    if model.startswith("lumi.vibration"):
        return "vibration"
    if model.startswith("lumi.weather") or model.startswith("lumi.sensor_ht"):
        return "temperaturesensor"
    if "sensor_magnet" in model:
        return "doorsensor"
    if "sensor_motion" in model:
        return "motion"

    # Tuya TS0201 is typically temperature/humidity
    if model == "ts0201":
        return "temperaturesensor"

    # Tuya TS0601 is a generic Tuya device class; in your environment the ones that
    # show presence/target_distance are semantically "radar/presence". You asked for
    # specific identification; we'll call these radar.
    if model == "ts0601" and ("_tze" in manuf or "_tz" in manuf):
        return "radar"

    # --- Cluster-based inference (if clusters are available) ---
    cls = clusters_from_endpoints(d.endpoints)

    if 0x0402 in cls or 0x0405 in cls:
        # Temp/humidity sensor
        return "temperaturesensor"
    if 0x0406 in cls:
        return "motion"
    if 0x0500 in cls:
        return "doorsensor"
    if 0x0006 in cls and 0x0008 in cls:
        return "button"

    # --- Keyword fallback from old_name (least reliable) ---
    if "radar" in old or "presence" in old:
        return "radar"
    if "pir" in old or "motion" in old:
        return "motion"
    if "temp" in old or "temperature" in old or "humidity" in old:
        return "temperaturesensor"
    if "door" in old or "window" in old or "contact" in old:
        return "doorsensor"
    if "blind" in old or "shade" in old:
        return "blinds"
    if "vibration" in old or "tilt" in old or "shake" in old:
        return "vibration"
    if "switch" in old or "button" in old or "remote" in old:
        return "button"
    if "plug" in old:
        return "plug"

    return "unknown"


# ---------------------------------------------------------------------------
# Name composition rules
# ---------------------------------------------------------------------------

def compose_name(d: Dev, kind: str, hx: str, loc: str, raw_ieee_only: bool) -> str:
    """
    Apply your naming rules:

    EndDevice:
      - if kind known:   <kind>_<hex>_<location>
      - if kind unknown: enddevice_unknown_<hex>

    Router:
      - router_<hex>_<location>
      - or router_<kind>_<hex>_<location> if kind meaningful (plug/radar/etc.)

    Other:
      - unknown + raw IEEE: dev_<hex>
      - else: <kind|device>_<hex>_<location>
    """
    role = (d.role or "").lower()

    if role == "enddevice":
        if kind != "unknown":
            if not loc:
                loc = "unknown"
            return norm_token(f"{kind}_{hx}_{loc}")
        # unknown enddevice
        return norm_token(f"enddevice_unknown_{hx}")

    if role == "router":
        # allow semantic override: if kind is radar/plug etc, keep it
        if not loc:
            loc = "unknown"
        if kind in ("router", "unknown"):
            return norm_token(f"router_{hx}_{loc}")
        return norm_token(f"router_{kind}_{hx}_{loc}")

    # other role
    if kind == "unknown" and raw_ieee_only:
        return norm_token(f"dev_{hx}")

    if not loc:
        loc = "unknown"
    if kind == "unknown":
        kind = "device"
    return norm_token(f"{kind}_{hx}_{loc}")


# ---------------------------------------------------------------------------
# Main logic: plan and apply
# ---------------------------------------------------------------------------

def main():
    ap = argparse.ArgumentParser(
        description="Batch rename Zigbee2MQTT devices to <type>_<hex>_<location> using database.db (modelId/manufName/endpoints) + your name table."
    )
    ap.add_argument("--db", default="/opt/zigbee2mqtt/data/database.db",
                    help="Path to Zigbee2MQTT database.db (NDJSON).")
    ap.add_argument("--map", default="/srv/z2m_vault_2026-01-25/device_exports_v3/DEVICE_RENAME_MAP_keep65.json",
                    help="Path to IEEE->friendly_name JSON (your authoritative table).")
    ap.add_argument("--broker", default="10.10.20.11",
                    help="MQTT broker host/IP.")
    ap.add_argument("--base-topic", default="zigbee2mqtt",
                    help="Zigbee2MQTT base topic.")
    ap.add_argument("--dry-run", action="store_true", default=True,
                    help="Dry-run (default): print plan only.")
    ap.add_argument("--apply", action="store_true",
                    help="Apply renames via MQTT.")
    ap.add_argument("--force", action="store_true", default=False,
                    help="Rename even if old name is not raw IEEE or dev_*. (Be careful.)")
    ap.add_argument("--only-ieee-names", action="store_true", default=False,
                    help="Only rename devices whose current name (from your map) is exactly 0x....")
    ap.add_argument("--keep-0x", action="store_true", default=True,
                    help="Keep 0x prefix in <hex> token (default).")
    ap.add_argument("--no-0x", dest="keep_0x", action="store_false",
                    help="Remove 0x prefix in <hex> token.")
    args = ap.parse_args()

    if args.apply:
        args.dry_run = False

    db = Path(args.db)
    mp = Path(args.map)
    if not db.exists():
        raise SystemExit(f"DB not found: {db}")
    if not mp.exists():
        raise SystemExit(f"Map not found: {mp}")

    name_map = load_name_map(mp)
    devs = load_devices_from_db(db, name_map)

    planned: List[Tuple[str, str, str, str, str, str]] = []
    # (ieee, role, manuf, model, old, new) - printed as a table

    for ieee, d in devs.items():
        old = (d.old_name or ieee).strip()
        raw_ieee = is_raw_ieee_name(old)

        if args.only_ieee_names and not raw_ieee:
            continue

        kind = infer_kind(d)
        if kind == "skip":
            continue

        hx = ieee_hex(ieee, args.keep_0x)
        loc = extract_location(old)
        new = compose_name(d, kind, hx, loc, raw_ieee)

        # Safety: unless forced, only rename if old is raw IEEE or starts with dev_
        if not args.force:
            if not (raw_ieee or old.lower().startswith("dev_")):
                continue

        if new != old:
            planned.append((ieee, d.role, d.manuf, d.model, old, new))

    # Print plan
    print(f"Devices in DB:   {len(devs)}")
    print(f"Map entries:     {len(name_map)}")
    print(f"Planned renames: {len(planned)}\n")

    print(f"{'IEEE':<20}  {'ROLE':<10}  {'MANUF':<18}  {'MODEL':<18}  {'FROM':<45}  {'TO'}")
    print("-" * 190)
    for ieee, role, manuf, model, old, new in planned:
        print(f"{ieee:<20}  {role:<10}  {manuf[:18]:<18}  {model[:18]:<18}  {old:<45}  {new}")

    if args.dry_run:
        print("\nDry-run only. Re-run with --apply to publish rename requests.")
        return

    # Apply renames via MQTT, using IEEE as "from" (stable, works in your setup)
    topic = f"{args.base_topic}/bridge/request/device/rename"
    for ieee, role, manuf, model, old, new in planned:
        payload = json.dumps({"from": ieee, "to": new})
        mqtt_pub(args.broker, topic, payload, dry_run=False)

    print("\nApplied. Refresh Zigbee2MQTT UI; changes may take a few seconds.")


if __name__ == "__main__":
    main()


Typisches Output:

Devices in DB:      21
Mapping entries:    70
Planned renames:    21

IEEE                  ROLE        KIND                FROM                                           TO

0xe406bffffe215ed1    Router      router              dev_0xe406bffffe215ed1Kitchen                  router_0xe406bffffe215ed1_Kitchen
0x00124b00336c4c12    Router      router              0x00124b00336c4c12                             router_0x00124b00336c4c12_unknown
0xe406bffffe216090    Router      router              dev_0xe406bffffe216090BathroomUpstairs         router_0xe406bffffe216090_BathroomUpstairs
0x94b216fffe3cbd0a    Router      router              0x94b216fffe3cbd0a                             router_0x94b216fffe3cbd0a_unknown
0x94b216fffe3cc0d8    Router      router              0x94b216fffe3cc0d8                             router_0x94b216fffe3cc0d8_unknown
0x00158d000af2ae8a    EndDevice   button              dev_0x00158d000af2ae8aOfficeDesk               button_0x00158d000af2ae8a_OfficeDesk
0x00158d008bbdfe39    EndDevice   button              0x00158d008bbdfe39                             button_0x00158d008bbdfe39_unknown
0x00158d008bbe0254    EndDevice   button              OfficeHeaterSwitch0x00158d008bbe0254           button_0x00158d008bbe0254_OfficeHeaterSwitch
0x00158d00087b8447    EndDevice   button              BedsideSwitch0x00158d00087b8447                button_0x00158d00087b8447_BedsideSwitch
0xa4c1389b2ba77c40    EndDevice   unknown             Ankleide0xa4c1389b2ba77c40                     enddevice_unknown_0xa4c1389b2ba77c40
0xa4c1383c9457544b    Router      router              dev_0xa4c1383c9457544bBathOG                   router_0xa4c1383c9457544b_BathOG
0x00158d0007c0b727    EndDevice   button              switch cosmetic mirror 0x00158d0007c0b727      button_0x00158d0007c0b727_cosmetic_mirror
0xa4c13838f0429b33    Router      router              dev_0xa4c13838f0429b33Kitchen                  router_0xa4c13838f0429b33_Kitchen
0xa4c138b81761a155    EndDevice   unknown             WashRoom_0xa4c138b81761a155                    enddevice_unknown_0xa4c138b81761a155
0xa4c1384874f54605    EndDevice   unknown             0xa4c1384874f54605                             enddevice_unknown_0xa4c1384874f54605
0x00158d00067a11da    EndDevice   vibration           dev_0x00158d00067a11daStatinBox                vibration_0x00158d00067a11da_StatinBox
0xa4c138d821780388    EndDevice   unknown             BathroomFloor0xa4c138d821780388                enddevice_unknown_0xa4c138d821780388
0xa4c138eaff76cf8a    EndDevice   unknown             dev_0xa4c138eaff76cf8aBedroomOG                enddevice_unknown_0xa4c138eaff76cf8a
0x00158d00067bee32    EndDevice   unknown             dev_0x00158d00067bee32Ultimaker                enddevice_unknown_0x00158d00067bee32
0xa4c13879363e2414    EndDevice   unknown             ShelterFloor_0xa4c13879363e2414                enddevice_unknown_0xa4c13879363e2414
0xa4c138deb45e5834    EndDevice   unknown             TerraceFloor_0xa4c138deb45e5834                enddevice_unknown_0xa4c138deb45e5834