diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b0cdde2..752a174 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,6 +40,9 @@ jobs: - name: v0.8 fixes (positional pairing, ref-merge, range names) run: python tests/test_v08_fixes.py + - name: Cover step/stop pairing stays within the ETS Function (issue #11) + run: python tests/test_cover_pairing.py + - name: Console script is installed run: | python -c "import importlib.metadata as m; print('entry points:', [e.name for e in m.entry_points(group='console_scripts') if e.name == 'nickol-knx-mcp'])" diff --git a/nickol_knx_mcp/generate_ha.py b/nickol_knx_mcp/generate_ha.py index 2875d68..91e1f94 100644 --- a/nickol_knx_mcp/generate_ha.py +++ b/nickol_knx_mcp/generate_ha.py @@ -127,6 +127,18 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: slat_addrs = {g.address for g in project.gas.values() if g.category == "shutter" and _is_slat(g.name)} + # Map each GA address to the ETS Function(s) that own it. A cover's step/stop + # and position/status must never be paired across an ETS Function boundary: + # in a room with both a window and an awning the names share a zone token + # ("Room A"), so name-token matching alone grabs the wrong shutter's step/stop + # (issue #11). ETS Function membership is the authoritative grouping. + addr_to_fns: dict[str, set[str]] = {} + for fkey, fn in (project.functions or {}).items(): + for a_key, ref in (fn.get("group_addresses", {}) or {}).items(): + a = ref.get("address") or a_key # dict key is the address fallback (cf. pairing.py) + if a: + addr_to_fns.setdefault(a, set()).add(fkey) + def status_for(cmd: GARecord): # 1. ETS Function role pairing is authoritative (names not needed). paired = fpairs.get(cmd.address) @@ -222,14 +234,23 @@ def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]: continue entity = {"name": ga.name, "move_long_address": ga.address} ptoks = _ident_tokens(ga.name) + my_fns = addr_to_fns.get(ga.address, set()) for sib in same_main_gas(ga): if sib.address in consumed or sib.address == ga.address: continue if sib.category != "shutter": continue - # only attach siblings of the SAME shutter when both carry a zone - # identity (prevents cross-wiring multiple blinds in one main group) - if ptoks and _ident_tokens(sib.name) and not (ptoks & _ident_tokens(sib.name)): + 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. + if sib_fns and not same_fn: + continue + # 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)): 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: diff --git a/tests/test_cover_pairing.py b/tests/test_cover_pairing.py new file mode 100644 index 0000000..e544146 --- /dev/null +++ b/tests/test_cover_pairing.py @@ -0,0 +1,126 @@ +"""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). + +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. + +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. +""" +from nickol_knx_mcp.project import build_loaded_from_raw +from nickol_knx_mcp.generate_ha import generate_ha_yaml + + +def _ga(addr, name, dmain, dsub): + return {"name": name, "identifier": f"GA-{addr}", "raw_address": 0, "address": addr, + "project_uid": None, "dpt": {"main": dmain, "sub": dsub}, "data_secure": False, + "communication_object_ids": [], "description": "", "comment": ""} + + +def _fn(name, *addrs): + 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", +} + + +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") + + 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", + "group_ranges": {}}}} + return build_loaded_from_raw(raw, "cover-pairing.knxproj") + + +def main(): + project = _build() + res = generate_ha_yaml(project) + covers = {c["move_long_address"]: c for c in + [e for e in _covers(res)]} + + # Each shutter must pair to ITS OWN step/stop and status, never a sibling's. + 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) + } + 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") + assert got == step, f"{mv}: move_short {got!r}, expected {step!r}" + if spos is not None: + 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). + 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 [] + + +if __name__ == "__main__": + main()