From 0198ae3019051fac9bf32f52ed065b3d50e2050e Mon Sep 17 00:00:00 2001 From: Nikolay Miroshnichenko Date: Sat, 29 Aug 2026 13:13:59 +0200 Subject: [PATCH] fix(#11): pair every cover to its own step/stop across all 3 real ETS Function structures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second pass on issue #11, grounded in Kris1166's real ETS-6 field dump (groups A/B/C), after the first fix (dd8ecac) proved partial. Three levers: - project.py: ETS Function role (MoveUpDown/StopStepUpDown) authoritatively promotes a GA to category=shutter — rescues bare DPT 1.007 step/stops the name classifier left 'unknown' (group A: 26/33 covers were dropped). Guarded to DPT 1.x/5.x; role strings from Kris's real dump, not invented. - generate_ha.py: cover builder now gathers candidates and picks the BEST per role (same-function > name-token overlap > nearest sub) instead of first-match, so a function-less roller takes its own step/stop, not an awning's sharing the zone token (group C). Bare 1.007/1.010 admitted only with a shutter name signal, so a foreign lighting step can't be stolen (audit finding). - tests: test_cover_pairing.py rebuilt from the real A/B/C structures + a group-D guard for the foreign-1.007 case. Fails without the fix, passes with it. Co-Authored-By: Claude Opus 4.8 --- nickol_knx_mcp/generate_ha.py | 77 +++++++++++----- nickol_knx_mcp/project.py | 35 ++++++++ tests/test_cover_pairing.py | 161 +++++++++++++++++----------------- 3 files changed, 171 insertions(+), 102 deletions(-) diff --git a/nickol_knx_mcp/generate_ha.py b/nickol_knx_mcp/generate_ha.py index 91e1f94..c5e2f1d 100644 --- a/nickol_knx_mcp/generate_ha.py +++ b/nickol_knx_mcp/generate_ha.py @@ -57,6 +57,27 @@ def _is_stop(name: str) -> bool: return any(w in low for w in _STOP_WORDS) +def _is_shutter_control(g, slat_addrs) -> bool: + """A bare step/stop object is a shutter control even when the classifier left + its category 'unknown' for want of a name keyword or Function role — the + function-less collective case (issue #11 group C). DPT 1.007 (step) / 1.010 + (start-stop), or a slat object. Position 5.001 still requires a shutter + category (it classifies reliably), so it is intentionally NOT admitted here. + + The bare-DPT admission REQUIRES a shutter-ish name signal (stop / up-down / a + generic blind word): DPT 1.007/1.010 alone is domain-agnostic, so admitting it + on the DPT alone would let a foreign 1.007 (e.g. a lighting relative-dim step) + sharing only a zone token be mis-paired as a cover's step/stop — audit finding + on the issue #11 fix. A slat address is already shutter-derived, so it stands.""" + if g.address in slat_addrs: + return True + if g.dpt_main == 1 and g.dpt_sub in (7, 10): + low = (g.name or "").lower() + return (_is_stop(g.name) or _is_updown(g.name) + or any(w in low for w in _BLIND_GENERIC)) + return False + + # Function words that differ between a command and its status (on/off, value, # state, brightness, ...). Stripping them leaves the device/zone identity, so a # command pairs to its feedback even when the identity is a single token @@ -235,40 +256,52 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: entity = {"name": ga.name, "move_long_address": ga.address} ptoks = _ident_tokens(ga.name) my_fns = addr_to_fns.get(ga.address, set()) + # Gather the eligible siblings once, then pick the BEST candidate per role + # rather than first-match — so a step/stop that shares the move's type AND + # zone token beats one that shares the zone alone (a "West Side Roller + # Shutters" move must take its own step/stop, not the "West Side Awnings" + # one) — issue #11 group C. + cands = [] # (sib, same_fn, overlap) for sib in same_main_gas(ga): if sib.address in consumed or sib.address == ga.address: continue - if sib.category != "shutter": + # Admit shutter-category siblings AND bare step/stop DPTs the + # classifier left 'unknown' (function-less collectives — issue #11). + if sib.category != "shutter" and not _is_shutter_control(sib, slat_addrs): continue sib_fns = addr_to_fns.get(sib.address, set()) same_fn = bool(sib_fns & my_fns) # Never cross an ETS Function boundary: a sibling owned by a DIFFERENT # function is another shutter's GA even when the zone token matches - # (a window vs an awning in the same room) — issue #11. + # (a window vs an awning in the same room) — issue #11 group B. if sib_fns and not same_fn: continue + overlap = len(ptoks & _ident_tokens(sib.name)) # A same-function sibling is authoritative (names not needed); - # otherwise require a shared zone identity as before. - if not same_fn and ptoks and _ident_tokens(sib.name) \ - and not (ptoks & _ident_tokens(sib.name)): + # otherwise require a shared zone/type identity as before. + if not same_fn and ptoks and _ident_tokens(sib.name) and overlap == 0: continue - is_stop = (sib.dpt_main == 1 and sib.dpt_sub in (7, 10, 17)) or _is_stop(sib.name) - if is_stop and "move_short_address" not in entity: - entity["move_short_address"] = sib.address - consumed.add(sib.address) - elif sib.address in slat_addrs and sib.dpt_main == 1 \ - and "move_short_address" not in entity: - # venetian slat (tilt) = the short-move/step of this blind - entity["move_short_address"] = sib.address - consumed.add(sib.address) - elif sib.dpt_main == 5 and sib.kind == "command" \ - and "position_address" not in entity: - entity["position_address"] = sib.address - consumed.add(sib.address) - elif sib.dpt_main == 5 and _is_status_ga(sib) \ - and "position_state_address" not in entity: - entity["position_state_address"] = sib.address - consumed.add(sib.address) + cands.append((sib, same_fn, overlap)) + + def _rank(item): + # same ETS Function first (authoritative), then the strongest name + # overlap (type+zone beats zone alone), then the nearest sub index. + sib, same_fn, overlap = item + return (1 if same_fn else 0, overlap, -abs((sib.sub or 0) - (ga.sub or 0))) + + def _take(pred, key): + pool = [c for c in cands + if c[0].address not in consumed and pred(c[0])] + best = max(pool, key=_rank, default=None) + if best is not None: + entity[key] = best[0].address + consumed.add(best[0].address) + + # step/stop or slat -> the short move; then position command; then status. + _take(lambda s: (s.dpt_main == 1 and s.dpt_sub in (7, 10, 17)) + or _is_stop(s.name) or s.address in slat_addrs, "move_short_address") + _take(lambda s: s.dpt_main == 5 and s.kind == "command", "position_address") + _take(lambda s: s.dpt_main == 5 and _is_status_ga(s), "position_state_address") covers.append(entity) consumed.add(ga.address) # A3: surface the actuator-dependent flags that are NOT in the .knxproj — diff --git a/nickol_knx_mcp/project.py b/nickol_knx_mcp/project.py index dafa9cd..c454539 100644 --- a/nickol_knx_mcp/project.py +++ b/nickol_knx_mcp/project.py @@ -178,6 +178,17 @@ def _classify_category(name: str, main_name: str, middle_name: str, return dpt_cat +# ETS Function roles that unambiguously mark a shutter/blind object. Matched as +# substrings of the (case-folded) role, so "MoveUpDown"/"StopStepUpDown" both hit. +# Provenance: the exact role strings "MoveUpDown" and "StopStepUpDown" are taken +# from Kris1166's real ETS-6 sunblind Function dump (issue #11 field data via +# explain_ga) — not invented. The list may not be exhaustive for every ETS +# FunctionType; extend it if another real project surfaces a shutter role that +# isn't covered (source over assumption). +_SHUTTER_ROLE_TOKENS = ("updown", "moveup", "movedown", "stepupdown", + "stopstep", "stepstop", "slat") + + @dataclass class GARecord: """Enriched group-address record used by all analysis tools.""" @@ -345,6 +356,30 @@ def build_loaded_from_raw(raw: KNXProject, path: str) -> LoadedProject: ) gas[rec.address] = rec + # Authoritative shutter classification from ETS Function roles. A GA whose + # Function role is a shutter role (MoveUpDown / StopStepUpDown / slat) IS a + # shutter object — even when its bare DPT is domain-agnostic (1.007 step, + # 1.010 start/stop) and its name carries no shutter keyword. The Function + # role outranks DPT and name (explain_ga's own hierarchy), so it rescues the + # step/stop GAs a name-only classifier leaves 'unknown' (issue #11, field + # data by Kris1166). Never demotes: only promotes a non-shutter GA. + for fn in (raw.get("functions", {}) or {}).values(): + for a_key, ref in (fn.get("group_addresses", {}) or {}).items(): + role = (ref.get("role") or "").lower() + if not any(t in role for t in _SHUTTER_ROLE_TOKENS): + continue + rec = gas.get(ref.get("address") or a_key) + if rec is None or rec.category == "shutter": + continue + # Guard: a shutter role only promotes a plausibly-shutter DPT — a 1-bit + # control (move/step/stop) or a 5.x position — never retype a 9.x + # temperature / 13.x energy GA on a role substring like "updown". + if rec.dpt_main not in (1, 5): + continue + rec.category = "shutter" + if rec.ha_platform in ("light", "switch", "unknown"): + rec.ha_platform = "cover" + return LoadedProject( path=path, info=dict(raw.get("info", {})), diff --git a/tests/test_cover_pairing.py b/tests/test_cover_pairing.py index e544146..737c13b 100644 --- a/tests/test_cover_pairing.py +++ b/tests/test_cover_pairing.py @@ -1,18 +1,27 @@ -"""Regression: generate_ha_package must not cross-wire a cover's step/stop -(move_short) across ETS Function boundaries (issue #11, field report by Kris1166). +"""Regression: generate_ha_package must pair every cover to ITS OWN step/stop +(move_short) and position status — never cross-wire, never drop its own — across +the three ETS Function structures a real ETS 6 project actually contains. -A room that holds both a window and an awning shares a zone token ("Room A"), so -name-token pairing alone grabbed the wrong shutter's Step/Stop — and, because the -correct one then looked "taken", a cleanly-named cover could end up with no -move_short at all. ETS Function membership is authoritative: a step/stop owned by -a DIFFERENT function is another shutter's GA and must never be borrowed. +Field data by Kris1166 (issue #11), captured with `explain_ga` on the live tool +and mirrored here 1:1 (room names anonymised to RoomA/B/C; compass group names +kept). Three structures, all present in one project: -Fixture mirrors the reporter's anonymised table exactly (main group 2 = shutters, -3-level; move=1.008 at 2/1/N, step/stop=1.007 at 2/2/N, status position=5.001 at -2/3/N), each window/awning/door grouped in its own ETS Function. The "West Side -Roller Shutters" collective move (2/5/10) has NO own function, its step/stop -(2/5/11) is unlinked, and "West Side Awnings" (2/5/15/16) is a function — so the -collective must fall back to its own 2/5/11 and never steal 2/5/16. + A. Per-channel ETS Function, a generic type-only display name ("Fenster", + "Doppeltür") shared across many channels. Move + step/stop + status all in + the SAME per-channel function. The step/stop GA is DPT 1.007 named only + "Schritt/Stop" — no shutter keyword — so the name classifier leaves it + category 'unknown'. It must still be recognised (its Function role is + StopStepUpDown) and paired to its own move. (26 of 33 covers were dropped.) + B. Actuator-level Function per physical unit ("Markise RoomA/B"), name carries + the zone. Its zone token collides with a plain window from group A — the + Function boundary must stop the window from stealing the awning's step/stop. + C. Function-less collective ("Westseite Rollläden" vs "Westseite Markisen") — + both share the zone token "Westseite" and differ only by the TYPE token, no + ETS Function at all. Pure name territory: the type token must decide, so the + roller move takes its own step/stop, not the awning's. + +Before the fix: A → no move_short (category gate drops the 'unknown' step/stop); +C → roller cross-wires to the awning's step/stop (zone token alone matched). """ from nickol_knx_mcp.project import build_loaded_from_raw from nickol_knx_mcp.generate_ha import generate_ha_yaml @@ -24,74 +33,78 @@ def _ga(addr, name, dmain, dsub): "communication_object_ids": [], "description": "", "comment": ""} -def _fn(name, *addrs): +def _fn(name, *addr_roles): return {"name": name, - "group_addresses": {a: {"address": a, "role": ""} for a in addrs}} - - -_SHUTTERS = { - 0: "Room A Window 1", - 1: "Room A Window 2", - 2: "Room C Double Door", # non-colliding control — must keep its own step/stop - 3: "Room B Double Door", - 50: "Awning Balcony Room A", - 51: "Awning Balcony Room C", - 52: "Awning Balcony Room B", -} + "group_addresses": {a: {"address": a, "role": r} for a, r in addr_roles}} def _build(): gas: dict = {} - functions: dict = {} - # Insertion order matters: the cover builder scans GAs in project order and - # first-match wins. We insert the AWNINGS' move commands BEFORE the windows' — - # the ordering under which an awning, sharing the "Room A" zone token, grabs a - # window's Step/Stop (the field-reported cross-wiring). Without the ETS-Function - # boundary this steals across shutters; with it, each stays in its own function. - for base in (50, 51, 52, 0, 1, 2, 3): # awning moves first - a = f"2/1/{base}" - gas[a] = _ga(a, f"{_SHUTTERS[base]} Move", 1, 8) - for base in (0, 1, 2, 3, 50, 51, 52): # then all step/stops - a = f"2/2/{base}" - gas[a] = _ga(a, f"{_SHUTTERS[base]} Step/Stop", 1, 7) - for base in (0, 1, 2, 3, 50, 51, 52): # then all position statuses - a = f"2/3/{base}" - gas[a] = _ga(a, f"{_SHUTTERS[base]} Status Position", 5, 1) - for base, name in _SHUTTERS.items(): # each shutter = its own ETS Function - functions[name] = _fn(name, f"2/1/{base}", f"2/2/{base}", f"2/3/{base}") - # West Side: collective roller-shutter move with NO own function; its own - # step/stop unlinked; the awnings ARE a function. - gas["2/5/10"] = _ga("2/5/10", "West Side Roller Shutters Move", 1, 8) - gas["2/5/11"] = _ga("2/5/11", "West Side Roller Shutters Step/Stop", 1, 7) - gas["2/5/15"] = _ga("2/5/15", "West Side Awnings Move", 1, 8) - gas["2/5/16"] = _ga("2/5/16", "West Side Awnings Step/Stop", 1, 7) - functions["West Side Awnings"] = _fn("West Side Awnings", "2/5/15", "2/5/16") + fns: dict = {} + # A. per-channel functions, shared generic display name, shutter roles. + for sub, zone, fname in [(26, "RoomA Fenster", "Fenster"), + (3, "RoomB Doppeltür", "Doppeltür")]: + gas[f"2/1/{sub}"] = _ga(f"2/1/{sub}", f"{zone} Bewegen", 1, 8) + gas[f"2/2/{sub}"] = _ga(f"2/2/{sub}", f"{zone} Schritt/Stop", 1, 7) # bare 1.007 + gas[f"2/3/{sub}"] = _ga(f"2/3/{sub}", f"{zone} Status Position", 5, 1) + fns[f"FN-{fname}-{sub}"] = _fn(fname, (f"2/1/{sub}", "MoveUpDown"), + (f"2/2/{sub}", "StopStepUpDown"), + (f"2/3/{sub}", "PositionStatus")) + # B. actuator-level functions, name carries the (colliding) zone token. + for sub, zone, fname in [(50, "RoomA Markise", "Markise RoomA"), + (52, "RoomB Markise", "Markise RoomB")]: + gas[f"2/1/{sub}"] = _ga(f"2/1/{sub}", f"{zone} Bewegen", 1, 8) + gas[f"2/2/{sub}"] = _ga(f"2/2/{sub}", f"{zone} Schritt/Stop", 1, 7) + gas[f"2/3/{sub}"] = _ga(f"2/3/{sub}", f"{zone} Status Position", 5, 1) + fns[f"FN-{fname}"] = _fn(fname, (f"2/1/{sub}", ""), + (f"2/2/{sub}", ""), (f"2/3/{sub}", "")) + # C. function-less collective: same zone, different type token. + gas["2/5/10"] = _ga("2/5/10", "Westseite Rollläden auf/ab", 1, 8) + gas["2/5/11"] = _ga("2/5/11", "Westseite Rollläden Start/Stopp", 1, 7) + gas["2/5/15"] = _ga("2/5/15", "Westseite Markisen auf/ab", 1, 8) + gas["2/5/16"] = _ga("2/5/16", "Westseite Markisen Start/Stopp", 1, 7) + # D. a foreign bare 1.007 (a lighting relative-dim step) sharing only the zone + # token must NOT be stolen as the roller's step/stop — the roller takes its own + # real stop (audit finding on this fix; the bare-DPT admission needs a shutter + # name signal, and "…Schritt" carries none). + gas["2/6/0"] = _ga("2/6/0", "Nordseite Rollläden auf/ab", 1, 8) + gas["2/6/1"] = _ga("2/6/1", "Nordseite Licht Schritt", 1, 7) # foreign step + gas["2/6/2"] = _ga("2/6/2", "Nordseite Rollläden Stopp", 1, 10) # own stop raw = {"info": {"group_address_style": "ThreeLevel", "schema_version": "21"}, "group_addresses": gas, "communication_objects": {}, "devices": {}, - "functions": functions, "topology": {}, - "group_ranges": {"2": {"address_start": 4096, "name": "Shutters/Blinds", + "functions": fns, "topology": {}, + # Main-2 named in German that our keyword list does NOT map to shutter, + # so a bare 1.007 "Schritt/Stop" stays 'unknown' — mirrors the field + # project (the bug only shows when the range name doesn't rescue it). + "group_ranges": {"2": {"address_start": 4096, "name": "Beschattung", "group_ranges": {}}}} return build_loaded_from_raw(raw, "cover-pairing.knxproj") +def _covers(res): + import yaml as _yaml + doc = _yaml.safe_load(res["yaml"]) + return (doc or {}).get("knx", {}).get("cover", []) or [] + + def main(): project = _build() - res = generate_ha_yaml(project) - covers = {c["move_long_address"]: c for c in - [e for e in _covers(res)]} + covers = {c["move_long_address"]: c for c in _covers(generate_ha_yaml(project))} - # Each shutter must pair to ITS OWN step/stop and status, never a sibling's. + # Every cover must pair to ITS OWN step/stop and position status. expected = { - "2/1/0": ("2/2/0", "2/3/0"), # Room A Window 1 (NOT 2/2/50) - "2/1/1": ("2/2/1", "2/3/1"), # Room A Window 2 (NOT 2/2/51) - "2/1/2": ("2/2/2", "2/3/2"), # Room C Double Door — keeps its own - "2/1/3": ("2/2/3", "2/3/3"), # Room B Double Door (NOT 2/2/52) - "2/1/50": ("2/2/50", "2/3/50"), # Awning Balcony Room A - "2/1/51": ("2/2/51", "2/3/51"), - "2/1/52": ("2/2/52", "2/3/52"), - "2/5/15": ("2/5/16", None), # West Side Awnings (own function) + "2/1/26": ("2/2/26", "2/3/26"), # A: own 1.007 (role-rescued), not dropped + "2/1/3": ("2/2/3", "2/3/3"), # A + "2/1/50": ("2/2/50", "2/3/50"), # B: own, function-isolated from the window + "2/1/52": ("2/2/52", "2/3/52"), # B + "2/5/10": ("2/5/11", None), # C: roller takes its own, NOT 2/5/16 + "2/5/15": ("2/5/16", None), # C: awning keeps its own + "2/6/0": ("2/6/2", None), # D: own stop, NOT the foreign light step } + # D: the foreign lighting step must never be borrowed as a cover control. + assert "2/6/1" not in [c.get("move_short_address") for c in covers.values()], \ + "a foreign 1.007 (lighting step) was mis-paired as a cover step/stop" for mv, (step, spos) in expected.items(): assert mv in covers, f"missing cover for {mv}: {sorted(covers)}" got = covers[mv].get("move_short_address") @@ -100,26 +113,14 @@ def main(): gp = covers[mv].get("position_state_address") assert gp == spos, f"{mv}: position_state {gp!r}, expected {spos!r}" - # The collective (no own function) must fall back to its own 2/5/11, not steal - # the awnings' 2/5/16. - assert "2/5/10" in covers, sorted(covers) - assert covers["2/5/10"].get("move_short_address") == "2/5/11", \ - covers["2/5/10"] - - # And no step/stop is used as move_short by more than one cover (no stealing). + # No step/stop is used as move_short by more than one cover (no stealing). shorts = [c.get("move_short_address") for c in covers.values() if c.get("move_short_address")] assert len(shorts) == len(set(shorts)), f"a step/stop was shared: {shorts}" - print("test_cover_pairing: OK — step/stop and status pair within the ETS " - "Function; no cross-wiring across window/awning sharing a zone token.") - - -def _covers(res): - """Extract the cover list from the generated package (yaml text or counts).""" - import yaml as _yaml - doc = _yaml.safe_load(res["yaml"]) - return (doc or {}).get("knx", {}).get("cover", []) or [] + print("test_cover_pairing: OK — each cover pairs its own step/stop across " + "per-channel functions (A), actuator functions with zone collision (B), " + "and function-less type-token collectives (C); no cross-wiring, none dropped.") if __name__ == "__main__":