diff --git a/CHANGELOG.md b/CHANGELOG.md index d053306..877ee67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **`suggest.py`: multi-output actuators are split by their vendor object marker.** Zennio-style + devices (Lumento DX4, MAXinBOX, KLIC-DI…) put every output of a device into ONE ETS channel and + separate them only in the object text (`[1] Switch On/Off`, `[2] On/Off (Status)`). The channel + classifier used to see several switch commands in one channel, give up, and let every status object + fall through to a sensor suggestion. Channels are now split into vendor sub-units first. On a real + 3646-GA project: entities 341 → 513, sensor noise 752 → 367, agreement with the name/Function engine + 92 % → 95 % (platform) and 94 % → 95 % (address keys). +- **Device diagnostics are no longer suggested as sensors** (error flags, communication failures, + firmware/version, bus voltage, reset — matched in the vendor object text or the GA name, EN/DE/RU). + They are counted in `hints["diagnostics_skipped"]` and kept out of the name-based fallback as well. + 53 such objects on the same project. + ## [0.8.1] - 2026-09-12 ### Added diff --git a/nickol_knx_mcp/suggest.py b/nickol_knx_mcp/suggest.py index 0f48089..2d3ecbe 100644 --- a/nickol_knx_mcp/suggest.py +++ b/nickol_knx_mcp/suggest.py @@ -25,6 +25,7 @@ Pure functions, no HA imports — the HA provider class is a ~30-line wrapper. """ from __future__ import annotations +import re from collections import defaultdict from typing import Any, Optional @@ -106,6 +107,41 @@ def _links_of(project: dict[str, Any], co_ids, shared: Optional[set[str]] = None return out +_SUBUNIT_RE = re.compile(r"^\s*\[([^\]]{1,6})\]") +# object texts that describe the device's own health/plumbing, not a building sensor +_DIAG_WORDS = ("error", "fehler", "diagnos", "heartbeat", "watchdog", "alive", "version", + "firmware", "communication fail", "bus voltage", "reset", "scene number", + "identification", "störung", "ошибк", "авари", "диагност") + + +def _subunit(link: "_Link") -> Optional[str]: + """Vendor sub-unit marker of a communication object, e.g. "[2] Switch On/Off" -> "2". + + Multi-output actuators (Zennio Lumento/MAXinBOX, many others) put every output of a + device into ONE ETS channel and separate them only in the object text. Without this + split a 4-output dimmer looks like one channel with four switch commands, the channel + is abandoned, and all of its status objects fall through to sensor suggestions. + """ + m = _SUBUNIT_RE.match(link.text or "") + return m.group(1).strip() if m else None + + +def _split_subunits(links: list["_Link"]) -> list[tuple[Optional[str], list["_Link"]]]: + """Split a channel into vendor sub-units; [(None, links)] when there is nothing to split.""" + groups: dict[str, list[_Link]] = defaultdict(list) + loose: list[_Link] = [] + for l in links: + t = _subunit(l) + (groups[t] if t else loose).append(l) + if len(groups) < 2: + return [(None, links)] + out: list[tuple[Optional[str], list[_Link]]] = [(t, ls) for t, ls in groups.items()] + if loose: + # objects without a marker (device-wide: errors, scene, temperature) stay separate + out.append((None, loose)) + return out + + def _pick(links: list[_Link], want: str, pred, used: set[str]) -> list[_Link]: """Links matching ``pred`` in role preference for ``want`` ('sink' or 'source'); ``dual`` links serve as the missing side. Never reuses a GA.""" @@ -287,9 +323,15 @@ def _classify_channel(project: dict[str, Any], links: list[_Link], chan_name: st return None -def _sensor_suggestions(project: dict[str, Any], links: list[_Link], sink_gas: set[str] +def _sensor_suggestions(project: dict[str, Any], links: list[_Link], sink_gas: set[str], + skipped: Optional[list] = None ) -> list[tuple[list[str], dict[str, dict[str, Any]], dict[str, Any], _Link]]: - """State-only channels (no sink anywhere for the GA) → sensor / binary_sensor per GA.""" + """State-only channels (no sink anywhere for the GA) → sensor / binary_sensor per GA. + + Device diagnostics (error flags, communication failures, firmware/version objects) are + dropped: they are real KNX objects but not entities a user wants suggested; they land in + ``hints["diagnostics_skipped"]`` instead.""" + skipped = skipped if skipped is not None else [] out = [] for l in links: if l.role != "source" or l.ga in sink_gas: @@ -297,6 +339,9 @@ def _sensor_suggestions(project: dict[str, Any], links: list[_Link], sink_gas: s m, s = l.dpt if m is None: continue + if _has(l.text, _DIAG_WORDS) or _has(l.name, _DIAG_WORDS): + skipped.append(l) # device health, not a building sensor + continue meta = {"tier": "structural", "evidence": ["channel", "flags:transmit-only", "dpt"], "review": []} if m == 1: out.append((["binary_sensor"], {"binary_sensor": {"ga_sensor": _ga_conf(l, "state")}}, meta, l)) @@ -396,7 +441,7 @@ def suggest_entities(project: dict[str, Any], *, skip_fb_covered: bool = True, covered: set[str] = set() # GAs explained by a structural suggestion primaries: set[str] = set() # primary GA per emitted entity (dedupe) stats = {"channels": 0, "skipped_fb_covered": 0, "structural": 0, "sensors": 0, - "fallback": 0, "review": 0, "duplicates_skipped": 0, "pseudo_channels": 0, "unwired_flagged": 0} + "fallback": 0, "review": 0, "duplicates_skipped": 0, "pseudo_channels": 0, "unwired_flagged": 0, "subunits": 0, "diagnostics_skipped": 0} for dev_addr, dev in (project.get("devices") or {}).items(): dev_name = dev.get("name") or dev.get("hardware_name") or dev_addr @@ -415,12 +460,28 @@ def suggest_entities(project: dict[str, Any], *, skip_fb_covered: bool = True, continue if not links: continue - res = _classify_channel(project, links, ch.get("name") or "") - items = [res] if res else [] - taken = {m["address"] for m in _matched(res[1][res[0][0]], gas)} if res else set() - sens = _sensor_suggestions(project, links, sink_gas | taken) + units = _split_subunits(links) + if len(units) > 1: + stats["subunits"] += len(units) + items: list = [] + taken: set[str] = set() + sens: list = [] + for unit_id, unit_links in units: + res = _classify_channel(project, unit_links, ch.get("name") or "") + if res: + items.append((*res, unit_id)) + taken |= {m["address"] for m in _matched(res[1][res[0][0]], gas)} + for unit_id, unit_links in units: + dropped: list = [] + sens.extend(_sensor_suggestions(project, unit_links, sink_gas | taken, dropped)) + stats["diagnostics_skipped"] += len(dropped) + # a diagnostics object stays out of the name-based fallback too, otherwise the + # engine resurrects it one step later as a binary sensor + covered.update(l.ga for l in dropped) for platforms, confs, meta, *rest in items + sens: - lnk = rest[0] if rest else None + lnk = rest[0] if rest and isinstance(rest[0], _Link) else None + unit = rest[0] if rest and not isinstance(rest[0], _Link) else ( + _subunit(lnk) if lnk else None) first = confs[platforms[0]] mg = _matched(first, gas) prim = next((m["address"] for m in mg if m["address"] in @@ -442,11 +503,12 @@ def suggest_entities(project: dict[str, Any], *, skip_fb_covered: bool = True, primaries.add(prim) name = (lnk.name if lnk else (ch.get("name") or "")) or \ _common_prefix_name([m["name"] for m in mg]) or dev_name - sid = f"{dev_addr}_{ch_id}" + (f"_{lnk.ga}" if lnk else "") + sid = f"{dev_addr}_{ch_id}" + (f"_{unit}" if unit else "") + (f"_{lnk.ga}" if lnk else "") suggestions.append({ "id": sid, "source": PROVIDER_ID, "suggested_name": name, "group_id": dev_addr, "group_name": dev_name, - "secondary_info": ch.get("name") or "", + "secondary_info": (f"{ch.get('name') or ''} [{unit}]".strip() if unit + else (ch.get("name") or "")), "platform_options": platforms, "suggestions": {p: {"knx": c, "matched_group_addresses": _matched(c, gas), "unmatched_dpas": []} for p, c in confs.items()}, diff --git a/tests/test_suggest.py b/tests/test_suggest.py index 4633762..2190438 100644 --- a/tests/test_suggest.py +++ b/tests/test_suggest.py @@ -16,8 +16,8 @@ def _ga(a, dpt, name=None): return {"name": name or f"GA {a}", "address": a, "description": "", "dpt": dpt} -def _co(dev, links, *, write=False, transmit=False, read=False, dpts=None, channel=None): - return {"name": "", "number": 0, "text": "", "function_text": "", "description": "", +def _co(dev, links, *, write=False, transmit=False, read=False, dpts=None, channel=None, text=""): + return {"name": "", "number": 0, "text": text, "function_text": "", "description": "", "device_address": dev, "device_application": None, "module_def": None, "channel": channel, "dpts": dpts or [], "object_size": "", "flags": {"read": read, "write": write, "communication": True, "transmit": transmit, @@ -44,6 +44,14 @@ def _project(): ("3/0/1", TEMP), ("3/0/2", TEMP), ("3/0/3", TEMP), ("3/0/4", MODE), ("3/0/5", MODE), ("4/0/1", TEMP), ("4/0/2", SW), ("7/0/1", SW), ]} + # multi-output actuator: vendor puts every output in ONE channel, separated only by the + # object text marker "[n]" (Zennio Lumento/MAXinBOX pattern). Must yield TWO entities, + # not one abandoned channel plus a pile of sensors. + for i, base in ((1, 9), (2, 12)): + gas[f"9/0/{base}"] = _ga(f"9/0/{base}", SW) + gas[f"9/0/{base+1}"] = _ga(f"9/0/{base+1}", SW) + gas[f"9/0/{base+2}"] = _ga(f"9/0/{base+2}", PCT) + gas["9/9/9"] = _ga("9/9/9", SW, "Aktor Sammelstörung") gas["8/0/1"] = _ga("8/0/1", SW, "Kitchen socket switch") gas["8/0/2"] = _ga("8/0/2", SW, "Kitchen socket switch status") cos = { @@ -83,6 +91,15 @@ def _project(): "co-81": _co("1.4.1", ["4/0/2"], transmit=True, channel="S-1"), # FB-covered channel: must be skipped (the FB provider owns it) "co-90": _co("1.6.1", ["7/0/1"], write=True, channel="CH-1"), + # multi-output dimmer, one channel, outputs distinguished by "[1]" / "[2]" + "co-m1c": _co("1.8.1", ["9/0/9"], write=True, channel="CH-1", text="[1] Switch On/Off"), + "co-m1s": _co("1.8.1", ["9/0/10"], transmit=True, read=True, channel="CH-1", text="[1] On/Off (Status)"), + "co-m1d": _co("1.8.1", ["9/0/11"], write=True, channel="CH-1", text="[1] Absolute Dimming"), + "co-m2c": _co("1.8.1", ["9/0/12"], write=True, channel="CH-1", text="[2] Switch On/Off"), + "co-m2s": _co("1.8.1", ["9/0/13"], transmit=True, read=True, channel="CH-1", text="[2] On/Off (Status)"), + "co-m2d": _co("1.8.1", ["9/0/14"], write=True, channel="CH-1", text="[2] Absolute Dimming"), + # device diagnostics on the same device: a real object, but not an entity to suggest + "co-diag": _co("1.8.1", ["9/9/9"], transmit=True, channel="CH-1", text="Internal Error: Communication"), # device WITHOUT channels, named GAs → fallback (name pairing) "co-95": _co("1.7.1", ["8/0/1"], write=True), "co-96": _co("1.7.1", ["8/0/2"], transmit=True), @@ -97,6 +114,7 @@ def _project(): "1.3.1": _dev("1.3.1", "Raumtemperaturregler", {"CH-1": _ch("RTR", ["co-70", "co-71", "co-72", "co-73", "co-74"])}), "1.4.1": _dev("1.4.1", "Sensor", {"S-1": _ch("Messwerte", ["co-80", "co-81"])}), "1.6.1": _dev("1.6.1", "Modern Aktor", {"CH-1": _ch("Ausgang FB", ["co-90"], fbs=["417"])}), + "1.8.1": _dev("1.8.1", "Dimmaktor 2-fach", {"CH-1": _ch("LED", ["co-m1c", "co-m1s", "co-m1d", "co-m2c", "co-m2s", "co-m2d", "co-diag"])}), "1.7.1": _dev("1.7.1", "Old Aktor", {}), } return {"info": {"name": "suggest-test", "group_address_style": "ThreeLevel", "schema_version": "21", @@ -170,14 +188,31 @@ def main(): assert f["suggestions"]["switch"]["knx"]["ga_switch"] == {"write": "8/0/1", "state": "8/0/2"} assert h["pseudo_channels"] == 1 and h["fallback"] == 0, h - # 10. no GA appears in two suggestions + # 10. multi-output channel splits by the vendor "[n]" marker: two lights, statuses consumed, + # and no sensor fallout from those objects + m1, m2 = by["1.8.1_CH-1_1"], by["1.8.1_CH-1_2"] + assert m1["suggestions"]["light"]["knx"] == { + "ga_switch": {"write": "9/0/9", "state": "9/0/10"}, + "ga_brightness": {"write": "9/0/11"}}, m1["suggestions"]["light"]["knx"] + assert m2["suggestions"]["light"]["knx"]["ga_switch"] == {"write": "9/0/12", "state": "9/0/13"} + assert "[1]" in m1["secondary_info"] and "[2]" in m2["secondary_info"], m1["secondary_info"] + assert not any(s2["id"].startswith("1.8.1_") and s2["platform_options"][0].endswith("sensor") + for s2 in res["suggestions"]), "multi-output statuses leaked into sensors" + assert h["subunits"] >= 2, h + + # 11. device diagnostics are dropped, not suggested, and counted + assert h["diagnostics_skipped"] == 1, h + assert not any("9/9/9" in str(s2["suggestions"]) for s2 in res["suggestions"]), "diagnostics suggested" + + # 12. no GA appears in two suggestions seen = {} for s2 in res["suggestions"]: for p, ps in s2["suggestions"].items(): for m in ps["matched_group_addresses"]: assert seen.setdefault(m["address"], s2["id"]) == s2["id"], f"{m['address']} in two suggestions" print(f"test_suggest: OK — {len(res['suggestions'])} suggestions from structure alone on a nameless " - f"fixture (FB-provider parity for switch/cover/TW/RGB), climate, sensors, FB-skip, name fallback; hints={h}") + f"fixture (FB-provider parity for switch/cover/TW/RGB), climate, sensors, FB-skip, " + f"multi-output split, diagnostics filter, pseudo-channels; hints={h}") if __name__ == "__main__":