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

fix(#11): pair every cover to its own step/stop across all 3 real ETS Function structures

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Nikolay Miroshnichenko
2026-08-29 13:13:59 +02:00
co-authored by Claude Opus 4.8
parent dd8ecac090
commit 0198ae3019
3 changed files with 171 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nickol_knx_mcp/generate_ha.py
+55 -22
View File
@@ -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 —
nickol_knx_mcp/project.py
+35
View File
@@ -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", {})),
tests/test_cover_pairing.py
+81 -80
View File
@@ -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__":