Wire up ETS Function role pairing for command/status

Implement pairing.function_status_pairs(): pair command<->status GAs from ETS
Function roles (SwitchOnOff<->InfoOnOff etc.), used by check_missing_status and
the HA generator. Fixes false "missing status" on function-paired commands and
gives HA entities the correct state_address even when the feedback GA is named
only "Status". Add a regression test.

Found via real-ETS6 fixture testing: Functions were previously ignored despite
the README claiming they were the primary pairing signal.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Nikolay Miroshnichenko
2026-06-28 10:35:34 +02:00
co-authored by Claude Opus 4.8
parent 1b0fe566f7
commit 29d39dc147
5 changed files with 106 additions and 4 deletions
+9
View File
@@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **ETS Function role pairing (the headline feature) is now actually wired up.**
command↔status pairs are taken from ETS Function roles (e.g. `SwitchOnOff`↔`InfoOnOff`)
in `pairing.function_status_pairs()`, used by both `check_missing_status` and the HA
generator. A feedback GA named only "Status" now pairs correctly (name tokens not needed),
HA entities get the right `state_address`, and function-paired commands are no longer
false-flagged as missing a status. Found because real ETS6 projects exposed that Functions
were previously ignored despite the README.
### Fixed
- **Critical:** `load_project` MCP tool recursed infinitely (`RecursionError`) on every
real `.knxproj` because the server tool function shadowed the imported project loader.
+4 -3
View File
@@ -15,7 +15,7 @@ from collections import defaultdict
from typing import Any, Optional
from .project import LoadedProject, GARecord, STATUS_KEYWORDS
from .pairing import find_status
from .pairing import find_status, function_status_pairs
SEVERITY_ERROR = "error"
SEVERITY_WARN = "warning"
@@ -168,8 +168,9 @@ def detect_missing_status(project: LoadedProject) -> list[dict[str, Any]]:
"""
findings = _function_role_status(project)
# addresses already explained by a function finding -> skip in heuristic
covered: set[str] = set()
# Commands already paired to a status by an ETS Function are satisfied
# (authoritative). Plus any address a function-missing-status finding covers.
covered: set[str] = set(function_status_pairs(project).keys())
for f in findings:
covered.update(f.get("addresses", []))
+7 -1
View File
@@ -15,14 +15,20 @@ import yaml
from .project import LoadedProject, GARecord
from .analyze import _is_status_ga
from .pairing import find_status, base_tokens
from .pairing import find_status, base_tokens, function_status_pairs
def generate_ha_yaml(project: LoadedProject) -> dict[str, Any]:
"""Return {'yaml': str, 'review': [...], 'counts': {...}}."""
status_gas = [g for g in project.gas.values() if _is_status_ga(g)]
fpairs = function_status_pairs(project) # authoritative cmd-addr -> status-addr
def status_for(cmd: GARecord):
# 1. ETS Function role pairing is authoritative (names not needed).
paired = fpairs.get(cmd.address)
if paired and paired in project.gas:
return project.gas[paired]
# 2. fall back to name-token pairing.
same_main = [s for s in status_gas if s.main == cmd.main]
return find_status(cmd, same_main) or find_status(cmd, status_gas)
+60
View File
@@ -60,3 +60,63 @@ def find_status(command: GARecord, candidates: Iterable[GARecord]) -> Optional[G
if score > best_score:
best, best_score = c, score
return best
# --------------------------------------------------------------------------- #
# Authoritative pairing from ETS Functions (roles)
# --------------------------------------------------------------------------- #
_FN_STATUS_TOKENS = ("info", "status", "state", "feedback", "rueck", "rück")
_FN_STATUS_PREFIXES = ("Info", "Status", "State", "Feedback")
def _role_is_status(role: str) -> bool:
r = (role or "").lower()
return any(t in r for t in _FN_STATUS_TOKENS)
def _status_suffix(role: str) -> str:
"""'InfoOnOff' -> 'onoff', 'StatusValue' -> 'value' (strip status prefix)."""
r = role or ""
for p in _FN_STATUS_PREFIXES:
if r.startswith(p):
r = r[len(p):]
break
return r.lower()
def function_status_pairs(project) -> dict[str, str]:
"""Map command-GA address -> status-GA address using ETS Function roles.
This is the authoritative signal: ETS Functions group the GAs of one logical
function and tag each with a role (e.g. ``SwitchOnOff`` for the command and
``InfoOnOff`` for the feedback). Within each function we pair every command
role to a status role, preferring a matching role suffix
(``InfoOnOff`` <-> ``SwitchOnOff``) and otherwise falling back to the single
status GA in that function. Names are never needed, so a feedback GA called
just "Status" still pairs correctly.
"""
pairs: dict[str, str] = {}
for fn in (project.functions or {}).values():
roles = fn.get("group_addresses", {}) or {}
cmds: list[tuple[str, str]] = []
stats: list[tuple[str, str]] = []
for key, ref in roles.items():
addr = ref.get("address", key)
role = ref.get("role") or ""
(stats if _role_is_status(role) else cmds).append((addr, role))
valid_stats = [(a, r) for a, r in stats if a in project.gas]
if not cmds or not valid_stats:
continue
for caddr, crole in cmds:
if caddr not in project.gas:
continue
partner = None
for saddr, srole in valid_stats:
suf = _status_suffix(srole)
if suf and suf in crole.lower():
partner = saddr
break
if partner is None:
partner = valid_stats[0][0]
pairs[caddr] = partner
return pairs
+26
View File
@@ -144,3 +144,29 @@ except Exception as exc:
print(f"OK: delegated, raised {type(exc).__name__} (expected, not RecursionError)")
else:
print("OK: delegated (no error)")
# --------------------------------------------------------------------------- #
# Regression: ETS Function roles must pair command <-> status (the headline
# feature). A status GA named only "Status" pairs via the InfoOnOff role even
# though name-token pairing alone would miss it.
# --------------------------------------------------------------------------- #
print("\n=== REGRESSION: ETS Function role pairing ===")
from nickol_knx_mcp.pairing import function_status_pairs
class _StubProject:
functions = {
"F-1": {
"name": "LivingroomLight",
"group_addresses": {
"0/0/1": {"address": "0/0/1", "role": "SwitchOnOff"},
"0/0/2": {"address": "0/0/2", "role": "InfoOnOff"},
},
}
}
gas = {"0/0/1": object(), "0/0/2": object()}
_pairs = function_status_pairs(_StubProject())
assert _pairs == {"0/0/1": "0/0/2"}, f"function pairing regression: {_pairs}"
print("OK: SwitchOnOff <-> InfoOnOff paired via ETS Function role")