public_git/nickol-knx-mcp
1
0
Fork 0
mirror of https://github.com/NickoScope/nickol-knx-mcp.git synced 2026-09-30 03:41:58 +02:00
Code Issues Packages Projects Releases Wiki Activity

suggest: split multi-output channels by vendor object marker; drop device diagnostics

Browse Source 
- _subunit()/_split_subunits(): '[1] Switch On/Off' style markers separate the outputs a vendor
  packs into one ETS channel; each sub-unit is classified on its own (ids and secondary_info carry it)
- diagnostics (error/communication/firmware/version/bus voltage/reset, EN+DE+RU) never become sensors
  and are excluded from the name-based fallback; counted in hints.diagnostics_skipped
- real-project effect (3646 GAs): entities 341->513, sensors 752->367, platform agreement 92->95 %,
  address-key agreement 94->95 %
- tests: multi-output split (two lights, statuses consumed, no sensor fallout) + diagnostics filter
This commit is contained in:
Nikolay Miroshnichenko
2026-09-12 09:09:08 +02:00
parent 80fc279ac1
commit 07352cdb61
3 changed files with 125 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+14
View File
@@ -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
nickol_knx_mcp/suggest.py
+72 -10
View File
@@ -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()},
tests/test_suggest.py
+39 -4
View File
@@ -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__":